Coverage for python/lsst/meas/extensions/scarlet/io/hierarchical_blend_data.py: 94%
83 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 02:24 -0700
1# This file is part of meas_extensions_scarlet.
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# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22from __future__ import annotations
24from dataclasses import dataclass
25from typing import Any
27import numpy as np
28from numpy.typing import DTypeLike
30import lsst.scarlet.lite as scl
31from lsst.scarlet.lite import Box
33from .source_data import _decode_span_array, _encode_span_array
35__all__ = ["LsstHierarchicalBlendData"]
37CURRENT_SCHEMA = "1.0.0"
38BLEND_TYPE = "lsst_hierarchical"
39scl.io.migration.MigrationRegistry.set_current(BLEND_TYPE, CURRENT_SCHEMA)
41# `LsstHierarchicalBlendData` superceeds scarlet_lite's
42# `HierarchicalBlendData`. We keep track of the legacy types so that we can
43# migrate them to the new type when reading from disk.
44LEGACY_HIERARCHICAL_TYPES = ("hierarchical", "hierarchical_blend")
47@dataclass(kw_only=True)
48class LsstHierarchicalBlendData(scl.io.ScarletBlendBaseData):
49 """A meas-owned hierarchical blend that carries the detection footprint.
51 The LSST-pipeline replacement for scarlet_lite's
52 `~lsst.scarlet.lite.io.HierarchicalBlendData`, with attributes specific
53 to the LSST science pipelines needs.
55 Attributes
56 ----------
57 children
58 Map from blend IDs to child blends.
59 span_array
60 The detected-parent footprint mask (``True`` inside the footprint).
61 origin
62 The ``(y, x)`` origin of ``span_array`` in observation coordinates.
63 legacy_spans
64 ``True`` when ``span_array`` was reconstructed by a migration rather
65 than carried from the original detection. In that case the
66 spans are *not* the true detection footprint, but approximated
67 by the union of child footprints if possible. If not then the
68 spans are a filled rectangle over the child bounding boxes.
69 """
71 blend_type: str = BLEND_TYPE
72 version: str = CURRENT_SCHEMA
73 children: dict[int, scl.io.ScarletBlendBaseData]
74 span_array: np.ndarray
75 origin: tuple[int, int]
76 legacy_spans: bool = False
78 @property
79 def shape(self) -> tuple[int, int]:
80 """The ``(height, width)`` of the footprint span mask."""
81 return self.span_array.shape[0], self.span_array.shape[1]
83 @property
84 def bbox(self) -> Box:
85 """The bounding box of the detected-parent footprint."""
86 return Box(self.span_array.shape, origin=self.origin)
88 def as_dict(self) -> dict[str, Any]:
89 """Return the object encoded into a dict for JSON serialization.
91 Returns
92 -------
93 result : dict[str, Any]
94 The object encoded as a JSON-compatible dict.
95 """
96 result: dict[str, Any] = {
97 "blend_type": self.blend_type,
98 "children": {bid: child.as_dict() for bid, child in self.children.items()},
99 "span_array": _encode_span_array(self.span_array),
100 "shape": self.shape,
101 "origin": tuple(int(o) for o in self.origin),
102 "legacy_spans": bool(self.legacy_spans),
103 "version": self.version,
104 }
105 if self.metadata is not None:
106 result["metadata"] = scl.io.utils.encode_metadata(self.metadata)
107 return result
109 @classmethod
110 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> LsstHierarchicalBlendData:
111 """Reconstruct `LsstHierarchicalBlendData` from a JSON compatible dict.
113 Parameters
114 ----------
115 data : dict
116 Dictionary representation of the object.
117 dtype : DTypeLike
118 Datatype of the resulting model.
120 Returns
121 -------
122 result : LsstHierarchicalBlendData
123 The reconstructed object.
124 """
125 data = scl.io.migration.MigrationRegistry.migrate(BLEND_TYPE, data)
126 children: dict[int, scl.io.ScarletBlendBaseData] = {}
127 for blend_id, child in data["children"].items():
128 try:
129 children[int(blend_id)] = scl.io.ScarletBlendBaseData.from_dict(child, dtype=dtype)
130 except KeyError:
131 raise scl.io.utils.PersistenceError(
132 f"Unknown blend type: {child.get('blend_type')} for blend ID: {blend_id}"
133 )
134 shape = tuple(int(s) for s in data["shape"])
135 # Spans are a boolean mask, unaffected by the model ``dtype``.
136 span_array = _decode_span_array(data["span_array"], shape, bool)
137 origin = tuple(int(o) for o in data["origin"])
138 legacy_spans = bool(data.get("legacy_spans", False))
139 metadata = scl.io.utils.decode_metadata(data.get("metadata", None))
140 return cls(
141 children=children,
142 span_array=span_array,
143 origin=origin,
144 legacy_spans=legacy_spans,
145 metadata=metadata,
146 )
148 @classmethod
149 def convert_from_hierarchical(cls, blend_dict: dict) -> dict:
150 """Promote a legacy scarlet_lite ``hierarchical`` blend dict to the
151 ``lsst_hierarchical`` shape.
153 Reads the legacy on-disk format directly (not through scarlet_lite's
154 ``hierarchical`` migration chain, so it is unaffected by future
155 ``HierarchicalBlendData`` changes in scarlet_lite).
156 ``children`` dicts pass through untouched and migrate individually
157 in :meth:`from_dict`.
159 Real ``metadata["spans"]`` are promoted verbatim
160 (``legacy_spans=False``).
161 When absent, a filled rectangle over the children's bounding box is
162 synthesized with ``legacy_spans=True``.
164 Parameters
165 ----------
166 blend_dict : dict
167 A legacy scarlet_lite ``hierarchical`` blend dict.
169 Returns
170 -------
171 result : dict
172 An ``lsst_hierarchical`` blend dict ready for
173 :meth:`from_dict` dispatch.
174 """
175 children = blend_dict["children"]
176 metadata = scl.io.utils.decode_metadata(blend_dict.get("metadata", None))
178 if metadata is not None and "spans" in metadata:
179 span_array = np.asarray(metadata["spans"], dtype=bool)
180 origin = tuple(int(o) for o in metadata["origin"])
181 legacy_spans = False
182 residual = {k: v for k, v in metadata.items() if k not in ("spans", "origin")}
183 else:
184 origin, shape = cls._bbox_from_children(children)
185 span_array = np.ones(shape, dtype=bool)
186 legacy_spans = True
187 residual = dict(metadata) if metadata else {}
189 result: dict[str, Any] = {
190 "blend_type": BLEND_TYPE,
191 "version": CURRENT_SCHEMA,
192 "children": children,
193 "span_array": _encode_span_array(span_array),
194 "shape": tuple(int(s) for s in span_array.shape),
195 "origin": tuple(int(o) for o in origin),
196 "legacy_spans": legacy_spans,
197 }
198 if residual: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 result["metadata"] = scl.io.utils.encode_metadata(residual)
200 return result
202 @staticmethod
203 def _bbox_from_children(children: dict) -> tuple[tuple[int, int], tuple[int, int]]:
204 """Compute the ``(origin, shape)`` covering all child blend dicts.
206 This should only be used by the legacy-spans synthesis path,
207 which supports only the flat `~lsst.scarlet.lite.io.ScarletBlendData`
208 children the deblender produces.
210 Raises
211 ------
212 ValueError
213 If ``children`` is empty.
214 """
215 if not children: 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true
216 raise ValueError(
217 "Cannot synthesize legacy spans for a hierarchical blend with no children."
218 )
219 origins = []
220 shapes = []
221 for child in children.values():
222 if "origin" not in child or "shape" not in child:
223 raise NotImplementedError(
224 "Legacy span synthesis only supports flat child blends with "
225 f"'origin'/'shape'; got blend_type {child.get('blend_type')!r}."
226 )
227 origins.append(tuple(int(o) for o in child["origin"]))
228 shapes.append(tuple(int(s) for s in child["shape"]))
229 min_y = min(o[0] for o in origins)
230 min_x = min(o[1] for o in origins)
231 max_y = max(o[0] + s[0] for o, s in zip(origins, shapes))
232 max_x = max(o[1] + s[1] for o, s in zip(origins, shapes))
233 origin = (min_y, min_x)
234 shape = (max_y - min_y, max_x - min_x)
235 return origin, shape
238LsstHierarchicalBlendData.register()