Coverage for python / lsst / scarlet / lite / io / blend.py: 57%
64 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:28 +0000
1from __future__ import annotations
3import logging
4from dataclasses import dataclass
5from functools import cached_property
6from typing import Any
8import numpy as np
9from deprecated.sphinx import deprecated # type: ignore
10from numpy.typing import DTypeLike
12from ..bbox import Box
13from ..blend import Blend
14from ..observation import Observation
15from .blend_base import ScarletBlendBaseData
16from .migration import PRE_SCHEMA, MigrationRegistry, migration
17from .source import ScarletSourceBaseData
18from .utils import decode_metadata, encode_metadata, extract_from_metadata
20__all__ = ["ScarletBlendData"]
22CURRENT_SCHEMA = "1.0.0"
23BLEND_TYPE = "blend"
24MigrationRegistry.set_current(BLEND_TYPE, CURRENT_SCHEMA)
25logger = logging.getLogger(__name__)
28@dataclass(kw_only=True)
29class ScarletBlendData(ScarletBlendBaseData):
30 """Data for an entire blend.
32 Attributes
33 ----------
34 blend_type :
35 The type of blend being stored.
36 metadata :
37 Metadata associated with the blend,
38 for example the order of bands, the PSF, etc.
39 origin :
40 The lower bound of the blend's bounding box.
41 shape :
42 The shape of the blend's bounding box.
43 sources :
44 Data for the sources contained in the blend,
45 indexed by the source id.
46 version :
47 The schema version of the stored data.
48 """
50 blend_type: str = BLEND_TYPE
51 origin: tuple[int, int]
52 shape: tuple[int, int]
53 sources: dict[Any, ScarletSourceBaseData]
54 version: str = CURRENT_SCHEMA
56 @cached_property
57 def bbox(self) -> Box:
58 """The bounding box of the blend"""
59 return Box(self.shape, origin=self.origin)
61 def as_dict(self) -> dict:
62 """Return the object encoded into a dict for JSON serialization
64 Returns
65 -------
66 result :
67 The object encoded as a JSON compatible dict
68 """
69 result: dict[str, Any] = {
70 "blend_type": self.blend_type,
71 "origin": self.origin,
72 "shape": self.shape,
73 "sources": {bid: source.as_dict() for bid, source in self.sources.items()},
74 "version": self.version,
75 }
76 if self.metadata is not None:
77 result["metadata"] = encode_metadata(self.metadata)
78 return result
80 @classmethod
81 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletBlendData:
82 """Reconstruct `ScarletBlendData` from JSON compatible
83 dict.
85 Parameters
86 ----------
87 data :
88 Dictionary representation of the object
89 dtype :
90 Datatype of the resulting model.
92 Returns
93 -------
94 result :
95 The reconstructed object
96 """
97 data = MigrationRegistry.migrate(BLEND_TYPE, data)
98 metadata = data.get("metadata", None)
100 return cls(
101 origin=tuple(data["origin"]), # type: ignore
102 shape=tuple(data["shape"]), # type: ignore
103 sources={
104 bid: ScarletSourceBaseData.from_dict(source, dtype=dtype)
105 for bid, source in data["sources"].items()
106 },
107 metadata=decode_metadata(metadata),
108 )
110 def minimal_data_to_blend(
111 self,
112 model_psf: np.ndarray | None = None,
113 psf: np.ndarray | None = None,
114 bands: tuple[str] | None = None,
115 dtype: DTypeLike = np.float32,
116 ) -> Blend:
117 """Convert the storage data model into a scarlet lite blend
119 Parameters
120 ----------
121 model_psf :
122 PSF in model space (usually a nyquist sampled circular Gaussian).
123 psf :
124 The PSF of the observation.
125 If not provided, the PSF stored in the blend data is used.
126 bands :
127 The bands in the blend model.
128 If not provided, the bands stored in the blend data are used.
129 dtype :
130 The data type of the model that is generated.
132 Returns
133 -------
134 blend :
135 A scarlet blend model extracted from persisted data.
136 """
138 _model_psf: np.ndarray = extract_from_metadata(model_psf, self.metadata, "model_psf")
139 _psf: np.ndarray = extract_from_metadata(psf, self.metadata, "psf")
140 _bands: tuple[str] = extract_from_metadata(bands, self.metadata, "bands")
141 model_box = self.bbox
142 observation = Observation.empty(
143 bands=_bands,
144 psfs=_psf,
145 model_psf=_model_psf,
146 bbox=model_box,
147 dtype=dtype,
148 )
149 return self.to_blend(observation)
151 def to_blend(self, observation: Observation) -> Blend:
152 """Convert the storage data model into a scarlet lite blend
154 Parameters
155 ----------
156 observation :
157 The observation that contains the blend.
158 If `observation` is ``None`` then an `Observation` containing
159 no image data is initialized.
161 Returns
162 -------
163 blend :
164 A scarlet blend model extracted from persisted data.
165 """
166 sources = []
167 for source_data in self.sources.values():
168 source = source_data.to_source(observation)
169 sources.append(source)
171 return Blend(sources=sources, observation=observation, metadata=self.metadata)
173 @staticmethod
174 @deprecated(
175 reason="ScarletBlendData.from_blend is deprecated. Use blend.to_data() instead.",
176 version="v30.0",
177 category=FutureWarning,
178 )
179 def from_blend(blend: Blend) -> ScarletBlendData:
180 """Deprecated: Convert a scarlet lite blend into a storage data model.
182 Parameters
183 ----------
184 blend :
185 The blend to convert.
186 Returns
187 -------
188 result :
189 The storage data model representing the blend.
190 """
191 return blend.to_data()
194ScarletBlendData.register()
197@migration(BLEND_TYPE, PRE_SCHEMA)
198def _to_1_0_0(data: dict) -> dict:
199 """Migrate a pre-schema blend to schema version 1.0.0
201 Parameters
202 ----------
203 data :
204 The data to migrate.
206 Returns
207 -------
208 result :
209 The migrated data.
210 """
211 # Support legacy models before metadata was used
212 if "metadata" not in data and "psf" in data:
213 data["metadata"] = {
214 "psf": data["psf"],
215 "psf_shape": data["psf_shape"],
216 "bands": tuple(data["bands"]),
217 "array" "_keys": ["psf"],
218 }
219 data["version"] = "1.0.0"
220 return data