Coverage for python / lsst / images / serialization / _common.py: 87%
38 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-30 09:07 +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 "JsonRef",
19 "MetadataValue",
20 "OpaqueArchiveMetadata",
21 "ReadResult",
22 "no_header_updates",
23)
25import operator
26from typing import TYPE_CHECKING, Any, NamedTuple, Protocol, Self
28import astropy.table
29import astropy.units
30import pydantic
32from .._geom import Box
33from ..utils import is_none
35try:
36 from lsst.daf.butler import DatasetProvenance, SerializedDatasetRef
37except ImportError:
38 type DatasetProvenance = Any # type: ignore[no-redef]
39 type SerializedDatasetRef = Any # type: ignore[no-redef]
41if TYPE_CHECKING:
42 import astropy.io.fits
45type MetadataValue = (
46 pydantic.StrictInt | pydantic.StrictFloat | pydantic.StrictStr | pydantic.StrictBool | None
47)
50class ButlerInfo(pydantic.BaseModel):
51 """Information about a butler dataset."""
53 dataset: SerializedDatasetRef
54 provenance: DatasetProvenance = pydantic.Field(default_factory=DatasetProvenance)
57class JsonRef(pydantic.BaseModel, serialize_by_alias=True):
58 """Pydantic model for JSON Reference / Pointer (IETF RFC 6901).
60 Notes
61 -----
62 This model does not do any of the escaping or special-character
63 interpretation required by the spec; it assumes that's already been done,
64 so its job is *just* putting a ``$ref`` field inside another model.
65 """
67 ref: str = pydantic.Field(alias="$ref")
70class ArchiveTree(
71 pydantic.BaseModel, ser_json_inf_nan="constants", ser_json_bytes="base64", val_json_bytes="base64"
72):
73 """An intermediate base class of `pydantic.BaseModel` that should be used
74 for all objects that may be used as the top-level tree models written to
75 archives.
76 """
78 metadata: dict[str, MetadataValue] = pydantic.Field(
79 default_factory=dict, description="Additional unstructured metadata.", exclude_if=operator.not_
80 )
81 butler_info: ButlerInfo | None = pydantic.Field(
82 default=None,
83 description="Information about the butler dataset backed by this file.",
84 exclude_if=is_none,
85 )
86 indirect: list[Any] = pydantic.Field(
87 default_factory=list,
88 description="Serialized nested objects that may be saved or read more than once.",
89 exclude_if=operator.not_,
90 )
93class ReadResult[T: Any](NamedTuple):
94 """A struct that can be used to return both a deserialized object and
95 metadata associated with it, even when the in-memory type cannot hold
96 metadata.
97 """
99 deserialized: T
100 """The deserialized object itself."""
102 metadata: dict[str, MetadataValue]
103 """Additional flexible metadata stored with the object."""
105 butler_info: ButlerInfo | None
106 """Butler provenance information for the dataset this file backs."""
109class ArchiveReadError(RuntimeError):
110 """Exception raised when the contents of an archive cannot be read."""
113class OpaqueArchiveMetadata(Protocol):
114 """Interface for opaque archive metadata.
116 In addition to implementing the methods defined here, all implementations
117 must be pickleable.
118 """
120 def copy(self) -> Self | None:
121 """Copy, reference, or discard metadata when its holding object is
122 copied.
123 """
124 ...
126 def subset(self, bbox: Box) -> Self | None:
127 """Copy, reference, or discard metadata when a subset of its its
128 holding object is extracted.
129 """
130 ...
133def no_header_updates(header: astropy.io.fits.Header) -> None:
134 """Do not make any modifications to the given FITS header."""