Coverage for tests / test_fitsRawFormatter.py: 38%
98 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-07 08:21 +0000
1# This file is part of obs_base.
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/>.
22import unittest
24import astropy.units as u
25from astro_metadata_translator import FitsTranslator, StubTranslator
26from astro_metadata_translator.translators.helpers import tracking_from_degree_headers
27from astropy.coordinates import Angle
29import lsst.afw.geom
30import lsst.afw.math
31import lsst.daf.base
32import lsst.daf.butler
33import lsst.geom
34import lsst.resources
35import lsst.utils.tests
36from lsst.afw.cameraGeom import makeUpdatedDetector
37from lsst.afw.cameraGeom.testUtils import CameraWrapper, DetectorWrapper
38from lsst.obs.base import (
39 FilterDefinition,
40 FilterDefinitionCollection,
41 FitsRawFormatterBase,
42 MakeRawVisitInfoViaObsInfo,
43)
44from lsst.obs.base.tests import make_ramp_exposure_untrimmed
45from lsst.obs.base.utils import InitialSkyWcsError, createInitialSkyWcs
48class SimpleTestingTranslator(FitsTranslator, StubTranslator):
49 """Simple `~astro_metadata_translator.MetadataTranslator` used for
50 testing.
51 """
53 _const_map = {
54 "boresight_rotation_angle": Angle(90 * u.deg),
55 "boresight_rotation_coord": "sky",
56 "detector_exposure_id": 12345,
57 # The following are defined to prevent warnings about
58 # undefined translators
59 "dark_time": 0.0 * u.s,
60 "exposure_time": 0.0 * u.s,
61 "physical_filter": "u",
62 "detector_num": 0,
63 "detector_name": "0",
64 "detector_group": "",
65 "detector_unique_name": "0",
66 "detector_serial": "",
67 "observation_id": "--",
68 "science_program": "unknown",
69 "object": "unknown",
70 "exposure_id": 0,
71 "visit_id": 0,
72 "relative_humidity": 30.0,
73 "pressure": 0.0 * u.MPa,
74 "temperature": 273 * u.K,
75 "altaz_begin": None,
76 }
77 _trivial_map = {"boresight_airmass": "AIRMASS", "observation_type": "OBSTYPE"}
79 def to_tracking_radec(self):
80 radecsys = ("RADESYS",)
81 radecpairs = (("RA", "DEC"),)
82 return tracking_from_degree_headers(self, radecsys, radecpairs, unit=(u.deg, u.deg))
85class MakeTestingRawVisitInfo(MakeRawVisitInfoViaObsInfo):
86 """Test class for VisitInfo creation."""
88 metadataTranslator = SimpleTestingTranslator
91class SimpleFitsRawFormatter(FitsRawFormatterBase):
92 """Simple test formatter for datastore interaction."""
94 filterDefinitions = FilterDefinitionCollection(FilterDefinition(physical_filter="u", band="u"))
96 @property
97 def translatorClass(self):
98 return SimpleTestingTranslator
100 def getDetector(self, id):
101 """Use CameraWrapper to create a fake detector that can map from
102 PIXELS to FIELD_ANGLE.
104 Always return Detector #10, so all the tests are self-consistent, and
105 make sure it is in "assembled" form, since that's what the base
106 formatter implementations assume.
107 """
108 return makeUpdatedDetector(CameraWrapper().camera.get(10))
111class FitsRawFormatterTestCase(lsst.utils.tests.TestCase):
112 """Test that we can read and write FITS files with butler."""
114 def setUp(self):
115 # The FITS WCS and VisitInfo coordinates in this header are
116 # intentionally different, to make comparisons between them more
117 # obvious.
118 self.boresight = lsst.geom.SpherePoint(10.0, 20.0, lsst.geom.degrees)
119 self.header = {
120 "TELESCOP": "TEST",
121 "INSTRUME": "UNKNOWN",
122 "AIRMASS": 1.2,
123 "RADESYS": "ICRS",
124 "OBSTYPE": "science",
125 "EQUINOX": 2000,
126 "OBSGEO-X": "-5464588.84421314",
127 "OBSGEO-Y": "-2493000.19137644",
128 "OBSGEO-Z": "2150653.35350771",
129 "RA": self.boresight.getLatitude().asDegrees(),
130 "DEC": self.boresight.getLongitude().asDegrees(),
131 "CTYPE1": "RA---SIN",
132 "CTYPE2": "DEC--SIN",
133 "CRPIX1": 5,
134 "CRPIX2": 6,
135 "CRVAL1": self.boresight.getLatitude().asDegrees() + 1,
136 "CRVAL2": self.boresight.getLongitude().asDegrees() + 1,
137 "CD1_1": 1e-5,
138 "CD1_2": 0,
139 "CD2_2": 1e-5,
140 "CD2_1": 0,
141 }
142 # make a property list of the above, for use by the formatter.
143 self.metadata = lsst.daf.base.PropertyList()
144 self.metadata.update(self.header)
146 maker = MakeTestingRawVisitInfo()
147 self.visitInfo = maker(self.header)
149 self.metadataSkyWcs = lsst.afw.geom.makeSkyWcs(self.metadata, strip=False)
150 self.boresightSkyWcs = createInitialSkyWcs(self.visitInfo, CameraWrapper().camera.get(10))
152 # set this to `contextlib.nullcontext()` to print the log warnings
153 self.warnContext = self.assertLogs(level="WARNING")
155 # Make a ref to pass to the formatter.
156 universe = lsst.daf.butler.DimensionUniverse()
157 dataId = lsst.daf.butler.DataCoordinate.standardize(
158 instrument="Cam1", exposure=2, detector=10, physical_filter="u", band="u", universe=universe
159 )
160 datasetType = lsst.daf.butler.DatasetType(
161 "dummy",
162 dimensions=("instrument", "exposure", "detector"),
163 storageClass=lsst.daf.butler.StorageClass(),
164 universe=universe,
165 )
166 ref = lsst.daf.butler.DatasetRef(dataId=dataId, datasetType=datasetType, run="test")
168 # We have no file in these tests, so make an empty descriptor.
169 fileDescriptor = lsst.daf.butler.FileDescriptor(None, None)
170 self.formatter = SimpleFitsRawFormatter(fileDescriptor, ref=ref)
171 # Force the formatter's metadata to be what we've created above.
172 self.formatter._metadata = self.metadata
174 def test_makeWcs(self):
175 detector = self.formatter.getDetector(1)
176 wcs = self.formatter.makeWcs(self.visitInfo, detector)
177 self.assertNotEqual(wcs, self.metadataSkyWcs)
178 self.assertEqual(wcs, self.boresightSkyWcs)
180 def test_makeWcs_if_metadata_is_bad(self):
181 """Always use the VisitInfo WCS if available."""
182 detector = self.formatter.getDetector(1)
183 self.metadata.remove("CTYPE1")
184 wcs = self.formatter.makeWcs(self.visitInfo, detector)
185 self.assertNotEqual(wcs, self.metadataSkyWcs)
186 self.assertEqual(wcs, self.boresightSkyWcs)
188 def test_makeWcs_warn_if_visitInfo_is_None(self):
189 """If VisitInfo is None, log a warning and use the metadata WCS."""
190 detector = self.formatter.getDetector(1)
191 with self.warnContext:
192 wcs = self.formatter.makeWcs(None, detector)
193 self.assertEqual(wcs, self.metadataSkyWcs)
194 self.assertNotEqual(wcs, self.boresightSkyWcs)
196 def test_makeWcs_fail_if_visitInfo_is_None(self):
197 """If VisitInfo is None and metadata failed, raise an exception."""
198 detector = self.formatter.getDetector(1)
199 self.metadata.remove("CTYPE1")
200 with self.warnContext, self.assertRaises(InitialSkyWcsError):
201 self.formatter.makeWcs(None, detector)
203 def test_makeWcs_fail_if_detector_is_bad(self):
204 """If Detector is broken, raise an exception."""
205 # This detector doesn't know about FIELD_ANGLE, so can't be used to
206 # make a SkyWcs.
207 detector = DetectorWrapper().detector
208 with self.assertRaises(InitialSkyWcsError):
209 self.formatter.makeWcs(self.visitInfo, detector)
211 def test_amp_parameter(self):
212 """Test loading subimages with the 'amp' parameter."""
213 with lsst.utils.tests.getTempFilePath(".fits") as tmpFile:
214 # Get a detector; this must be the same one that's baked into the
215 # simple formatter at the top of this file, so that's how we get
216 # it.
217 detector = self.formatter.getDetector(1)
218 # Make full exposure with ramp values and save just the image to
219 # the temp file (with metadata), so it looks like a raw.
220 full = make_ramp_exposure_untrimmed(detector)
221 full.image.writeFits(tmpFile, metadata=self.metadata)
222 # Loop over amps and try to read them via the formatter.
223 for n, amp in enumerate(detector):
224 for amp_parameter in [amp, amp.getName(), n]:
225 for parameters in [{"amp": amp_parameter}, {"amp": amp_parameter, "detector": detector}]:
226 with self.subTest(parameters=repr(parameters)):
227 # Make a new formatter that points at the new file
228 # and has the right parameters.
229 formatter = SimpleFitsRawFormatter(
230 lsst.daf.butler.FileDescriptor(
231 lsst.daf.butler.Location(None, path=lsst.resources.ResourcePath(tmpFile)),
232 lsst.daf.butler.StorageClassFactory().getStorageClass("ExposureI"),
233 parameters=parameters,
234 ),
235 ref=self.formatter.dataset_ref,
236 )
237 subexp = formatter.read()
238 self.assertImagesEqual(subexp.image, full[amp.getRawBBox()].image)
239 self.assertEqual(len(subexp.getDetector()), 1)
240 self.assertAmplifiersEqual(subexp.getDetector()[0], amp)
241 self.assertEqual(subexp.visitInfo.id, 2)
242 # We could try transformed amplifiers here that involve flips
243 # and offsets, but:
244 # - we already test the low-level code that does that in afw;
245 # - we test very similar high-level code (which calls that
246 # same afw code) in the non-raw Exposure formatter, in
247 # test_butlerFits.py;
248 # - the only instruments that actually have those kinds of
249 # amplifiers are those in obs_lsst, and that has a different
250 # raw formatter implementation that we need to test there
251 # anyway;
252 # - these are kind of expensive tests.
255class MemoryTester(lsst.utils.tests.MemoryTestCase):
256 """Check for file leaks."""
259def setup_module(module):
260 """Initialize pytest."""
261 lsst.utils.tests.init()
264if __name__ == "__main__": 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 lsst.utils.tests.init()
266 unittest.main()