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

28 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-18 09:00 +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. 

11 

12from __future__ import annotations 

13 

14__all__ = ( 

15 "ArchiveReadError", 

16 "ArchiveTree", 

17 "ButlerInfo", 

18 "MetadataValue", 

19 "OpaqueArchiveMetadata", 

20 "no_header_updates", 

21) 

22 

23import operator 

24from typing import TYPE_CHECKING, Any, Protocol, Self 

25 

26import astropy.table 

27import astropy.units 

28import pydantic 

29 

30from .._geom import Box 

31from ..utils import is_none 

32 

33try: 

34 from lsst.daf.butler import DatasetProvenance, SerializedDatasetRef 

35except ImportError: 

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

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

38 

39if TYPE_CHECKING: 

40 import astropy.io.fits 

41 

42 

43type MetadataValue = ( 

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

45) 

46 

47 

48class ButlerInfo(pydantic.BaseModel): 

49 """Information about a butler dataset.""" 

50 

51 dataset: SerializedDatasetRef 

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

53 

54 

55class ArchiveTree( 

56 pydantic.BaseModel, ser_json_inf_nan="constants", ser_json_bytes="base64", val_json_bytes="base64" 

57): 

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

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

60 archives. 

61 """ 

62 

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

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

65 ) 

66 butler_info: ButlerInfo | None = pydantic.Field( 

67 default=None, 

68 description="Information aobut the butler dataset backed by this file.", 

69 exclude_if=is_none, 

70 ) 

71 

72 

73class ArchiveReadError(RuntimeError): 

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

75 

76 

77class OpaqueArchiveMetadata(Protocol): 

78 """Interface for opaque archive metadata. 

79 

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

81 must be pickleable. 

82 """ 

83 

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

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

86 copied. 

87 """ 

88 ... 

89 

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

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

92 holding object is extracted. 

93 """ 

94 ... 

95 

96 

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

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