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", "InputArchive") 

15 

16from abc import ABC, abstractmethod 

17from collections.abc import Callable 

18from contextlib import AbstractContextManager 

19from types import EllipsisType 

20from typing import 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 SCHEMA_URL_HOST, ArchiveTree, OpaqueArchiveMetadata, no_header_updates 

32from ._tables import TableModel 

33 

34if TYPE_CHECKING: 

35 from .._transforms import FrameSet 

36 

37 

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

39# autodoc-typehints plugin. 

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

41 

42 

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

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

45 

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

47 """ 

48 

49 schema_url: str 

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

51 

52 schema_name: str 

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

54 

55 schema_version: str 

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

57 

58 format_version: int | None 

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

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

61 

62 @classmethod 

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

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

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

66 

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

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

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

70 """ 

71 parsed = ResourcePath(schema_url) 

72 if parsed.netloc != SCHEMA_URL_HOST: 

73 raise ValueError( 

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

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

76 ) 

77 tail = parsed.basename() 

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

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

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

81 if not name or not version: 

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

83 return cls( 

84 schema_url=schema_url, 

85 schema_name=name, 

86 schema_version=version, 

87 format_version=format_version, 

88 ) 

89 

90 

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

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

93 

94 Notes 

95 ----- 

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

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

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

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

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

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

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

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

104 """ 

105 

106 @classmethod 

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

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

109 without deserializing pixel data. 

110 

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

112 """ 

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

114 

115 @classmethod 

116 def open_tree( 

117 cls, 

118 path: ResourcePathExpression, 

119 *, 

120 partial: bool = True, 

121 **backend_kwargs: Any, 

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

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

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

125 

126 Parameters 

127 ---------- 

128 path 

129 File to be opened. Can be local or remote. 

130 partial 

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

132 Can be ignored by a backend where not relevant. 

133 **backend_kwargs 

134 Any keyword parameters that should be forwarded to the backend 

135 open. 

136 

137 Raises 

138 ------ 

139 ArchiveReadError 

140 If the file's schema is not registered. 

141 

142 Notes 

143 ----- 

144 Each concrete backend implements this. 

145 """ 

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

147 

148 @abstractmethod 

149 def deserialize_pointer[U: ArchiveTree, V]( 

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

151 ) -> V: 

152 """Deserialize an object that was saved by 

153 `~lsst.serialization.OutputArchive.serialize_pointer`. 

154 

155 Parameters 

156 ---------- 

157 pointer 

158 JSON Pointer model to dereference. 

159 model_type 

160 Pydantic model type that the pointer should dereference to. 

161 deserializer 

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

163 archive, and returns the deserialized object. 

164 

165 Returns 

166 ------- 

167 V 

168 The deserialized object. 

169 

170 Notes 

171 ----- 

172 Implementations are required to remember previously-deserialized 

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

174 times. 

175 

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

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

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

179 of its Pydantic tree. 

180 """ 

181 raise NotImplementedError() 

182 

183 @abstractmethod 

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

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

186 

187 Parameters 

188 ---------- 

189 ref 

190 Implementation-specific reference to the frame set. 

191 

192 Returns 

193 ------- 

194 FrameSet 

195 Loaded frame set. 

196 """ 

197 raise NotImplementedError() 

198 

199 @abstractmethod 

200 def get_array( 

201 self, 

202 model: ArrayReferenceModel | InlineArrayModel, 

203 *, 

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

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

206 ) -> np.ndarray: 

207 """Load an array from the archive. 

208 

209 Parameters 

210 ---------- 

211 model 

212 A Pydantic model that references or holds the array. 

213 slices 

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

215 strip_header 

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

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

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

219 """ 

220 raise NotImplementedError() 

221 

222 @abstractmethod 

223 def get_table( 

224 self, 

225 model: TableModel, 

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

227 ) -> astropy.table.Table: 

228 """Load a table from the archive. 

229 

230 Parameters 

231 ---------- 

232 model 

233 A Pydantic model that references or holds the table. 

234 strip_header 

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

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

237 `~lsst.serialization.OutputArchive.add_table`. 

238 

239 Returns 

240 ------- 

241 astropy.table.Table 

242 The loaded table. 

243 """ 

244 raise NotImplementedError() 

245 

246 @abstractmethod 

247 def get_structured_array( 

248 self, 

249 model: TableModel, 

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

251 ) -> np.ndarray: 

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

253 

254 Parameters 

255 ---------- 

256 model 

257 A Pydantic model that references or holds the table. 

258 strip_header 

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

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

261 `~lsst.serialization.OutputArchive.add_structured_array`. 

262 

263 Returns 

264 ------- 

265 numpy.ndarray 

266 The loaded table as a structured array. 

267 """ 

268 raise NotImplementedError() 

269 

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

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

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

273 

274 Returns 

275 ------- 

276 OpaqueArchiveMetadata 

277 Opaque metadata specific to this archive type that should be 

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

279 """ 

280 return None