Coverage for python/lsst/images/serialization/_io.py: 91%

87 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-07-09 02:27 -0700

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"""Generic ``read_archive`` / ``write_archive`` dispatchers and the 

12schema-name registry. 

13""" 

14 

15from __future__ import annotations 

16 

17__all__ = ( 

18 "class_for_schema", 

19 "parameterize_tree", 

20 "public_type_for_schema", 

21 "read_archive", 

22 "register_schema_class", 

23 "tree_class_for_info", 

24 "write_archive", 

25) 

26 

27import importlib 

28import importlib.metadata 

29from typing import IO, TYPE_CHECKING, Any, overload 

30 

31from lsst.resources import ResourcePathExpression 

32 

33from ._backends import backend_for_path 

34from ._common import ArchiveReadError, ArchiveTree 

35 

36if TYPE_CHECKING: 

37 from ._input_archive import ArchiveInfo 

38 

39_REGISTRY: dict[str, type[ArchiveTree]] = {} 

40"""Map of ``SCHEMA_NAME`` to the registered ``ArchiveTree`` subclass. 

41 

42The registry is keyed by name only. Schema-version compatibility is 

43enforced when the selected tree's ``model_validate*`` runs, via 

44``min_read_version``. 

45""" 

46 

47_SCHEMA_ENTRY_POINT_GROUP = "lsst.images.schemas" 

48"""Entry point group for third-party serialization-model providers.""" 

49 

50_BUILTIN_SCHEMA_PROVIDERS: dict[str, str] = { 

51 "cell_aperture_correction_map": ( 

52 "lsst.images.cells._aperture_corrections:CellApertureCorrectionMapSerializationModel" 

53 ), 

54 "cell_coadd": "lsst.images.cells._coadd:CellCoaddSerializationModel", 

55 "cell_psf": "lsst.images.cells._psf:CellPointSpreadFunctionSerializationModel", 

56 "coadd_provenance": "lsst.images.cells._provenance:CoaddProvenanceSerializationModel", 

57} 

58"""Schema providers owned by this package but not imported by ``lsst.images``. 

59 

60These duplicate the package's own ``lsst.images.schemas`` entry points so 

61source-tree use via ``PYTHONPATH=python`` has the same lazy-import behavior as 

62an installed distribution with entry point metadata. 

63 

64Schemas whose model classes are imported unconditionally by ``lsst.images`` do 

65not need built-in providers or entry points: their ``ArchiveTree`` subclass 

66hooks register them before this lazy path is needed. 

67""" 

68 

69 

70def class_for_schema(schema_name: str) -> type[ArchiveTree] | None: 

71 """Return the registered ``ArchiveTree`` subclass for ``schema_name``. 

72 

73 If no class is already registered, this attempts schema-specific lazy 

74 imports from built-in providers and then from entry points in the 

75 ``lsst.images.schemas`` group before returning `None`. 

76 

77 Parameters 

78 ---------- 

79 schema_name 

80 Schema name (e.g. ``"visit_image"``). 

81 """ 

82 if (cls := _REGISTRY.get(schema_name)) is not None: 

83 return cls 

84 _load_builtin_schema_provider(schema_name) 

85 if (cls := _REGISTRY.get(schema_name)) is not None: 

86 return cls 

87 _load_schema_entry_points(schema_name) 

88 return _REGISTRY.get(schema_name) 

89 

90 

91def register_schema_class(cls: type[ArchiveTree]) -> None: 

92 """Register ``cls`` under ``cls.SCHEMA_NAME``. 

93 

94 No-op when the same class is registered again (re-import during 

95 tests). Raises `RuntimeError` when a *different* class is 

96 registered under an existing name. 

97 

98 Intended to be called from ``ArchiveTree.__pydantic_init_subclass__``; 

99 not part of the public API. 

100 """ 

101 key = cls.SCHEMA_NAME 

102 existing = _REGISTRY.get(key) 

103 if existing is cls: 

104 return 

105 if existing is not None: 

106 raise RuntimeError( 

107 f"Schema {cls.SCHEMA_NAME!r} is already registered to " 

108 f"{existing.__qualname__}; refusing to replace it with " 

109 f"{cls.__qualname__}." 

110 ) 

111 _REGISTRY[key] = cls 

112 

113 

114def _load_builtin_schema_provider(schema_name: str) -> None: 

115 """Import a package-local provider for ``schema_name``, if one exists.""" 

116 provider = _BUILTIN_SCHEMA_PROVIDERS.get(schema_name) 

117 if provider is None: 

118 return 

119 try: 

120 obj = _load_provider_object(provider) 

121 except Exception as err: 

122 raise ArchiveReadError( 

123 f"Could not load built-in schema provider {provider!r} for schema {schema_name!r}: {err}" 

124 ) from err 

125 _register_provider_object(obj) 

126 if schema_name not in _REGISTRY: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 raise ArchiveReadError( 

128 f"Built-in schema provider {provider!r} did not register schema {schema_name!r}." 

129 ) 

130 

131 

132def _load_schema_entry_points(schema_name: str) -> None: 

133 """Load entry points named ``schema_name`` from ``lsst.images.schemas``.""" 

134 loaded: list[str] = [] 

135 for entry_point in importlib.metadata.entry_points( 

136 group=_SCHEMA_ENTRY_POINT_GROUP, 

137 name=schema_name, 

138 ): 

139 loaded.append(entry_point.value) 

140 try: 

141 obj = entry_point.load() 

142 except Exception as err: 

143 raise ArchiveReadError( 

144 f"Could not load schema provider entry point {entry_point.value!r} " 

145 f"for schema {schema_name!r}: {err}" 

146 ) from err 

147 _register_provider_object(obj) 

148 if loaded and schema_name not in _REGISTRY: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true

149 raise ArchiveReadError( 

150 f"Schema provider entry point(s) for {schema_name!r} did not register that schema: {loaded}." 

151 ) 

152 

153 

154def _load_provider_object(provider: str) -> object: 

155 """Load ``module[:attribute[.nested]]`` provider specifications.""" 

156 module_name, _, attr_path = provider.partition(":") 

157 obj: object = importlib.import_module(module_name) 

158 if attr_path: 158 ↛ 161line 158 didn't jump to line 161 because the condition on line 158 was always true

159 for attr in attr_path.split("."): 

160 obj = getattr(obj, attr) 

161 return obj 

162 

163 

164def _register_provider_object(obj: object) -> None: 

165 """Register ``obj`` if a provider returned an ``ArchiveTree`` subclass.""" 

166 if isinstance(obj, type) and issubclass(obj, ArchiveTree): 166 ↛ exitline 166 didn't return from function '_register_provider_object' because the condition on line 166 was always true

167 register_schema_class(obj) 

168 

169 

170def tree_class_for_info(info: ArchiveInfo, path: ResourcePathExpression | IO[bytes]) -> type[ArchiveTree]: 

171 """Return the registered `ArchiveTree` subclass for ``info``'s schema. 

172 

173 Parameters 

174 ---------- 

175 info 

176 Basic archive info whose ``schema_name`` selects the tree class. 

177 path 

178 Path or stream being opened, used only for the error message. 

179 

180 Raises 

181 ------ 

182 ArchiveReadError 

183 If no class is registered for the schema. 

184 """ 

185 tree_cls = class_for_schema(info.schema_name) 

186 if tree_cls is None: 

187 raise ArchiveReadError(f"No registered schema {info.schema_name!r}; cannot open {path!r}.") 

188 return tree_cls 

189 

190 

191def parameterize_tree( 

192 tree_cls: type[ArchiveTree], 

193 pointer_type: type[Any], 

194) -> type[ArchiveTree]: 

195 """Parameterise ``tree_cls`` over ``pointer_type`` if it is generic. 

196 

197 Some `ArchiveTree` subclasses (e.g. ``SumFieldSerializationModel``) 

198 take no type parameters; their ``_get_archive_tree_type`` returns 

199 the class itself. Match that behaviour here so per-backend 

200 ``open_tree`` implementations can call this uniformly. 

201 

202 Parameters 

203 ---------- 

204 tree_cls 

205 Archive tree class to parameterise. 

206 pointer_type 

207 Pointer type to parameterise ``tree_cls`` over when it is generic. 

208 """ 

209 if not getattr(tree_cls, "__parameters__", ()): 

210 return tree_cls 

211 return tree_cls[pointer_type] # type: ignore[index] 

212 

213 

214def public_type_for_schema(schema_name: str) -> type | None: 

215 """Return the in-memory Python class produced when reading an archive 

216 whose top-level tree has schema name ``schema_name``. 

217 

218 Looks the schema name up in the registry and returns the registered 

219 tree's ``PUBLIC_TYPE`` ClassVar (the type its ``deserialize`` produces). 

220 Returns `None` when nothing is registered for ``schema_name``. 

221 

222 Parameters 

223 ---------- 

224 schema_name 

225 Schema name (e.g. ``"visit_image"``). 

226 """ 

227 tree_cls = class_for_schema(schema_name) 

228 if tree_cls is None: 

229 return None 

230 return getattr(tree_cls, "PUBLIC_TYPE", None) 

231 

232 

233@overload 

234def read_archive[T]( 

235 path: ResourcePathExpression | IO[bytes], cls: type[T], *, format: str | None = ..., **kwargs: Any 

236) -> T: ... 

237@overload 

238def read_archive( 238 ↛ exitline 238 didn't return from function 'read_archive' because

239 path: ResourcePathExpression | IO[bytes], cls: None = ..., *, format: str | None = ..., **kwargs: Any 

240) -> Any: ... 

241def read_archive( 

242 path: ResourcePathExpression | IO[bytes], 

243 cls: type[Any] | None = None, 

244 *, 

245 format: str | None = None, 

246 **kwargs: Any, 

247) -> Any: 

248 """Read an archive whose in-memory type is inferred from its schema. 

249 

250 Dispatches to the appropriate backend based on ``path``'s extension (or, 

251 for stream input, its leading bytes), resolves the registered in-memory 

252 type from the file's schema, and returns the fully deserialized object. 

253 Schema-version compatibility is enforced when the model validates the 

254 on-disk tree, via ``min_read_version``. 

255 A path with a ``.gz``/``.zst`` compression suffix is decompressed 

256 transparently; stream input must already be decompressed. 

257 

258 This is the convenient way to read a whole object. To read individual 

259 components, or to reach the metadata and butler info stored alongside the 

260 object, use `open_archive` instead. 

261 

262 Parameters 

263 ---------- 

264 path 

265 File to read; convertible to `lsst.resources.ResourcePath`, or a 

266 seekable binary stream containing the file's content (e.g. 

267 ``io.BytesIO(data)`` for in-memory bytes). 

268 cls 

269 Optional expected in-memory type. 

270 When given, the file's schema is checked against ``cls`` and the 

271 deserialized object is validated with ``isinstance`` (raising 

272 `TypeError` otherwise), and the static return type is ``T``. 

273 format 

274 Optional backend name (``"fits"``, ``"ndf"``, or ``"json"``) 

275 forcing the backend, instead of dispatching on the path's extension 

276 or the stream's leading bytes. 

277 **kwargs 

278 Type-specific keyword arguments forwarded to the object's 

279 ``deserialize`` (e.g. ``bbox`` for an image subset read). 

280 Mis-targeted arguments surface as ``TypeError``. 

281 Backend-specific open options (e.g. ``page_size``) are not accepted 

282 here; use `open_archive` for those. 

283 

284 Returns 

285 ------- 

286 object 

287 The deserialized object. 

288 

289 Raises 

290 ------ 

291 ValueError 

292 Raised by `backend_for_path` if the file extension is not 

293 recognized, by `backend_for_stream` if a stream's leading bytes are 

294 not recognized, or by `backend_for_name` if ``format`` is not a 

295 known backend name. 

296 ArchiveReadError 

297 Raised when the file's ``schema_name`` is not registered, or 

298 propagated from the model's ``min_read_version`` check on 

299 ``model_validate*``. 

300 TypeError 

301 Raised when ``cls`` is given and the file's schema or the 

302 deserialized object is not compatible with it. 

303 """ 

304 # Imported here to break the _io <-> _reader import cycle: _reader imports 

305 # class_for_schema / public_type_for_schema from this module at load time. 

306 from ._reader import open_archive 

307 

308 # A subset read (any deserialize kwarg with a value) reads incrementally; 

309 # a plain whole-object read may slurp the file up front. This mirrors the 

310 # ``partial`` default the per-backend readers used. 

311 partial = any(value is not None for value in kwargs.values()) 

312 with open_archive(path, cls, format=format, partial=partial) as reader: 

313 return reader.read(**kwargs) 

314 

315 

316def write_archive(obj: Any, path: str, **kwargs: Any) -> Any: 

317 """Write ``obj`` to ``path``, dispatching by file extension. 

318 

319 Forwards ``**kwargs`` to the per-backend ``write`` (e.g. 

320 ``compression_options`` for FITS). No registry lookup is performed: 

321 the per-backend ``write`` already accepts any object with a 

322 ``serialize`` method. 

323 

324 Parameters 

325 ---------- 

326 obj 

327 Object to write; must implement ``serialize`` like the per-backend 

328 write functions expect. 

329 path 

330 Destination path. The extension selects the backend. 

331 **kwargs 

332 Forwarded verbatim to the backend's ``write``. 

333 

334 Returns 

335 ------- 

336 Any 

337 Whatever the per-backend ``write`` returns (the serialised 

338 archive tree). 

339 """ 

340 return backend_for_path(path).write(obj, path, **kwargs)