LSST Applications g04e9c324dd+8c5ae1fdc5,g134cb467dc+1b3060144d,g18429d2f64+f642bf4753,g199a45376c+0ba108daf9,g1fd858c14a+2dcf163641,g262e1987ae+7b8c96d2ca,g29ae962dfc+3bd6ecb08a,g2cef7863aa+aef1011c0b,g35bb328faa+8c5ae1fdc5,g3fd5ace14f+53e1a9e7c5,g4595892280+fef73a337f,g47891489e3+2efcf17695,g4d44eb3520+642b70b07e,g53246c7159+8c5ae1fdc5,g67b6fd64d1+2efcf17695,g67fd3c3899+b70e05ef52,g74acd417e5+317eb4c7d4,g786e29fd12+668abc6043,g87389fa792+8856018cbb,g89139ef638+2efcf17695,g8d7436a09f+3be3c13596,g8ea07a8fe4+9f5ccc88ac,g90f42f885a+a4e7b16d9b,g97be763408+ad77d7208f,g9dd6db0277+b70e05ef52,ga681d05dcb+a3f46e7fff,gabf8522325+735880ea63,gac2eed3f23+2efcf17695,gb89ab40317+2efcf17695,gbf99507273+8c5ae1fdc5,gd8ff7fe66e+b70e05ef52,gdab6d2f7ff+317eb4c7d4,gdc713202bf+b70e05ef52,gdfd2d52018+b10e285e0f,ge365c994fd+310e8507c4,ge410e46f29+2efcf17695,geaed405ab2+562b3308c0,gffca2db377+8c5ae1fdc5,w.2025.35
LSST Data Management Base Package
Loading...
Searching...
No Matches
io.py
Go to the documentation of this file.
1from __future__ import annotations
2
3import json
4import logging
5from abc import ABC, abstractmethod
6from dataclasses import dataclass
7from typing import Any, Callable, ClassVar
8
9import numpy as np
10from numpy.typing import DTypeLike
11
12from .bbox import Box
13from .blend import Blend
14from .component import Component, FactorizedComponent
15from .image import Image
16from .observation import Observation
17from .parameters import FixedParameter
18from .source import Source
19
20__all__ = [
21 "ScarletComponentData",
22 "ScarletFactorizedComponentData",
23 "ScarletSourceData",
24 "ScarletBlendData",
25 "ScarletModelData",
26 "ComponentCube",
27]
28
29logger = logging.getLogger(__name__)
30
31
32def _numpy_to_json(arr: np.ndarray) -> dict[str, Any]:
33 """
34 Encode a numpy array as JSON-serializable dictionary.
35
36 Parameters
37 ----------
38 arr :
39 The numpy array to encode
40
41 Returns
42 -------
43 result :
44 A JSON formatted dictionary containing the dtype, shape,
45 and data of the array.
46 """
47 # Convert to native Python types for JSON serialization
48 flattened = arr.flatten()
49
50 # Convert numpy scalars to native Python types
51 if np.issubdtype(arr.dtype, np.integer):
52 data: list = [int(x) for x in flattened]
53 elif np.issubdtype(arr.dtype, np.floating):
54 data = [float(x) for x in flattened]
55 elif np.issubdtype(arr.dtype, np.complexfloating):
56 data = [complex(x) for x in flattened]
57 elif np.issubdtype(arr.dtype, np.bool_):
58 data = [bool(x) for x in flattened]
59 else:
60 # For other types (strings, objects, etc.), convert to string
61 data = [str(x) for x in flattened]
62
63 return {"dtype": str(arr.dtype), "shape": tuple(arr.shape), "data": data}
64
65
66def _json_to_numpy(encoded_dict: dict[str, Any]) -> np.ndarray:
67 """
68 Decode a JSON dictionary back to a numpy array.
69
70 Parameters
71 ----------
72 encoded_dict :
73 Dictionary with 'dtype', 'shape', and 'data' keys.
74
75 Returns
76 -------
77 result :
78 The reconstructed numpy array.
79 """
80 if "dtype" not in encoded_dict or "shape" not in encoded_dict or "data" not in encoded_dict:
81 raise ValueError("Encoded dictionary must contain 'dtype', 'shape', and 'data' keys.")
82 return np.array(encoded_dict["data"], dtype=encoded_dict["dtype"]).reshape(encoded_dict["shape"])
83
84
85def _encode_metadata(metadata: dict[str, Any] | None) -> dict[str, Any] | None:
86 """Pack metadata into a JSON compatible format.
87
88 Parameters
89 ----------
90 metadata :
91 The metadata to be packed.
92
93 Returns
94 -------
95 result :
96 The packed metadata.
97 """
98 if metadata is None:
99 return None
100 encoded = {}
101 array_keys = []
102 for key, value in metadata.items():
103 if isinstance(value, np.ndarray):
104 _encoded = _numpy_to_json(value)
105 encoded[key] = _encoded["data"]
106 encoded[f"{key}_shape"] = _encoded["shape"]
107 encoded[f"{key}_dtype"] = _encoded["dtype"]
108 array_keys.append(key)
109 else:
110 encoded[key] = value
111 if len(array_keys) > 0:
112 encoded["array_keys"] = array_keys
113 return encoded
114
115
116def _decode_metadata(metadata: dict[str, Any] | None) -> dict[str, Any] | None:
117 """Unpack metadata from a JSON compatible format.
118
119 Parameters
120 ----------
121 metadata :
122 The metadata to be unpacked.
123
124 Returns
125 -------
126 result :
127 The unpacked metadata.
128 """
129 if metadata is None:
130 return None
131 if "array_keys" in metadata:
132 for key in metadata["array_keys"]:
133 # Default dtype is float32 to support legacy models
134 dtype = metadata.pop(f"{key}_dtype", "float32")
135 shape = metadata.pop(f"{key}_shape", None)
136 if shape is None and f"{key}Shape" in metadata:
137 # Support legacy models that use `keyShape`
138 shape = metadata[f"{key}Shape"]
139 decoded = _json_to_numpy({"dtype": dtype, "shape": shape, "data": metadata[key]})
140 metadata[key] = decoded
141 # Remove the array keys after decoding
142 del metadata["array_keys"]
143 return metadata
144
145
147 data: Any,
148 metadata: dict[str, Any] | None,
149 key: str,
150) -> Any:
151 """Extract relevant information from the metadata.
152
153 Parameters
154 ----------
155 data :
156 The data to extract information from.
157 metadata :
158 The metadata to extract information from.
159 key :
160 The key to extract from the metadata.
161
162 Returns
163 -------
164 result :
165 A tuple containing the extracted data and metadata.
166 """
167 if data is not None:
168 return data
169 if metadata is None:
170 raise ValueError("Both data and metadata cannot be None")
171 if key not in metadata:
172 raise ValueError(f"'{key}' not found in metadata")
173 return metadata[key]
174
175
176@dataclass(kw_only=True)
178 """Base data for a scarlet component"""
179
180 component_registry: ClassVar[dict[str, type[ScarletComponentBaseData]]] = {}
181 component_type: str
182
183 @classmethod
184 def register(cls) -> None:
185 """Register a new component type"""
186 ScarletComponentBaseData.component_registry[cls.component_type] = cls
187
188 @abstractmethod
189 def to_component(self, observation: Observation) -> Component:
190 """Convert the storage data model into a scarlet Component
191
192 Parameters
193 ----------
194 observation :
195 The observation that the component is associated with
196
197 Returns
198 -------
199 component :
200 A scarlet component extracted from persisted data.
201 """
202
203 @abstractmethod
204 def as_dict(self) -> dict:
205 """Return the object encoded into a dict for JSON serialization
206
207 Returns
208 -------
209 result :
210 The object encoded as a JSON compatible dict
211 """
212
213 @staticmethod
214 def from_dict(data: dict, dtype: DTypeLike | None = None) -> ScarletComponentBaseData:
215 """Reconstruct `ScarletComponentBaseData` from JSON compatible
216 dict.
217
218 Parameters
219 ----------
220 data :
221 Dictionary representation of the object
222 dtype :
223 Datatype of the resulting model.
224
225 Returns
226 -------
227 result :
228 The reconstructed object
229 """
230 component_type = data["component_type"]
231 cls = ScarletComponentBaseData.component_registry[component_type]
232 return cls.from_dict(data, dtype=dtype)
233
234 @staticmethod
235 def from_component(component: Component) -> ScarletComponentBaseData:
236 """Reconstruct `ScarletComponentBaseData` from a scarlet Component.
237
238 Parameters
239 ----------
240 component :
241 The scarlet component to be converted.
242
243 Returns
244 -------
245 result :
246 The reconstructed object
247 """
248 if isinstance(component, FactorizedComponent):
249 return ScarletFactorizedComponentData._from_component(component)
250 else:
251 return ScarletComponentData._from_component(component)
252
253
254@dataclass(kw_only=True)
256 """Data for a component expressed as a 3D data cube
257
258 This is used for scarlet component models that are not factorized,
259 storing their entire model as a 3D data cube (bands, y, x).
260
261 Attributes
262 ----------
263 origin :
264 The lower bound of the components bounding box.
265 peak :
266 The peak of the component.
267 model :
268 The model for the component.
269 """
270
271 origin: tuple[int, int]
272 peak: tuple[float, float]
273 model: np.ndarray
274 component_type: str = "component"
275
276 @property
277 def shape(self):
278 return self.model.shape[-2:]
279
280 def to_component(self, observation: Observation) -> ComponentCube:
281 """Convert the storage data model into a scarlet Component
282
283 Parameters
284 ----------
285 observation :
286 The observation that the component is associated with
287
288 Returns
289 -------
290 component :
291 A scarlet component extracted from persisted data.
292 """
293 bbox = Box(self.shape, origin=self.origin)
294 model = self.model
295 if self.peak is None:
296 peak = None
297 else:
298 peak = (int(np.round(self.peak[0])), int(np.round(self.peak[0])))
299 assert peak is not None
300 component = ComponentCube(
301 bands=observation.bands,
302 bbox=bbox,
303 model=Image(model, yx0=bbox.origin, bands=observation.bands), # type: ignore
304 peak=peak,
305 )
306 return component
307
308 def as_dict(self) -> dict:
309 """Return the object encoded into a dict for JSON serialization
310
311 Returns
312 -------
313 result :
314 The object encoded as a JSON compatible dict
315 """
316 return {
317 "origin": self.origin,
318 "shape": self.model.shape,
319 "peak": self.peak,
320 "model": tuple(self.model.flatten().astype(float)),
321 "component_type": "component",
322 }
323
324 @classmethod
325 def from_dict(cls, data: dict, dtype: DTypeLike | None = None) -> ScarletComponentData:
326 """Reconstruct `ScarletComponentData` from JSON compatible dict
327
328 Parameters
329 ----------
330 data :
331 Dictionary representation of the object
332 dtype :
333 Datatype of the resulting model.
334
335 Returns
336 -------
337 result :
338 The reconstructed object
339 """
340 if data["component_type"] != "component":
341 raise ValueError(f"Invalid component type: {data['component_type']}")
342 shape = tuple(data["shape"])
343
344 return cls(
345 origin=tuple(data["origin"]), # type: ignore
346 peak=data["peak"],
347 model=np.array(data["model"]).reshape(shape).astype(dtype),
348 )
349
350 @staticmethod
351 def _from_component(component: Component) -> ScarletComponentData:
352 """Reconstruct `ScarletComponentData` from a scarlet Component.
353
354 Parameters
355 ----------
356 component :
357 The scarlet component to be converted.
358
359 Returns
360 -------
361 result :
362 The reconstructed object
363 """
365 origin=component.bbox.origin, # type: ignore
366 peak=component.peak, # type: ignore
367 model=component.get_model().data,
368 )
369
370
371@dataclass(kw_only=True)
373 """Data for a factorized component
374
375 Attributes
376 ----------
377 origin :
378 The lower bound of the component's bounding box.
379 peak :
380 The ``(y, x)`` peak of the component.
381 spectrum :
382 The SED of the component.
383 morph :
384 The 2D morphology of the component.
385 """
386
387 component_type: str = "factorized"
388 origin: tuple[int, int]
389 peak: tuple[float, float]
390 spectrum: np.ndarray
391 morph: np.ndarray
392
393 @property
394 def shape(self):
395 return self.morph.shape
396
397 def to_component(self, observation: Observation) -> FactorizedComponent:
398 """Convert the storage data model into a scarlet FactorizedComponent
399
400 Parameters
401 ----------
402 observation :
403 The observation that the component is associated with
404
405 Returns
406 -------
407 factorized_component :
408 A scarlet factorized component extracted from persisted data.
409 """
410 bbox = Box(self.shape, origin=self.origin)
411 spectrum = self.spectrum
412 morph = self.morph
413 if self.peak is None:
414 peak = None
415 else:
416 peak = (int(np.round(self.peak[0])), int(np.round(self.peak[1])))
417 assert peak is not None
418 # Note: since we aren't fitting a model, we don't need to
419 # set the RMS of the background.
420 # We set it to NaN just to be safe.
421 component = FactorizedComponent(
422 bands=observation.bands,
423 spectrum=FixedParameter(spectrum),
424 morph=FixedParameter(morph),
425 peak=peak,
426 bbox=bbox,
427 bg_rms=np.full((len(observation.bands),), np.nan),
428 )
429 return component
430
431 def as_dict(self) -> dict:
432 """Return the object encoded into a dict for JSON serialization
433
434 Returns
435 -------
436 result :
437 The object encoded as a JSON compatible dict
438 """
439 return {
440 "origin": tuple(int(o) for o in self.origin),
441 "shape": tuple(int(s) for s in self.morph.shape),
442 "peak": tuple(int(p) for p in self.peak),
443 "spectrum": tuple(self.spectrum.astype(float)),
444 "morph": tuple(self.morph.flatten().astype(float)),
445 "component_type": "factorized",
446 }
447
448 @classmethod
449 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletFactorizedComponentData:
450 """Reconstruct `ScarletFactorizedComponentData` from JSON compatible
451 dict.
452
453 Parameters
454 ----------
455 data :
456 Dictionary representation of the object
457 dtype :
458 Datatype of the resulting model.
459
460 Returns
461 -------
462 result :
463 The reconstructed object
464 """
465 shape = tuple(data["shape"])
466
467 return cls(
468 origin=tuple(data["origin"]), # type: ignore
469 peak=data["peak"],
470 spectrum=np.array(data["spectrum"]).astype(dtype),
471 morph=np.array(data["morph"]).reshape(shape).astype(dtype),
472 )
473
474 @staticmethod
475 def _from_component(component: FactorizedComponent) -> ScarletFactorizedComponentData:
476 """Reconstruct `ScarletFactorizedComponentData` from a scarlet
477 FactorizedComponent.
478
479 Parameters
480 ----------
481 component :
482 The scarlet component to be converted.
483
484 Returns
485 -------
486 result :
487 The reconstructed object
488 """
490 origin=component.bbox.origin, # type: ignore
491 peak=component.peak, # type: ignore
492 spectrum=component.spectrum,
493 morph=component.morph,
494 )
495
496
497# Register the component types
498ScarletComponentData.register()
499ScarletFactorizedComponentData.register()
500
501
502@dataclass(kw_only=True)
504 """Data for a scarlet source
505
506 Attributes
507 ----------
508 components :
509 The components contained in the source that are not factorized.
510 factorized_components :
511 The components contained in the source that are factorized.
512 peak_id :
513 The peak ID of the source in it's parent's footprint peak catalog.
514 """
515
516 components: list[ScarletComponentBaseData]
517 peak_id: int
518
519 def as_dict(self) -> dict:
520 """Return the object encoded into a dict for JSON serialization
521
522 Returns
523 -------
524 result :
525 The object encoded as a JSON compatible dict
526 """
527 result = {
528 "components": [component.as_dict() for component in self.components],
529 "peak_id": self.peak_id,
530 }
531 return result
532
533 @classmethod
534 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletSourceData:
535 """Reconstruct `ScarletSourceData` from JSON compatible
536 dict.
537
538 Parameters
539 ----------
540 data :
541 Dictionary representation of the object
542 dtype :
543 Datatype of the resulting model.
544
545 Returns
546 -------
547 result :
548 The reconstructed object
549 """
550 # Check for legacy models
551 if "factorized" in data:
552 components: list[ScarletComponentBaseData] = [
553 ScarletFactorizedComponentData.from_dict(component, dtype=dtype)
554 for component in data["factorized"]
555 ]
556 else:
557 components = [
558 ScarletComponentBaseData.from_dict(component, dtype=dtype) for component in data["components"]
559 ]
560 return cls(components=components, peak_id=int(data["peak_id"]))
561
562 @classmethod
563 def from_source(cls, source: Source) -> ScarletSourceData:
564 """Reconstruct `ScarletSourceData` from a scarlet Source.
565
566 Parameters
567 ----------
568 source :
569 The scarlet source to be converted.
570
571 Returns
572 -------
573 result :
574 The reconstructed object
575 """
576 components = [ScarletComponentBaseData.from_component(component) for component in source.components]
577 return cls(components=components, peak_id=source.peak_id) # type: ignore
578
579
580@dataclass(kw_only=True)
582 """Data for an entire blend.
583
584 Attributes
585 ----------
586 origin :
587 The lower bound of the blend's bounding box.
588 shape :
589 The shape of the blend's bounding box.
590 sources :
591 Data for the sources contained in the blend,
592 indexed by the source id.
593 psf :
594 The PSF of the observation.
595 """
596
597 blend_type: str = "blend"
598 origin: tuple[int, int]
599 shape: tuple[int, int]
600 sources: dict[int, ScarletSourceData]
601 metadata: dict[str, Any] | None = None
602
603 def as_dict(self) -> dict:
604 """Return the object encoded into a dict for JSON serialization
605
606 Returns
607 -------
608 result :
609 The object encoded as a JSON compatible dict
610 """
611 result: dict[str, Any] = {
612 "blend_type": self.blend_type,
613 "origin": self.origin,
614 "shape": self.shape,
615 "sources": {bid: source.as_dict() for bid, source in self.sources.items()},
616 }
617 if self.metadata is not None:
618 result["metadata"] = _encode_metadata(self.metadata)
619 return result
620
621 @classmethod
622 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletBlendData:
623 """Reconstruct `ScarletBlendData` from JSON compatible
624 dict.
625
626 Parameters
627 ----------
628 data :
629 Dictionary representation of the object
630 dtype :
631 Datatype of the resulting model.
632
633 Returns
634 -------
635 result :
636 The reconstructed object
637 """
638 if "metadata" not in data and "psf" in data:
639 # Support legacy models before metadata was used
640 metadata: dict[str, Any] | None = {
641 "psf": data["psf"],
642 "psf_shape": data["psf_shape"],
643 "bands": tuple(data["bands"]),
644 "array_keys": ["psf"],
645 }
646 else:
647 metadata = data.get("metadata", None)
648
649 return cls(
650 origin=tuple(data["origin"]), # type: ignore
651 shape=tuple(data["shape"]), # type: ignore
652 sources={
653 int(bid): ScarletSourceData.from_dict(source, dtype=dtype)
654 for bid, source in data["sources"].items()
655 },
656 metadata=_decode_metadata(metadata),
657 )
658
660 self,
661 model_psf: np.ndarray | None = None,
662 psf: np.ndarray | None = None,
663 bands: tuple[str] | None = None,
664 dtype: DTypeLike = np.float32,
665 ) -> Blend:
666 """Convert the storage data model into a scarlet lite blend
667
668 Parameters
669 ----------
670 model_psf :
671 PSF in model space (usually a nyquist sampled circular Gaussian).
672 psf :
673 The PSF of the observation.
674 If not provided, the PSF stored in the blend data is used.
675 bands :
676 The bands in the blend model.
677 If not provided, the bands stored in the blend data are used.
678 dtype :
679 The data type of the model that is generated.
680
681 Returns
682 -------
683 blend :
684 A scarlet blend model extracted from persisted data.
685 """
686 _model_psf: np.ndarray = _extract_from_metadata(model_psf, self.metadata, "model_psf")
687 _psf: np.ndarray = _extract_from_metadata(psf, self.metadata, "psf")
688 _bands: tuple[str] = _extract_from_metadata(bands, self.metadata, "bands")
689 model_box = Box(self.shape, origin=self.origin)
690 observation = Observation.empty(
691 bands=_bands,
692 psfs=_psf,
693 model_psf=_model_psf,
694 bbox=model_box,
695 dtype=dtype,
696 )
697 return self.to_blend(observation)
698
699 def to_blend(self, observation: Observation) -> Blend:
700 """Convert the storage data model into a scarlet lite blend
701
702 Parameters
703 ----------
704 observation :
705 The observation that contains the blend.
706 If `observation` is ``None`` then an `Observation` containing
707 no image data is initialized.
708
709 Returns
710 -------
711 blend :
712 A scarlet blend model extracted from persisted data.
713 """
714 sources = []
715 for source_id, source_data in self.sources.items():
716 components: list[Component] = [
717 component.to_component(observation) for component in source_data.components
718 ]
719
720 source = Source(components=components)
721 # Store identifiers for the source
722 source.record_id = source_id # type: ignore
723 source.peak_id = source_data.peak_id # type: ignore
724 sources.append(source)
725
726 return Blend(sources=sources, observation=observation, metadata=self.metadata)
727
728 @staticmethod
730 blend: Blend,
731 ) -> ScarletBlendData:
732 """Convert a scarlet lite blend into a persistable data object
733
734 Parameters
735 ----------
736 blend :
737 The blend that is being persisted.
738
739 Returns
740 -------
741 blend_data :
742 The data model for a single blend.
743 """
744 sources = {}
745 for source in blend.sources:
746 sources[source.record_id] = ScarletSourceData.from_source(source) # type: ignore
747
748 blend_data = ScarletBlendData(
749 origin=blend.bbox.origin, # type: ignore
750 shape=blend.bbox.shape, # type: ignore
751 sources=sources,
752 metadata=blend.metadata,
753 )
754
755 return blend_data
756
757
758@dataclass(kw_only=True)
760 """Data for a hierarchical blend.
761
762 Attributes
763 ----------
764 children :
765 Map from blend IDs to
766 """
767
768 blend_type: str = "hierarchical_blend"
769 children: dict[int, ScarletBlendData | HierarchicalBlendData]
770 metadata: dict[str, Any] | None = None
771
772 def as_dict(self) -> dict:
773 """Return the object encoded into a dict for JSON serialization
774
775 Returns
776 -------
777 result :
778 The object encoded as a JSON compatible dict
779 """
780 result: dict[str, Any] = {
781 "blend_type": self.blend_type,
782 "children": {bid: child.as_dict() for bid, child in self.children.items()},
783 }
784 if self.metadata is not None:
785 result["metadata"] = _encode_metadata(self.metadata)
786 return result
787
788 @classmethod
789 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> HierarchicalBlendData:
790 """Reconstruct `HierarchicalBlendData` from JSON compatible dict.
791
792 Parameters
793 ----------
794 data :
795 Dictionary representation of the object
796 dtype :
797 Datatype of the resulting model.
798
799 Returns
800 -------
801 result :
802 The reconstructed object
803 """
804 children: dict[int, ScarletBlendData | HierarchicalBlendData] = {}
805 for blend_id, child in data["children"].items():
806 if child["blend_type"] == "hierarchical_blend":
807 children[int(blend_id)] = HierarchicalBlendData.from_dict(child, dtype=dtype)
808 elif child["blend_type"] == "blend":
809 children[int(blend_id)] = ScarletBlendData.from_dict(child, dtype=dtype)
810 else:
811 raise ValueError(f"Unknown blend type: {child['blend_type']} for blend ID: {blend_id}")
812
813 metadata = _decode_metadata(data.get("metadata", None))
814 return cls(children=children, metadata=metadata)
815
816
818 """A container that propagates scarlet models for an entire catalog.
819
820 Attributes
821 ----------
822 blends :
823 Map from parent IDs in the source catalog
824 to scarlet model data for each parent ID (blend).
825 metadata :
826 Metadata associated with the model,
827 for example the order of bands.
828 """
829
830 blends: dict[int, ScarletBlendData | HierarchicalBlendData]
831 metadata: dict[str, Any] | None
832
834 self,
835 blends: dict[int, ScarletBlendData | HierarchicalBlendData] | None = None,
836 metadata: dict[str, Any] | None = None,
837 ):
838 """Initialize an instance"""
839 self.metadata = metadata
840 if blends is None:
841 blends = {}
842 self.blends = blends
843
844 def as_dict(self) -> dict:
845 """Return the object encoded into a dict for JSON serialization
846
847 Returns
848 -------
849 result :
850 The object encoded as a JSON compatible dict
851 """
852 result = {
853 "blends": {bid: blend.as_dict() for bid, blend in self.blends.items()},
854 "metadata": _encode_metadata(self.metadata),
855 }
856 return result
857
858 def json(self) -> str:
859 """Serialize the data model to a JSON formatted string
860
861 Returns
862 -------
863 result : `str`
864 The result of the object converted into a JSON format
865 """
866 result = self.as_dict()
867 return json.dumps(result)
868
869 @classmethod
870 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletModelData:
871 """Reconstruct `ScarletModelData` from JSON compatible dict.
872
873 Parameters
874 ----------
875 data :
876 Dictionary representation of the object
877 dtype :
878 Datatype of the resulting model.
879
880 Returns
881 -------
882 result :
883 The reconstructed object
884 """
885 if "psfShape" in data:
886 # Support legacy models before metadata was used
887 metadata: dict[str, Any] | None = {
888 "model_psf": data["psf"],
889 "model_psf_shape": data["psfShape"],
890 "array_keys": ["model_psf"],
891 }
892 else:
893 metadata = data.get("metadata", None)
894
895 blends: dict[int, ScarletBlendData | HierarchicalBlendData] | None = {}
896 for bid, blend in data.get("blends", {}).items():
897 if "blend_type" not in blend:
898 # Assume that this is a legacy model
899 blend["blend_type"] = "blend"
900 blend_data: ScarletBlendData | HierarchicalBlendData
901 if blend["blend_type"] == "hierarchical_blend":
902 blend_data = HierarchicalBlendData.from_dict(blend, dtype=dtype)
903 elif blend["blend_type"] == "blend":
904 blend_data = ScarletBlendData.from_dict(blend, dtype=dtype)
905 else:
906 raise ValueError(f"Unknown blend type: {blend['blend_type']} for blend ID: {bid}")
907 blends[int(bid)] = blend_data # type: ignore
908
909 return cls(
910 blends=blends,
911 metadata=_decode_metadata(metadata),
912 )
913
914 @classmethod
915 def parse_obj(cls, data: dict) -> ScarletModelData:
916 """Construct a ScarletModelData from python decoded JSON object.
917
918 Parameters
919 ----------
920 data :
921 The result of json.load(s) on a JSON persisted ScarletModelData
922
923 Returns
924 -------
925 result :
926 The `ScarletModelData` that was loaded the from the input object
927 """
928 return cls.from_dict(data, dtype=np.float32)
929
930
932 """Dummy component for a component cube.
933
934 This is duck-typed to a `lsst.scarlet.lite.Component` in order to
935 generate a model from the component.
936
937 If scarlet lite ever implements a component as a data cube,
938 this class can be removed.
939 """
940
941 def __init__(self, bands: tuple[Any, ...], bbox: Box, model: Image, peak: tuple[int, int]):
942 """Initialization
943
944 Parameters
945 ----------
946 bands :
947 model :
948 The 3D (bands, y, x) model of the component.
949 peak :
950 The `(y, x)` peak of the component.
951 bbox :
952 The bounding box of the component.
953 """
954 super().__init__(bands, bbox)
955 self._model = model
956 self.peak = peak
957
958 def get_model(self) -> Image:
959 """Generate the model for the source
960
961 Returns
962 -------
963 model :
964 The model as a 3D `(band, y, x)` array.
965 """
966 return self._model
967
968 def resize(self, model_box: Box) -> bool:
969 """Test whether or not the component needs to be resized"""
970 return False
971
972 def update(self, it: int, input_grad: np.ndarray) -> None:
973 """Implementation of unused abstract method"""
974
975 def parameterize(self, parameterization: Callable) -> None:
976 """Implementation of unused abstract method"""
A class to represent a 2-dimensional array of pixels.
Definition Image.h:51
__init__(self, tuple[Any,...] bands, Box bbox, Image model, tuple[int, int] peak)
Definition io.py:941
None parameterize(self, Callable parameterization)
Definition io.py:975
bool resize(self, Box model_box)
Definition io.py:968
None update(self, int it, np.ndarray input_grad)
Definition io.py:972
HierarchicalBlendData from_dict(cls, dict data, DTypeLike dtype=np.float32)
Definition io.py:789
Blend to_blend(self, Observation observation)
Definition io.py:699
Blend minimal_data_to_blend(self, np.ndarray|None model_psf=None, np.ndarray|None psf=None, tuple[str]|None bands=None, DTypeLike dtype=np.float32)
Definition io.py:665
ScarletBlendData from_dict(cls, dict data, DTypeLike dtype=np.float32)
Definition io.py:622
ScarletBlendData from_blend(Blend blend)
Definition io.py:731
ScarletComponentBaseData from_component(Component component)
Definition io.py:235
Component to_component(self, Observation observation)
Definition io.py:189
ScarletComponentBaseData from_dict(dict data, DTypeLike|None dtype=None)
Definition io.py:214
ScarletComponentData from_dict(cls, dict data, DTypeLike|None dtype=None)
Definition io.py:325
ComponentCube to_component(self, Observation observation)
Definition io.py:280
ScarletComponentData _from_component(Component component)
Definition io.py:351
FactorizedComponent to_component(self, Observation observation)
Definition io.py:397
ScarletFactorizedComponentData _from_component(FactorizedComponent component)
Definition io.py:475
ScarletFactorizedComponentData from_dict(cls, dict data, DTypeLike dtype=np.float32)
Definition io.py:449
ScarletModelData from_dict(cls, dict data, DTypeLike dtype=np.float32)
Definition io.py:870
ScarletModelData parse_obj(cls, dict data)
Definition io.py:915
__init__(self, dict[int, ScarletBlendData|HierarchicalBlendData]|None blends=None, dict[str, Any]|None metadata=None)
Definition io.py:837
ScarletSourceData from_source(cls, Source source)
Definition io.py:563
ScarletSourceData from_dict(cls, dict data, DTypeLike dtype=np.float32)
Definition io.py:534
dict[str, Any]|None _encode_metadata(dict[str, Any]|None metadata)
Definition io.py:85
dict[str, Any] _numpy_to_json(np.ndarray arr)
Definition io.py:32
np.ndarray _json_to_numpy(dict[str, Any] encoded_dict)
Definition io.py:66
Any _extract_from_metadata(Any data, dict[str, Any]|None metadata, str key)
Definition io.py:150
dict[str, Any]|None _decode_metadata(dict[str, Any]|None metadata)
Definition io.py:116