Source code for spectral_indices.transformations.base

from __future__ import annotations

from abc import ABC
from dataclasses import asdict, dataclass, fields
from typing import Any, Dict, List

from xarray import DataArray




@dataclass
class Transformation(ABC):
    """Base class for transformations on xarrays."""

    @property
    def name(self) -> str:
        """Return class name."""
        return self.__class__.__name__



    def planify(self, array: DataArray) -> DataArray:
        """Lazy transformation on xarray. This function plan some modifications that are added to dask graph."""
        pass




    def to_dict(self) -> Dict[str, Any]:
        """Return attributes as dict."""
        return asdict(self)




    @classmethod
    def from_dict(cls, transformation_dict: Dict[str, Any]) -> Transformation:
        """Build transformation from attributes dict.

        Args:
            trasnformation_dict (``Dict[str, Any]``): Dict of trasnformation attributes.

        Returns:
            ``Transformation``:
                - Return an instance of transformation.
        """
        name = transformation_dict["name"]
        params = transformation_dict["params"]
        transformation = TransformationFactory.build(name, **params)
        return transformation


    def __repr__(self):
        params = [f"{param_name}={val}" for param_name, val in self.to_dict().items()]
        repr = f"{self.name}({', '.join(params)})"
        return repr



class TransformationFactory:
    """Factory to store & build dynamically transformations."""

    _transformations: List[Transformation] = {}

    @classmethod
    def register(cls, transformation_class: Transformation) -> Transformation:
        """Register a transformation into factory. This function can be used as a decorator.

        Args:
            transformation_class (``Transformation``): TRansformation class to register.

        Returns:
            ``Transformation``:
                - Return Transformation builder.
        """
        cls._transformations[transformation_class.__name__] = transformation_class
        return transformation_class

    @classmethod
    def build(cls, transformation_name: str, **kwargs) -> Transformation:
        """Build a transformation from factory.

        Args:
            transformation_name (``str``): Name of transformation to build.

        Returns:
            ``Transformation``:
                - Transformation.
        """
        transformation = cls._transformations[transformation_name]
        return transformation(**kwargs)