Re-architect data calculators (#319)

* Re-architect calculators

* Complete

.pre-commit-config.yaml

@@ -48,7 +48,7 @@ repos:
           - toml
         files: ^(ecowitt2mqtt|tests)/.+\.py$
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.960
+    rev: v0.982
     hooks:
       - id: mypy
         files: ^ecowitt2mqtt/.+\.py$

ecowitt2mqtt/const.py

@@ -93,8 +93,7 @@
 DATA_POINT_SOLARRADIATION: Final = "solarradiation"
 DATA_POINT_SOLARRADIATION_LUX: Final = "solarradiation_lux"
 DATA_POINT_SOLARRADIATION_PERCEIVED: Final = "solarradiation_perceived"
-DATA_POINT_TEMPF: Final = "tempf"
-DATA_POINT_TEMPINF: Final = "tempinf"
+DATA_POINT_TEMP: Final = "temp"
 DATA_POINT_TF_CO2: Final = "tf_co2"
 DATA_POINT_THERMAL_PERCEPTION: Final = "thermalperception"
 DATA_POINT_TOTAL_RAIN: Final = "totalrain"
@@ -111,7 +110,7 @@
 DATA_POINT_WH90BATT_PC: Final = "wh90battpc"
 DATA_POINT_WH90CAP_VOLT: Final = "ws90cap_volt"
 DATA_POINT_WINDCHILL: Final = "windchill"
-DATA_POINT_WINDSPEEDMPH: Final = "windspeedmph"
+DATA_POINT_WINDSPEED: Final = "windspeed"
 DATA_POINT_WRAIN_PIEZO: Final = "wrain_piezo"
 DATA_POINT_WS90_VER: Final = "ws90_ver"
 DATA_POINT_YEARLY_RAIN: Final = "yearlyrain"

ecowitt2mqtt/helpers/calculator/__init__.py

@@ -1,11 +1,28 @@
 """Define various calculators."""
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any
+from functools import wraps
+from typing import TYPE_CHECKING, Any, Callable, Dict, TypeVar
 
-from ecowitt2mqtt.helpers.typing import DataValueType
+from ecowitt2mqtt.errors import EcowittError
+from ecowitt2mqtt.helpers.typing import CalculatedValueType, PreCalculatedValueType
+
+if TYPE_CHECKING:
+    from ecowitt2mqtt.config import Config
+
+_CalculatorType = TypeVar("_CalculatorType", bound="Calculator")
+_CalculateFromPayloadFuncType = Callable[
+    [_CalculatorType, Dict[str, PreCalculatedValueType]], "CalculatedDataPoint"
+]
+
+
+class CalculationKeysMissingError(EcowittError):
+    """Define an error when keys required for a calculated data point are missing."""
+
+    pass
 
 
 class DataPointType(Enum):
@@ -20,7 +37,85 @@ class CalculatedDataPoint:
     """Define a calculated data point."""
 
     data_point_key: str
-    value: DataValueType
+    value: CalculatedValueType
     unit: str | None = None
     attributes: dict[str, Any] = field(default_factory=dict)
     data_type: DataPointType = DataPointType.NON_BOOLEAN
+
+
+class Calculator:
+    """Define a calculator."""
+
+    def __init__(self, config: Config, payload_key: str, data_point_key: str) -> None:
+        """Initialize."""
+        self._config = config
+        self._data_point_key = data_point_key
+        self._payload_key = payload_key
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        return None
+
+    def calculate_from_value(
+        self, value: PreCalculatedValueType
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+
+    def calculate_from_payload(
+        self, payload: dict[str, PreCalculatedValueType]
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+
+    def get_calculated_data_point(
+        self,
+        value: CalculatedValueType,
+        *,
+        attributes: dict[str, Any] | None = None,
+        data_type: DataPointType | None = None,
+    ) -> CalculatedDataPoint:
+        """Get the output unit for this calculation."""
+        data_point = CalculatedDataPoint(
+            data_point_key=self._data_point_key, value=value, unit=self.output_unit
+        )
+
+        if attributes:
+            data_point.attributes = attributes
+        if data_type:
+            data_point.data_type = data_type
+
+        return data_point
+
+    @staticmethod
+    def requires_keys(
+        *keys: Iterable[str],
+    ) -> Callable[[_CalculateFromPayloadFuncType], _CalculateFromPayloadFuncType]:
+        """Define a decorator that requires certain payload keys to exist."""
+
+        def decorator(
+            func: _CalculateFromPayloadFuncType,
+        ) -> _CalculateFromPayloadFuncType:
+            """Decorate."""
+
+            @wraps(func)
+            def wrapper(
+                calculator: _CalculatorType, payload: dict[str, PreCalculatedValueType]
+            ) -> CalculatedDataPoint:
+                """Wrap."""
+                if not all(k in payload for k in keys):
+                    raise CalculationKeysMissingError
+                return func(calculator, payload)
+
+            return wrapper
+
+        return decorator
+
+
+class SimpleCalculator(Calculator):
+    """Define a calculator that returns a value as-is (with an added unit)."""
+
+    def calculate_from_value(
+        self, value: PreCalculatedValueType
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+        return self.get_calculated_data_point(value)

ecowitt2mqtt/helpers/calculator/battery.py

@@ -1,4 +1,4 @@
-"""Define battery utilities."""
+"""Define battery calculators."""
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
@@ -24,7 +24,12 @@
     ELECTRIC_POTENTIAL_VOLT,
     PERCENTAGE,
 )
-from ecowitt2mqtt.helpers.calculator import CalculatedDataPoint, DataPointType
+from ecowitt2mqtt.helpers.calculator import (
+    CalculatedDataPoint,
+    Calculator,
+    DataPointType,
+)
+from ecowitt2mqtt.helpers.typing import PreCalculatedValueType
 from ecowitt2mqtt.util import glob_search
 
 if TYPE_CHECKING:
@@ -66,40 +71,6 @@ class BooleanBatteryState(StrEnum):
 }
 
 
-def calculate_battery(
-    config: Config, payload_key: str, data_point_key: str, value: float
-) -> CalculatedDataPoint:
-    """Calculate a battery value."""
-    strategy = get_battery_strategy(config, payload_key)
-
-    if strategy == BatteryStrategy.NUMERIC:
-        return CalculatedDataPoint(
-            data_point_key=data_point_key, value=value, unit=ELECTRIC_POTENTIAL_VOLT
-        )
-    if strategy == BatteryStrategy.PERCENTAGE:
-        # Percentage batteries occur in "steps":
-        #   * 1 = 20%
-        #   * 2 = 40%
-        #   * 3 = 60%
-        #   * 4 = 80%
-        #   * 5 = 100%
-        #   * 6 = 120% (plugged into mains voltage)
-        return CalculatedDataPoint(
-            data_point_key=data_point_key, value=value * 20, unit=PERCENTAGE
-        )
-    if value == 0.0:
-        return CalculatedDataPoint(
-            data_point_key=data_point_key,
-            value=BooleanBatteryState.OFF,
-            data_type=DataPointType.BOOLEAN,
-        )
-    return CalculatedDataPoint(
-        data_point_key=data_point_key,
-        value=BooleanBatteryState.ON,
-        data_type=DataPointType.BOOLEAN,
-    )
-
-
 def get_battery_strategy(config: Config, key: str) -> BatteryStrategy:
     """Get the battery strategy for a particular key."""
     strategies = [config.battery_overrides.get(key)]
@@ -116,3 +87,49 @@ def get_battery_strategy(config: Config, key: str) -> BatteryStrategy:
             return strategy
 
     return config.default_battery_strategy
+
+
+class BatteryCalculator(Calculator):
+    """Define a battery calculator."""
+
+    def __init__(self, config: Config, payload_key: str, data_point_key: str) -> None:
+        """Initialize."""
+        super().__init__(config, payload_key, data_point_key)
+
+        self._battery_strategy = get_battery_strategy(config, payload_key)
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        if self._battery_strategy == BatteryStrategy.NUMERIC:
+            return ELECTRIC_POTENTIAL_VOLT
+        if self._battery_strategy == BatteryStrategy.PERCENTAGE:
+            return PERCENTAGE
+        return None
+
+    def calculate_from_value(
+        self, value: PreCalculatedValueType
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+        assert isinstance(value, float)
+
+        if self._battery_strategy == BatteryStrategy.NUMERIC:
+            return self.get_calculated_data_point(value)
+
+        if self._battery_strategy == BatteryStrategy.PERCENTAGE:
+            # Percentage batteries occur in "steps":
+            #   * 1 = 20%
+            #   * 2 = 40%
+            #   * 3 = 60%
+            #   * 4 = 80%
+            #   * 5 = 100%
+            #   * 6 = 120% (plugged into mains voltage)
+            return self.get_calculated_data_point(value * 20)
+
+        if value == 0.0:
+            return self.get_calculated_data_point(
+                BooleanBatteryState.OFF, data_type=DataPointType.BOOLEAN
+            )
+        return self.get_calculated_data_point(
+            BooleanBatteryState.ON, data_type=DataPointType.BOOLEAN
+        )

ecowitt2mqtt/helpers/calculator/humidity.py

@@ -0,0 +1,59 @@
+"""Define humidity calculators."""
+from __future__ import annotations
+
+from ecowitt2mqtt.const import (
+    DATA_POINT_HUMIDITY,
+    DATA_POINT_TEMP,
+    PERCENTAGE,
+    UNIT_SYSTEM_IMPERIAL,
+    WATER_VAPOR_GRAMS_PER_CUBIC_METER,
+    WATER_VAPOR_POUNDS_PER_CUBIC_FOOT,
+)
+from ecowitt2mqtt.helpers.calculator import (
+    CalculatedDataPoint,
+    Calculator,
+    SimpleCalculator,
+)
+from ecowitt2mqtt.helpers.typing import PreCalculatedValueType
+from ecowitt2mqtt.util.meteo import (
+    get_absolute_humidity,
+    get_temperature_meteocalc_object,
+)
+
+
+class AbsoluteHumidityCalculator(Calculator):
+    """Define an absolute humidity calculator."""
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        if self._config.output_unit_system == UNIT_SYSTEM_IMPERIAL:
+            return WATER_VAPOR_POUNDS_PER_CUBIC_FOOT
+        return WATER_VAPOR_GRAMS_PER_CUBIC_METER
+
+    @Calculator.requires_keys(DATA_POINT_TEMP, DATA_POINT_HUMIDITY)
+    def calculate_from_payload(
+        self, payload: dict[str, PreCalculatedValueType]
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+        assert isinstance(payload[DATA_POINT_TEMP], float)
+        assert isinstance(payload[DATA_POINT_HUMIDITY], float)
+
+        temp_obj = get_temperature_meteocalc_object(
+            payload[DATA_POINT_TEMP], self._config.input_unit_system
+        )
+        value = get_absolute_humidity(temp_obj, payload[DATA_POINT_HUMIDITY])
+
+        if self._config.output_unit_system == UNIT_SYSTEM_IMPERIAL:
+            value /= 16018.46592051
+
+        return self.get_calculated_data_point(value)
+
+
+class RelativeHumidityCalculator(SimpleCalculator):
+    """Define a boolean leak calculator."""
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        return PERCENTAGE

ecowitt2mqtt/helpers/calculator/illuminance.py

@@ -0,0 +1,73 @@
+"""Define illuminance calculators."""
+from __future__ import annotations
+
+import math
+
+from ecowitt2mqtt.const import (
+    DATA_POINT_SOLARRADIATION,
+    ILLUMINANCE_WATTS_PER_SQUARE_METER,
+    LIGHT_LUX,
+    PERCENTAGE,
+)
+from ecowitt2mqtt.helpers.calculator import (
+    CalculatedDataPoint,
+    Calculator,
+    SimpleCalculator,
+)
+from ecowitt2mqtt.helpers.typing import PreCalculatedValueType
+
+
+class IlluminanceLuxCalculator(Calculator):
+    """Define a illuminance calculator (lux)."""
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        return LIGHT_LUX
+
+    @Calculator.requires_keys(DATA_POINT_SOLARRADIATION)
+    def calculate_from_payload(
+        self, payload: dict[str, PreCalculatedValueType]
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+        assert isinstance(payload[DATA_POINT_SOLARRADIATION], float)
+
+        return self.get_calculated_data_point(
+            round(float(payload[DATA_POINT_SOLARRADIATION]) / 0.0079, 1)
+        )
+
+
+class IlluminancePerceivedCalculator(Calculator):
+    """Define a illuminance calculator (perceived)."""
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        return PERCENTAGE
+
+    @Calculator.requires_keys(DATA_POINT_SOLARRADIATION)
+    def calculate_from_payload(
+        self, payload: dict[str, PreCalculatedValueType]
+    ) -> CalculatedDataPoint:
+        """Perform the calculation."""
+        assert isinstance(payload[DATA_POINT_SOLARRADIATION], float)
+
+        lux_value = round(float(payload[DATA_POINT_SOLARRADIATION]) / 0.0079, 1)
+
+        try:
+            final_value = round(math.log10(lux_value) / 5, 2) * 100
+        except ValueError:
+            # If we've approached negative infinity, we'll get a math domain error; in
+            # that case, return 0.0:
+            final_value = 0.0
+
+        return self.get_calculated_data_point(final_value)
+
+
+class IlluminanceWM2Calculator(SimpleCalculator):
+    """Define a illuminance calculator (W/m²)."""
+
+    @property
+    def output_unit(self) -> str | None:
+        """Get the output unit of measurement for this calculation."""
+        return ILLUMINANCE_WATTS_PER_SQUARE_METER