Coverage for python/lsst/images/_transforms/_ast.py: 4%

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. 

11 

12from __future__ import annotations 

13 

14from collections.abc import Iterable 

15from typing import TYPE_CHECKING, Any, ClassVar, Self 

16 

17import numpy as np 

18 

19__all__ = ( 

20 "USING_STARLINK_PYAST", 

21 "Channel", 

22 "CmpFrame", 

23 "CmpMap", 

24 "FitsChan", 

25 "Frame", 

26 "FrameSet", 

27 "Mapping", 

28 "MatrixMap", 

29 "PolyMap", 

30 "ShiftMap", 

31 "SkyFrame", 

32 "StringStream", 

33 "UnitMap", 

34 "ZoomMap", 

35) 

36 

37if TYPE_CHECKING: 

38 import starlink.Ast 

39 

40 USING_STARLINK_PYAST = True 

41else: 

42 try: 

43 from astshim import ( 

44 Channel, 

45 CmpFrame, 

46 CmpMap, 

47 FitsChan, 

48 Frame, 

49 FrameDict, 

50 FrameSet, 

51 Mapping, 

52 MatrixMap, 

53 Object, 

54 PolyMap, 

55 ShiftMap, 

56 SkyFrame, 

57 StringStream, 

58 UnitMap, 

59 ZoomMap, 

60 ) 

61 except ImportError: 

62 import starlink.Ast 

63 

64 USING_STARLINK_PYAST = True 

65 else: 

66 USING_STARLINK_PYAST = False 

67 

68 

69if USING_STARLINK_PYAST: 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was never true

70 

71 class StringStream: 

72 """A bridge object that mimics both astshim.StringStream and the 

73 interface expected by starlink.Ast.Channel. 

74 

75 Notes 

76 ----- 

77 This object can be constructed like an `astshim.StringStream`, but its 

78 `astsink` and `astsource` methods actually correspond to the 

79 `starlink.Ast` interface, so we can use `starlink.Ast` objects to 

80 implement the `FitsChan` classes in this module 

81 """ 

82 

83 def __init__(self, text: str = "") -> None: 

84 if "\n" in text or "\r" in text: 

85 self._lines = text.splitlines() 

86 elif text and len(text) % 80 == 0: 

87 # Astropy WCS.to_header_string() yields a single concatenated 

88 # FITS header block; FitsChan expects one card per source line. 

89 self._lines = [text[n : n + 80] for n in range(0, len(text), 80)] 

90 else: 

91 self._lines = text.splitlines() 

92 

93 def astsource(self) -> str | None: 

94 if not self._lines: 

95 return None 

96 return self._lines.pop(0) 

97 

98 def astsink(self, line: str) -> None: 

99 self._lines.append(line) 

100 

101 def to_string(self) -> str: 

102 if not self._lines: 

103 return "" 

104 return "\n".join(self._lines) + "\n" 

105 

106 def getSinkData(self) -> str: 

107 return self.to_string() 

108 

109 class Object: 

110 """Bridge class that exposes the `astshim.Object` interface while 

111 being backed by an `astshim.Ast.Object`. 

112 """ 

113 

114 def __init__(self, impl: starlink.Ast.Object) -> None: 

115 if not isinstance(impl, self._IMPL_TYPE): 

116 raise TypeError(f"{type(self).__name__} cannot wrap {type(impl).__name__}.") 

117 self._impl = impl 

118 

119 _IMPL_TYPE: ClassVar[type[starlink.Ast.Object]] = starlink.Ast.Object 

120 

121 def show(self, showComments: bool = True) -> str: 

122 sink = StringStream() 

123 options = "" if showComments else "Comment=0" 

124 chan = starlink.Ast.Channel(None, sink, options=options) 

125 chan.write(self._impl) 

126 return sink.to_string() 

127 

128 def __eq__(self, other: object) -> bool: 

129 if not isinstance(other, Object) or type(self) is not type(other): 

130 return NotImplemented 

131 if self is other: 

132 # Bypass stringification if they are the same object. 

133 return True 

134 # ``astshim.Object`` ships a structural ``__eq__``; mirror that on 

135 # the starlink-pyast wrapper by comparing the AST channel 

136 # serialisation, which is the canonical content-faithful 

137 # representation for AST objects. Strip comments so cosmetic 

138 # changes between equivalent objects do not break equality. 

139 return self.show(showComments=False) == other.show(showComments=False) 

140 

141 __hash__ = None # type: ignore[assignment] 

142 

143 @classmethod 

144 def fromString(cls, serialized: str) -> Self: 

145 source = StringStream(serialized) 

146 channel = starlink.Ast.Channel(source) 

147 return cls._wrap(channel.read()) 

148 

149 @classmethod 

150 def _wrap(cls, impl: starlink.Ast.Object) -> Self: 

151 subcls = cls._most_derived_type(impl) 

152 result = object.__new__(subcls) 

153 Object.__init__(result, impl) 

154 return result 

155 

156 @classmethod 

157 def _most_derived_type(cls, impl: starlink.Ast.Object) -> type[Self]: 

158 for subcls in cls.__subclasses__(): 

159 if isinstance(impl, subcls._IMPL_TYPE): 

160 return subcls._most_derived_type(impl) 

161 return cls 

162 

163 class Mapping(Object): 

164 _IMPL_TYPE: ClassVar[type[starlink.Ast.Mapping]] = starlink.Ast.Mapping 

165 

166 def simplified(self) -> Mapping: 

167 return Mapping._wrap(self._impl.simplify()) 

168 

169 def applyForward(self, xy: Any) -> Any: 

170 return self._impl.tran(xy, True) 

171 

172 def applyInverse(self, xy: Any) -> Any: 

173 return self._impl.tran(xy, False) 

174 

175 def then(self, other: Mapping) -> Mapping: 

176 return Mapping._wrap(starlink.Ast.CmpMap(self._impl, other._impl, True)) 

177 

178 def inverted(self) -> Mapping: 

179 copy = self._impl.copy() 

180 copy.invert() 

181 return Mapping._wrap(copy) 

182 

183 def linearApprox(self, lbnd: Any, ubnd: Any, tol: float) -> np.ndarray: 

184 """Best linear approximation to this mapping over a hyper-box. 

185 

186 Parameters 

187 ---------- 

188 lbnd, ubnd 

189 Per-axis lower / upper input-coordinate bounds of the 

190 box over which the approximation is required. 

191 tol 

192 Maximum permitted deviation from linearity, expressed 

193 as a positive Cartesian displacement in the output 

194 coordinate system. 

195 

196 Returns 

197 ------- 

198 fit : `numpy.ndarray` 

199 A ``(1 + Nout, Nin)`` array whose first row holds the 

200 per-output constant offsets and whose remaining rows hold 

201 the Jacobian (``J[i][j] = ∂out_i/∂in_j``). 

202 

203 Raises 

204 ------ 

205 RuntimeError 

206 Raised if no linear approximation within ``tol`` exists 

207 over the requested box. 

208 

209 Notes 

210 ----- 

211 This matches ``astshim.Mapping.linearApprox``. starlink-pyast 

212 instead returns ``(success_flag, flat_coeffs)`` with the 

213 coefficients in the same row-major-by-output ordering as the 

214 astshim flat buffer, so reshaping recovers astshim's layout. 

215 """ 

216 success, coeffs = self._impl.linearapprox(lbnd, ubnd, tol) 

217 if not success: 

218 raise RuntimeError("Mapping not sufficiently linear") 

219 nin = self._impl.Nin 

220 nout = self._impl.Nout 

221 return np.asarray(coeffs, dtype=float).reshape(1 + nout, nin) 

222 

223 class UnitMap(Mapping): 

224 def __init__(self, n_coord: int) -> None: 

225 super().__init__(starlink.Ast.UnitMap(n_coord)) 

226 

227 _IMPL_TYPE: ClassVar[type[starlink.Ast.UnitMap]] = starlink.Ast.UnitMap 

228 

229 class ShiftMap(Mapping): 

230 def __init__(self, shift: Iterable[float]) -> None: 

231 super().__init__(starlink.Ast.ShiftMap(list(shift))) 

232 

233 _IMPL_TYPE: ClassVar[type[starlink.Ast.ShiftMap]] = starlink.Ast.ShiftMap 

234 

235 class CmpMap(Mapping): 

236 def __init__(self, map_a: Mapping, map_b: Mapping, series: bool) -> None: 

237 super().__init__(starlink.Ast.CmpMap(map_a._impl, map_b._impl, series)) 

238 

239 _IMPL_TYPE: ClassVar[type[starlink.Ast.CmpMap]] = starlink.Ast.CmpMap 

240 

241 class ZoomMap(Mapping): 

242 def __init__(self, n_coord: int, zoom: float) -> None: 

243 super().__init__(starlink.Ast.ZoomMap(n_coord, zoom)) 

244 

245 _IMPL_TYPE: ClassVar[type[starlink.Ast.ZoomMap]] = starlink.Ast.ZoomMap 

246 

247 class PolyMap(Mapping): 

248 def __init__(self, coeff_f: Any, coeff_i_or_nout: Any, options: str = "") -> None: 

249 # astshim's PolyMap takes ``nout`` as the second positional; 

250 # starlink.Ast.PolyMap requires an explicit inverse-coefficient 

251 # array. Adapt to both by synthesizing an empty inverse when 

252 # an integer ``nout`` is supplied. 

253 coeff_f_arr = np.asarray(coeff_f, dtype=float) 

254 if isinstance(coeff_i_or_nout, int): 

255 nin = coeff_f_arr.shape[1] - 2 

256 coeff_i = np.zeros((0, 2 + nin), dtype=float) 

257 else: 

258 coeff_i = np.asarray(coeff_i_or_nout, dtype=float) 

259 super().__init__(starlink.Ast.PolyMap(coeff_f_arr, coeff_i, options)) 

260 

261 _IMPL_TYPE: ClassVar[type[starlink.Ast.PolyMap]] = starlink.Ast.PolyMap 

262 

263 class MatrixMap(Mapping): 

264 def __init__(self, matrix: np.ndarray, options: str = ""): 

265 super().__init__(starlink.Ast.MatrixMap(matrix, options)) 

266 

267 _IMPL_TYPE: ClassVar[type[starlink.Ast.MatrixMap]] = starlink.Ast.MatrixMap 

268 

269 class Frame(Mapping): 

270 def __init__(self, n_axes: int, options: str = "") -> None: 

271 super().__init__(starlink.Ast.Frame(n_axes, options)) 

272 

273 _IMPL_TYPE: ClassVar[type[starlink.Ast.Frame]] = starlink.Ast.Frame 

274 

275 @property 

276 def ident(self) -> str: 

277 return self._impl.Ident 

278 

279 @property 

280 def domain(self) -> str: 

281 return self._impl.Domain 

282 

283 @domain.setter 

284 def domain(self, value: str) -> None: 

285 self._impl.Domain = value 

286 

287 def setUnit(self, axis: int, unit: str) -> None: 

288 setattr(self._impl, f"Unit_{axis}", unit) 

289 

290 def getUnit(self, axis: int) -> str: 

291 return getattr(self._impl, f"Unit_{axis}") 

292 

293 def setLabel(self, axis: int, label: str) -> None: 

294 setattr(self._impl, f"Label_{axis}", label) 

295 

296 def getBottom(self, axis: int) -> float: 

297 return getattr(self._impl, f"Bottom_{axis}") 

298 

299 def getTop(self, axis: int) -> float: 

300 return getattr(self._impl, f"Top_{axis}") 

301 

302 class SkyFrame(Frame): 

303 def __init__(self, options: str = "") -> None: 

304 Object.__init__(self, starlink.Ast.SkyFrame(options)) 

305 

306 _IMPL_TYPE: ClassVar[type[starlink.Ast.SkyFrame]] = starlink.Ast.SkyFrame 

307 

308 class CmpFrame(Frame): 

309 def __init__(self, frame_a: Frame, frame_b: Frame) -> None: 

310 Object.__init__(self, starlink.Ast.CmpFrame(frame_a._impl, frame_b._impl, "")) 

311 

312 _IMPL_TYPE: ClassVar[type[starlink.Ast.CmpFrame]] = starlink.Ast.CmpFrame 

313 

314 class FrameSet(Frame): 

315 def __init__(self, base_frame: Frame) -> None: 

316 Object.__init__(self, starlink.Ast.FrameSet(base_frame._impl)) 

317 

318 BASE: ClassVar[int] = 1 

319 _IMPL_TYPE: ClassVar[type[starlink.Ast.FrameSet]] = starlink.Ast.FrameSet 

320 

321 @property 

322 def nFrame(self) -> int: 

323 return self._impl.Nframe 

324 

325 @property 

326 def base(self) -> int: 

327 return self._impl.Base 

328 

329 @base.setter 

330 def base(self, value: int) -> None: 

331 self._impl.Base = value 

332 

333 @property 

334 def current(self) -> int: 

335 return self._impl.Current 

336 

337 @current.setter 

338 def current(self, value: int) -> None: 

339 self._impl.Current = value 

340 

341 def addFrame(self, iframe: int, mapping: Mapping, frame: Frame) -> None: 

342 self._impl.addframe(iframe, mapping._impl, frame._impl) 

343 

344 def getFrame(self, iframe: int, copy: bool = True) -> Frame: 

345 result = self._impl.getframe(iframe) 

346 if copy: 

347 result = result.copy() 

348 return Frame._wrap(result) 

349 

350 def getMapping(self, iframe1: int | None = None, iframe2: int | None = None) -> Mapping: 

351 if iframe1 is None: 

352 iframe1 = self.base 

353 if iframe2 is None: 

354 iframe2 = self.current 

355 return Mapping._wrap(self._impl.getmapping(iframe1, iframe2)) 

356 

357 class FrameDict(FrameSet): 

358 def __init__(self, obj: Object) -> None: 

359 Object.__init__(self, obj._impl) 

360 

361 class FitsChan(Object): 

362 def __init__(self, stream: StringStream | None = None, options: str = "") -> None: 

363 source = stream if stream is not None else None 

364 sink = stream if stream is not None else None 

365 super().__init__(starlink.Ast.FitsChan(source, sink, options)) 

366 

367 _IMPL_TYPE: ClassVar[type[starlink.Ast.FitsChan]] = starlink.Ast.FitsChan 

368 

369 def read(self) -> Any: 

370 return Object._wrap(self._impl.read()) 

371 

372 def write(self, obj: Any) -> int: 

373 return self._impl.write(obj._impl) 

374 

375 def setFitsI(self, keyword: str, value: int) -> None: 

376 self._impl.setfitsI(keyword, value, "", 1) 

377 

378 def __iter__(self) -> Any: 

379 return iter(self._impl) 

380 

381 class Channel(Object): 

382 def __init__(self, stream: StringStream, options: str = "") -> None: 

383 super().__init__(starlink.Ast.Channel(None, stream, options)) 

384 

385 _IMPL_TYPE: ClassVar[type[starlink.Ast.Channel]] = starlink.Ast.Channel 

386 

387 def write(self, obj: Object) -> int: 

388 return self._impl.write(obj._impl)