From 4fef17df741a1908ec396de426a04e7d9f4dd7ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Jaworski?= Date: Tue, 16 Apr 2024 02:17:21 +0200 Subject: [PATCH] Feature: smart meter (#168) * Smart Meter: General Integration * feat: fold power consumption into generic sensor with some shims * fix: accidental mutation of methods dict * fix: regression in wind sensor * fix: use old-style typing (for now) --------- Co-authored-by: Mikolaj Pastucha --- blebox_uniapi/box_types.py | 60 +++++++-- blebox_uniapi/sensor.py | 270 +++++++++++++++++++------------------ 2 files changed, 182 insertions(+), 148 deletions(-) diff --git a/blebox_uniapi/box_types.py b/blebox_uniapi/box_types.py index a2c80bc..8388b8f 100644 --- a/blebox_uniapi/box_types.py +++ b/blebox_uniapi/box_types.py @@ -276,9 +276,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -297,9 +297,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -318,9 +318,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -341,9 +341,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -376,9 +376,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -408,9 +408,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -440,9 +440,9 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: [ "switchBox.energy", { - "energy": "powerMeasuring/powerConsumption/[0]/value", + "powerConsumption": lambda x: "powerMeasuring/powerConsumption/[0]/value", "periodS": "powerMeasuring/powerConsumption/[0]/periodS", - "measurment_enabled": "powerMeasuring/enabled", + "measurement_enabled": "powerMeasuring/enabled", }, ] ], @@ -671,5 +671,37 @@ def get_latest_api_level(product_type: str) -> Union[dict, int]: ] ], }, + 20230606: { + "api_path": "/state", + "extended_state_path": "/state/extended", + "sensors": [ + [ + "multiSensor", + { + "frequency": lambda x: f"multiSensor/sensors/[id={x}]/value", + "current": lambda x: f"multiSensor/sensors/[id={x}]/value", + "voltage": lambda x: f"multiSensor/sensors/[id={x}]/value", + "apparentPower": lambda x: f"multiSensor/sensors/[id={x}]/value", + "activePower": lambda x: f"multiSensor/sensors/[id={x}]/value", + "reactivePower": lambda x: f"multiSensor/sensors/[id={x}]/value", + "reverseActiveEnergy": lambda x: f"multiSensor/sensors/[id={x}]/value", + "forwardActiveEnergy": lambda x: f"multiSensor/sensors/[id={x}]/value", + "illuminance": lambda x: f"multiSensor/sensors/[id={x}]/value", + "temperature": lambda x: f"multiSensor/sensors/[id={x}]/value", + "wind": lambda x: f"multiSensor/sensors/[id={x}]/value", + "humidity": lambda x: f"multiSensor/sensors/[id={x}]/value", + }, + ] + ], + "binary_sensors": [ + [ + "multiSensor", + { + "rain": lambda x: f"multiSensor/sensors/[id={x}]/value", + "flood": lambda x: f"multiSensor/sensors/[id={x}]/value", + }, + ] + ], + }, }, } diff --git a/blebox_uniapi/sensor.py b/blebox_uniapi/sensor.py index bfa2d2c..5921e4c 100644 --- a/blebox_uniapi/sensor.py +++ b/blebox_uniapi/sensor.py @@ -1,4 +1,7 @@ import datetime +import numbers +from functools import partial + from .feature import Feature from typing import TYPE_CHECKING, Union, Optional @@ -7,70 +10,97 @@ class SensorFactory: - type_class_mapper: dict[str, type] = {} + device_constructors: dict[str, type] = {} @classmethod - def register(cls, sensor_type: str): - def decorator(subclass: type): - cls.type_class_mapper[sensor_type] = subclass - return subclass + def register(cls, sensor_type: str, **kwargs): + if sensor_type in cls.device_constructors: + raise RuntimeError("Can't register same sensor type twice") + + def decorator(registrable: type): + constructor = registrable + if kwargs: + constructor = partial(registrable, sensor_type=sensor_type, **kwargs) + + cls.device_constructors[sensor_type] = constructor + # note: returning unmodified, so we can register registrable + # multiple times under different names and with different kwargs + return registrable return decorator + @staticmethod + def _sensor_states(extended_state: dict): + """Read potential sensor states from extended state dictionary""" + # note: probably we should iterate extended state in future if there + # are other api flavours other than multiSensor that provide sensors + states = extended_state.get("multiSensor", {}).get("sensors", []) + + # note: power measuring feature predates multiSensor API, so we need a small + # shim to adapt older shape of power measuring schema to the new sensor API + if "powerMeasuring" in extended_state: + power_states = extended_state["powerMeasuring"].get("powerConsumption", []) + # note: be careful of names as this has been historically named differently + # in home-assistant + states.extend({"type": "powerConsumption", **s} for s in power_states) + return states + @classmethod def many_from_config(cls, product, box_type_config, extended_state): - type_class_mapper = cls.type_class_mapper if extended_state: object_list = [] - alias, methods = box_type_config[0] - sensor_list = extended_state.get("multiSensor", {}).get("sensors", []) - for sensor in sensor_list: - sensor_type = sensor.get("type") + # note: first item was historically an alias, but it has been since + # abandoned. We still keep it in the box config. + _, methods = box_type_config[0] + + for sensor in cls._sensor_states(extended_state): + device_class = sensor.get("type") sensor_id = sensor.get("id") - if type_class_mapper.get(sensor_type): - value_method = {sensor_type: methods[sensor_type](sensor_id)} - object_list.append( - type_class_mapper[sensor_type]( - product=product, - alias=f"{sensor_type}_{str(sensor_id)}", - methods=value_method, - ) - ) - if "powerConsumption" in str(extended_state): - consumption_meters = extended_state.get("powerMeasuring", {}).get( - "powerConsumption", [] - ) - for _ in consumption_meters: - object_list.append( - Energy( - product=product, alias="powerConsumption", methods=methods - ) + alias = device_class + if sensor_id is not None: + alias = f"{device_class}_{sensor_id}" + + if constructor := cls.device_constructors.get(device_class): + # note: methods for sensor readings are provided as template + # functions (lambdas) in the box config. We need to "materialize" + # them to make sure they are properly indexed by sensor ID + materialized_methods = { + **methods, + device_class: methods[device_class](sensor_id), + } + + feature = constructor( + product=product, alias=alias, methods=materialized_methods ) + object_list.append(feature) return object_list + + # legacy handling of some old device API that do not provide extended state + alias, methods = box_type_config[0] + if alias.endswith("air"): + method_list = [method for method in methods if "value" in method] + return [ + AirQuality(product=product, alias=method.split(".")[0], methods=methods) + for method in method_list + ] + if alias.endswith("temperature"): + return [Temperature(product=product, alias=alias, methods=methods)] else: - alias, methods = box_type_config[0] - if alias.endswith("air"): - method_list = [method for method in methods if "value" in method] - return [ - AirQuality( - product=product, alias=method.split(".")[0], methods=methods - ) - for method in method_list - ] - if alias.endswith("temperature"): - return [Temperature(product=product, alias=alias, methods=methods)] - else: - return [] + return [] class BaseSensor(Feature): _unit: str _device_class: str _native_value: Union[float, int, str] + _sensor_type: Optional[str] - def __init__(self, product: "Box", alias: str, methods: dict): + def __init__( + self, product: "Box", alias: str, methods: dict, sensor_type: str = None + ): + self._sensor_type = sensor_type super().__init__(product, alias, methods) @property @@ -89,32 +119,79 @@ def native_value(self): def many_from_config(cls, product, box_type_config, extended_state): raise NotImplementedError("Please use SensorFactory") + def __str__(self): + return f"<{self.__class__.__name__} sensor_type={self._sensor_type}, alias={self._alias}>" + + +@SensorFactory.register("frequency", unit="Hz", scale=1_000) +@SensorFactory.register("current", unit="mA", scale=10) +@SensorFactory.register("voltage", unit="V", scale=10) +@SensorFactory.register("apparentPower", unit="va") +@SensorFactory.register("reactivePower", unit="var") +@SensorFactory.register("activePower", unit="W") +@SensorFactory.register("reverseActiveEnergy", unit="kWh") +@SensorFactory.register("forwardActiveEnergy", unit="kWh") +@SensorFactory.register("illuminance", unit="lx", scale=100) +@SensorFactory.register("humidity", unit="percentage", scale=100) +@SensorFactory.register("wind", unit="m/s", scale=10) +class GenericSensor(BaseSensor): + def __init__( + # base sensor params + self, + product: "Box", + alias: str, + methods: dict, + *, + # generalization params + sensor_type: str, + unit: str, + scale: float = 1, + precision: Optional[int] = None, + ): + super().__init__(product, alias, methods) + self._unit = unit + self._scale = scale + self._precision = precision + # note: this seems redundant but there is at least one sensor type that + # has different mapping in home assistant (wind/wind_speed). Should be + # fixed in upstream first. + self._device_class = sensor_type + self._sensor_type = sensor_type + + def after_update(self): + product = self._product + if product.last_data is None: + return + + raw = self.raw_value(self._device_class) + if not isinstance(raw, numbers.Number): + raw = float("nan") -@SensorFactory.register("illuminance") -class Illuminance(BaseSensor): - _current: Union[float, int, None] + native = raw / self._scale + if self._precision: + native = round(native, self._precision) - def __init__(self, product: "Box", alias: str, methods: dict): - super().__init__(product, alias, methods) - self._unit = "lx" - self._device_class = "illuminance" + self._native_value = native + +@SensorFactory.register("powerConsumption", unit="kWh") +class PowerConsumption(GenericSensor): + # note: almost the same as typical generic sensor but also provides extra property + # to read last reset value @property - def current(self) -> Union[float, int, None]: - return self._current + def last_reset(self): + return datetime.datetime.now() - datetime.timedelta( + seconds=self._read_period_of_measurement() + ) - def _read_illuminance(self): + def _read_period_of_measurement(self) -> int: product = self._product if product.last_data is not None: - raw = self.raw_value("illuminance") + raw = self.raw_value("periodS") if raw is not None: alias = self._alias - return round(product.expect_int(alias, raw, 10000000, 0) / 100.0, 1) - return None - - def after_update(self) -> None: - self._native_value = self._read_illuminance() - self._current = self._read_illuminance() + return product.expect_int(alias, raw, 3600, 0) + return 0 @SensorFactory.register("temperature") @@ -164,78 +241,3 @@ def _pm_value(self, name: str) -> Optional[int]: def after_update(self) -> None: self._native_value = self._pm_value(f"{self.device_class}.value") - - -@SensorFactory.register("humidity") -class Humidity(BaseSensor): - def __init__(self, product: "Box", alias: str, methods: dict): - super().__init__(product, alias, methods) - self._unit = "percentage" - self._device_class = "humidity" - - def _read_humidity(self, field: str) -> Optional[int]: - product = self._product - if product.last_data is not None: - raw = self.raw_value(field) - if raw is not None: - alias = self._alias - return round(product.expect_int(alias, raw, 10000, 0) / 100.0, 1) - - return None - - def after_update(self) -> None: - self._native_value = self._read_humidity(f"{self.device_class}") - - -class Energy(BaseSensor): - def __init__(self, product: "Box", alias: str, methods: dict): - super().__init__(product, alias, methods) - self._unit = "kWh" - self._device_class = "powerMeasurement" - - @property - def last_reset(self): - return datetime.datetime.now() - datetime.timedelta( - seconds=self._read_period_of_measurement() - ) - - def _read_period_of_measurement(self) -> int: - product = self._product - if product.last_data is not None: - raw = self.raw_value("periodS") - if raw is not None: - alias = self._alias - return product.expect_int(alias, raw, 3600, 0) - return 0 - - def _read_power_measurement(self): - product = self._product - if product.last_data is not None: - raw = float(self.raw_value("energy")) - return raw - return None - - def after_update(self) -> None: - self._native_value = self._read_power_measurement() - - -@SensorFactory.register("wind") -class Wind(BaseSensor): - def __init__(self, product: "Box", alias: str, methods: dict): - super().__init__(product, alias, methods) - self._unit = "m/s" - self._device_class = "wind_speed" - - def _read_wind_speed(self): - product = self._product - if product.last_data is not None: - raw = self.raw_value("wind") - if raw is not None: - alias = self._alias - # wind value unit in API is "0.1 m/s" so to get m/s we need to divide by 10 - # min value = 0, max value for sure not bigger than 200km/h so about 60m/s so 600 in API - return round(product.expect_int(alias, raw, 600, 0) / 10.0, 1) - return None - - def after_update(self) -> None: - self._native_value = self._read_wind_speed()