Coverage for python / lsst / atmospec / centroiding.py: 27%

1# This file is part of atmospec. 

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 

22import lsst.afw.image as afwImage 

23import lsst.pipe.base as pipeBase 

24 

25from lsst.meas.algorithms import LoadReferenceObjectsConfig, ReferenceObjectLoader 

26from lsst.meas.astrom import AstrometryTask, FitAffineWcsTask 

27from lsst.pipe.tasks.quickFrameMeasurement import QuickFrameMeasurementTask 

28from lsst.pipe.tasks.peekExposure import PeekExposureTask 

29from lsst.pipe.base.task import TaskError 

30import lsst.pipe.base.connectionTypes as cT 

31import lsst.pex.config as pexConfig 

32 

33from .utils import getTargetCentroidFromWcs 

34 

35__all__ = ['SingleStarCentroidTaskConfig', 'SingleStarCentroidTask'] 

36 

37 

38class SingleStarCentroidTaskConnections(pipeBase.PipelineTaskConnections, 

39 dimensions=("instrument", "visit", "detector")): 

40 inputExp = cT.Input( 

41 name="icExp", 

42 doc="Image-characterize output exposure", 

43 storageClass="ExposureF", 

44 dimensions=("instrument", "visit", "detector"), 

45 multiple=False, 

46 ) 

47 inputSources = cT.Input( 

48 name="icSrc", 

49 doc="Image-characterize output sources.", 

50 storageClass="SourceCatalog", 

51 dimensions=("instrument", "visit", "detector"), 

52 multiple=False, 

53 ) 

54 astromRefCat = cT.PrerequisiteInput( 

55 doc="Reference catalog to use for astrometry", 

56 name="the_monster_20250219", 

57 storageClass="SimpleCatalog", 

58 dimensions=("skypix",), 

59 deferLoad=True, 

60 multiple=True, 

61 ) 

62 atmospecCentroid = cT.Output( 

63 name="atmospecCentroid", 

64 doc="The main star centroid in yaml format.", 

65 storageClass="StructuredDataDict", 

66 dimensions=("instrument", "visit", "detector"), 

67 ) 

68 

69 

70class SingleStarCentroidTaskConfig(pipeBase.PipelineTaskConfig, 

71 pipelineConnections=SingleStarCentroidTaskConnections): 

72 """Configuration parameters for ProcessStarTask.""" 

73 astromRefObjLoader = pexConfig.ConfigField( 

74 dtype=LoadReferenceObjectsConfig, 

75 doc="Reference object loader config for astrometric calibration.", 

76 ) 

77 astrometry = pexConfig.ConfigurableField( 

78 target=AstrometryTask, 

79 doc="Task to perform astrometric calibration to refine the WCS", 

80 ) 

81 centroidingFallbackTask = pexConfig.ConfigurableField( 

82 target=PeekExposureTask, 

83 doc="The task to run find the brightest star if astrometry fails", 

84 ) 

85 referenceFilterOverride = pexConfig.Field( 

86 dtype=str, 

87 doc="Which filter in the reference catalog to match to?", 

88 default="phot_g_mean" 

89 ) 

90 

91 def setDefaults(self): 

92 super().setDefaults() 

93 self.astromRefObjLoader.pixelMargin = 1000 

94 

95 self.astrometry.wcsFitter.retarget(FitAffineWcsTask) 

96 

97 # Use magnitude limits for the reference catalog 

98 self.astrometry.referenceSelector.doMagLimit = True 

99 self.astrometry.referenceSelector.magLimit.minimum = 1 

100 self.astrometry.referenceSelector.magLimit.maximum = 15 

101 self.astrometry.referenceSelector.magLimit.fluxField = "phot_g_mean_flux" 

102 self.astrometry.matcher.maxRotationDeg = 5.99 

103 self.astrometry.matcher.maxOffsetPix = 3000 

104 

105 # Use a SNR limit for the science catalog 

106 self.astrometry.sourceSelector["science"].doSignalToNoise = True 

107 self.astrometry.sourceSelector["science"].signalToNoise.minimum = 10 

108 self.astrometry.sourceSelector["science"].signalToNoise.fluxField = "slot_PsfFlux_instFlux" 

109 self.astrometry.sourceSelector["science"].signalToNoise.errField = "slot_PsfFlux_instFluxErr" 

110 self.astrometry.sourceSelector["science"].doRequirePrimary = False 

111 self.astrometry.sourceSelector["science"].doIsolated = False 

112 

113 def validate(self): 

114 super().validate() 

115 task = self.centroidingFallbackTask 

116 # note these aren't instantiated yet, so we can't check the type 

117 # of the instance, just the target. _DefaultName is a class attribute 

118 # that definitely exists, but has a lower case first letter. 

119 if task.target._DefaultName not in ('quickFrameMeasurementTask', 'peekExposureTask'): 

120 raise ValueError(f"centroidingFallbackTask is of unknown type {task.target}") 

121 

122 

123class SingleStarCentroidTask(pipeBase.PipelineTask): 

124 """XXX Docs here 

125 """ 

126 

127 ConfigClass = SingleStarCentroidTaskConfig 

128 _DefaultName = 'singleStarCentroid' 

129 

130 def __init__(self, initInputs=None, **kwargs): 

131 super().__init__(**kwargs) 

132 

133 self.makeSubtask("astrometry", refObjLoader=None) 

134 self.makeSubtask('centroidingFallbackTask') 

135 

136 def runQuantum(self, butlerQC, inputRefs, outputRefs): 

137 inputs = butlerQC.get(inputRefs) 

138 refObjLoader = ReferenceObjectLoader(dataIds=[ref.datasetRef.dataId 

139 for ref in inputRefs.astromRefCat], 

140 refCats=inputs.pop('astromRefCat'), 

141 name=self.config.connections.astromRefCat, 

142 config=self.config.astromRefObjLoader, log=self.log) 

143 

144 refObjLoader.pixelMargin = 1000 

145 self.astrometry.setRefObjLoader(refObjLoader) 

146 

147 # See L603 (def runQuantum(self, butlerQC, inputRefs, outputRefs):) 

148 # in calibrate.py to put photocal back in 

149 

150 outputs = self.run(**inputs) 

151 butlerQC.put(outputs, outputRefs) 

152 

153 def runFallbackTask(self, exp): 

154 task = self.centroidingFallbackTask 

155 if isinstance(task, QuickFrameMeasurementTask): 

156 result = task.run(exp) 

157 return result.brightestObjCentroid # tuple 

158 elif isinstance(task, PeekExposureTask): 

159 result = task.run(exp) 

160 centroid = result.brightestCentroid # Point2D 

161 return (centroid[0], centroid[1]) 

162 else: 

163 raise ValueError(f"Unsupported fallback task: {task}") 

164 

165 def run(self, inputExp, inputSources): 

166 """XXX Docs 

167 """ 

168 

169 # TODO: Change this to doing this the proper way 

170 referenceFilterName = self.config.referenceFilterOverride 

171 referenceFilterLabel = afwImage.FilterLabel(physical=referenceFilterName, band=referenceFilterName) 

172 # there's a better way of doing this with the task I think 

173 originalFilterLabel = inputExp.getFilter() 

174 inputExp.setFilter(referenceFilterLabel) 

175 originalWcs = inputExp.getWcs() 

176 

177 successfulFit = False 

178 try: 

179 astromResult = self.astrometry.run(sourceCat=inputSources, exposure=inputExp) 

180 scatter = astromResult.scatterOnSky.asArcseconds() 

181 inputExp.setFilter(originalFilterLabel) 

182 if scatter < 1: 

183 successfulFit = True 

184 except (RuntimeError, TaskError, IndexError, ValueError, AttributeError) as e: 

185 # IndexError raised for low source counts: 

186 # index 0 is out of bounds for axis 0 with size 0 

187 

188 # ValueError: negative dimensions are not allowed 

189 # seen when refcat source count is low (but non-zero) 

190 

191 # AttributeError: 'NoneType' object has no attribute 'asArcseconds' 

192 # when the result is a failure as the wcs is set to None on failure 

193 self.log.warning(f"Solving failed: {e}") 

194 inputExp.setWcs(originalWcs) # this is set to None when the fit fails, so restore it 

195 finally: 

196 inputExp.setFilter(originalFilterLabel) # always restore this 

197 

198 centroid = None 

199 if successfulFit: 

200 target = inputExp.visitInfo.object 

201 centroid = getTargetCentroidFromWcs(inputExp, target, logger=self.log) 

202 if not centroid: 

203 successfulFit = False 

204 self.log.warning(f'Failed to find target centroid for {target} via WCS') 

205 if not centroid: 

206 centroid = self.runFallbackTask(inputExp) 

207 

208 centroidTuple = (centroid[0], centroid[1]) # unify Point2D or tuple to tuple 

209 self.log.info(f"Centroid of main star found at {centroidTuple} found" 

210 f" via {'astrometry' if successfulFit else 'QuickFrameMeasurement'}") 

211 result = pipeBase.Struct(atmospecCentroid={'centroid': centroidTuple, 

212 'astrometricMatch': successfulFit}) 

213 return result