Coverage for tests/test_processStar_extra.py: 99%

1#!/usr/bin/env python 

2# This file is part of atmospec. 

3# 

4# Developed for the LSST Data Management System. 

5# This product includes software developed by the LSST Project 

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

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

8# for details of code ownership. 

9# 

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

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

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

13# (at your option) any later version. 

14# 

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

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

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

18# GNU General Public License for more details. 

19# 

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

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

22"""Coverage-oriented test cases for lsst.atmospec.processStar.""" 

23 

24import types 

25import unittest 

26from unittest import mock 

27 

28import numpy as np 

29 

30import lsst.utils 

31import lsst.utils.tests 

32import lsst.afw.image as afwImage 

33import lsst.geom as geom 

34import lsst.afw.detection as afwDetect 

35from lsst.afw.geom.ellipses import Quadrupole 

36from lsst.pex.config import FieldValidationError 

37 

38from lsst.atmospec.processStar import ( 

39 ProcessStarTask, 

40 ProcessStarTaskConfig, 

41 ProcessStarTaskConnections, 

42) 

43 

44MODULE = "lsst.atmospec.processStar" 

45 

46 

47def makeSourceExposure(width=100, height=100): 

48 """Build a small ExposureF containing three gaussian sources. 

49 

50 Returns the exposure. The sources are: 

51 - a round, faint-ish source at (30, 30) 

52 - an elongated, bright source at (70, 70) 

53 - a round, very faint source at (50, 20) 

54 """ 

55 exp = afwImage.ExposureF(width, height) 

56 exp.variance.array[:] = 1.0 

57 arr = exp.image.array 

58 yy, xx = np.mgrid[0:height, 0:width] 

59 

60 def addGaussian(x, y, flux, sx, sy): 

61 g = np.exp(-(((xx - x)**2) / (2 * sx**2) 

62 + ((yy - y)**2) / (2 * sy**2))) 

63 g *= flux / g.sum() 

64 arr[:] = arr + g 

65 

66 addGaussian(30, 30, 10000, 2, 2) 

67 addGaussian(70, 70, 50000, 2, 6) 

68 addGaussian(50, 20, 500, 2, 2) 

69 return exp 

70 

71 

72def fakeFindConfig(): 

73 """A namespace supplying the object-finding config fields. 

74 

75 These fields are referenced by findObjects/findMainSource but are not 

76 present on the real config, so tests inject this fake instead. 

77 """ 

78 return types.SimpleNamespace( 

79 mainStarNpixMin=5, 

80 mainStarNsigma=5, 

81 mainStarGrow=0, 

82 mainStarGrowIsotropic=True, 

83 mainSourceFindingMethod='ROUNDEST', 

84 mainStarFluxCut=1e-15, 

85 mainStarRoundnessCut=1e9, 

86 ) 

87 

88 

89class ConfigTestCase(lsst.utils.tests.TestCase): 

90 """Tests for ProcessStarTaskConfig and connections.""" 

91 

92 def testDefaultsAndSetDefaults(self): 

93 config = ProcessStarTaskConfig() 

94 # setDefaults was applied during construction. 

95 self.assertFalse(config.charImage.doWriteExposure) 

96 self.assertFalse(config.charImage.doApCorr) 

97 self.assertFalse(config.charImage.doMeasurePsf) 

98 self.assertEqual(config.charImage.detection.includeThresholdMultiplier, 3) 

99 

100 def testValidatePasses(self): 

101 config = ProcessStarTaskConfig() 

102 # doFitAtmosphere is False by default so this should not raise. 

103 config.validate() 

104 

105 def testValidateFitAtmosphereRaises(self): 

106 config = ProcessStarTaskConfig() 

107 config.doFitAtmosphere = True 

108 with mock.patch(f"{MODULE}.shutil.which", return_value=None): 

109 with self.assertRaises(FieldValidationError): 

110 config.validate() 

111 

112 def testValidateFitAtmosphereOnSpectrogramRaises(self): 

113 config = ProcessStarTaskConfig() 

114 config.doFitAtmosphereOnSpectrogram = True 

115 with mock.patch(f"{MODULE}.shutil.which", return_value=None): 

116 with self.assertRaises(FieldValidationError): 

117 config.validate() 

118 

119 def testValidateWithUvspecPresent(self): 

120 config = ProcessStarTaskConfig() 

121 config.doFitAtmosphere = True 

122 config.doFitAtmosphereOnSpectrogram = True 

123 with mock.patch(f"{MODULE}.shutil.which", return_value="/usr/bin/uvspec"): 

124 config.validate() 

125 

126 def testConnectionsDefault(self): 

127 config = ProcessStarTaskConfig() 

128 connections = ProcessStarTaskConnections(config=config) 

129 self.assertIn("spectrumForwardModelFitParameters", connections.outputs) 

130 self.assertNotIn("spectrumLibradtranFitParameters", connections.outputs) 

131 self.assertNotIn("spectrogramLibradtranFitParameters", connections.outputs) 

132 

133 def testConnectionsToggles(self): 

134 config = ProcessStarTaskConfig() 

135 config.doFullForwardModelDeconvolution = False 

136 connections = ProcessStarTaskConnections(config=config) 

137 self.assertNotIn("spectrumForwardModelFitParameters", connections.outputs) 

138 

139 

140class ProcessStarTaskTestCase(lsst.utils.tests.TestCase): 

141 """Tests for ProcessStarTask methods.""" 

142 

143 def setUp(self): 

144 self.config = ProcessStarTaskConfig() 

145 self.task = ProcessStarTask(config=self.config) 

146 

147 def testInit(self): 

148 self.assertTrue(hasattr(self.task, "isr")) 

149 self.assertTrue(hasattr(self.task, "charImage")) 

150 

151 def testGetEllipticity(self): 

152 # A round shape has (near) zero ellipticity. 

153 roundShape = Quadrupole(4.0, 4.0, 0.0) 

154 self.assertAlmostEqual(self.task._getEllipticity(roundShape), 0.0) 

155 # An elongated shape has a positive ellipticity. 

156 elongated = Quadrupole(10.0, 2.0, 0.0) 

157 self.assertGreater(self.task._getEllipticity(elongated), 0.0) 

158 

159 def testFindObjects(self): 

160 exp = makeSourceExposure() 

161 self.task.config = fakeFindConfig() 

162 fpSet = self.task.findObjects(exp) 

163 self.assertIsInstance(fpSet, afwDetect.FootprintSet) 

164 self.assertGreaterEqual(len(fpSet.getFootprints()), 1) 

165 

166 def testFindObjectsWithSigmaAndGrow(self): 

167 exp = makeSourceExposure() 

168 cfg = fakeFindConfig() 

169 cfg.mainStarGrow = 2 

170 self.task.config = cfg 

171 # Pass explicit nSigma; grow taken from config (>0 branch). 

172 fpSet = self.task.findObjects(exp, nSigma=5) 

173 self.assertIsInstance(fpSet, afwDetect.FootprintSet) 

174 

175 def testGetRoundestObject(self): 

176 exp = makeSourceExposure() 

177 self.task.config = fakeFindConfig() 

178 fpSet = self.task.findObjects(exp) 

179 source = self.task.getRoundestObject(fpSet, exp, fluxCut=1e-15) 

180 self.assertIsInstance(source, afwDetect.Footprint) 

181 

182 def testGetBrightestObject(self): 

183 exp = makeSourceExposure() 

184 self.task.config = fakeFindConfig() 

185 fpSet = self.task.findObjects(exp) 

186 source = self.task.getBrightestObject(fpSet, exp, roundnessCut=1e9) 

187 self.assertIsInstance(source, afwDetect.Footprint) 

188 

189 def testFindMainSourceRoundest(self): 

190 exp = makeSourceExposure() 

191 cfg = fakeFindConfig() 

192 cfg.mainSourceFindingMethod = 'ROUNDEST' 

193 self.task.config = cfg 

194 centroid = self.task.findMainSource(exp) 

195 self.assertEqual(len(centroid), 2) 

196 

197 def testFindMainSourceBrightest(self): 

198 exp = makeSourceExposure() 

199 cfg = fakeFindConfig() 

200 cfg.mainSourceFindingMethod = 'BRIGHTEST' 

201 self.task.config = cfg 

202 centroid = self.task.findMainSource(exp) 

203 self.assertEqual(len(centroid), 2) 

204 

205 def testFindMainSourceInvalid(self): 

206 exp = makeSourceExposure() 

207 cfg = fakeFindConfig() 

208 cfg.mainSourceFindingMethod = 'NONSENSE' 

209 self.task.config = cfg 

210 with self.assertRaises(RuntimeError): 

211 self.task.findMainSource(exp) 

212 

213 def testUpdateMetadataWithCentroid(self): 

214 exp = afwImage.ExposureF(10, 10) 

215 vi = afwImage.VisitInfo(boresightAirmass=1.5) 

216 exp.getInfo().setVisitInfo(vi) 

217 self.task.updateMetadata(exp, centroid=(12.0, 34.0)) 

218 md = exp.getMetadata() 

219 self.assertEqual(md['OBJECTX'], 12.0) 

220 self.assertEqual(md['OBJECTY'], 34.0) 

221 self.assertEqual(md['AIRMASS'], 1.5) 

222 self.assertIn('HA', md) 

223 

224 def testUpdateMetadataWithoutCentroid(self): 

225 exp = afwImage.ExposureF(10, 10) 

226 vi = afwImage.VisitInfo(boresightAirmass=1.2) 

227 exp.getInfo().setVisitInfo(vi) 

228 self.task.updateMetadata(exp) 

229 md = exp.getMetadata() 

230 self.assertIsNone(md['OBJECTX']) 

231 self.assertIsNone(md['OBJECTY']) 

232 

233 def testGetNormalizedTargetNameMapped(self): 

234 # 'ETA1DOR' is mapped in nameMappings.txt. 

235 result = self.task.getNormalizedTargetName('spec:ETA1DOR') 

236 self.assertEqual(result, 'HD42525') 

237 

238 def testGetNormalizedTargetNameUnmapped(self): 

239 result = self.task.getNormalizedTargetName('HD999999') 

240 self.assertEqual(result, 'HD999999') 

241 

242 def testGetSpectractorTargetSettingAutoExact(self): 

243 self.task.config = types.SimpleNamespace(targetCentroidMethod='auto') 

244 result = self.task._getSpectractorTargetSetting({'astrometricMatch': True}) 

245 self.assertEqual(result, 'guess') 

246 

247 def testGetSpectractorTargetSettingAutoFit(self): 

248 self.task.config = types.SimpleNamespace(targetCentroidMethod='auto') 

249 result = self.task._getSpectractorTargetSetting({'astrometricMatch': False}) 

250 self.assertEqual(result, 'fit') 

251 

252 def testGetSpectractorTargetSettingExact(self): 

253 self.task.config = types.SimpleNamespace(targetCentroidMethod='exact') 

254 self.assertEqual(self.task._getSpectractorTargetSetting({}), 'guess') 

255 

256 def testGetSpectractorTargetSettingFallthrough(self): 

257 self.task.config = types.SimpleNamespace(targetCentroidMethod='WCS') 

258 self.assertEqual(self.task._getSpectractorTargetSetting({}), 'WCS') 

259 

260 def testLoadStarNames(self): 

261 names = self.task.loadStarNames() 

262 self.assertIsInstance(names, list) 

263 self.assertTrue(len(names) > 0) 

264 

265 def testFlatfield(self): 

266 exp = afwImage.ExposureF(5, 5) 

267 self.assertIs(self.task.flatfield(exp, None), exp) 

268 

269 def testRepairCosmics(self): 

270 exp = afwImage.ExposureF(5, 5) 

271 self.assertIs(self.task.repairCosmics(exp, None), exp) 

272 

273 def testMeasureSpectrum(self): 

274 self.task.extraction = mock.MagicMock() 

275 sentinel = object() 

276 self.task.extraction.getFluxBasic.return_value = sentinel 

277 result = self.task.measureSpectrum("exp", "cen", "bbox", "disp") 

278 self.assertIs(result, sentinel) 

279 self.task.extraction.initialise.assert_called_once() 

280 

281 def testCalcSpectrumBBoxPositiveOrder(self): 

282 exp = afwImage.ExposureF(1000, 6000) 

283 bbox = self.task.calcSpectrumBBox(exp, (500, 500), 40, order='+1') 

284 self.assertIsInstance(bbox, geom.Box2I) 

285 self.assertEqual(bbox.getMinY(), 600) 

286 

287 def testCalcSpectrumBBoxNegativeOrder(self): 

288 exp = afwImage.ExposureF(1000, 6000) 

289 bbox = self.task.calcSpectrumBBox(exp, (500, 5500), 40, order='-1') 

290 self.assertIsInstance(bbox, geom.Box2I) 

291 # yStart is clamped to a value >= 0. 

292 self.assertGreaterEqual(bbox.getMinY(), 0) 

293 

294 def testPauseDefault(self): 

295 self.task.debug = types.SimpleNamespace(pauseOnDisplay=False) 

296 self.assertIsNone(self.task.pause()) 

297 

298 def testPauseWithInput(self): 

299 self.task.debug = types.SimpleNamespace(pauseOnDisplay=True) 

300 with mock.patch("builtins.input", return_value=""): 

301 self.assertIsNone(self.task.pause()) 

302 

303 def testRunQuantum(self): 

304 butlerQC = mock.MagicMock() 

305 inputRefs = mock.MagicMock() 

306 outputRefs = mock.MagicMock() 

307 butlerQC.get.return_value = { 

308 'inputExp': 'exp', 

309 'inputCentroid': 'cen', 

310 } 

311 self.task.run = mock.MagicMock(return_value="outputs") 

312 self.task.runQuantum(butlerQC, inputRefs, outputRefs) 

313 self.task.run.assert_called_once() 

314 _, kwargs = self.task.run.call_args 

315 self.assertIn('dataIdDict', kwargs) 

316 butlerQC.put.assert_called_once_with("outputs", outputRefs) 

317 

318 def _makeRunExposure(self, objectName='HD12345'): 

319 exp = afwImage.ExposureF(10, 10) 

320 vi = afwImage.VisitInfo(boresightAirmass=1.5, object=objectName) 

321 exp.getInfo().setVisitInfo(vi) 

322 return exp 

323 

324 def _runWithMocks(self, exp, inputCentroid, grating='other'): 

325 with mock.patch(f"{MODULE}.isDispersedExp", return_value=True), \ 

326 mock.patch(f"{MODULE}.getLinearStagePosition", return_value=10.0), \ 

327 mock.patch(f"{MODULE}.getFilterAndDisperserFromExp", 

328 return_value=('SDSSr', grating)), \ 

329 mock.patch(f"{MODULE}.SpectractorShim") as shim: 

330 instance = shim.return_value 

331 instance.run.return_value = mock.MagicMock() 

332 result = self.task.run(inputExp=exp, inputCentroid=inputCentroid, 

333 dataIdDict={'visit': 1}) 

334 return result, shim, instance 

335 

336 def testRunNotDispersedRaises(self): 

337 exp = self._makeRunExposure() 

338 with mock.patch(f"{MODULE}.isDispersedExp", return_value=False): 

339 with self.assertRaises(RuntimeError): 

340 self.task.run(inputExp=exp, inputCentroid={}, dataIdDict={}) 

341 

342 def testRunBadObjectRaises(self): 

343 exp = self._makeRunExposure(objectName='Test') 

344 with mock.patch(f"{MODULE}.isDispersedExp", return_value=True), \ 

345 mock.patch(f"{MODULE}.getLinearStagePosition", return_value=10.0), \ 

346 mock.patch(f"{MODULE}.getFilterAndDisperserFromExp", 

347 return_value=('SDSSr', 'other')): 

348 with self.assertRaises(ValueError): 

349 self.task.run(inputExp=exp, 

350 inputCentroid={'astrometricMatch': True, 

351 'centroid': (1, 2)}, 

352 dataIdDict={}) 

353 

354 def testRunWithDictCentroid(self): 

355 exp = self._makeRunExposure() 

356 inputCentroid = {'astrometricMatch': True, 'centroid': (5.0, 6.0)} 

357 result, shim, instance = self._runWithMocks(exp, inputCentroid) 

358 shim.assert_called_once() 

359 instance.run.assert_called_once() 

360 self.assertTrue(hasattr(result, "spectractorSpectrum")) 

361 

362 def testRunWithTupleCentroid(self): 

363 # A raw tuple centroid is only meaningful when the centroid method is 

364 # not 'auto' (auto expects the dict form), so use an 'exact' config. 

365 config = ProcessStarTaskConfig() 

366 config.targetCentroidMethod = 'exact' 

367 self.task = ProcessStarTask(config=config) 

368 exp = self._makeRunExposure() 

369 inputCentroid = (5.0, 6.0) 

370 result, shim, instance = self._runWithMocks(exp, inputCentroid) 

371 instance.run.assert_called_once() 

372 

373 def testRunHologramGrating(self): 

374 exp = self._makeRunExposure() 

375 inputCentroid = {'astrometricMatch': False, 'centroid': (5.0, 6.0)} 

376 result, shim, instance = self._runWithMocks(exp, inputCentroid, 

377 grating='holo4_003') 

378 # 4mm window added to the linear stage position of 10.0 

379 _, kwargs = shim.call_args 

380 self.assertEqual(kwargs['paramOverrides']['DISTANCE2CCD'], 14.0) 

381 

382 def testRunForceObjectName(self): 

383 config = ProcessStarTaskConfig() 

384 config.forceObjectName = 'HD54321' 

385 task = ProcessStarTask(config=config) 

386 exp = self._makeRunExposure() 

387 inputCentroid = {'astrometricMatch': True, 'centroid': (5.0, 6.0)} 

388 with mock.patch(f"{MODULE}.isDispersedExp", return_value=True), \ 

389 mock.patch(f"{MODULE}.getLinearStagePosition", return_value=10.0), \ 

390 mock.patch(f"{MODULE}.getFilterAndDisperserFromExp", 

391 return_value=('SDSSr', 'other')), \ 

392 mock.patch(f"{MODULE}.SpectractorShim") as shim: 

393 shim.return_value.run.return_value = mock.MagicMock() 

394 task.run(inputExp=exp, inputCentroid=inputCentroid, 

395 dataIdDict={'visit': 1}) 

396 args, _ = shim.return_value.run.call_args 

397 # target is the final positional argument passed to shim.run 

398 self.assertEqual(args[3], 'HD54321') 

399 

400 def testRunAstrometrySuccess(self): 

401 exp = afwImage.ExposureF(10, 10) 

402 with mock.patch(f"{MODULE}.ReferenceObjectLoader"), \ 

403 mock.patch(f"{MODULE}.FitAffineWcsTask"), \ 

404 mock.patch(f"{MODULE}.AstrometryTask") as astromTask: 

405 solver = astromTask.return_value 

406 astromResult = mock.MagicMock() 

407 astromResult.scatterOnSky.asArcseconds.return_value = 0.5 

408 solver.run.return_value = astromResult 

409 result = self.task.runAstrometry(butler=None, exp=exp, icSrc=None) 

410 self.assertIs(result, astromResult) 

411 

412 def testRunAstrometryPoorScatter(self): 

413 exp = afwImage.ExposureF(10, 10) 

414 with mock.patch(f"{MODULE}.ReferenceObjectLoader"), \ 

415 mock.patch(f"{MODULE}.FitAffineWcsTask"), \ 

416 mock.patch(f"{MODULE}.AstrometryTask") as astromTask: 

417 solver = astromTask.return_value 

418 astromResult = mock.MagicMock() 

419 astromResult.scatterOnSky.asArcseconds.return_value = 5.0 

420 solver.run.return_value = astromResult 

421 result = self.task.runAstrometry(butler=None, exp=exp, icSrc=None) 

422 self.assertIsNone(result) 

423 

424 def testRunAstrometryFailure(self): 

425 exp = afwImage.ExposureF(10, 10) 

426 with mock.patch(f"{MODULE}.ReferenceObjectLoader"), \ 

427 mock.patch(f"{MODULE}.FitAffineWcsTask"), \ 

428 mock.patch(f"{MODULE}.AstrometryTask") as astromTask: 

429 solver = astromTask.return_value 

430 solver.run.side_effect = RuntimeError("boom") 

431 result = self.task.runAstrometry(butler=None, exp=exp, icSrc=None) 

432 self.assertIsNone(result) 

433 

434 

435class InitDebugTestCase(lsst.utils.tests.TestCase): 

436 """Exercise the debug-display branches of ProcessStarTask.__init__.""" 

437 

438 def _makeDebugInfo(self, **kwargs): 

439 defaults = dict(enabled=False, display=False, displayBackend='virtualDevice', 

440 notHeadless=False, pauseOnDisplay=False) 

441 defaults.update(kwargs) 

442 return types.SimpleNamespace(**defaults) 

443 

444 def testInitDebugEnabledNoDisplay(self): 

445 debug = self._makeDebugInfo(enabled=True, display=False, notHeadless=True) 

446 with mock.patch(f"{MODULE}.lsstDebug.Info", return_value=debug): 

447 task = ProcessStarTask(config=ProcessStarTaskConfig()) 

448 self.assertTrue(task.debug.enabled) 

449 

450 def testInitDebugDisplaySuccess(self): 

451 import lsst.afw.display as afwDisp 

452 debug = self._makeDebugInfo(enabled=True, display=True, notHeadless=False) 

453 with mock.patch(f"{MODULE}.lsstDebug.Info", return_value=debug), \ 

454 mock.patch.object(afwDisp, "setDefaultBackend"), \ 

455 mock.patch.object(afwDisp, "setDefaultMaskTransparency"), \ 

456 mock.patch.object(afwDisp, "Display") as displayMock: 

457 task = ProcessStarTask(config=ProcessStarTaskConfig()) 

458 self.assertTrue(task.debug.display) 

459 displayMock.assert_called() 

460 

461 def testInitDebugDisplayNameError(self): 

462 # Force the display setup to raise NameError, hitting the except block. 

463 import lsst.afw.display as afwDisp 

464 debug = self._makeDebugInfo(enabled=True, display=True, notHeadless=False) 

465 with mock.patch(f"{MODULE}.lsstDebug.Info", return_value=debug), \ 

466 mock.patch.object(afwDisp, "setDefaultBackend", 

467 side_effect=NameError("boom")): 

468 task = ProcessStarTask(config=ProcessStarTaskConfig()) 

469 # display is disabled after the NameError. 

470 self.assertFalse(task.debug.display) 

471 

472 

473class TestMemory(lsst.utils.tests.MemoryTestCase): 

474 pass 

475 

476 

477def setup_module(module): 

478 lsst.utils.tests.init() 

479 

480 

481if __name__ == "__main__": 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true

482 lsst.utils.tests.init() 

483 unittest.main()