Coverage for tests / test_mergeResults.py: 16%
77 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-25 08:56 +0000
1# This file is part of cp_verify.
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/>.
21import numpy as np
22import unittest
24from astropy.table import Table
26import lsst.utils.tests
27import lsst.afw.cameraGeom as cameraGeom
28import lsst.cp.verify as cpVerify
31class MergeResultsTestCase(lsst.utils.tests.TestCase):
32 """Unit test for merge code.
33 """
35 def setUp(self):
36 # Generate a camera.
37 self.camera = cameraGeom.testUtils.CameraWrapper(isLsstLike=False).camera
39 # Fix the seed so we can expect consistent results.
40 np.random.seed(12345)
42 # Make first exposure
43 self.detectorResults1 = list()
44 self.detectorDimensions1 = list()
45 for detector in self.camera:
46 dimensions = {'exposure': 1, 'detector': detector.getId(), 'instrument': 'testCam'}
47 self.detectorDimensions1.append(dimensions)
48 self.detectorResults1.append(self.generateDetectorResults(detector))
50 # Make second exposure
51 self.detectorResults2 = list()
52 self.detectorDimensions2 = list()
53 for detector in self.camera:
54 dimensions = {'exposure': 2, 'detector': detector.getId(), 'instrument': 'testCam'}
55 self.detectorDimensions2.append(dimensions)
56 self.detectorResults2.append(self.generateDetectorResults(detector, forceFailure=True))
58 def test_merging(self):
59 """Generate simulated verify statistics, and prove they merge
60 correctly.
61 """
62 # Generate table versions of the detector results.
63 verifyStatsTask = cpVerify.CpVerifyStatsTask()
64 expTables1 = [Table(verifyStatsTask.repackStats(rr, dd)[0]) for rr, dd in
65 zip(self.detectorResults1, self.detectorDimensions1)]
66 expTables2 = [Table(verifyStatsTask.repackStats(rr, dd)[0]) for rr, dd in
67 zip(self.detectorResults2, self.detectorDimensions2)]
69 # Ensure we're merging the tables as well.
70 expMergeConfig = cpVerify.CpVerifyExpMergeTask.ConfigClass()
71 expMergeConfig.hasInputResults = True
73 # Do the merge.
74 expMergeTask = cpVerify.CpVerifyExpMergeTask(config=expMergeConfig)
75 expResults1 = expMergeTask.run(self.detectorResults1, self.detectorDimensions1,
76 self.camera, inputResults=expTables1)
77 expResults2 = expMergeTask.run(self.detectorResults2, self.detectorDimensions2,
78 self.camera, inputResults=expTables2)
79 self.assertTrue(expResults1.outputStats['SUCCESS'])
80 self.assertFalse(expResults2.outputStats['SUCCESS']) # Expected to have one failure
81 self.assertEqual(expResults2.outputStats['R:1,0 S:1,0']['FAILURES'][0], "C:0,2 SIGMA_TEST")
83 # Prep inputs to full calibration run merge.
84 inputStats = [expResults1.outputStats, expResults2.outputStats]
85 inputDims = [{'exposure': 1}, {'exposure': 2}]
87 runMergeTask = cpVerify.CpVerifyRunMergeTask()
89 runResults = runMergeTask.run(inputStats, inputDims, self.camera)
90 self.assertFalse(runResults.outputStats['SUCCESS'])
92 # We know this one failed.
93 failureList = runResults.outputStats[2]['FAILURES']
94 self.assertEqual(len(failureList), 1)
95 self.assertEqual(failureList[0], "R:1,0 S:1,0 C:0,2 SIGMA_TEST")
97 def generateDetectorResults(self, detector, mean=10.0, sigma=1.2, forceFailure=False):
98 """Make the simulated verify statistics.
100 Parameters
101 ----------
102 detector : `lsst.afw.cameraGeom.Detector`
103 Detector geometry to make statistics for.
104 mean : `float`
105 Center of the random normal distribution to pull
106 statistics from.
107 sigma : `float`
108 Sigma of the normal distribution.
109 forceFailure : `bool`
110 If True, force the "verify" to fail.
112 Returns
113 -------
114 detectorStats : `dict` [`str`, `dict`]
115 Nested dictionary of verify statistics.
116 """
118 valueNames = ['MEAN', 'SIGMA', 'VALUE']
120 ampDict = dict()
121 detDict = dict()
122 catDict = dict()
123 success = True
124 for amp in detector.getAmplifiers():
125 ampName = amp.getName()
126 ampDict[ampName] = dict()
127 ampDict[ampName]['VECTOR'] = []
128 for value in valueNames:
129 ampDict[ampName][value] = np.random.normal(10.0, 1.2)
130 ampDict[ampName]['VECTOR'].append(np.random.uniform(size=len(detector)))
132 ampVerify = dict()
133 detVerify = dict()
134 catVerify = dict()
135 expVerify = dict()
136 for ampName in ampDict:
137 ampVerify[ampName] = dict()
138 ampSuccess = True
139 for value in valueNames:
140 if forceFailure and (value == "SIGMA"
141 and ampName == "C:0,2"
142 and detector.getName() == 'R:1,0 S:1,0'):
143 ampVerify[ampName][value + "_TEST"] = False
144 ampSuccess = False
145 success = False
146 else:
147 ampVerify[ampName][value + "_TEST"] = True
148 ampVerify[ampName]['SUCCESS'] = ampSuccess
150 return {'SUCCESS': success, 'AMP': ampDict, 'DET': detDict, 'CATALOG': catDict,
151 'VERIFY': {'AMP': ampVerify, 'DET': detVerify, 'CATALOG': catVerify, 'EXP': expVerify}}
154class MemoryTester(lsst.utils.tests.MemoryTestCase):
155 pass
158def setup_module(module):
159 lsst.utils.tests.init()
162if __name__ == "__main__": 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 lsst.utils.tests.init()
164 unittest.main()