Coverage for python/lsst/images/ndf/_common.py: 100%
34 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-27 08:32 +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 "HdsNameShrinker",
16 "NdfPointerModel",
17 "archive_path_to_hdf5_path",
18 "archive_path_to_hdf5_path_components",
19)
21import pydantic
23from ._hds import DAT__SZNAM
26class NdfPointerModel(pydantic.BaseModel):
27 """Reference to an NDF-archive sub-tree by HDF5 path.
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 """
34 path: str
35 """HDF5 absolute path (e.g. ``/MORE/LSST/PSF``)."""
38class HdsNameShrinker:
39 """Shrink HDS component names to fit the HDS object-name length limit.
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.
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.
52 Parameters
53 ----------
54 max_length
55 Maximum component length, by default the HDS limit (``DAT__SZNAM``).
56 """
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
63 def shrink(self, name: str, reserve: int = 0) -> str:
64 """Shrink a component name to fit in ``max_length - reserve``.
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).
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
94 def shrink_versioned(self, base: str, version: int) -> str:
95 """Shrink a component while preserving a visible version suffix.
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.
103 Parameters
104 ----------
105 base
106 Component name to shrink.
107 version
108 Version number whose suffix is preserved when greater than one.
109 """
110 suffix = f"_{version}" if version > 1 else ""
111 return self.shrink(base, reserve=len(suffix)) + suffix
114def archive_path_to_hdf5_path(archive_path: str, shrinker: HdsNameShrinker) -> str:
115 """Translate a serialization archive path to an NDF HDF5 path.
117 The empty path maps to the main JSON tree at ``/MORE/LSST/JSON``.
118 Any non-empty path is uppercased and kept hierarchical under
119 ``/MORE/LSST/``. This mirrors the serialization path while keeping HDS
120 component names within the HDS object-name limit (``DAT__SZNAM``).
122 Parameters
123 ----------
124 archive_path
125 Serialization archive path to translate.
126 shrinker
127 Name shrinker used to keep components within the HDS name limit.
128 """
129 if not archive_path:
130 return "/MORE/LSST/JSON"
131 components = archive_path_to_hdf5_path_components(archive_path, shrinker)
132 return "/MORE/LSST/" + "/".join(components)
135def archive_path_to_hdf5_path_components(archive_path: str, shrinker: HdsNameShrinker) -> list[str]:
136 """Return HDS-compatible path components for an archive path.
138 Each component is uppercased; components longer than the HDS object-name
139 limit (``DAT__SZNAM``) are shrunk by ``shrinker``.
141 Parameters
142 ----------
143 archive_path
144 Serialization archive path to split into components.
145 shrinker
146 Name shrinker used to keep components within the HDS name limit.
147 """
148 return [shrinker.shrink(component) for component in archive_path.strip("/").split("/") if component]