Coverage for python/lsst/images/_transforms/_camera_frame_set.py: 57%
107 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 08:55 +0000
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
14__all__ = ("CameraFrameSet", "CameraFrameSetSerializationModel")
16from typing import Any, ClassVar
18import astropy.units as u
19import pydantic
21from .._geom import Bounds, Box
22from ..serialization import ArchiveTree, InputArchive, InvalidParameterError, OutputArchive
23from . import _ast as astshim
24from . import _frames # use this import style to facilitate pattern matching
25from ._frame_set import FrameLookupError, FrameSet
26from ._transform import Transform
29class CameraFrameSet(FrameSet):
30 """A `FrameSet` that manages the coordinate systems of a camera.
32 The `CameraFrameSet` class constructor is considered a private
33 implementation detail. At present, instances can only be obtained by
34 loading them from storage (via
35 `~CameraFrameSetSerializationModel.deserialize`) or converting a legacy
36 `lsst.afw.cameraGeom` object (`from_legacy`).
38 Parameters
39 ----------
40 instrument
41 Short (butler dimension) name of the instrument.
42 ast
43 AST frame set describing the camera's coordinate systems.
44 """
46 # This constructor is kept private while we support both the astshim
47 # and starlink-pyast AST wrappers. For now:
48 # 'instrument': the short (butler dimension) name.
49 # 'ast': an astshim.FrameSet as returned by
50 # lsst.afw.cameraGeom.TransformMap.makeFrameSet.
51 # Should have frames with Ident values FOCAL_PLANE, FIELD_ANGLE
52 # and DETECTOR_${ID}, and the focal plane frame must know its
53 # units.
54 def __init__(self, instrument: str, ast: astshim.FrameSet) -> None:
55 self._ast = ast
56 self._focal_plane_frame_id: int = 0
57 self._field_angle_frame_id: int = 0
58 self._detector_frame_ids: dict[int, int] = {}
59 for frame_id in range(1, self._ast.nFrame + 1):
60 ast_frame = self._ast.getFrame(frame_id, copy=False)
61 match ast_frame.ident:
62 case "FOCAL_PLANE":
63 self._focal_plane_frame_id = frame_id
64 case "FIELD_ANGLE":
65 self._field_angle_frame_id = frame_id
66 case str(s) if s.startswith("DETECTOR_"): 66 ↛ 69line 66 didn't jump to line 69 because the pattern on line 66 always matched
67 detector_id = int(s.removeprefix("DETECTOR_"))
68 self._detector_frame_ids[detector_id] = frame_id
69 case _:
70 raise ValueError(f"Unexpected frame in camera AST FrameSet:\n{ast_frame.show()}.")
71 if self._focal_plane_frame_id == 0: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 raise ValueError("No FOCAL_PLANE frame in camera AST FrameSet.")
73 self._focal_plane_frame = _frames.FocalPlaneFrame(
74 instrument=instrument,
75 unit=u.Unit(self._ast.getFrame(self._focal_plane_frame_id, copy=False).getUnit(1)),
76 )
77 self._field_angle_frame = _frames.FieldAngleFrame(instrument=instrument)
78 if self._field_angle_frame_id == 0: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 raise ValueError("No FIELD_ANGLE frame in camera AST FrameSet.")
81 def __eq__(self, other: object) -> bool:
82 if type(other) is not CameraFrameSet: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 return NotImplemented
84 # Every cached attribute on this class is derived from ``_ast`` and
85 # the instrument name. Compare the *simplified* AST serialisations
86 # explicitly rather than relying on ``Object.__eq__``: what we care
87 # about is that the two frame sets describe the same transforms, and
88 # simplifying first makes the check about behaviour rather than the
89 # particular intermediate representation.
90 return (
91 self.instrument == other.instrument
92 and self._ast.simplified().show() == other._ast.simplified().show()
93 )
95 __hash__ = None # type: ignore[assignment]
97 @property
98 def instrument(self) -> str:
99 """Name of the instrument (`str`)."""
100 return self._focal_plane_frame.instrument
102 def focal_plane(self, visit: int | None = None) -> _frames.FocalPlaneFrame:
103 """Return a focal plane frame for this instrument.
105 Parameters
106 ----------
107 visit
108 ID for the visit this frame will correspond to. This only needs
109 to be provided in contexts where camera frames will be related to
110 the sky via a `SkyProjection`.
111 """
112 if visit is None:
113 return self._focal_plane_frame
114 else:
115 return self._focal_plane_frame.model_copy(update={"visit": visit})
117 def field_angle(self, visit: int | None = None) -> _frames.FieldAngleFrame:
118 """Return a field angle frame for this instrument.
120 Parameters
121 ----------
122 visit
123 ID for the visit this frame will correspond to. This only needs
124 to be provided in contexts where camera frames will be related to
125 the sky via a `SkyProjection`.
126 """
127 if visit is None:
128 return self._field_angle_frame
129 else:
130 return self._field_angle_frame.model_copy(update={"visit": visit})
132 def detector(self, detector: int, *, visit: int | None = None) -> _frames.DetectorFrame:
133 """Return a detector pixel-coordinate frame for this instrument.
135 Parameters
136 ----------
137 detector
138 ID of the detector.
139 visit
140 ID for the visit this frame will correspond to. This only needs
141 to be provided in contexts where camera frames will be related to
142 the sky via a `SkyProjection`.
143 """
144 try:
145 frame_id = self._detector_frame_ids[detector]
146 except KeyError:
147 raise FrameLookupError(
148 f"No frame for detector {detector!r} in camera for {self.instrument!r}."
149 ) from None
150 ast_frame = self._ast.getFrame(frame_id, copy=False)
151 bbox = Box.factory[
152 int(ast_frame.getBottom(2)) : int(ast_frame.getTop(2)),
153 int(ast_frame.getBottom(1)) : int(ast_frame.getTop(1)),
154 ]
155 return _frames.DetectorFrame(instrument=self.instrument, detector=detector, visit=visit, bbox=bbox)
157 def __contains__(self, frame: _frames.Frame) -> bool:
158 try:
159 self._parse_frame_arg(frame)
160 return True
161 except FrameLookupError:
162 return False
164 def __getitem__[I: _frames.Frame, O: _frames.Frame](self, key: tuple[I, O]) -> Transform[I, O]:
165 in_frame, out_frame = key
166 in_frame_id, in_bounds = self._parse_frame_arg(in_frame)
167 out_frame_id, out_bounds = self._parse_frame_arg(out_frame)
168 return Transform(
169 in_frame,
170 out_frame,
171 self._ast.getMapping(in_frame_id, out_frame_id),
172 in_bounds=in_bounds,
173 out_bounds=out_bounds,
174 )
176 def _parse_frame_arg(self, frame: _frames.Frame) -> tuple[int, Bounds | None]:
177 bounds: Bounds | None = None
178 match frame:
179 case _frames.DetectorFrame(instrument=self.instrument, detector=detector_id):
180 try:
181 frame_id = self._detector_frame_ids[detector_id]
182 except KeyError:
183 raise FrameLookupError(
184 f"No frame for detector {detector_id!r} in camera for {self.instrument!r}."
185 ) from None
186 bounds = frame.bbox
187 case _frames.FocalPlaneFrame(instrument=self.instrument):
188 frame_id = self._focal_plane_frame_id
189 case _frames.FieldAngleFrame(instrument=self.instrument):
190 frame_id = self._field_angle_frame_id
191 case _:
192 raise FrameLookupError(f"Invalid frame for camera {self.instrument}: {frame!r}.")
193 return frame_id, bounds
195 def serialize(self, archive: OutputArchive[Any]) -> CameraFrameSetSerializationModel:
196 """Serialize the frame set to an archive.
198 Parameters
199 ----------
200 archive
201 Archive to serialize to.
203 Returns
204 -------
205 `CameraFrameSetSerializationModel`
206 Serialized form of the frame set.
207 """
208 return CameraFrameSetSerializationModel(instrument=self.instrument, ast=self._ast.show())
210 @staticmethod
211 def _get_archive_tree_type(
212 pointer_type: type[pydantic.BaseModel],
213 ) -> type[CameraFrameSetSerializationModel]:
214 """Return the serialization model type for this object for an archive
215 type that uses the given pointer type.
216 """
217 return CameraFrameSetSerializationModel
219 @classmethod
220 def from_legacy(cls, camera: Any) -> CameraFrameSet:
221 """Construct a transform from a legacy `lsst.afw.cameraGeom.Camera`.
223 Parameters
224 ----------
225 camera
226 An `lsst.afw.cameraGeom.Camera` instance to convert.
227 """
228 transform_map = camera.getTransformMap()
229 ast_frame_set = transform_map.makeFrameSet(list(camera))
230 return CameraFrameSet("HSC", ast_frame_set)
233class CameraFrameSetSerializationModel(ArchiveTree):
234 """Serialization model for `CameraFrameSet`."""
236 SCHEMA_NAME: ClassVar[str] = "camera_frame_set"
237 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
238 MIN_READ_VERSION: ClassVar[int] = 1
239 PUBLIC_TYPE: ClassVar[type] = CameraFrameSet
241 instrument: str = pydantic.Field(description="Name of the instrument.")
242 ast: str = pydantic.Field(
243 description="A serialized Starlink AST FrameSet, using the AST native encoding."
244 )
246 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> CameraFrameSet:
247 """Deserialize a frame set from an archive.
249 Parameters
250 ----------
251 archive
252 Archive to read from.
253 **kwargs
254 Unsupported keyword arguments are accepted only to provide better
255 error messages (raising `serialization.InvalidParameterError`).
256 """
257 if kwargs: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 raise InvalidParameterError(f"Unrecognized parameters for CameraFrameSet: {set(kwargs.keys())}.")
259 return CameraFrameSet(self.instrument, astshim.FrameSet.fromString(self.ast))