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

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 tree_cls: type[ArchiveTree], 

120 *, 

121 partial: bool = True, 

122 **backend_kwargs: Any, 

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

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

125 ``(archive, tree)`` as a context manager. 

126 

127 ``tree_cls`` is the un-parameterized `ArchiveTree` subclass; each 

128 backend parameterizes it with its own pointer model. Backend-specific 

129 open options (e.g. ``page_size`` for FITS) are accepted via 

130 ``**backend_kwargs``; ``partial`` is honoured where meaningful. 

131 

132 Each concrete backend implements this. 

133 """ 

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

135 

136 @abstractmethod 

137 def deserialize_pointer[U: ArchiveTree, V]( 

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

139 ) -> V: 

140 """Deserialize an object that was saved by 

141 `~lsst.serialization.OutputArchive.serialize_pointer`. 

142 

143 Parameters 

144 ---------- 

145 pointer 

146 JSON Pointer model to dereference. 

147 model_type 

148 Pydantic model type that the pointer should dereference to. 

149 deserializer 

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

151 archive, and returns the deserialized object. 

152 

153 Returns 

154 ------- 

155 V 

156 The deserialized object. 

157 

158 Notes 

159 ----- 

160 Implementations are required to remember previously-deserialized 

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

162 times. 

163 

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

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

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

167 of its Pydantic tree. 

168 """ 

169 raise NotImplementedError() 

170 

171 @abstractmethod 

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

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

174 

175 Parameters 

176 ---------- 

177 ref 

178 Implementation-specific reference to the frame set. 

179 

180 Returns 

181 ------- 

182 FrameSet 

183 Loaded frame set. 

184 """ 

185 raise NotImplementedError() 

186 

187 @abstractmethod 

188 def get_array( 

189 self, 

190 model: ArrayReferenceModel | InlineArrayModel, 

191 *, 

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

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

194 ) -> np.ndarray: 

195 """Load an array from the archive. 

196 

197 Parameters 

198 ---------- 

199 model 

200 A Pydantic model that references or holds the array. 

201 slices 

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

203 strip_header 

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

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

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

207 """ 

208 raise NotImplementedError() 

209 

210 @abstractmethod 

211 def get_table( 

212 self, 

213 model: TableModel, 

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

215 ) -> astropy.table.Table: 

216 """Load a table from the archive. 

217 

218 Parameters 

219 ---------- 

220 model 

221 A Pydantic model that references or holds the table. 

222 strip_header 

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

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

225 `~lsst.serialization.OutputArchive.add_table`. 

226 

227 Returns 

228 ------- 

229 astropy.table.Table 

230 The loaded table. 

231 """ 

232 raise NotImplementedError() 

233 

234 @abstractmethod 

235 def get_structured_array( 

236 self, 

237 model: TableModel, 

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

239 ) -> np.ndarray: 

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

241 

242 Parameters 

243 ---------- 

244 model 

245 A Pydantic model that references or holds the table. 

246 strip_header 

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

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

249 `~lsst.serialization.OutputArchive.add_structured_array`. 

250 

251 Returns 

252 ------- 

253 numpy.ndarray 

254 The loaded table as a structured array. 

255 """ 

256 raise NotImplementedError() 

257 

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

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

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

261 

262 Returns 

263 ------- 

264 OpaqueArchiveMetadata 

265 Opaque metadata specific to this archive type that should be 

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

267 """ 

268 return None