Coverage for python / lsst / images / serialization / _common.py: 88%

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__ = ( 

15 "ArchiveReadError", 

16 "ArchiveTree", 

17 "ButlerInfo", 

18 "JsonRef", 

19 "MetadataValue", 

20 "OpaqueArchiveMetadata", 

21 "ReadResult", 

22 "no_header_updates", 

23) 

24 

25import operator 

26from abc import ABC, abstractmethod 

27from typing import TYPE_CHECKING, Any, NamedTuple, Protocol, Self 

28 

29import astropy.table 

30import astropy.units 

31import pydantic 

32 

33from .._geom import Box 

34from ..utils import is_none 

35 

36try: 

37 from lsst.daf.butler import DatasetProvenance, SerializedDatasetRef 

38except ImportError: 

39 type DatasetProvenance = Any # type: ignore[no-redef] 

40 type SerializedDatasetRef = Any # type: ignore[no-redef] 

41 

42if TYPE_CHECKING: 

43 import astropy.io.fits 

44 

45 from ._input_archive import InputArchive 

46 

47 

48type MetadataValue = ( 

49 pydantic.StrictInt | pydantic.StrictFloat | pydantic.StrictStr | pydantic.StrictBool | None 

50) 

51 

52 

53class ButlerInfo(pydantic.BaseModel): 

54 """Information about a butler dataset.""" 

55 

56 dataset: SerializedDatasetRef 

57 provenance: DatasetProvenance = pydantic.Field(default_factory=DatasetProvenance) 

58 

59 

60class JsonRef(pydantic.BaseModel, serialize_by_alias=True): 

61 """Pydantic model for JSON Reference / Pointer (IETF RFC 6901). 

62 

63 Notes 

64 ----- 

65 This model does not do any of the escaping or special-character 

66 interpretation required by the spec; it assumes that's already been done, 

67 so its job is *just* putting a ``$ref`` field inside another model. 

68 """ 

69 

70 ref: str = pydantic.Field(alias="$ref") 

71 

72 

73class ArchiveTree( 

74 pydantic.BaseModel, ABC, ser_json_inf_nan="constants", ser_json_bytes="base64", val_json_bytes="base64" 

75): 

76 """An intermediate base class of `pydantic.BaseModel` that should be used 

77 for all objects that may be used as the top-level tree models written to 

78 archives. 

79 """ 

80 

81 metadata: dict[str, MetadataValue] = pydantic.Field( 

82 default_factory=dict, description="Additional unstructured metadata.", exclude_if=operator.not_ 

83 ) 

84 butler_info: ButlerInfo | None = pydantic.Field( 

85 default=None, 

86 description="Information about the butler dataset backed by this file.", 

87 exclude_if=is_none, 

88 ) 

89 indirect: list[Any] = pydantic.Field( 

90 default_factory=list, 

91 description="Serialized nested objects that may be saved or read more than once.", 

92 exclude_if=operator.not_, 

93 ) 

94 

95 @abstractmethod 

96 def deserialize(self, archive: InputArchive[Any]) -> Any: 

97 """Return the in-memory object that was serialized to this tree.""" 

98 raise NotImplementedError() 

99 

100 

101class ReadResult[T: Any](NamedTuple): 

102 """A struct that can be used to return both a deserialized object and 

103 metadata associated with it, even when the in-memory type cannot hold 

104 metadata. 

105 """ 

106 

107 deserialized: T 

108 """The deserialized object itself.""" 

109 

110 metadata: dict[str, MetadataValue] 

111 """Additional flexible metadata stored with the object.""" 

112 

113 butler_info: ButlerInfo | None 

114 """Butler provenance information for the dataset this file backs.""" 

115 

116 

117class ArchiveReadError(RuntimeError): 

118 """Exception raised when the contents of an archive cannot be read.""" 

119 

120 

121class OpaqueArchiveMetadata(Protocol): 

122 """Interface for opaque archive metadata. 

123 

124 In addition to implementing the methods defined here, all implementations 

125 must be pickleable. 

126 """ 

127 

128 def copy(self) -> Self | None: 

129 """Copy, reference, or discard metadata when its holding object is 

130 copied. 

131 """ 

132 ... 

133 

134 def subset(self, bbox: Box) -> Self | None: 

135 """Copy, reference, or discard metadata when a subset of its its 

136 holding object is extracted. 

137 """ 

138 ... 

139 

140 

141def no_header_updates(header: astropy.io.fits.Header) -> None: 

142 """Do not make any modifications to the given FITS header."""