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

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 @classmethod 

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

122 source = StringStream(serialized) 

123 channel = starlink.Ast.Channel(source) 

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

125 

126 @classmethod 

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

128 subcls = cls._most_derived_type(impl) 

129 result = object.__new__(subcls) 

130 Object.__init__(result, impl) 

131 return result 

132 

133 @classmethod 

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

135 for subcls in cls.__subclasses__(): 

136 if isinstance(impl, subcls._IMPL_TYPE): 

137 return subcls._most_derived_type(impl) 

138 return cls 

139 

140 class Mapping(Object): 

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

142 

143 def simplified(self) -> Mapping: 

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

145 

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

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

148 

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

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

151 

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

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

154 

155 def inverted(self) -> Mapping: 

156 copy = self._impl.copy() 

157 copy.invert() 

158 return Mapping._wrap(copy) 

159 

160 class UnitMap(Mapping): 

161 def __init__(self, n_coord: int): 

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

163 

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

165 

166 class ShiftMap(Mapping): 

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

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

169 

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

171 

172 class CmpMap(Mapping): 

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

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

175 

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

177 

178 class Frame(Mapping): 

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

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

181 

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

183 

184 @property 

185 def ident(self) -> str: 

186 return self._impl.Ident 

187 

188 @property 

189 def domain(self) -> str: 

190 return self._impl.Domain 

191 

192 @domain.setter 

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

194 self._impl.Domain = value 

195 

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

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

198 

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

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

201 

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

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

204 

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

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

207 

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

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

210 

211 class SkyFrame(Frame): 

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

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

214 

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

216 

217 class CmpFrame(Frame): 

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

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

220 

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

222 

223 class FrameSet(Frame): 

224 def __init__(self, base_frame: Frame): 

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

226 

227 BASE: ClassVar[int] = 1 

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

229 

230 @property 

231 def nFrame(self) -> int: 

232 return self._impl.Nframe 

233 

234 @property 

235 def base(self) -> int: 

236 return self._impl.Base 

237 

238 @base.setter 

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

240 self._impl.Base = value 

241 

242 @property 

243 def current(self) -> int: 

244 return self._impl.Current 

245 

246 @current.setter 

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

248 self._impl.Current = value 

249 

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

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

252 

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

254 result = self._impl.getframe(iframe) 

255 if copy: 

256 result = result.copy() 

257 return Frame._wrap(result) 

258 

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

260 if iframe1 is None: 

261 iframe1 = self.base 

262 if iframe2 is None: 

263 iframe2 = self.current 

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

265 

266 class FrameDict(FrameSet): 

267 def __init__(self, obj: Object): 

268 Object.__init__(self, obj._impl) 

269 

270 class FitsChan(Object): 

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

272 source = stream if stream is not None else None 

273 sink = stream if stream is not None else None 

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

275 

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

277 

278 def read(self) -> Any: 

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

280 

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

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

283 

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

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

286 

287 def __iter__(self) -> Any: 

288 return iter(self._impl) 

289 

290 class Channel(Object): 

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

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

293 

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

295 

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

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