PySDM.products.impl.product

  1"""
  2logic around the `PySDM.products.impl.product.Product` - parent class for all products
  3"""
  4
  5import inspect
  6from abc import abstractmethod
  7from typing import Optional
  8
  9import pint
 10
 11from PySDM.physics.constants import PPB, PPM, PPT
 12from PySDM.impl.camel_case import camel_case_to_words
 13
 14_UNIT_REGISTRY = pint.UnitRegistry()
 15
 16
 17class Product:
 18    def __init__(self, *, unit: str, name: Optional[str] = None):
 19        self.name = name or camel_case_to_words(self.__class__.__name__)
 20
 21        self._unit = self._parse_unit(unit)
 22        self.unit_magnitude_in_base_units = self._unit.to_base_units().magnitude
 23        self.__check_unit()
 24
 25        self.shape = None
 26        self.buffer = None
 27        self.particulator = None
 28        self.formulae = None
 29
 30    def register(self, builder):
 31        self.particulator = builder.particulator
 32        self.formulae = self.particulator.formulae
 33        self.shape = self.particulator.mesh.grid
 34
 35    def set_buffer(self, buffer):
 36        self.buffer = buffer
 37
 38    def _download_to_buffer(self, storage):
 39        storage.download(self.buffer.ravel())
 40
 41    @staticmethod
 42    def _parse_unit(unit: str):
 43        if unit in ("%", "percent"):
 44            return 0.01 * _UNIT_REGISTRY.dimensionless
 45        if unit in ("PPB", "ppb"):
 46            return PPB * _UNIT_REGISTRY.dimensionless
 47        if unit in ("PPM", "ppm"):
 48            return PPM * _UNIT_REGISTRY.dimensionless
 49        if unit in ("PPT", "ppt"):
 50            return PPT * _UNIT_REGISTRY.dimensionless
 51        return _UNIT_REGISTRY.parse_expression(unit)
 52
 53    def __check_unit(self):
 54        init = inspect.signature(self.__init__)
 55        if "unit" not in init.parameters:
 56            raise AssertionError(
 57                f"method __init__ of class {type(self).__name__}"
 58                f" is expected to have a unit parameter"
 59            )
 60
 61        default_unit_arg = init.parameters["unit"].default
 62
 63        if (
 64            default_unit_arg is inspect._empty
 65            or default_unit_arg is None
 66            or str(default_unit_arg).strip() == ""
 67        ):
 68            raise AssertionError(
 69                f"unit parameter of {type(self).__name__}.__init__"
 70                f" is expected to have a non-empty default value"
 71            )
 72
 73        default_unit = self._parse_unit(default_unit_arg)
 74
 75        if default_unit.to_base_units().magnitude != 1:
 76            raise AssertionError(
 77                f'default value "{default_unit_arg}"'
 78                f" of {type(self).__name__}.__init__() unit parameter"
 79                f" is not a base SI unit"
 80            )
 81
 82        if self._unit.dimensionality != default_unit.dimensionality:
 83            raise AssertionError(
 84                f"provided unit ({self._unit}) has different dimensionality"
 85                f" ({self._unit.dimensionality}) than the default one"
 86                f" ({default_unit.dimensionality})"
 87                f" for product {type(self).__name__}"
 88            )
 89
 90    @property
 91    def unit(self):
 92        return str(self._unit)
 93
 94    @abstractmethod
 95    def _impl(self, **kwargs):
 96        raise NotImplementedError()
 97
 98    def get(self, **kwargs):
 99        result = self._impl(**kwargs)
100        result /= self.unit_magnitude_in_base_units
101        return result