Coverage for python/lsst/images/serialization/_input_archive.py: 100%

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 

14__all__ = ("ArchiveInfo", "DetachedArchive", "InputArchive") 

15 

16from abc import ABC, abstractmethod 

17from collections.abc import Callable 

18from contextlib import AbstractContextManager 

19from types import EllipsisType 

20from typing import IO, TYPE_CHECKING, Any, TypeVar 

21 

22import astropy.io.fits 

23import astropy.table 

24import astropy.units 

25import numpy as np 

26import pydantic 

27 

28from lsst.resources import ResourcePath, ResourcePathExpression 

29 

30from ._asdf_utils import ArrayReferenceModel, InlineArrayModel 

31from ._common import ( 

32 SCHEMA_URL_HOST, 

33 ArchiveAccessRequiredError, 

34 ArchiveTree, 

35 OpaqueArchiveMetadata, 

36 no_header_updates, 

37) 

38from ._tables import TableModel 

39 

40if TYPE_CHECKING: 

41 from .._transforms import FrameSet 

42 

43 

44# This pre-python-3.12 declaration is needed by Sphinx (probably the 

45# autodoc-typehints plugin. 

46P = TypeVar("P", bound=pydantic.BaseModel) 

47 

48 

49class ArchiveInfo(pydantic.BaseModel, frozen=True): 

50 """Basic identifying information about an on-disk archive. 

51 

52 Read from a file's headers/metadata without deserializing pixel data. 

53 """ 

54 

55 schema_url: str 

56 """Canonical schema URL of the top-level tree.""" 

57 

58 schema_name: str 

59 """Schema name parsed from ``schema_url``.""" 

60 

61 schema_version: str 

62 """Schema version parsed from ``schema_url``.""" 

63 

64 format_version: int | None 

65 """Container layout version (FITS ``FMTVER`` / NDF ``FORMAT_VERSION``); 

66 `None` for formats with no separate container version (JSON).""" 

67 

68 @classmethod 

69 def from_schema_url(cls, schema_url: str, *, format_version: int | None) -> ArchiveInfo: 

70 """Build an `ArchiveInfo` by parsing a schema URL of the form 

71 ``https://images.lsst.io/schemas/{name}-{version}``. 

72 

73 The URL is parsed with `~lsst.resources.ResourcePath` and its 

74 hostname must be ``images.lsst.io``, so a ``DATAMODL`` header written 

75 by an unrelated tool cannot steer reads toward an arbitrary schema. 

76 

77 Parameters 

78 ---------- 

79 schema_url 

80 Schema URL to parse for the schema name and version. 

81 format_version 

82 Container layout version, or `None` for formats with no 

83 separate container version. 

84 """ 

85 parsed = ResourcePath(schema_url) 

86 if parsed.netloc != SCHEMA_URL_HOST: 

87 raise ValueError( 

88 f"Schema URL {schema_url!r} is not hosted at {SCHEMA_URL_HOST!r}; " 

89 "this file was not written by lsst.images." 

90 ) 

91 tail = parsed.basename() 

92 # Split on the last hyphen: schema names may contain hyphens; the 

93 # version (after the final hyphen) is assumed not to. 

94 name, _, version = tail.rpartition("-") 

95 if not name or not version: 

96 raise ValueError(f"Cannot parse schema name/version from URL {schema_url!r}.") 

97 return cls( 

98 schema_url=schema_url, 

99 schema_name=name, 

100 schema_version=version, 

101 format_version=format_version, 

102 ) 

103 

104 

105class InputArchive[P: pydantic.BaseModel](ABC): 

106 """Abstract interface for reading from a file format. 

107 

108 Notes 

109 ----- 

110 An input archive instance is assumed to be paired with a Pydantic model 

111 that represents a JSON tree, with the archive used to deserialize data that 

112 is not native JSON from data that is (which may just be a reference to 

113 binary data stored elsewhere in the file). The archive doesn't actually 

114 hold that model instance because we'd prefer to avoid making the input 

115 archive generic over the model type. It is expected that most concrete 

116 archive implementations will provide a method to load the paired model from 

117 a file, but this is not part of the base class interface. 

118 """ 

119 

120 @classmethod 

121 def get_basic_info(cls, path: ResourcePathExpression) -> ArchiveInfo: 

122 """Return basic identifying information for the archive at ``path`` 

123 without deserializing pixel data. 

124 

125 Each concrete backend reads only the headers/metadata it needs. 

126 

127 Parameters 

128 ---------- 

129 path 

130 Path to the archive to read. 

131 """ 

132 raise NotImplementedError(f"{cls.__name__} does not implement get_basic_info.") 

133 

134 @classmethod 

135 def open_tree( 

136 cls, 

137 path: ResourcePathExpression | IO[bytes], 

138 *, 

139 partial: bool = True, 

140 **backend_kwargs: Any, 

141 ) -> AbstractContextManager[tuple[InputArchive[P], ArchiveTree, ArchiveInfo]]: 

142 """Open ``path``, load and validate its top-level tree, and yield 

143 ``(archive, tree, info)`` as a context manager. 

144 

145 Parameters 

146 ---------- 

147 path 

148 File to be opened (local or remote), or a seekable binary 

149 stream containing the file's content. 

150 partial 

151 Whether the file should be opened for incremental reads or not. 

152 Can be ignored by a backend where not relevant. 

153 **backend_kwargs 

154 Any keyword parameters that should be forwarded to the backend 

155 open. 

156 

157 Raises 

158 ------ 

159 ArchiveReadError 

160 If the file's schema is not registered. 

161 

162 Notes 

163 ----- 

164 Each concrete backend implements this. 

165 """ 

166 raise NotImplementedError(f"{cls.__name__} does not implement open_tree.") 

167 

168 @abstractmethod 

169 def deserialize_pointer[U: ArchiveTree, V]( 

170 self, pointer: P, model_type: type[U], deserializer: Callable[[U, InputArchive[P]], V] 

171 ) -> V: 

172 """Deserialize an object that was saved by 

173 `~lsst.serialization.OutputArchive.serialize_pointer`. 

174 

175 Parameters 

176 ---------- 

177 pointer 

178 JSON Pointer model to dereference. 

179 model_type 

180 Pydantic model type that the pointer should dereference to. 

181 deserializer 

182 Callable that takes an instance of ``model_type`` and an input 

183 archive, and returns the deserialized object. 

184 

185 Returns 

186 ------- 

187 V 

188 The deserialized object. 

189 

190 Notes 

191 ----- 

192 Implementations are required to remember previously-deserialized 

193 objects and return them when the same pointer is passed in multiple 

194 times. 

195 

196 There is no ``deserialize_direct`` (to pair with 

197 `~lsst.serialization.OutputArchive.serialize_direct`) because the 

198 caller can just call a deserializer function directly on a sub-model 

199 of its Pydantic tree. 

200 """ 

201 raise NotImplementedError() 

202 

203 @abstractmethod 

204 def get_frame_set(self, ref: P) -> FrameSet: 

205 """Return an already-deserialized frame set from the archive. 

206 

207 Parameters 

208 ---------- 

209 ref 

210 Implementation-specific reference to the frame set. 

211 

212 Returns 

213 ------- 

214 FrameSet 

215 Loaded frame set. 

216 """ 

217 raise NotImplementedError() 

218 

219 @abstractmethod 

220 def get_array( 

221 self, 

222 model: ArrayReferenceModel | InlineArrayModel, 

223 *, 

224 slices: tuple[slice, ...] | EllipsisType = ..., 

225 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

226 ) -> np.ndarray: 

227 """Load an array from the archive. 

228 

229 Parameters 

230 ---------- 

231 model 

232 A Pydantic model that references or holds the array. 

233 slices 

234 Slices that specify a subset of the original array to read. 

235 strip_header 

236 A callable that strips out any FITS header cards added by the 

237 ``update_header`` argument in the corresponding call to 

238 `~lsst.images.serialization.OutputArchive.add_array`. 

239 """ 

240 raise NotImplementedError() 

241 

242 @abstractmethod 

243 def get_table( 

244 self, 

245 model: TableModel, 

246 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

247 ) -> astropy.table.Table: 

248 """Load a table from the archive. 

249 

250 Parameters 

251 ---------- 

252 model 

253 A Pydantic model that references or holds the table. 

254 strip_header 

255 A callable that strips out any FITS header cards added by the 

256 ``update_header`` argument in the corresponding call to 

257 `~lsst.serialization.OutputArchive.add_table`. 

258 

259 Returns 

260 ------- 

261 astropy.table.Table 

262 The loaded table. 

263 """ 

264 raise NotImplementedError() 

265 

266 @abstractmethod 

267 def get_structured_array( 

268 self, 

269 model: TableModel, 

270 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

271 ) -> np.ndarray: 

272 """Load a table from the archive as a structured array. 

273 

274 Parameters 

275 ---------- 

276 model 

277 A Pydantic model that references or holds the table. 

278 strip_header 

279 A callable that strips out any FITS header cards added by the 

280 ``update_header`` argument in the corresponding call to 

281 `~lsst.serialization.OutputArchive.add_structured_array`. 

282 

283 Returns 

284 ------- 

285 numpy.ndarray 

286 The loaded table as a structured array. 

287 """ 

288 raise NotImplementedError() 

289 

290 def get_opaque_metadata(self) -> OpaqueArchiveMetadata | None: 

291 """Return opaque metadata loaded from the file that should be saved if 

292 another version of the object is saved to the same file format. 

293 

294 Returns 

295 ------- 

296 OpaqueArchiveMetadata 

297 Opaque metadata specific to this archive type that should be 

298 round-tripped if it is saved in the same format. 

299 """ 

300 return None 

301 

302 

303class DetachedArchive(InputArchive[Any]): 

304 """An input archive that is not attached to any file. 

305 

306 Every method that would read data from a file raises 

307 `ArchiveAccessRequiredError`. 

308 

309 Notes 

310 ----- 

311 Passing an instance to `ArchiveTree.deserialize_component` probes 

312 whether a component can be deserialized from the tree alone: success 

313 means no file access was needed, while `ArchiveAccessRequiredError` 

314 means the caller must use a live archive instead. Instances hold no 

315 state, so a single instance can be shared by any number of probes. 

316 """ 

317 

318 def deserialize_pointer[U: ArchiveTree, V]( 

319 self, pointer: Any, model_type: type[U], deserializer: Callable[[U, InputArchive[Any]], V] 

320 ) -> V: 

321 # Docstring inherited. 

322 raise ArchiveAccessRequiredError("Dereferencing an archive pointer requires file access.") 

323 

324 def get_frame_set(self, ref: Any) -> FrameSet: 

325 # Docstring inherited. 

326 raise ArchiveAccessRequiredError("Reading a frame set requires file access.") 

327 

328 def get_array( 

329 self, 

330 model: ArrayReferenceModel | InlineArrayModel, 

331 *, 

332 slices: tuple[slice, ...] | EllipsisType = ..., 

333 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

334 ) -> np.ndarray: 

335 # Docstring inherited. 

336 raise ArchiveAccessRequiredError("Reading an array requires file access.") 

337 

338 def get_table( 

339 self, 

340 model: TableModel, 

341 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

342 ) -> astropy.table.Table: 

343 # Docstring inherited. 

344 raise ArchiveAccessRequiredError("Reading a table requires file access.") 

345 

346 def get_structured_array( 

347 self, 

348 model: TableModel, 

349 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

350 ) -> np.ndarray: 

351 # Docstring inherited. 

352 raise ArchiveAccessRequiredError("Reading a structured array requires file access.")