Source code for eradiate.scenes.spectra._core

from __future__ import annotations

import typing as t
from abc import ABC, abstractmethod
from functools import singledispatchmethod

import attrs
import pint

from ..core import NodeSceneElement
from ..._factory import Factory
from ...attrs import documented, parse_docs
from ...spectral.ckd import BinSet
from ...spectral.index import (
from ...spectral.mono import WavelengthSet
from ...units import PhysicalQuantity

[docs] class SpectrumFactory(Factory):
[docs] def converter(self, quantity: str | PhysicalQuantity) -> t.Callable[[t.Any], t.Any]: """ Generate a converter wrapping :meth:`SpectrumFactory.convert` to handle defaults for shortened spectrum definitions. The produced converter processes a parameter ``value`` as follows: * if ``value`` is an int, a float or a :class:`pint.Quantity`, the converter calls itself using a dictionary ``{"type": "uniform", "quantity": quantity, "value": value}``; * if ``value`` is a dictionary, it adds a ``"quantity": quantity`` entry for the following values of the ``"type"`` entry: * ``"uniform"``; * ``"interpolated"``; * ``"multi_delta"``; * otherwise, it forwards ``value`` to :meth:`.SpectrumFactory.convert`. Parameters ---------- quantity : str or .PhysicalQuantity Quantity specifier (converted by :class:`.PhysicalQuantity`). See :meth:`.PhysicalQuantity.spectrum` for suitable values. Returns ------- callable Generated converter. """ def f(value): if isinstance(value, (int, float, pint.Quantity)): # Convert unitless values with dict wrapping and recursive call return f({"type": "uniform", "quantity": quantity, "value": value}) if isinstance(value, dict): # If generic spectrum is requested without a specified # "quantity" field, add pre-configured quantity and attempt # conversion using regular conversion protocol try: if ( value["type"] in {"uniform", "interpolated", "multi_delta"} and "quantity" not in value ): return self.convert({**value, "quantity": quantity}) except KeyError: # Note: A missing "type" field will also run this case, and # the ill-formed dict will be correctly reported upon # regular conversion pass if isinstance(value, Spectrum) and value.quantity is None: # If quantity is unspecified, apply the expected quantity return self.convert(attrs.evolve(value, quantity=quantity)) # Regular conversion (happens if value is neither int, float nor # dict without "quantity" field) return self.convert(value) return f
spectrum_factory = SpectrumFactory() spectrum_factory.register_lazy_batch( [ ( "_air_scattering_coefficient.AirScatteringCoefficientSpectrum", "air_scattering_coefficient", {}, ), ( "_interpolated.InterpolatedSpectrum", "interpolated", {}, ), ( "_solar_irradiance.SolarIrradianceSpectrum", "solar_irradiance", {}, ), ( "_uniform.UniformSpectrum", "uniform", {}, ), ( "_multi_delta.MultiDeltaSpectrum", "multi_delta", {}, ), ], cls_prefix="eradiate.scenes.spectra", )
[docs] @parse_docs @attrs.define(eq=False, slots=False) class Spectrum(NodeSceneElement, ABC): """ Spectrum interface. Notes ----- * This class is to be used as a mixin. * Subclasses must implement :meth:`eval_mono`, :meth:`eval_ckd` and :meth:`integral`. Warnings -------- Leaving the ``quantity`` field unset is discouraged and permitted only to allow for more convenience in some parts of the code (*e.g.* to increase the flexibility of spectrum conversion protocols). """ quantity: PhysicalQuantity | None = documented( attrs.field( default=None, converter=attrs.converters.optional(PhysicalQuantity), repr=lambda x: "None" if x is None else x.value.upper(), ), doc="Physical quantity which the spectrum represents. The specified " "quantity must be one which varies with wavelength. " "See :meth:`.PhysicalQuantity.spectrum` for allowed values.\n" "\n" "Child classes should implement value units validation and conversion " "based on ``quantity``.", type=".PhysicalQuantity or None", init_type=".PhysicalQuantity or str, optional", default="None", ) @quantity.validator def _quantity_validator(self, attribute, value): if value is None: return if value not in PhysicalQuantity.spectrum(): raise ValueError( f"while validating {}: " f"got value '{value}', expected one of " f"{str(PhysicalQuantity.spectrum())}" )
[docs] @singledispatchmethod def eval(self, si: SpectralIndex) -> pint.Quantity: """ Evaluate spectrum at a given spectral index. Parameters ---------- si : :class:`.SpectralIndex` Spectral index. Returns ------- value : quantity Evaluated spectrum. Notes ----- This method dispatches evaluation to specialized methods depending on the spectral index type. """ raise NotImplementedError
@eval.register(MonoSpectralIndex) def _(self, si) -> pint.Quantity: return self.eval_mono(w=si.w) @eval.register(CKDSpectralIndex) def _(self, si) -> pint.Quantity: return self.eval_ckd(w=si.w, g=si.g)
[docs] @abstractmethod def eval_mono(self, w: pint.Quantity) -> pint.Quantity: """ Evaluate spectrum in monochromatic modes. Parameters ---------- w : quantity Wavelength. Returns ------- value : quantity Evaluated spectrum as an array with the same shape as ``w``. """ raise NotImplementedError
[docs] @abstractmethod def eval_ckd(self, w: pint.Quantity, g: float) -> pint.Quantity: """ Evaluate spectrum in CKD modes. Parameters ---------- w : quantity Spectral bin center wavelength. g : float Absorption coefficient cumulative probability. Returns ------- value : quantity Evaluated spectrum as an array with shape ``w``. Notes ----- It is assumed that ``w`` and ``g`` have the same shape. In CKD mode, it is assumed that all spectra—except that of the absorption coefficient—are uniform over the spectral bin. These spectra are evaluated at the spectral bin center wavelength. """ pass
[docs] def integral(self, wmin: pint.Quantity, wmax: pint.Quantity) -> pint.Quantity: """ Compute the integral of the spectrum on a given interval. Parameters ---------- wmin : quantity Integration interval's lower bound. wmax : quantity Integration interval's upper bound. Returns ------- value : quantity or ndarray Computed integral value, in units consistent with the specified quantity. Warnings -------- If the ``quantity`` field is unset (*i.e.* left to its default value ``None``), the output of this method will inherit the units of value fields (the actual policy depends on the implementation, and unitless values are intepreted as dimnesionless). Note that leaving the ``quantity`` field unset is discouraged. """ raise NotImplementedError
[docs] @singledispatchmethod def select_in(self, spectral_set) -> BinSet | WavelengthSet: """ Select a subset of a spectral set. Parameters ---------- spectral_set : :class:`.BinSet` or :class:`.WavelengthSet` Spectral set. Returns ------- subset : :class:`.BinSet` or :class:`.WavelengthSet` Subset of the spectral set. Notes ----- This method is only relevant to subclasses used to represent spectral response function (the default implementation raises a :class:`NotImplementedError`). In this context, the spectral response function acts as a sort of filter that *selects* a subset of a spectral set, e.g. where the response is non-zero. This method is useful for :class:`.Experiment` objects to determine which spectral set is relevant for a given sensor. """ raise NotImplementedError
@select_in.register(WavelengthSet) def _(self, spectral_set) -> WavelengthSet: return self.select_in_wavelength_set(spectral_set) @select_in.register(BinSet) def _(self, spectral_set) -> BinSet: return self.select_in_bin_set(spectral_set) def select_in_wavelength_set(self, wset) -> WavelengthSet: raise NotImplementedError def select_in_bin_set(self, binset) -> BinSet: raise NotImplementedError