Coverage for python / lsst / pipe / tasks / snapCombine.py: 34%

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 

22__all__ = ["SnapCombineConfig", "SnapCombineTask"] 

23 

24import collections.abc 

25 

26import lsst.afw.image 

27import lsst.pex.config as pexConfig 

28import lsst.daf.base as dafBase 

29import lsst.afw.image as afwImage 

30import lsst.pipe.base as pipeBase 

31from lsst.coadd.utils import addToCoadd, setCoaddEdgeBits 

32from lsst.utils.timer import timeMethod 

33 

34 

35class SnapCombineConfig(pexConfig.Config): 

36 bad_mask_planes = pexConfig.ListField( 

37 dtype=str, 

38 doc="Mask planes that, if set, the associated pixels are not included in the combined exposure.", 

39 default=(), 

40 ) 

41 

42 

43class SnapCombineTask(pipeBase.Task): 

44 """Combine one or two snaps into a single visit image. 

45 """ 

46 

47 ConfigClass = SnapCombineConfig 

48 _DefaultName = "snapCombine" 

49 

50 def __init__(self, *args, **kwargs): 

51 pipeBase.Task.__init__(self, *args, **kwargs) 

52 

53 @timeMethod 

54 def run(self, exposures): 

55 """Combine one or two snaps, returning the combined image. 

56 

57 Parameters 

58 ---------- 

59 exposures : `lsst.afw.image.Exposure` or `list` [`lsst.afw.image.Exposure`] 

60 One or two exposures to combine as snaps. 

61 

62 Returns 

63 ------- 

64 result : `lsst.pipe.base.Struct` 

65 Results as a struct with attributes: 

66 

67 ``exposure`` 

68 Snap-combined exposure. 

69 

70 Raises 

71 ------ 

72 RuntimeError 

73 Raised if input argument does not contain either 1 or 2 exposures. 

74 """ 

75 if isinstance(exposures, lsst.afw.image.Exposure): 

76 return pipeBase.Struct(exposure=exposures) 

77 

78 if isinstance(exposures, collections.abc.Sequence) and not isinstance(exposures, str): 

79 match len(exposures): 

80 case 1: 

81 return pipeBase.Struct(exposure=exposures[0]) 

82 case 2: 

83 return self.combine(exposures[0], exposures[1]) 

84 case n: 

85 raise RuntimeError(f"Can only process 1 or 2 snaps, not {n}.") 

86 else: 

87 raise RuntimeError("`exposures` must be either an afw Exposure (single snap visit), or a " 

88 "list/tuple of one or two of them.") 

89 

90 def combine(self, snap0, snap1): 

91 """Combine two snaps, returning the combined image. 

92 

93 Parameters 

94 ---------- 

95 snap0, snap1 : `lsst.afw.image.Exposure` 

96 Exposures to combine. 

97 

98 Returns 

99 ------- 

100 result : `lsst.pipe.base.Struct` 

101 Results as a struct with attributes: 

102 

103 ``exposure`` 

104 Snap-combined exposure. 

105 """ 

106 self.log.info("Merging two snaps with exposure ids: %s, %s", snap0.visitInfo.id, snap1.visitInfo.id) 

107 combined = self._add_snaps(snap0, snap1) 

108 

109 return pipeBase.Struct( 

110 exposure=combined, 

111 ) 

112 

113 def _add_snaps(self, snap0, snap1): 

114 """Add two snap exposures together, returning a new exposure. 

115 

116 Parameters 

117 ---------- 

118 snap0 : `lsst.afw.image.Exposure` 

119 Snap exposure 0. 

120 snap1 : `lsst.afw.image.Exposure` 

121 Snap exposure 1. 

122 

123 Returns 

124 ------- 

125 combined : `lsst.afw.image.Exposure` 

126 Combined exposure. 

127 """ 

128 combined = snap0.Factory(snap0, True) 

129 combined.maskedImage.set(0) 

130 

131 weights = combined.maskedImage.image.Factory(combined.maskedImage.getBBox()) 

132 weight = 1.0 

133 bad_mask = afwImage.Mask.getPlaneBitMask(self.config.bad_mask_planes) 

134 addToCoadd(combined.maskedImage, weights, snap0.maskedImage, bad_mask, weight) 

135 addToCoadd(combined.maskedImage, weights, snap1.maskedImage, bad_mask, weight) 

136 

137 # pre-scaling the weight map instead of post-scaling the combined.maskedImage saves a bit of time 

138 # because the weight map is a simple Image instead of a MaskedImage 

139 weights *= 0.5 # so result is sum of both images, instead of average 

140 combined.maskedImage /= weights 

141 setCoaddEdgeBits(combined.maskedImage.getMask(), weights) 

142 

143 combined.info.setVisitInfo(self._merge_visit_info(snap0.visitInfo, snap1.visitInfo)) 

144 

145 return combined 

146 

147 def _merge_visit_info(self, info0, info1): 

148 """Merge the visitInfo values from the two exposures. 

149 

150 In particular: 

151 * id will be the id of snap 0. 

152 * date will be the average of the dates. 

153 * exposure time will be the sum of the times. 

154 

155 Parameters 

156 ---------- 

157 info0, info1 : `lsst.afw.image.VisitInfo` 

158 Metadata to combine. 

159 

160 Returns 

161 ------- 

162 info : `lsst.afw.image.VisitInfo` 

163 Combined metadata. 

164 

165 """ 

166 time = info0.exposureTime + info1.exposureTime 

167 date = (info0.date.get() + info1.date.get()) / 2.0 

168 result = info0.copyWith(exposureTime=time, 

169 date=dafBase.DateTime(date) 

170 ) 

171 return result