Coverage for python/lsst/images/_transforms/_ast.py: 4%
200 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-25 01:29 -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
14from collections.abc import Iterable
15from typing import TYPE_CHECKING, Any, ClassVar, Self
17import numpy as np
19__all__ = (
20 "USING_STARLINK_PYAST",
21 "Channel",
22 "CmpFrame",
23 "CmpMap",
24 "FitsChan",
25 "Frame",
26 "FrameSet",
27 "Mapping",
28 "PolyMap",
29 "ShiftMap",
30 "SkyFrame",
31 "StringStream",
32 "UnitMap",
33 "ZoomMap",
34)
36if TYPE_CHECKING:
37 import starlink.Ast
39 USING_STARLINK_PYAST = True
40else:
41 try:
42 from astshim import (
43 Channel,
44 CmpFrame,
45 CmpMap,
46 FitsChan,
47 Frame,
48 FrameDict,
49 FrameSet,
50 Mapping,
51 Object,
52 PolyMap,
53 ShiftMap,
54 SkyFrame,
55 StringStream,
56 UnitMap,
57 ZoomMap,
58 )
59 except ImportError:
60 import starlink.Ast
62 USING_STARLINK_PYAST = True
63 else:
64 USING_STARLINK_PYAST = False
67if USING_STARLINK_PYAST: 67 ↛ 69line 67 didn't jump to line 69 because the condition on line 67 was never true
69 class StringStream:
70 """A bridge object that mimics both astshim.StringStream and the
71 interface expected by starlink.Ast.Channel.
73 Notes
74 -----
75 This object can be constructed like an `astshim.StringStream`, but its
76 `astsink` and `astsource` methods actually correspond to the
77 `starlink.Ast` interface, so we can use `starlink.Ast` objects to
78 implement the `FitsChan` classes in this module
79 """
81 def __init__(self, text: str = "") -> None:
82 if "\n" in text or "\r" in text:
83 self._lines = text.splitlines()
84 elif text and len(text) % 80 == 0:
85 # Astropy WCS.to_header_string() yields a single concatenated
86 # FITS header block; FitsChan expects one card per source line.
87 self._lines = [text[n : n + 80] for n in range(0, len(text), 80)]
88 else:
89 self._lines = text.splitlines()
91 def astsource(self) -> str | None:
92 if not self._lines:
93 return None
94 return self._lines.pop(0)
96 def astsink(self, line: str) -> None:
97 self._lines.append(line)
99 def to_string(self) -> str:
100 if not self._lines:
101 return ""
102 return "\n".join(self._lines) + "\n"
104 def getSinkData(self) -> str:
105 return self.to_string()
107 class Object:
108 """Bridge class that exposes the `astshim.Object` interface while
109 being backed by an `astshim.Ast.Object`.
110 """
112 def __init__(self, impl: starlink.Ast.Object) -> None:
113 if not isinstance(impl, self._IMPL_TYPE):
114 raise TypeError(f"{type(self).__name__} cannot wrap {type(impl).__name__}.")
115 self._impl = impl
117 _IMPL_TYPE: ClassVar[type[starlink.Ast.Object]] = starlink.Ast.Object
119 def show(self, showComments: bool = True) -> str:
120 sink = StringStream()
121 options = "" if showComments else "Comment=0"
122 chan = starlink.Ast.Channel(None, sink, options=options)
123 chan.write(self._impl)
124 return sink.to_string()
126 def __eq__(self, other: object) -> bool:
127 if not isinstance(other, Object) or type(self) is not type(other):
128 return NotImplemented
129 if self is other:
130 # Bypass stringification if they are the same object.
131 return True
132 # ``astshim.Object`` ships a structural ``__eq__``; mirror that on
133 # the starlink-pyast wrapper by comparing the AST channel
134 # serialisation, which is the canonical content-faithful
135 # representation for AST objects. Strip comments so cosmetic
136 # changes between equivalent objects do not break equality.
137 return self.show(showComments=False) == other.show(showComments=False)
139 __hash__ = None # type: ignore[assignment]
141 @classmethod
142 def fromString(cls, serialized: str) -> Self:
143 source = StringStream(serialized)
144 channel = starlink.Ast.Channel(source)
145 return cls._wrap(channel.read())
147 @classmethod
148 def _wrap(cls, impl: starlink.Ast.Object) -> Self:
149 subcls = cls._most_derived_type(impl)
150 result = object.__new__(subcls)
151 Object.__init__(result, impl)
152 return result
154 @classmethod
155 def _most_derived_type(cls, impl: starlink.Ast.Object) -> type[Self]:
156 for subcls in cls.__subclasses__():
157 if isinstance(impl, subcls._IMPL_TYPE):
158 return subcls._most_derived_type(impl)
159 return cls
161 class Mapping(Object):
162 _IMPL_TYPE: ClassVar[type[starlink.Ast.Mapping]] = starlink.Ast.Mapping
164 def simplified(self) -> Mapping:
165 return Mapping._wrap(self._impl.simplify())
167 def applyForward(self, xy: Any) -> Any:
168 return self._impl.tran(xy, True)
170 def applyInverse(self, xy: Any) -> Any:
171 return self._impl.tran(xy, False)
173 def then(self, other: Mapping) -> Mapping:
174 return Mapping._wrap(starlink.Ast.CmpMap(self._impl, other._impl, True))
176 def inverted(self) -> Mapping:
177 copy = self._impl.copy()
178 copy.invert()
179 return Mapping._wrap(copy)
181 def linearApprox(self, lbnd: Any, ubnd: Any, tol: float) -> np.ndarray:
182 """Best linear approximation to this mapping over a hyper-box.
184 Parameters
185 ----------
186 lbnd, ubnd
187 Per-axis lower / upper input-coordinate bounds of the
188 box over which the approximation is required.
189 tol
190 Maximum permitted deviation from linearity, expressed
191 as a positive Cartesian displacement in the output
192 coordinate system.
194 Returns
195 -------
196 fit : `numpy.ndarray`
197 A ``(1 + Nout, Nin)`` array whose first row holds the
198 per-output constant offsets and whose remaining rows hold
199 the Jacobian (``J[i][j] = ∂out_i/∂in_j``).
201 Raises
202 ------
203 RuntimeError
204 Raised if no linear approximation within ``tol`` exists
205 over the requested box.
207 Notes
208 -----
209 This matches ``astshim.Mapping.linearApprox``. starlink-pyast
210 instead returns ``(success_flag, flat_coeffs)`` with the
211 coefficients in the same row-major-by-output ordering as the
212 astshim flat buffer, so reshaping recovers astshim's layout.
213 """
214 success, coeffs = self._impl.linearapprox(lbnd, ubnd, tol)
215 if not success:
216 raise RuntimeError("Mapping not sufficiently linear")
217 nin = self._impl.Nin
218 nout = self._impl.Nout
219 return np.asarray(coeffs, dtype=float).reshape(1 + nout, nin)
221 class UnitMap(Mapping):
222 def __init__(self, n_coord: int) -> None:
223 super().__init__(starlink.Ast.UnitMap(n_coord))
225 _IMPL_TYPE: ClassVar[type[starlink.Ast.UnitMap]] = starlink.Ast.UnitMap
227 class ShiftMap(Mapping):
228 def __init__(self, shift: Iterable[float]) -> None:
229 super().__init__(starlink.Ast.ShiftMap(list(shift)))
231 _IMPL_TYPE: ClassVar[type[starlink.Ast.ShiftMap]] = starlink.Ast.ShiftMap
233 class CmpMap(Mapping):
234 def __init__(self, map_a: Mapping, map_b: Mapping, series: bool) -> None:
235 super().__init__(starlink.Ast.CmpMap(map_a._impl, map_b._impl, series))
237 _IMPL_TYPE: ClassVar[type[starlink.Ast.CmpMap]] = starlink.Ast.CmpMap
239 class ZoomMap(Mapping):
240 def __init__(self, n_coord: int, zoom: float) -> None:
241 super().__init__(starlink.Ast.ZoomMap(n_coord, zoom))
243 _IMPL_TYPE: ClassVar[type[starlink.Ast.ZoomMap]] = starlink.Ast.ZoomMap
245 class PolyMap(Mapping):
246 def __init__(self, coeff_f: Any, coeff_i_or_nout: Any, options: str = "") -> None:
247 # astshim's PolyMap takes ``nout`` as the second positional;
248 # starlink.Ast.PolyMap requires an explicit inverse-coefficient
249 # array. Adapt to both by synthesizing an empty inverse when
250 # an integer ``nout`` is supplied.
251 coeff_f_arr = np.asarray(coeff_f, dtype=float)
252 if isinstance(coeff_i_or_nout, int):
253 nin = coeff_f_arr.shape[1] - 2
254 coeff_i = np.zeros((0, 2 + nin), dtype=float)
255 else:
256 coeff_i = np.asarray(coeff_i_or_nout, dtype=float)
257 super().__init__(starlink.Ast.PolyMap(coeff_f_arr, coeff_i, options))
259 _IMPL_TYPE: ClassVar[type[starlink.Ast.PolyMap]] = starlink.Ast.PolyMap
261 class Frame(Mapping):
262 def __init__(self, n_axes: int, options: str = "") -> None:
263 super().__init__(starlink.Ast.Frame(n_axes, options))
265 _IMPL_TYPE: ClassVar[type[starlink.Ast.Frame]] = starlink.Ast.Frame
267 @property
268 def ident(self) -> str:
269 return self._impl.Ident
271 @property
272 def domain(self) -> str:
273 return self._impl.Domain
275 @domain.setter
276 def domain(self, value: str) -> None:
277 self._impl.Domain = value
279 def setUnit(self, axis: int, unit: str) -> None:
280 setattr(self._impl, f"Unit_{axis}", unit)
282 def getUnit(self, axis: int) -> str:
283 return getattr(self._impl, f"Unit_{axis}")
285 def setLabel(self, axis: int, label: str) -> None:
286 setattr(self._impl, f"Label_{axis}", label)
288 def getBottom(self, axis: int) -> float:
289 return getattr(self._impl, f"Bottom_{axis}")
291 def getTop(self, axis: int) -> float:
292 return getattr(self._impl, f"Top_{axis}")
294 class SkyFrame(Frame):
295 def __init__(self, options: str = "") -> None:
296 Object.__init__(self, starlink.Ast.SkyFrame(options))
298 _IMPL_TYPE: ClassVar[type[starlink.Ast.SkyFrame]] = starlink.Ast.SkyFrame
300 class CmpFrame(Frame):
301 def __init__(self, frame_a: Frame, frame_b: Frame) -> None:
302 Object.__init__(self, starlink.Ast.CmpFrame(frame_a._impl, frame_b._impl, ""))
304 _IMPL_TYPE: ClassVar[type[starlink.Ast.CmpFrame]] = starlink.Ast.CmpFrame
306 class FrameSet(Frame):
307 def __init__(self, base_frame: Frame) -> None:
308 Object.__init__(self, starlink.Ast.FrameSet(base_frame._impl))
310 BASE: ClassVar[int] = 1
311 _IMPL_TYPE: ClassVar[type[starlink.Ast.FrameSet]] = starlink.Ast.FrameSet
313 @property
314 def nFrame(self) -> int:
315 return self._impl.Nframe
317 @property
318 def base(self) -> int:
319 return self._impl.Base
321 @base.setter
322 def base(self, value: int) -> None:
323 self._impl.Base = value
325 @property
326 def current(self) -> int:
327 return self._impl.Current
329 @current.setter
330 def current(self, value: int) -> None:
331 self._impl.Current = value
333 def addFrame(self, iframe: int, mapping: Mapping, frame: Frame) -> None:
334 self._impl.addframe(iframe, mapping._impl, frame._impl)
336 def getFrame(self, iframe: int, copy: bool = True) -> Frame:
337 result = self._impl.getframe(iframe)
338 if copy:
339 result = result.copy()
340 return Frame._wrap(result)
342 def getMapping(self, iframe1: int | None = None, iframe2: int | None = None) -> Mapping:
343 if iframe1 is None:
344 iframe1 = self.base
345 if iframe2 is None:
346 iframe2 = self.current
347 return Mapping._wrap(self._impl.getmapping(iframe1, iframe2))
349 class FrameDict(FrameSet):
350 def __init__(self, obj: Object) -> None:
351 Object.__init__(self, obj._impl)
353 class FitsChan(Object):
354 def __init__(self, stream: StringStream | None = None, options: str = "") -> None:
355 source = stream if stream is not None else None
356 sink = stream if stream is not None else None
357 super().__init__(starlink.Ast.FitsChan(source, sink, options))
359 _IMPL_TYPE: ClassVar[type[starlink.Ast.FitsChan]] = starlink.Ast.FitsChan
361 def read(self) -> Any:
362 return Object._wrap(self._impl.read())
364 def write(self, obj: Any) -> int:
365 return self._impl.write(obj._impl)
367 def setFitsI(self, keyword: str, value: int) -> None:
368 self._impl.setfitsI(keyword, value, "", 1)
370 def __iter__(self) -> Any:
371 return iter(self._impl)
373 class Channel(Object):
374 def __init__(self, stream: StringStream, options: str = "") -> None:
375 super().__init__(starlink.Ast.Channel(None, stream, options))
377 _IMPL_TYPE: ClassVar[type[starlink.Ast.Channel]] = starlink.Ast.Channel
379 def write(self, obj: Object) -> int:
380 return self._impl.write(obj._impl)