Coverage for python / lsst / pipe / tasks / rgb2hips / _utils.py: 0%

1# This file is part of pipe_tasks. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <https://www.gnu.org/licenses/>. 

21 

22from __future__ import annotations 

23 

24__all__ = ("_write_hips_image",) 

25 

26from PIL import Image 

27import numpy as np 

28from numpy.typing import NDArray 

29 

30 

31from lsst.resources import ResourcePath 

32 

33 

34# allow PIL to work with really large images 

35Image.MAX_IMAGE_PIXELS = None 

36 

37 

38def _get_dir_number(pixel: int) -> int: 

39 """Compute the directory number from a pixel. 

40 

41 Parameters 

42 ---------- 

43 pixel : `int` 

44 HEALPix pixel number. 

45 

46 Returns 

47 ------- 

48 dir_number : `int` 

49 HiPS directory number. 

50 """ 

51 return (pixel // 10000) * 10000 

52 

53 

54def _write_hips_image( 

55 image_data: NDArray, 

56 pixel_id: int, 

57 hpx_level: int, 

58 hips_base_path: ResourcePath, 

59 file_extension: str, 

60 output_type: str, 

61) -> None: 

62 """Write a processed image to disk in the HealPix tile format. 

63 

64 This function takes processed image data, converts it to the specified output 

65 type, and saves it into the appropriate directory structure based on the HealPix 

66 pixel ID and order level. 

67 

68 Parameters 

69 ---------- 

70 image_data : `NDArray` 

71 The RGB image data array to be written as a HealPix tile. 

72 pixel_id : `int` 

73 The unique HealPix ID corresponding to the output tile. 

74 hpx_level : `int` 

75 The HealPix order level of the output tile. 

76 hips_base_path : `ResourcePath` 

77 Base directory path where the HealPix tiles will be stored. 

78 file_extension : `str` 

79 File extension (format) for saving the image ('png' or 'webp'). 

80 output_type : `str` 

81 Data type of the output array, which can be: 

82 - "uint8": 8-bit unsigned integers (0-255) 

83 - "uint16": 16-bit unsigned integers (0-65535) 

84 - "half": 16-bit floating-point numbers 

85 - "float": 32-bit floating-point numbers 

86 

87 """ 

88 # clip in case any of the warping caused values over 1 

89 image_data = np.clip(image_data, 0, 1) 

90 # remap the image_data to the chosen output_type 

91 match output_type: 

92 case "uint8": 

93 image_data = (image_data * 255.0).astype(np.uint8) 

94 case "uint16": 

95 image_data = (image_data * 65535.0).astype(np.uint16) 

96 case "half": 

97 image_data = image_data.astype(np.float16) 

98 case "float": 

99 pass 

100 

101 # mangle the URI where to write 

102 dir_number = _get_dir_number(pixel_id) 

103 hips_dir = hips_base_path.join(f"Norder{hpx_level}", forceDirectory=True).join( 

104 f"Dir{dir_number}", forceDirectory=True 

105 ) 

106 

107 # Create the file URI for saving 

108 uri = hips_dir.join(f"Npix{pixel_id}.{file_extension}") 

109 

110 # Convert numpy array to PIL Image and save with appropriate arguments 

111 im = Image.fromarray(image_data, mode="RGB") 

112 

113 extra_args = {} 

114 if file_extension == "webp": 

115 # Set WebP-specific parameters for lossless compression 

116 extra_args["lossless"] = True 

117 extra_args["quality"] = 80 

118 

119 # Save the image to a temporary file and transfer to final location 

120 with ResourcePath.temporary_uri(suffix=uri.getExtension()) as temporary_uri: 

121 im.save(temporary_uri.ospath, **extra_args) 

122 uri.transfer_from(temporary_uri, transfer="copy", overwrite=True)