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) -> str: 

96 sink = _StringStream() 

97 chan = StarAst.Channel(None, sink) 

98 chan.write(_unwrap_ast_object(obj)) 

99 return sink.to_string() 

100 

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

102 source = _StringStream(serialized) 

103 chan = StarAst.Channel(source) 

104 return chan.read() 

105 

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

107 match obj: 

108 case _AstWrapper(): 

109 return obj._obj 

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

111 return obj 

112 case _: 

113 if hasattr(obj, "show"): 

114 try: 

115 return _deserialize_ast_object(obj.show()) 

116 except Exception: 

117 return obj 

118 return obj 

119 

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

121 return _coerce_to_star_ast_object(obj) 

122 

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

124 match obj: 

125 case None: 

126 return obj 

127 case _AstWrapper(): 

128 return obj 

129 case StarAst.FrameSet(): 

130 return _FrameSet._from_obj(obj) 

131 case StarAst.SkyFrame(): 

132 return _SkyFrame._from_obj(obj) 

133 case StarAst.Frame(): 

134 return _Frame._from_obj(obj) 

135 case StarAst.Mapping(): 

136 return _Mapping._from_obj(obj) 

137 case _: 

138 coerced = _coerce_to_star_ast_object(obj) 

139 if coerced is not obj: 

140 return _wrap_ast_object(coerced) 

141 return obj 

142 

143 class _Mapping(_AstWrapper): 

144 @staticmethod 

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

146 obj = _deserialize_ast_object(serialized) 

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

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

149 return _Mapping._from_obj(obj) 

150 

151 def show(self) -> str: 

152 return _serialize_ast_object(self._obj) 

153 

154 def simplified(self) -> _Mapping: 

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

156 

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

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

159 

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

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

162 

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

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

165 

166 def inverted(self) -> _Mapping: 

167 copy = self._obj.copy() 

168 copy.invert() 

169 return _Mapping._from_obj(copy) 

170 

171 class _UnitMap(_Mapping): 

172 def __init__(self, n_coord: int): 

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

174 

175 class _ShiftMap(_Mapping): 

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

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

178 

179 class _Frame(_AstWrapper): 

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

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

182 

183 @property 

184 def ident(self) -> str: 

185 return self._obj.Ident 

186 

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

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

189 

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

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

192 

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

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

195 

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

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

198 

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

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

201 

202 def show(self) -> str: 

203 return _serialize_ast_object(self._obj) 

204 

205 class _SkyFrame(_Frame): 

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

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

208 

209 class _FrameSet(_AstWrapper): 

210 BASE = 1 

211 

212 def __init__(self, base_frame: _Frame): 

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

214 

215 @property 

216 def nFrame(self) -> int: 

217 return self._obj.Nframe 

218 

219 @property 

220 def base(self) -> int: 

221 return self._obj.Base 

222 

223 @base.setter 

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

225 self._obj.Base = value 

226 

227 @property 

228 def current(self) -> int: 

229 return self._obj.Current 

230 

231 @current.setter 

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

233 self._obj.Current = value 

234 

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

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

237 

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

239 del copy 

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

241 

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

243 if iframe1 is None: 

244 iframe1 = self.base 

245 if iframe2 is None: 

246 iframe2 = self.current 

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

248 

249 def show(self) -> str: 

250 return _serialize_ast_object(self._obj) 

251 

252 @staticmethod 

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

254 obj = _deserialize_ast_object(serialized) 

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

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

257 return _FrameSet._from_obj(obj) 

258 

259 class _FitsChan: 

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

261 source = stream if stream is not None else None 

262 sink = stream if stream is not None else None 

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

264 

265 def read(self) -> Any: 

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

267 

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

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

270 

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

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

273 

274 def __iter__(self) -> Any: 

275 return iter(self._obj) 

276 

277 StringStream = _StringStream 

278 FitsChan = _FitsChan 

279 Mapping = _Mapping 

280 UnitMap = _UnitMap 

281 ShiftMap = _ShiftMap 

282 Frame = _Frame 

283 SkyFrame = _SkyFrame 

284 FrameSet = _FrameSet 

285 

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

287 return _wrap_ast_object(mapping) 

288 

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

290 return _wrap_ast_object(frame_set) 

291 

292else: 

293 assert AstShim is not None 

294 

295 StringStream = AstShim.StringStream 

296 FitsChan = AstShim.FitsChan 

297 Mapping = AstShim.Mapping 

298 UnitMap = AstShim.UnitMap 

299 ShiftMap = AstShim.ShiftMap 

300 Frame = AstShim.Frame 

301 SkyFrame = AstShim.SkyFrame 

302 FrameSet = AstShim.FrameSet 

303 

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

305 return mapping 

306 

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

308 return frame_set