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 

17__all__ = ( 

18 "USING_STARLINK_PYAST", 

19 "Channel", 

20 "CmpFrame", 

21 "CmpMap", 

22 "FitsChan", 

23 "Frame", 

24 "FrameSet", 

25 "Mapping", 

26 "ShiftMap", 

27 "SkyFrame", 

28 "StringStream", 

29 "UnitMap", 

30) 

31 

32if TYPE_CHECKING: 

33 import starlink.Ast 

34 

35 USING_STARLINK_PYAST = True 

36else: 

37 try: 

38 from astshim import ( 

39 Channel, 

40 CmpFrame, 

41 CmpMap, 

42 FitsChan, 

43 Frame, 

44 FrameDict, 

45 FrameSet, 

46 Mapping, 

47 Object, 

48 ShiftMap, 

49 SkyFrame, 

50 StringStream, 

51 UnitMap, 

52 ) 

53 except ImportError: 

54 import starlink.Ast 

55 

56 USING_STARLINK_PYAST = True 

57 else: 

58 USING_STARLINK_PYAST = False 

59 

60 

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

62 

63 class StringStream: 

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

65 interface expected by starlink.Ast.Channel. 

66 

67 Notes 

68 ----- 

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

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

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

72 implement the `FitsChan` classes in this module 

73 """ 

74 

75 def __init__(self, text: str = ""): 

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

77 self._lines = text.splitlines() 

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

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

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

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

82 else: 

83 self._lines = text.splitlines() 

84 

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

86 if not self._lines: 

87 return None 

88 return self._lines.pop(0) 

89 

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

91 self._lines.append(line) 

92 

93 def to_string(self) -> str: 

94 if not self._lines: 

95 return "" 

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

97 

98 def getSinkData(self) -> str: 

99 return self.to_string() 

100 

101 class Object: 

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

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

104 """ 

105 

106 def __init__(self, impl: starlink.Ast.Object): 

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

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

109 self._impl = impl 

110 

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

112 

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

114 sink = StringStream() 

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

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

117 chan.write(self._impl) 

118 return sink.to_string() 

119 

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

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

122 return NotImplemented 

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

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

125 # serialisation, which is the canonical content-faithful 

126 # representation for AST objects. Strip comments so cosmetic 

127 # changes between equivalent objects do not break equality. 

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

129 

130 __hash__ = None # type: ignore[assignment] 

131 

132 @classmethod 

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

134 source = StringStream(serialized) 

135 channel = starlink.Ast.Channel(source) 

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

137 

138 @classmethod 

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

140 subcls = cls._most_derived_type(impl) 

141 result = object.__new__(subcls) 

142 Object.__init__(result, impl) 

143 return result 

144 

145 @classmethod 

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

147 for subcls in cls.__subclasses__(): 

148 if isinstance(impl, subcls._IMPL_TYPE): 

149 return subcls._most_derived_type(impl) 

150 return cls 

151 

152 class Mapping(Object): 

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

154 

155 def simplified(self) -> Mapping: 

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

157 

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

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

160 

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

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

163 

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

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

166 

167 def inverted(self) -> Mapping: 

168 copy = self._impl.copy() 

169 copy.invert() 

170 return Mapping._wrap(copy) 

171 

172 class UnitMap(Mapping): 

173 def __init__(self, n_coord: int): 

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

175 

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

177 

178 class ShiftMap(Mapping): 

179 def __init__(self, shift: Iterable[float]): 

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

181 

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

183 

184 class CmpMap(Mapping): 

185 def __init__(self, map_a: Mapping, map_b: Mapping, series: bool): 

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

187 

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

189 

190 class Frame(Mapping): 

191 def __init__(self, n_axes: int, options: str = ""): 

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

193 

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

195 

196 @property 

197 def ident(self) -> str: 

198 return self._impl.Ident 

199 

200 @property 

201 def domain(self) -> str: 

202 return self._impl.Domain 

203 

204 @domain.setter 

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

206 self._impl.Domain = value 

207 

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

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

210 

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

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

213 

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

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

216 

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

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

219 

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

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

222 

223 class SkyFrame(Frame): 

224 def __init__(self, options: str = ""): 

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

226 

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

228 

229 class CmpFrame(Frame): 

230 def __init__(self, frame_a: Frame, frame_b: Frame): 

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

232 

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

234 

235 class FrameSet(Frame): 

236 def __init__(self, base_frame: Frame): 

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

238 

239 BASE: ClassVar[int] = 1 

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

241 

242 @property 

243 def nFrame(self) -> int: 

244 return self._impl.Nframe 

245 

246 @property 

247 def base(self) -> int: 

248 return self._impl.Base 

249 

250 @base.setter 

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

252 self._impl.Base = value 

253 

254 @property 

255 def current(self) -> int: 

256 return self._impl.Current 

257 

258 @current.setter 

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

260 self._impl.Current = value 

261 

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

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

264 

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

266 result = self._impl.getframe(iframe) 

267 if copy: 

268 result = result.copy() 

269 return Frame._wrap(result) 

270 

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

272 if iframe1 is None: 

273 iframe1 = self.base 

274 if iframe2 is None: 

275 iframe2 = self.current 

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

277 

278 class FrameDict(FrameSet): 

279 def __init__(self, obj: Object): 

280 Object.__init__(self, obj._impl) 

281 

282 class FitsChan(Object): 

283 def __init__(self, stream: StringStream | None = None, options: str = ""): 

284 source = stream if stream is not None else None 

285 sink = stream if stream is not None else None 

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

287 

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

289 

290 def read(self) -> Any: 

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

292 

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

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

295 

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

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

298 

299 def __iter__(self) -> Any: 

300 return iter(self._impl) 

301 

302 class Channel(Object): 

303 def __init__(self, stream: StringStream, options: str = ""): 

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

305 

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

307 

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

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