Coverage for src/ufig/plugins/add_generic_stamp_flags.py: 92%

1# Copyright (c) 2016 ETH Zurich, Institute of Astronomy, Tomasz Kacprzak 

2# <tomasz.kacprzak@phys.ethz.ch> 

3""" 

4Created on Aug 5, 2016 

5@author: Tomasz Kacprzak 

6 

7""" 

8 

9import os 

10 

11import h5py 

12import numpy as np 

13from cosmic_toolbox import arraytools as at 

14from cosmic_toolbox import logger 

15from ivy.plugin.base_plugin import BasePlugin 

16 

17from ufig import mask_utils 

18from ufig.array_util import set_flag_bit 

19 

20LOGGER = logger.get_logger(__file__) 

21 

22FLAGBIT_EDGE_DUPLICATE = 1 

23FLAGBIT_COADD_BOUNDARY = 2 

24FLAGBIT_IMAGE_BOUNDARY = 3 

25FLAGBIT_SYSMAP_DELTA_WEIGHT = 4 

26FLAGBIT_NEARBY_BRIGHT_STAR = 5 

27FLAGBIT_SURVEY_MASK = 6 

28 

29 

30class Plugin(BasePlugin): 

31 def __call__(self): 

32 par = self.ctx.parameters 

33 

34 sexcat_name = None 

35 if os.path.exists(par.sextractor_catalog_name): 35 ↛ 37line 35 didn't jump to line 37 because the condition on line 35 was always true

36 sexcat_name = par.sextractor_catalog_name 

37 elif os.path.exists(par.sextractor_forced_photo_catalog_name): 

38 sexcat_name = par.sextractor_forced_photo_catalog_name 

39 

40 if sexcat_name is not None: 40 ↛ 48line 40 didn't jump to line 48 because the condition on line 40 was always true

41 add_all_stamp_flags( 

42 sexcat_name, 

43 par.filepath_overlapblock, 

44 par.sextractor_mask_name, 

45 par.sextractor_catalog_off_mask_radius, 

46 ) 

47 else: 

48 LOGGER.info( 

49 "No SExtractor catalog found, skipping adding flags to the catalog." 

50 ) 

51 

52 def __str__(self): 

53 return "tile overlap/coadd bounadry col" 

54 

55 

56def add_all_stamp_flags( 

57 filename_cat, 

58 filename_overlapblock, 

59 filename_mask, 

60 off_mask_radius, 

61): 

62 add_edge_duplicate_flag(filename_cat, filename_overlapblock) 

63 add_pixel_based_masks(filename_cat, filename_mask, off_mask_radius) 

64 

65 

66def load_flags_stamp(filename_cat, len_cat): 

67 with h5py.File(filename_cat, "r") as f: 

68 if "FLAGS_STAMP" in f: 

69 flags_stamp = np.array(f["FLAGS_STAMP"]) 

70 else: 

71 flags_stamp = np.zeros(len_cat, dtype=np.uint32) 

72 return flags_stamp 

73 

74 

75def save_flags_stamp(filename_cat, flags_stamp): 

76 with h5py.File(filename_cat, "a") as f: 

77 if "FLAGS_STAMP" in f: 

78 del f["FLAGS_STAMP"] 

79 f.create_dataset(name="FLAGS_STAMP", data=flags_stamp, compression="lzf") 

80 LOGGER.info(f"Saved flags to {filename_cat}") 

81 

82 

83def add_pixel_based_masks(filename_cat, filename_mask, off_mask_radius): 

84 cat = at.load_hdf_cols( 

85 filename_cat, columns=["XWIN_IMAGE", "YWIN_IMAGE", "FLUX_RADIUS"] 

86 ) 

87 flags_stamp = load_flags_stamp(filename_cat, len(cat)) 

88 

89 # Pixel mask 

90 with h5py.File(filename_mask, mode="r") as fh5: 

91 pixel_mask = fh5["mask"][...] 

92 

93 FLAGBITS = {"delta_weight": 0, "star": 1, "mask": 2} 

94 n_bits = len(FLAGBITS) 

95 

96 # Get masked pixels 

97 pixel_mask_sysmap_delta_weight = mask_utils.set_masked_pixels( 

98 pixel_mask, [FLAGBITS["delta_weight"]], n_bits=n_bits 

99 ) 

100 pixel_mask_nearby_star = mask_utils.set_masked_pixels( 

101 pixel_mask, [FLAGBITS["star"]], n_bits=n_bits 

102 ) 

103 pixel_mask_survey_mask = mask_utils.set_masked_pixels( 

104 pixel_mask, [FLAGBITS["mask"]], n_bits=n_bits 

105 ) 

106 

107 # Transform to catalog mask 

108 select_sysmap_delta_weight = ~mask_utils.pixel_mask_to_catalog_mask( 

109 pixel_mask_sysmap_delta_weight, cat, off_mask_radius 

110 ) 

111 select_mask_nearby_star = ~mask_utils.pixel_mask_to_catalog_mask( 

112 pixel_mask_nearby_star, cat, off_mask_radius 

113 ) 

114 select_survey_mask = ~mask_utils.pixel_mask_to_catalog_mask( 

115 pixel_mask_survey_mask, cat, off_mask_radius 

116 ) 

117 

118 # Set FLAGS_STAMP 

119 set_flag_bit( 

120 flags=flags_stamp, 

121 select=select_sysmap_delta_weight, 

122 field=FLAGBIT_SYSMAP_DELTA_WEIGHT, 

123 ) 

124 set_flag_bit( 

125 flags=flags_stamp, 

126 select=select_mask_nearby_star, 

127 field=FLAGBIT_NEARBY_BRIGHT_STAR, 

128 ) 

129 set_flag_bit( 

130 flags=flags_stamp, select=select_survey_mask, field=FLAGBIT_SURVEY_MASK 

131 ) 

132 

133 save_flags_stamp(filename_cat, flags_stamp) 

134 

135 

136def add_edge_duplicate_flag(filename_cat, filename_overlapblock): 

137 with h5py.File(filename_cat, "r") as f: 

138 x = np.int32(np.array(f["X_IMAGE"])) - 1 

139 y = np.int32(np.array(f["Y_IMAGE"])) - 1 

140 

141 with h5py.File(filename_overlapblock, "r") as f: 

142 img_mask = np.array(f["img_mask"], dtype=np.int8) 

143 

144 x[x < 0] = 0 

145 y[y < 0] = 0 

146 x[x >= img_mask.shape[1]] = img_mask.shape[1] 

147 y[y >= img_mask.shape[0]] = img_mask.shape[0] 

148 

149 is_duplicate = img_mask[y, x] == 1 

150 

151 flags_stamp = load_flags_stamp(filename_cat, len(is_duplicate)) 

152 set_flag_bit(flags=flags_stamp, select=is_duplicate, field=FLAGBIT_EDGE_DUPLICATE) 

153 

154 save_flags_stamp(filename_cat, flags_stamp)