Coverage for python/lsst/images/ndf/_common.py: 100%

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 "HdsNameShrinker", 

16 "NdfPointerModel", 

17 "archive_path_to_hdf5_path", 

18 "archive_path_to_hdf5_path_components", 

19) 

20 

21import pydantic 

22 

23from ._hds import DAT__SZNAM 

24 

25 

26class NdfPointerModel(pydantic.BaseModel): 

27 """Reference to an NDF-archive sub-tree by HDF5 path. 

28 

29 Used by `NdfOutputArchive`/`NdfInputArchive` to point to 

30 sub-trees that have been hoisted out of the main JSON tree into separate 

31 HDS components. 

32 """ 

33 

34 path: str 

35 """HDF5 absolute path (e.g. ``/MORE/LSST/PSF``).""" 

36 

37 

38class HdsNameShrinker: 

39 """Shrink HDS component names to fit the HDS object-name length limit. 

40 

41 Names are uppercased; names at or under the limit pass through unchanged. 

42 Each distinct over-long name is assigned a readable prefix followed by an 

43 underscore-separated hexadecimal counter that increments per assignment, 

44 so distinct names written through the same shrinker never collide. 

45 Assignments are remembered, so repeated requests for the same name return 

46 the same result. 

47 

48 A shrinker is scoped to a single output file: shrunk names are not stable 

49 across files, so readers must use the paths recorded in the file's JSON 

50 tree rather than recomputing them. 

51 

52 Parameters 

53 ---------- 

54 max_length 

55 Maximum component length, by default the HDS limit (``DAT__SZNAM``). 

56 """ 

57 

58 def __init__(self, max_length: int = DAT__SZNAM) -> None: 

59 self._max_length = max_length 

60 self._assigned: dict[tuple[str, int], str] = {} 

61 self._counter = 0 

62 

63 def shrink(self, name: str, reserve: int = 0) -> str: 

64 """Shrink a component name to fit in ``max_length - reserve``. 

65 

66 Parameters 

67 ---------- 

68 name 

69 The component name to shrink. 

70 reserve 

71 Number of characters to leave available for a suffix the caller 

72 will append (e.g. a version suffix). 

73 

74 Returns 

75 ------- 

76 `str` 

77 The uppercased name, unchanged if it fits, otherwise truncated 

78 and suffixed with ``_`` and an uppercase hexadecimal counter 

79 (at least three digits) so the result is exactly 

80 ``max_length - reserve`` characters. 

81 """ 

82 max_length = self._max_length - reserve 

83 name = name.upper() 

84 if len(name) <= max_length: 

85 return name 

86 key = (name, max_length) 

87 if (shrunk := self._assigned.get(key)) is None: 

88 self._counter += 1 

89 token = f"_{self._counter:03X}" 

90 shrunk = f"{name[: max_length - len(token)]}{token}" 

91 self._assigned[key] = shrunk 

92 return shrunk 

93 

94 def shrink_versioned(self, base: str, version: int) -> str: 

95 """Shrink a component while preserving a visible version suffix. 

96 

97 When ``version`` is greater than one a ``_{version}`` suffix is 

98 reserved at the tail and the ``base`` is shrunk into the remaining 

99 characters, so the version number stays readable in Starlink tools. 

100 Version one (the first occurrence) is shrunk exactly like an 

101 unversioned component. 

102 """ 

103 suffix = f"_{version}" if version > 1 else "" 

104 return self.shrink(base, reserve=len(suffix)) + suffix 

105 

106 

107def archive_path_to_hdf5_path(archive_path: str, shrinker: HdsNameShrinker) -> str: 

108 """Translate a serialization archive path to an NDF HDF5 path. 

109 

110 The empty path maps to the main JSON tree at ``/MORE/LSST/JSON``. 

111 Any non-empty path is uppercased and kept hierarchical under 

112 ``/MORE/LSST/``. This mirrors the serialization path while keeping HDS 

113 component names within the HDS object-name limit (``DAT__SZNAM``). 

114 """ 

115 if not archive_path: 

116 return "/MORE/LSST/JSON" 

117 components = archive_path_to_hdf5_path_components(archive_path, shrinker) 

118 return "/MORE/LSST/" + "/".join(components) 

119 

120 

121def archive_path_to_hdf5_path_components(archive_path: str, shrinker: HdsNameShrinker) -> list[str]: 

122 """Return HDS-compatible path components for an archive path. 

123 

124 Each component is uppercased; components longer than the HDS object-name 

125 limit (``DAT__SZNAM``) are shrunk by ``shrinker``. 

126 """ 

127 return [shrinker.shrink(component) for component in archive_path.strip("/").split("/") if component]