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

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 Any 

16 

17__all__ = ( 

18 "USING_STARLINK_PYAST", 

19 "FitsChan", 

20 "Frame", 

21 "FrameSet", 

22 "Mapping", 

23 "ShiftMap", 

24 "SkyFrame", 

25 "StringStream", 

26 "UnitMap", 

27 "wrap_frame_set", 

28 "wrap_mapping", 

29) 

30 

31try: 

32 import starlink.Ast as StarAst 

33except ImportError: 

34 StarAst = None 

35 import astshim as AstShim 

36 

37 USING_STARLINK_PYAST = False 

38else: 

39 AstShim = None 

40 USING_STARLINK_PYAST = True 

41 

42# Public backend-dependent bindings. 

43StringStream: Any 

44FitsChan: Any 

45Mapping: Any 

46UnitMap: Any 

47ShiftMap: Any 

48Frame: Any 

49SkyFrame: Any 

50FrameSet: Any 

51 

52 

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

54 assert StarAst is not None 

55 

56 class _StringStream: 

57 """A source/sink object compatible with starlink.Ast channel APIs.""" 

58 

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

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

61 self._lines = text.splitlines() 

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

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

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

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

66 else: 

67 self._lines = text.splitlines() 

68 

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

70 if not self._lines: 

71 return None 

72 return self._lines.pop(0) 

73 

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

75 self._lines.append(line) 

76 

77 def to_string(self) -> str: 

78 if not self._lines: 

79 return "" 

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

81 

82 class _AstWrapper: 

83 def __init__(self, obj: Any): 

84 self._obj = obj 

85 

86 @classmethod 

87 def _from_obj(cls, obj: Any) -> Any: 

88 self = cls.__new__(cls) 

89 _AstWrapper.__init__(self, obj) 

90 return self 

91 

92 def __getattr__(self, name: str) -> Any: 

93 return _wrap_ast_object(getattr(self._obj, name)) 

94 

95 def _serialize_ast_object(obj: Any, showComments: bool = True) -> str: 

96 sink = _StringStream() 

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

98 chan = StarAst.Channel(None, sink, options=options) 

99 chan.write(_unwrap_ast_object(obj)) 

100 return sink.to_string() 

101 

102 def _deserialize_ast_object(serialized: str) -> Any: 

103 source = _StringStream(serialized) 

104 chan = StarAst.Channel(source) 

105 return chan.read() 

106 

107 def _coerce_to_star_ast_object(obj: Any) -> Any: 

108 match obj: 

109 case _AstWrapper(): 

110 return obj._obj 

111 case StarAst.FrameSet() | StarAst.Mapping() | StarAst.Frame(): 

112 return obj 

113 case _: 

114 if hasattr(obj, "show"): 

115 try: 

116 return _deserialize_ast_object(obj.show()) 

117 except Exception: 

118 return obj 

119 return obj 

120 

121 def _unwrap_ast_object(obj: Any) -> Any: 

122 return _coerce_to_star_ast_object(obj) 

123 

124 def _wrap_ast_object(obj: Any) -> Any: 

125 match obj: 

126 case None: 

127 return obj 

128 case _AstWrapper(): 

129 return obj 

130 case StarAst.FrameSet(): 

131 return _FrameSet._from_obj(obj) 

132 case StarAst.SkyFrame(): 

133 return _SkyFrame._from_obj(obj) 

134 case StarAst.Frame(): 

135 return _Frame._from_obj(obj) 

136 case StarAst.Mapping(): 

137 return _Mapping._from_obj(obj) 

138 case _: 

139 coerced = _coerce_to_star_ast_object(obj) 

140 if coerced is not obj: 

141 return _wrap_ast_object(coerced) 

142 return obj 

143 

144 class _Mapping(_AstWrapper): 

145 @staticmethod 

146 def fromString(serialized: str) -> _Mapping: 

147 obj = _deserialize_ast_object(serialized) 

148 if not isinstance(obj, StarAst.Mapping): 

149 raise TypeError(f"Serialized object is not a Mapping: {type(obj)}") 

150 return _Mapping._from_obj(obj) 

151 

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

153 return _serialize_ast_object(self._obj, showComments=showComments) 

154 

155 def simplified(self) -> _Mapping: 

156 return _Mapping._from_obj(self._obj.simplify()) 

157 

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

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

160 

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

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

163 

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

165 return _Mapping._from_obj(StarAst.CmpMap(self._obj, _unwrap_ast_object(other), True)) 

166 

167 def inverted(self) -> _Mapping: 

168 copy = self._obj.copy() 

169 copy.invert() 

170 return _Mapping._from_obj(copy) 

171 

172 class _UnitMap(_Mapping): 

173 def __init__(self, n_coord: int): 

174 super().__init__(StarAst.UnitMap(n_coord)) 

175 

176 class _ShiftMap(_Mapping): 

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

178 super().__init__(StarAst.ShiftMap(list(shift))) 

179 

180 class _Frame(_AstWrapper): 

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

182 super().__init__(StarAst.Frame(n_axes, options)) 

183 

184 @property 

185 def ident(self) -> str: 

186 return self._obj.Ident 

187 

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

189 setattr(self._obj, f"Unit_{axis}", unit) 

190 

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

192 return getattr(self._obj, f"Unit_{axis}") 

193 

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

195 setattr(self._obj, f"Label_{axis}", label) 

196 

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

198 return getattr(self._obj, f"Bottom_{axis}") 

199 

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

201 return getattr(self._obj, f"Top_{axis}") 

202 

203 def show(self) -> str: 

204 return _serialize_ast_object(self._obj) 

205 

206 class _SkyFrame(_Frame): 

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

208 _AstWrapper.__init__(self, StarAst.SkyFrame(options)) 

209 

210 class _FrameSet(_AstWrapper): 

211 BASE = 1 

212 

213 def __init__(self, base_frame: _Frame): 

214 super().__init__(StarAst.FrameSet(_unwrap_ast_object(base_frame))) 

215 

216 @property 

217 def nFrame(self) -> int: 

218 return self._obj.Nframe 

219 

220 @property 

221 def base(self) -> int: 

222 return self._obj.Base 

223 

224 @base.setter 

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

226 self._obj.Base = value 

227 

228 @property 

229 def current(self) -> int: 

230 return self._obj.Current 

231 

232 @current.setter 

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

234 self._obj.Current = value 

235 

236 def addFrame(self, iframe: int, mapping: _Mapping, frame: _Frame) -> None: 

237 self._obj.addframe(iframe, _unwrap_ast_object(mapping), _unwrap_ast_object(frame)) 

238 

239 def getFrame(self, iframe: int, copy: bool = True) -> _Frame: 

240 del copy 

241 return _wrap_ast_object(self._obj.getframe(iframe)) 

242 

243 def getMapping(self, iframe1: int | None = None, iframe2: int | None = None) -> _Mapping: 

244 if iframe1 is None: 

245 iframe1 = self.base 

246 if iframe2 is None: 

247 iframe2 = self.current 

248 return _wrap_ast_object(self._obj.getmapping(iframe1, iframe2)) 

249 

250 def show(self) -> str: 

251 return _serialize_ast_object(self._obj) 

252 

253 @staticmethod 

254 def fromString(serialized: str) -> _FrameSet: 

255 obj = _deserialize_ast_object(serialized) 

256 if not isinstance(obj, StarAst.FrameSet): 

257 raise TypeError(f"Serialized object is not a FrameSet: {type(obj)}") 

258 return _FrameSet._from_obj(obj) 

259 

260 class _FitsChan: 

261 def __init__(self, stream: _StringStream | None = None, options: str = ""): 

262 source = stream if stream is not None else None 

263 sink = stream if stream is not None else None 

264 self._obj = StarAst.FitsChan(source, sink, options) 

265 

266 def read(self) -> Any: 

267 return _wrap_ast_object(self._obj.read()) 

268 

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

270 return self._obj.write(_unwrap_ast_object(obj)) 

271 

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

273 self._obj.setfitsI(keyword, value, "", 1) 

274 

275 def __iter__(self) -> Any: 

276 return iter(self._obj) 

277 

278 StringStream = _StringStream 

279 FitsChan = _FitsChan 

280 Mapping = _Mapping 

281 UnitMap = _UnitMap 

282 ShiftMap = _ShiftMap 

283 Frame = _Frame 

284 SkyFrame = _SkyFrame 

285 FrameSet = _FrameSet 

286 

287 def wrap_mapping(mapping: Any) -> Any: 

288 return _wrap_ast_object(mapping) 

289 

290 def wrap_frame_set(frame_set: Any) -> Any: 

291 return _wrap_ast_object(frame_set) 

292 

293else: 

294 assert AstShim is not None 

295 

296 StringStream = AstShim.StringStream 

297 FitsChan = AstShim.FitsChan 

298 Mapping = AstShim.Mapping 

299 UnitMap = AstShim.UnitMap 

300 ShiftMap = AstShim.ShiftMap 

301 Frame = AstShim.Frame 

302 SkyFrame = AstShim.SkyFrame 

303 FrameSet = AstShim.FrameSet 

304 

305 def wrap_mapping(mapping: Any) -> Any: 

306 return mapping 

307 

308 def wrap_frame_set(frame_set: Any) -> Any: 

309 return frame_set