Voltage-Compensated Dummy Load Power Measurement

utils/measure/.env.dist

@@ -93,4 +93,5 @@ MYSTROM_DEVICE_IP=x.x.x.x
 # MODEL_NAME=xx
 # DUMMY_LOAD=true
 # POWERMETER_ENTITY_ID=sensor.my_power
+# VOLTAGEMETER_ENTITY_ID=sensor.my_voltage
 # RESUME=true

utils/measure/measure/const.py

@@ -1,5 +1,5 @@
 import os
-from enum import Enum
+from enum import Enum, StrEnum
 from pathlib import Path
 
 QUESTION_GENERATE_MODEL_JSON = "generate_model_json"
@@ -21,3 +21,14 @@ class MeasureType(str, Enum):
     RECORDER = "Recorder"
     AVERAGE = "Average"
     CHARGING = "Charging device"
+
+
+class Trend(StrEnum):
+    INCREASING = "increasing"
+    DECREASING = "decreasing"
+    STEADY = "steady"
+
+
+dummy_load_measurement_count = 20
+dummy_load_measurements_duration = 30
+RETRY_COUNT_LIMIT = 100

utils/measure/measure/powermeter/const.py

@@ -14,3 +14,4 @@ class PowerMeterType(str, Enum):
 
 
 QUESTION_POWERMETER_ENTITY_ID = "powermeter_entity_id"
+QUESTION_VOLTAGEMETER_ENTITY_ID = "voltagemeter_entity_id"

utils/measure/measure/powermeter/dummy.py

@@ -7,8 +7,17 @@
 
 
 class DummyPowerMeter(PowerMeter):
-    def get_power(self) -> PowerMeasurementResult:
-        return PowerMeasurementResult(20.5, time.time())
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        if include_voltage:
+            return PowerMeasurementResult(
+                power=20.5,
+                voltage=233.0,
+                updated=time.time(),
+            )
+        return PowerMeasurementResult(power=20.5, updated=time.time())
+
+    def has_voltage_support(self) -> bool:
+        return True
 
     def process_answers(self, answers: dict[str, Any]) -> None:
         pass

utils/measure/measure/powermeter/errors.py

@@ -12,3 +12,7 @@ class ZeroReadingError(PowerMeterError):
 
 class ApiConnectionError(PowerMeterError):
     pass
+
+
+class UnsupportedFeatureError(PowerMeterError):
+    pass

utils/measure/measure/powermeter/hass.py

@@ -4,23 +4,27 @@
 from typing import Any
 
 import inquirer
-from homeassistant_api import Client
+from homeassistant_api import Client, Entity
 
-from measure.powermeter.const import QUESTION_POWERMETER_ENTITY_ID
-from measure.powermeter.errors import PowerMeterError
+from measure.const import QUESTION_DUMMY_LOAD
+from measure.powermeter.const import QUESTION_POWERMETER_ENTITY_ID, QUESTION_VOLTAGEMETER_ENTITY_ID
+from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
 from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class HassPowerMeter(PowerMeter):
     def __init__(self, api_url: str, token: str, call_update_entity: bool) -> None:
         self._call_update_entity = call_update_entity
         self._entity_id: str | None = None
+        self._voltage_entity_id: str | None = None
+        self._entities: list[Entity] | None = None
         try:
             self.client = Client(api_url, token, cache_session=False)
         except Exception as e:
             raise PowerMeterError(f"Failed to connect to HA API: {e}") from e
 
-    def get_power(self) -> PowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Get a new power reading from Hass-API. Optionally include voltage."""
         if self._call_update_entity:
             self.client.trigger_service(
                 "homeassistant",
@@ -33,24 +37,100 @@ def get_power(self) -> PowerMeasurementResult:
         if state == "unavailable":
             raise PowerMeterError(f"Power sensor {self._entity_id} unavailable")
         last_updated = state.last_updated.timestamp()
-        return PowerMeasurementResult(float(state.state), last_updated)
+        power_value = float(state.state)
+
+        if include_voltage and not self.has_voltage_support():
+            raise UnsupportedFeatureError("Voltage sensor entity not found.")
+
+        if include_voltage:
+            voltage_state = self.client.get_state(entity_id=self._voltage_entity_id)
+            if voltage_state == "unavailable":
+                raise PowerMeterError(f"Voltage sensor {self._voltage_entity_id} unavailable")
+
+            voltage_value = float(voltage_state.state)
+            return PowerMeasurementResult(
+                power=power_value,
+                voltage=voltage_value,
+                updated=last_updated,
+            )
+
+        return PowerMeasurementResult(power=power_value, updated=last_updated)
+
+    def has_voltage_support(self) -> bool:
+        if not self._voltage_entity_id:
+            return False
+
+        voltage_state = self.client.get_state(entity_id=self._voltage_entity_id)
+        if voltage_state == "unavailable":
+            raise PowerMeterError(f"Voltage sensor {self._voltage_entity_id} unavailable")
+        return True
+
+    def autodetect_voltage_entity(self, power_entity: str) -> bool:
+        """Try to find a matching voltage entity for the current power entity."""
+        matched_sensors = self.match_power_and_voltage_sensors()
+
+        if not matched_sensors or power_entity not in matched_sensors:
+            # no match found for our power sensor
+            return False
+
+        self._voltage_entity_id = matched_sensors.get(power_entity)
+        return True
 
     def get_questions(self) -> list[inquirer.questions.Question]:
-        power_sensor_list = self.get_power_sensors()
+        def _should_skip_voltage_sensor_question(answers: dict[str, Any]) -> bool:
+            """Determine if the voltage sensor question should be asked."""
+            if not answers.get(QUESTION_DUMMY_LOAD, False):
+                return True
+            return self.autodetect_voltage_entity(answers.get(QUESTION_POWERMETER_ENTITY_ID))
 
+        power_sensor_list = self.get_power_sensors()
         return [
             inquirer.List(
                 name=QUESTION_POWERMETER_ENTITY_ID,
                 message="Select the powermeter",
                 choices=power_sensor_list,
             ),
+            inquirer.List(
+                name=QUESTION_VOLTAGEMETER_ENTITY_ID,
+                message="Select the voltage sensor",
+                choices=lambda answers: self.get_voltage_sensors(),
+                ignore=_should_skip_voltage_sensor_question,
+            ),
         ]
 
     def get_power_sensors(self) -> list[str]:
-        entities = self.client.get_entities()
-        sensors = entities["sensor"].entities.values()
-        power_sensors = [entity.entity_id for entity in sensors if entity.state.attributes.get("unit_of_measurement") == "W"]
-        return sorted(power_sensors)
+        return self.get_entities_by_unit_of_measurement("W")
+
+    def get_voltage_sensors(self) -> list[str]:
+        return self.get_entities_by_unit_of_measurement("V")
+
+    def get_entities_by_unit_of_measurement(self, unit_of_measurement: str) -> list[str]:
+        return sorted(
+            [entity.entity_id for entity in self.get_entities() if entity.state.attributes.get("unit_of_measurement") == unit_of_measurement],
+        )
+
+    def get_entities(self) -> list[Entity]:
+        if not self._entities:
+            entities = self.client.get_entities()
+            self._entities = list(entities["sensor"].entities.values())
+        return self._entities
+
+    def match_power_and_voltage_sensors(self) -> dict[str, str]:
+        power_sensors = self.get_power_sensors()
+        voltage_sensors = self.get_voltage_sensors()
+
+        # Create mappings based on base names
+        power_map = {sensor.rsplit("_power", 1)[0]: sensor for sensor in power_sensors}
+        voltage_map = {sensor.rsplit("_voltage", 1)[0]: sensor for sensor in voltage_sensors}
+
+        matched_sensors = {}
+        for base_name, power_sensor in power_map.items():
+            if base_name in voltage_map:
+                matched_sensors[power_sensor] = voltage_map[base_name]
+
+        return matched_sensors
 
     def process_answers(self, answers: dict[str, Any]) -> None:
         self._entity_id = answers[QUESTION_POWERMETER_ENTITY_ID]
+        if QUESTION_VOLTAGEMETER_ENTITY_ID in answers:
+            self._voltage_entity_id = answers[QUESTION_VOLTAGEMETER_ENTITY_ID]

utils/measure/measure/powermeter/kasa.py

@@ -6,21 +6,31 @@
 
 from kasa import SmartPlug
 
+from measure.powermeter.errors import UnsupportedFeatureError
 from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class KasaPowerMeter(PowerMeter):
     def __init__(self, device_ip: str) -> None:
         self._smartplug = SmartPlug(device_ip)
 
-    def get_power(self) -> PowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Get a new power reading from the Kasa device. Optionally include voltage (FIXME: not yet implemented)."""
+        if include_voltage:
+            # FIXME: Not yet implemented # noqa: FIX001
+            raise UnsupportedFeatureError("Voltage measurement is not yet implemented for Kasa devices.")
+
         loop = asyncio.get_event_loop()
         power = loop.run_until_complete(self.async_read_power_meter())
-        return PowerMeasurementResult(power, time.time())
+
+        return PowerMeasurementResult(power=power, updated=time.time())
 
     async def async_read_power_meter(self) -> None:
         await self._smartplug.update()
         return self._smartplug.emeter_realtime["power"]
 
+    def has_voltage_support(self) -> bool:
+        return False
+
     def process_answers(self, answers: dict[str, Any]) -> None:
         pass

utils/measure/measure/powermeter/manual.py

@@ -7,9 +7,23 @@
 
 
 class ManualPowerMeter(PowerMeter):
-    def get_power(self) -> PowerMeasurementResult:
-        power = input("Input power measurement:")
-        return PowerMeasurementResult(float(power), time.time())
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Manually enter power readings. Optionally enter voltage readings as well."""
+        if include_voltage:
+            power = input("Input power measurement: ")
+            voltage = input("Input voltage measurement: ")
+
+            return PowerMeasurementResult(
+                power=float(power),
+                voltage=float(voltage),
+                updated=time.time(),
+            )
+
+        power = input("Input power measurement: ")
+        return PowerMeasurementResult(power=float(power), updated=time.time())
+
+    def has_voltage_support(self) -> bool:
+        return True
 
     def process_answers(self, answers: dict[str, Any]) -> None:
         pass

utils/measure/measure/powermeter/mystrom.py

@@ -5,15 +5,20 @@
 
 import requests
 
-from measure.powermeter.errors import PowerMeterError
+from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
 from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class MyStromPowerMeter(PowerMeter):
     def __init__(self, device_ip: str) -> None:
         self._device_ip = device_ip
 
-    def get_power(self) -> PowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Get a new power reading from the MyStrom device. Optionally include voltage (FIXME: not yet implemented)."""
+        if include_voltage:
+            # FIXME: Not yet implemented # noqa: FIX001
+            raise UnsupportedFeatureError("Voltage measurement is not yet implemented for MyStrom devices.")
+
         r = requests.get(
             f"http://{self._device_ip}/report",
             timeout=10,
@@ -25,7 +30,10 @@
         except KeyError as error:
             raise PowerMeterError("Unexpected JSON response format") from error
 
-        return PowerMeasurementResult(float(power), time.time())
+        return PowerMeasurementResult(power=float(power), updated=time.time())
+
+    def has_voltage_support(self) -> bool:
+        return False
 
     def process_answers(self, answers: dict[str, Any]) -> None:
         pass

utils/measure/measure/powermeter/ocr.py

@@ -3,6 +3,7 @@
 import os
 from typing import Any
 
+from measure.powermeter.errors import UnsupportedFeatureError
 from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
@@ -16,12 +17,16 @@
         self.file = open(filepath, "rb")  # noqa: SIM115
         super().__init__()
 
-    def get_power(self) -> PowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Get a new power reading via OCR."""
+        if include_voltage:
+            raise UnsupportedFeatureError("Voltage measurement are not supported for OCR mode.")
+
         last_line = self.read_last_line()
         (timestamp, power) = last_line.strip().split(";")
         power = float(power)
         timestamp = float(timestamp)
-        return PowerMeasurementResult(power, timestamp)
+        return PowerMeasurementResult(power=power, updated=timestamp)
 
     def read_last_line(self) -> str:
         try:
@@ -32,5 +37,8 @@
             self.file.seek(0)
         return self.file.readline().decode()
 
+    def has_voltage_support(self) -> bool:
+        return False
+
     def process_answers(self, answers: dict[str, Any]) -> None:
         pass

utils/measure/measure/powermeter/powermeter.py

@@ -8,8 +8,12 @@
 
 class PowerMeter(ABC):
     @abstractmethod
-    def get_power(self) -> PowerMeasurementResult:
-        """Get a power measurement from the meter"""
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
+        """Get a power measurement from the meter. Optionally include voltage readings."""
+
+    @abstractmethod
+    def has_voltage_support(self) -> bool:
+        """Returns bool depending on the powermeter capabilities to act as a voltmeter."""
 
     def get_questions(self) -> list[Question]:
         """Get questions to ask for the chosen powermeter"""
@@ -23,3 +27,4 @@ def process_answers(self, answers: dict[str, Any]) -> None:
 class PowerMeasurementResult(NamedTuple):
     power: float
     updated: float
+    voltage: Optional[float] = None  # noqa: F821