Coverage for tests/test_transforms.py: 32%
134 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-22 01:54 -0700
1# This file is part of lsst-images.
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# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14import dataclasses
15import functools
16import os
17import unittest
18from typing import Any, ClassVar
20import numpy as np
21import pydantic
23from lsst.images import (
24 Box,
25 CameraFrameSet,
26 CameraFrameSetSerializationModel,
27 DetectorFrame,
28 FocalPlaneFrame,
29 SkyProjection,
30 Transform,
31 TransformSerializationModel,
32)
33from lsst.images.fits import PointerModel
34from lsst.images.serialization import ArchiveTree, InputArchive, JsonRef, OutputArchive
35from lsst.images.tests import (
36 DP2_VISIT_DETECTOR_DATA_ID,
37 RoundtripFits,
38 RoundtripJson,
39 check_transform,
40 compare_sky_projection_to_legacy_wcs,
41 legacy_points_to_xy_array,
42)
44DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None)
47class TransformTestCase(unittest.TestCase):
48 """Tests for the Transform, SkyProjection, and FrameSet classes."""
50 def test_identity(self) -> None:
51 """Test an identity transform."""
52 frame = DetectorFrame(**DP2_VISIT_DETECTOR_DATA_ID, bbox=Box.factory[:5, :4])
53 xy = frame.bbox.meshgrid().map(np.ravel)
54 identity = Transform.identity(frame)
55 check_transform(self, identity, xy, xy, frame, frame)
56 self.assertEqual(identity.decompose(), [])
57 with RoundtripJson(self, identity) as roundtrip:
58 pass
59 check_transform(self, roundtrip.result, xy, xy, frame, frame)
61 @unittest.skipUnless(DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not in the environment.")
62 def test_camera(self) -> None:
63 """Test that we can:
65 - make a CameraFrameSet from the AST representation returned by afw;
66 - transform points and get the same result as afw;
67 - round-trip the CameraFrameSet through FITS serialization and still
68 do all of that;
69 - also roundtrip a Transform that can be obtained from the
70 CameraFrameSet, by referencing the mappings in the frame set.
72 This test is skipped if legacy modules cannot be imported.
74 This test provides coverage for the archive system's pointer and
75 frame-set reference machinery.
76 """
77 try:
78 from lsst.afw.cameraGeom import Camera
79 except ImportError:
80 raise unittest.SkipTest("'lsst.afw.cameraGeom' could not be imported.") from None
81 assert DATA_DIR is not None, "Guaranteed by decorator."
82 filename = os.path.join(DATA_DIR, "dp2", "legacy", "camera.fits")
83 legacy_camera = Camera.readFits(filename)
84 frame_set = CameraFrameSet.from_legacy(legacy_camera)
85 detector_id: int = DP2_VISIT_DETECTOR_DATA_ID["detector"]
86 self.compare_to_legacy_camera(legacy_camera, frame_set)
87 test_holder = FrameSetTestHolder(
88 frames=frame_set,
89 pixels_to_fp=frame_set[frame_set.detector(detector_id), frame_set.focal_plane()],
90 )
91 with RoundtripFits(self, test_holder) as roundtrip1:
92 self.assertEqual(len(roundtrip1.serialized.pixels_to_fp.frames), 2)
93 self.assertEqual(len(roundtrip1.serialized.pixels_to_fp.bounds), 2)
94 self.assertEqual(len(roundtrip1.serialized.pixels_to_fp.mappings), 1)
95 # Instead of storing the AST mapping directly, we should have
96 # stored a reference to the frame set:
97 self.assertIsInstance(roundtrip1.serialized.pixels_to_fp.mappings[0], PointerModel)
98 self.compare_to_legacy_camera(legacy_camera, roundtrip1.result.frames)
99 self.assertEqual(roundtrip1.result.pixels_to_fp.in_frame, frame_set.detector(detector_id))
100 self.assertEqual(roundtrip1.result.pixels_to_fp.out_frame, frame_set.focal_plane())
101 self.assertEqual(
102 roundtrip1.result.pixels_to_fp._ast_mapping.simplified().show(),
103 test_holder.pixels_to_fp._ast_mapping.simplified().show(),
104 )
105 with RoundtripJson(self, test_holder) as roundtrip2:
106 self.assertEqual(len(roundtrip2.serialized.pixels_to_fp.frames), 2)
107 self.assertEqual(len(roundtrip2.serialized.pixels_to_fp.bounds), 2)
108 self.assertEqual(len(roundtrip2.serialized.pixels_to_fp.mappings), 1)
109 # Instead of storing the AST mapping directly, we should have
110 # stored a reference to the frame set:
111 self.assertIsInstance(roundtrip2.serialized.pixels_to_fp.mappings[0], JsonRef)
112 raw_data = roundtrip2.inspect()
113 self.assertEqual(len(raw_data["indirect"]), 1)
114 self.assertEqual(raw_data["frames"], {"$ref": "#/indirect/0"})
115 self.compare_to_legacy_camera(legacy_camera, roundtrip2.result.frames)
116 self.assertEqual(roundtrip2.result.pixels_to_fp.in_frame, frame_set.detector(detector_id))
117 self.assertEqual(roundtrip2.result.pixels_to_fp.out_frame, frame_set.focal_plane())
118 self.assertEqual(
119 roundtrip2.result.pixels_to_fp._ast_mapping.simplified().show(),
120 test_holder.pixels_to_fp._ast_mapping.simplified().show(),
121 )
123 def compare_to_legacy_camera(self, legacy_camera: Any, frame_set: CameraFrameSet) -> None:
124 """Test the transforms extracted from a CameraFrameSet against the
125 legacy lsst.afw.cameraGeom implementations.
126 """
127 from lsst.afw.cameraGeom import FIELD_ANGLE, FOCAL_PLANE, PIXELS
128 from lsst.geom import Point2D
130 legacy_detector = legacy_camera[16]
131 pixel_legacy_points = [Point2D(50.0, 60.0), Point2D(801.2, 322.8), Point2D(33.5, 22.1)]
132 fp_legacy_points = [legacy_detector.transform(p, PIXELS, FOCAL_PLANE) for p in pixel_legacy_points]
133 fa_legacy_points = [legacy_detector.transform(p, PIXELS, FIELD_ANGLE) for p in pixel_legacy_points]
134 pixel_xy_array = legacy_points_to_xy_array(pixel_legacy_points)
135 fp_xy_array = legacy_points_to_xy_array(fp_legacy_points)
136 fa_xy_array = legacy_points_to_xy_array(fa_legacy_points)
137 # Test transforms extracted directly from the frame set.
138 pixel_to_fp = frame_set[frame_set.detector(16), frame_set.focal_plane()]
139 check_transform(
140 self, pixel_to_fp, pixel_xy_array, fp_xy_array, frame_set.detector(16), frame_set.focal_plane()
141 )
142 pixel_to_fa = frame_set[frame_set.detector(16), frame_set.field_angle()]
143 check_transform(
144 self, pixel_to_fa, pixel_xy_array, fa_xy_array, frame_set.detector(16), frame_set.field_angle()
145 )
146 fp_to_fa = frame_set[frame_set.focal_plane(), frame_set.field_angle()]
147 check_transform(
148 self, fp_to_fa, fp_xy_array, fa_xy_array, frame_set.focal_plane(), frame_set.field_angle()
149 )
150 # Test a composition.
151 pixel_to_fa_indirect = pixel_to_fp.then(fp_to_fa)
152 check_transform(
153 self,
154 pixel_to_fa_indirect,
155 pixel_xy_array,
156 fa_xy_array,
157 frame_set.detector(16),
158 frame_set.field_angle(),
159 )
160 pixel_to_fp_d, fp_to_fa_d = pixel_to_fa_indirect.decompose()
161 check_transform(
162 self, pixel_to_fp_d, pixel_xy_array, fp_xy_array, frame_set.detector(16), frame_set.focal_plane()
163 )
164 check_transform(
165 self, fp_to_fa_d, fp_xy_array, fa_xy_array, frame_set.focal_plane(), frame_set.field_angle()
166 )
167 fa_to_fp_d, fp_to_pixel_d = pixel_to_fa_indirect.inverted().decompose()
168 check_transform(
169 self, fa_to_fp_d, fa_xy_array, fp_xy_array, frame_set.field_angle(), frame_set.focal_plane()
170 )
171 check_transform(
172 self, fp_to_pixel_d, fp_xy_array, pixel_xy_array, frame_set.focal_plane(), frame_set.detector(16)
173 )
175 @unittest.skipUnless(DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not in the environment.")
176 def test_detector_wcs(self) -> None:
177 """Test the Transform/SkyProjection representation of a detector
178 WCS.
179 """
180 try:
181 from lsst.afw.image import ExposureFitsReader
182 except ImportError:
183 raise unittest.SkipTest("'lsst.afw.image' could not be imported.") from None
184 assert DATA_DIR is not None, "Guaranteed by decorator."
185 filename = os.path.join(DATA_DIR, "dp2", "legacy", "visit_image.fits")
186 reader = ExposureFitsReader(filename)
187 legacy_wcs = reader.readWcs()
188 wcs_bbox = Box.from_legacy(reader.readDetector().getBBox())
189 subimage_bbox = Box.from_legacy(reader.readBBox())
190 detector_frame = DetectorFrame(**DP2_VISIT_DETECTOR_DATA_ID, bbox=wcs_bbox)
191 sky_projection = SkyProjection.from_legacy(legacy_wcs, detector_frame)
192 assert sky_projection.fits_approximation is not None
193 compare_sky_projection_to_legacy_wcs(self, sky_projection, legacy_wcs, detector_frame, subimage_bbox)
194 # When we convert from a legacy SkyWcs, the internal AST Mapping needs
195 # to really be an AST FrameSet in order to be able to convert back.
196 self.assertIn("Begin FrameSet", sky_projection.show())
197 compare_sky_projection_to_legacy_wcs(
198 self, sky_projection, sky_projection.to_legacy(), detector_frame, subimage_bbox
199 )
200 self.assertIn("Begin FrameSet", sky_projection.fits_approximation.show())
201 compare_sky_projection_to_legacy_wcs(
202 self,
203 sky_projection.fits_approximation,
204 sky_projection.fits_approximation.to_legacy(),
205 detector_frame,
206 subimage_bbox,
207 is_fits=True,
208 )
209 with RoundtripJson(self, sky_projection, "SkyProjection") as roundtrip:
210 pass
211 compare_sky_projection_to_legacy_wcs(
212 self, roundtrip.result, legacy_wcs, detector_frame, subimage_bbox
213 )
214 # The AST FrameSet-ness needs to propagate through serialization.
215 self.assertIn("Begin FrameSet", roundtrip.result.show())
216 compare_sky_projection_to_legacy_wcs(
217 self, sky_projection, roundtrip.result.to_legacy(), detector_frame, subimage_bbox
218 )
219 with RoundtripJson(self, sky_projection.fits_approximation, "SkyProjection") as roundtrip:
220 pass
221 compare_sky_projection_to_legacy_wcs(
222 self,
223 roundtrip.result,
224 legacy_wcs.getFitsApproximation(),
225 detector_frame,
226 subimage_bbox,
227 is_fits=True,
228 )
229 self.assertIn("Begin FrameSet", roundtrip.result.show())
230 compare_sky_projection_to_legacy_wcs(
231 self,
232 sky_projection.fits_approximation,
233 roundtrip.result.to_legacy(),
234 detector_frame,
235 subimage_bbox,
236 is_fits=True,
237 )
240@dataclasses.dataclass
241class FrameSetTestHolder:
242 """A top-level object that holds a CameraFrameSet and a transform
243 extracted from it, for testing archive pointers and frame set references.
244 """
246 frames: CameraFrameSet
247 pixels_to_fp: Transform[DetectorFrame, FocalPlaneFrame]
249 def serialize[P: pydantic.BaseModel](self, archive: OutputArchive[P]) -> FrameSetTestHolderModel[P]:
250 frames_model = archive.serialize_frame_set(
251 "frames", self.frames, self.frames.serialize, key=id(self.frames)
252 )
253 pixels_to_fp_model = archive.serialize_direct(
254 "pixels_to_fp", functools.partial(self.pixels_to_fp.serialize, use_frame_sets=True)
255 )
256 return FrameSetTestHolderModel[P](frames=frames_model, pixels_to_fp=pixels_to_fp_model)
258 @staticmethod
259 def _get_archive_tree_type[P: pydantic.BaseModel](
260 pointer_type: type[P],
261 ) -> type[FrameSetTestHolderModel[P]]:
262 return FrameSetTestHolderModel[pointer_type] # type: ignore
265class FrameSetTestHolderModel[P: pydantic.BaseModel](ArchiveTree):
266 """The serialization model for FrameSetTestHolder."""
268 SCHEMA_NAME: ClassVar[str] = "_test_frame_set_holder"
269 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
270 MIN_READ_VERSION: ClassVar[int] = 1
271 PUBLIC_TYPE: ClassVar[type] = FrameSetTestHolder
273 frames: CameraFrameSetSerializationModel | P
274 pixels_to_fp: TransformSerializationModel[P]
276 def deserialize(self, archive: InputArchive[Any]) -> FrameSetTestHolder:
277 assert not isinstance(self.frames, CameraFrameSetSerializationModel), "Archive pointer expected."
278 frames = archive.deserialize_pointer(
279 self.frames, CameraFrameSetSerializationModel, CameraFrameSetSerializationModel.deserialize
280 )
281 pixels_to_fp = self.pixels_to_fp.deserialize(archive)
282 return FrameSetTestHolder(frames, pixels_to_fp)
285if __name__ == "__main__":
286 unittest.main()