Coverage for python/lsst/images/serialization/_common.py: 68%
53 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-05-29 08:40 +0000
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.
12from __future__ import annotations
14__all__ = (
15 "ArchiveReadError",
16 "ArchiveTree",
17 "ButlerInfo",
18 "InvalidComponentError",
19 "InvalidParameterError",
20 "JsonRef",
21 "MetadataValue",
22 "OpaqueArchiveMetadata",
23 "ReadResult",
24 "no_header_updates",
25)
27import operator
28from abc import ABC, abstractmethod
29from typing import TYPE_CHECKING, Any, NamedTuple, Protocol, Self
31import astropy.table
32import astropy.units
33import pydantic
35from .._geom import Box
36from ..utils import is_none
38try:
39 from lsst.daf.butler import DatasetProvenance, SerializedDatasetRef
40except ImportError:
41 type DatasetProvenance = Any # type: ignore[no-redef]
42 type SerializedDatasetRef = Any # type: ignore[no-redef]
44if TYPE_CHECKING:
45 import astropy.io.fits
47 from ._input_archive import InputArchive
50type MetadataValue = (
51 pydantic.StrictInt | pydantic.StrictFloat | pydantic.StrictStr | pydantic.StrictBool | None
52)
55class ButlerInfo(pydantic.BaseModel):
56 """Information about a butler dataset."""
58 dataset: SerializedDatasetRef
59 provenance: DatasetProvenance = pydantic.Field(default_factory=DatasetProvenance)
62class JsonRef(pydantic.BaseModel, serialize_by_alias=True):
63 """Pydantic model for JSON Reference / Pointer (IETF RFC 6901).
65 Notes
66 -----
67 This model does not do any of the escaping or special-character
68 interpretation required by the spec; it assumes that's already been done,
69 so its job is *just* putting a ``$ref`` field inside another model.
70 """
72 ref: str = pydantic.Field(alias="$ref")
75class ArchiveTree(
76 pydantic.BaseModel, ABC, ser_json_inf_nan="constants", ser_json_bytes="base64", val_json_bytes="base64"
77):
78 """An intermediate base class of `pydantic.BaseModel` that should be used
79 for all objects that may be used as the top-level tree models written to
80 archives.
81 """
83 metadata: dict[str, MetadataValue] = pydantic.Field(
84 default_factory=dict, description="Additional unstructured metadata.", exclude_if=operator.not_
85 )
86 butler_info: ButlerInfo | None = pydantic.Field(
87 default=None,
88 description="Information about the butler dataset backed by this file.",
89 exclude_if=is_none,
90 )
91 indirect: list[Any] = pydantic.Field(
92 default_factory=list,
93 description="Serialized nested objects that may be saved or read more than once.",
94 exclude_if=operator.not_,
95 )
97 @abstractmethod
98 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
99 """Return the in-memory object that was serialized to this tree.
101 Parameters
102 ----------
103 archive
104 The input archive to read from.
105 **kwargs
106 Additional keyword arguments specific to this type.
108 Raises
109 ------
110 ~lsst.images.serialization.InvalidParameterError
111 Raised for unsupported ``**kwargs``.
113 Notes
114 -----
115 Subclass implementations may take additional keyword-only arguments.
116 Callers that invoke this method without knowing what those might be
117 should catch `TypeError` and re-raise as
118 `~lsst.images.serialization.InvalidParameterError` if they pass
119 additional keyword arguments.
120 """
121 raise NotImplementedError()
123 def deserialize_component(self, component: str, archive: InputArchive[Any], **kwargs: Any) -> Any:
124 """Return a component in-memory object that was serialized to this
125 tree.
127 Parameters
128 ----------
129 component
130 Name of the component to read.
131 archive
132 The input archive to read from.
133 **kwargs
134 Additional keyword arguments specific to this type.
136 Raises
137 ------
138 ~lsst.images.serialization.InvalidComponentError
139 Raise if ``component`` is not recognized.
140 ~lsst.images.serialization.InvalidParameterError
141 Raised for unsupported ``**kwargs``.
143 Notes
144 -----
145 The default implementation for this method tries to get an attribute
146 with the component's name from ``self``, and then:
148 - returns `None` if it is `None`;
149 - calls `deserialize` on that object if it is also an
150 `~lsst.images.serialization.ArchiveTree`;
151 - returns it directly otherwise.
153 If there is no such attribute, it raises
154 `~lsst.images.serialization.InvalidComponentError`.
156 ``**kwargs`` are forwarded to component `deserialize` methods, but
157 are otherwise not checked. Subclasses are generally expected to
158 implement this method to do that checking and handle any components
159 for which the other will not work, and then delegate to `super` at
160 the end.
161 """
162 try:
163 component_model = getattr(self, component)
164 except AttributeError:
165 raise InvalidComponentError(
166 f"Component {component!r} is not recognized by {type(self).__name__}."
167 ) from None
168 if component_model is None:
169 return None
170 if isinstance(component_model, ArchiveTree):
171 return component_model.deserialize(archive, **kwargs)
172 return component_model
175class ReadResult[T: Any](NamedTuple):
176 """A struct that can be used to return both a deserialized object and
177 metadata associated with it, even when the in-memory type cannot hold
178 metadata.
179 """
181 deserialized: T
182 """The deserialized object itself."""
184 metadata: dict[str, MetadataValue]
185 """Additional flexible metadata stored with the object."""
187 butler_info: ButlerInfo | None
188 """Butler provenance information for the dataset this file backs."""
191class ArchiveReadError(RuntimeError):
192 """Exception raised when the contents of an archive cannot be read."""
195class InvalidParameterError(ArchiveReadError):
196 """Exception raised by `ArchiveTree.deserialize` or
197 `ArchiveTree.deserialize_component` when passed an invalid keyword
198 argument.
199 """
202class InvalidComponentError(ArchiveReadError):
203 """Exception `ArchiveTree.deserialize_component` when passed an invalid
204 component name.
205 """
208class OpaqueArchiveMetadata(Protocol):
209 """Interface for opaque archive metadata.
211 In addition to implementing the methods defined here, all implementations
212 must be pickleable.
213 """
215 def copy(self) -> Self | None:
216 """Copy, reference, or discard metadata when its holding object is
217 copied.
218 """
219 ...
221 def subset(self, bbox: Box) -> Self | None:
222 """Copy, reference, or discard metadata when a subset of its its
223 holding object is extracted.
224 """
225 ...
228def no_header_updates(header: astropy.io.fits.Header) -> None:
229 """Do not make any modifications to the given FITS header."""