source.py

Go to the documentation of this file.

from __future__ import annotations
 
import logging
from dataclasses import dataclass
from typing import Any
 
import numpy as np
from deprecated.sphinx import deprecated  # type: ignore
from numpy.typing import DTypeLike
 
from ..component import Component
from ..observation import Observation
from ..source import Source
from .component import ScarletComponentBaseData
from .migration import PRE_SCHEMA, MigrationRegistry, migration
from .source_base import ScarletSourceBaseData
from .utils import decode_metadata, encode_metadata
 
__all__ = ["ScarletSourceData"]
 
CURRENT_SCHEMA = "1.0.0"
SOURCE_TYPE = "source"
MigrationRegistry.set_current(SOURCE_TYPE, CURRENT_SCHEMA)
logger = logging.getLogger(__name__)
 
 
@dataclass(kw_only=True)
class ScarletSourceData(ScarletSourceBaseData):
    """Data for a scarlet source
 
    Attributes
    ----------
    components :
        The components contained in the source that are not factorized.
    metadata :
        Metadata associated with the source.
        If `metadata` contains the `id` key, it is used as the
        key for the source in a `Blend`'s dictionary of sources.
    source_type :
        The type of source being stored
    version :
        The schema version of the ScarletSourceData.
    """
 
    source_type: str = SOURCE_TYPE
    components: list[ScarletComponentBaseData]
    metadata: dict[str, Any] | None = None
    version: str = CURRENT_SCHEMA
 
    def as_dict(self) -> dict:
        """Return the object encoded into a dict for JSON serialization
 
        Returns
        -------
        result :
            The object encoded as a JSON compatible dict
        """
        result: dict[str, Any] = {
            "source_type": self.source_type,
            "components": [component.as_dict() for component in self.components],
            "version": self.version,
        }
        if self.metadata is not None:
            result["metadata"] = encode_metadata(self.metadata)
        return result
 
    @classmethod
    def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> ScarletSourceData:
        """Reconstruct `ScarletSourceData` from JSON compatible
        dict.
 
        Parameters
        ----------
        data :
            Dictionary representation of the object
        dtype :
            Datatype of the resulting model.
 
        Returns
        -------
        result :
            The reconstructed object
        """
        data = MigrationRegistry.migrate(SOURCE_TYPE, data)
        metadata = data.get("metadata", None)
        components = [
            ScarletComponentBaseData.from_dict(component, dtype=dtype) for component in data["components"]
        ]
        return cls(components=components, metadata=decode_metadata(metadata))
 
    def to_source(self, observation: Observation) -> Source:
        """Convert to a `Source` for use in scarlet
 
        Parameters
        ----------
        observation:
            The observation used to render the source.
 
        Returns
        -------
        source:
            The `Source` representation of this data.
        """
        components: list[Component] = [component.to_component(observation) for component in self.components]
        return Source(components=components, metadata=self.metadata)
 
    @staticmethod
    @deprecated(
        reason="from_source is deprecated and will be removed in a future release.",
        version="v30.0",
        category=FutureWarning,
    )
    def from_source(source: Source) -> ScarletSourceData:
        """Deprecated: Create a `ScarletSourceData` from a scarlet `Source`
 
        Parameters
        ----------
        source:
            The scarlet `Source` to convert.
 
        Returns
        -------
        result:
            The `ScarletSourceData` representation of the source.
        """
        return source.to_data()
 
 
ScarletSourceData.register()
 
 
@migration(SOURCE_TYPE, PRE_SCHEMA)
def _to_1_0_0(data: dict) -> dict:
    """Migrate a pre-schema source to schema version 1.0.0
 
    There were no changes to this data model in v1.0.0 but we need
    to provide a way to migrate pre-schema data.
 
    Parameters
    ----------
    data :
        The data to migrate.
    Returns
    -------
    result :
        The migrated data.
    """
    # Check for legacy models
    if "factorized" in data:
        data["data_type"] = "factorized"
        data["components"] = data["factorized"]
        del data["factorized"]
    data["version"] = "1.0.0"
    return data