Source code for eradiate.scenes.measure._core

from __future__ import annotations

import os
import typing as t
import warnings
from abc import ABC, abstractmethod

import attrs
import xarray as xr

import eradiate

from ..core import NodeSceneElement
from ..spectra import (
from ... import validators
from ..._factory import Factory
from ...attrs import documented, get_doc, parse_docs
from ...kernel import InitParameter
from ...srf_tools import convert as convert_srf
from ...units import PhysicalQuantity
from ...units import unit_registry as ureg

measure_factory = Factory()
            {"aliases": ["distantflux"]},
            {"aliases": ["hdistant"]},
            {"aliases": ["mdistant", "distant"]},
            {"aliases": ["mradiancemeter"]},

def _srf_converter(value: t.Any) -> Spectrum:
    Converter for :class:`.Measure` ``srf`` attribute.

    value : Any
        Value to convert.

        Converted value.

    The behaviour of this converter depends on the value type:
    * If ``value`` is not a string or path, it is passed to the
      :class:`.spectrum_factory`'s converter.
    * If ``value`` is a path, the converter tries to open the corresponding
        file on the hard drive; should that fail, it queries the Eradiate data
        store with that path.
    * If ``value`` is a string, it is interpreted as an SRF identifier:
      * If the identifier does not end with `-raw`, the converter looks for a
        prepared version of the SRF and loads it if it exists, else it loads the
        raw SRF.
      * If the identifier ends with `-raw`, the converter loads the raw SRF.
    if isinstance(value, (str, os.PathLike, xr.Dataset)):
        ds = convert_srf(value)
        w = ureg.Quantity(ds.w.values, ds.w.attrs["units"])
        srf = ds.data_vars["srf"].values
        return InterpolatedSpectrum(quantity="dimensionless", wavelengths=w, values=srf)
        converter = spectrum_factory.converter(quantity="dimensionless")
        converted = converter(value)
        if not isinstance(converted, (InterpolatedSpectrum, MultiDeltaSpectrum)):
            raise ValueError(
                f"SRF must be an InterpolatedSpectrum or MultiDeltaSpectrum, "
                f"got {converted}"
        return converted

def _str_summary_raw(x):
    if not x:
        return "{}"

    keys = list(x.keys())

    if len(keys) == 1:
        return f"dict<1 item>({{{keys[0]}: {{...}} }})"
        return f"dict<{len(keys)} items>({{{keys[0]}: {{...}} , ... }})"

[docs] @parse_docs @attrs.define(eq=False, slots=False) class Measure(NodeSceneElement, ABC): """ Abstract base class for all measure scene elements. See Also -------- :func:`.mitsuba_run` Notes ----- * This class is meant to be used as a mixin. * Raw results stored in the `results` field as nested dictionaries with the following structure: .. code:: python { spectral_key_0: dict_0, spectral_key_1: dict_1, ... } Keys are spectral loop indexes; values are nested dictionaries produced by :func:`.mitsuba_run`. """ # -------------------------------------------------------------------------- # Fields and properties # -------------------------------------------------------------------------- id: str | None = documented( attrs.field( default="measure", validator=attrs.validators.optional(attrs.validators.instance_of(str)), ), doc=get_doc(NodeSceneElement, "id", "doc"), type=get_doc(NodeSceneElement, "id", "type"), init_type=get_doc(NodeSceneElement, "id", "init_type"), default='"measure"', ) mi_results: dict = documented( attrs.field(factory=dict, repr=_str_summary_raw, init=False), doc="Storage for raw results yielded by the kernel.", type="dict", default="{}", ) srf: Spectrum = documented( attrs.field( factory=lambda: MultiDeltaSpectrum(wavelengths=550.0 * ureg.nm), converter=_srf_converter, validator=validators.has_quantity(PhysicalQuantity.DIMENSIONLESS), ), doc="Spectral response function (SRF). If a path is passed, it attempts " "to load a dataset from that location. If a keyword is passed, e.g., " "``'sentinel_2a-msi-4'`` it tries to serve the corresponding dataset " "from the Eradiate data store. By default, the *prepared* version of " "the SRF is served unless it does not exist in which case the *raw* " "version is served. To request that the raw version is served, append " "``'-raw'`` to the keyword, e.g., ``'sentinel_2a-msi-4-raw'``. " "Note that the prepared SRF provide a better speed versus accuracy " "trade-off, but for the best accuracy, the raw SRF should be used. " "Other types will be converted by :data:`.spectrum_factory`.", type=".Spectrum", init_type="Path or str or .Spectrum or dict", default=":class:`MultiDeltaSpectrum(wavelengths=550.0 * ureg.nm) " "<.MultiDeltaSpectrum>`", ) sampler: str = documented( attrs.field( default="independent", validator=attrs.validators.in_( {"independent", "stratified", "multijitter", "orthogonal", "ldsampler"} ), ), doc="Mitsuba sampler used to generate pseudo-random number sequences.", type="str", init_type='{"independent", "stratified", "multijitter", "orthogonal", ' '"ldsampler"}', default='"independent"', ) rfilter: str = documented( attrs.field( default="box", validator=attrs.validators.in_({"box", "gaussian"}), ), doc="Reconstruction filter used to scatter samples on sensor pixels. " "By default, using a box filter is recommended.", type="str", init_type='{"box", "gaussian"}', default='"box"', ) spp: int = documented( attrs.field(default=1000, converter=int, validator=validators.is_positive), doc="Number of samples per pixel.", type="int", default="1000", ) @spp.validator def _spp_validator(self, attribute, value): if eradiate.mode().is_single_precision and value > 100000: warnings.warn( f"Measure {} is defined with a sample count greater " "than 1e5, but the selected mode is single-precision: results " "may be incorrect." ) @property @abstractmethod def film_resolution(self) -> tuple[int, int]: """ tuple: Getter for film resolution as a (int, int) pair. """ pass # -------------------------------------------------------------------------- # Flag-style queries # --------------------------------------------------------------------------
[docs] def is_distant(self) -> bool: """ Return ``True`` iff measure records radiometric quantities at infinite distance. """ # Default implementation returns False return False
# -------------------------------------------------------------------------- # Kernel dictionary generation # -------------------------------------------------------------------------- @property def kernel_type(self) -> str: raise NotImplementedError @property def sensor_id(self) -> str: return @property def template(self) -> dict: result = { "type": self.kernel_type, "id": self.sensor_id, "film.type": "hdrfilm", "film.width": self.film_resolution[0], "film.height": self.film_resolution[1], "film.pixel_format": "luminance", "film.component_format": "float32", "film.rfilter.type": "box", "sampler.type": self.sampler, "sampler.sample_count": self.spp, "medium.type": InitParameter( lambda ctx: "ref" if f"{self.sensor_id}.atmosphere_medium_id" in ctx.kwargs else InitParameter.UNUSED, ), "": InitParameter( lambda ctx: ctx.kwargs[f"{self.sensor_id}.atmosphere_medium_id"] if f"{self.sensor_id}.atmosphere_medium_id" in ctx.kwargs else InitParameter.UNUSED, ), } return result # -------------------------------------------------------------------------- # Post-processing information # -------------------------------------------------------------------------- @property def var(self) -> tuple[str, dict]: """ str, dict: Post-processing variable field name and metadata. """ return "img", dict()