consolidate ExtendedPowerMeasurementResult and PowerMeasurementResult…

… types
bramstroker · Dec 19, 2024 · 4a42905 · 4a42905
1 parent 67674e6
commit 4a42905
Show file tree

Hide file tree

Showing 10 changed files with 27 additions and 37 deletions.
diff --git a/utils/measure/measure/powermeter/dummy.py b/utils/measure/measure/powermeter/dummy.py
@@ -3,13 +3,13 @@
 import time
 from typing import Any
 
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class DummyPowerMeter(PowerMeter):
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
         if include_voltage:
-            return ExtendedPowerMeasurementResult(20.5, 233.0, time.time())
+            return PowerMeasurementResult(20.5, 233.0, time.time())
         return PowerMeasurementResult(20.5, time.time())
 
     def has_voltage_support(self) -> bool:

diff --git a/utils/measure/measure/powermeter/hass.py b/utils/measure/measure/powermeter/hass.py
@@ -8,7 +8,7 @@
 
 from measure.powermeter.const import QUESTION_POWERMETER_ENTITY_ID
 from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class HassPowerMeter(PowerMeter):
@@ -21,7 +21,7 @@ def __init__(self, api_url: str, token: str, call_update_entity: bool) -> None:
         except Exception as e:
             raise PowerMeterError(f"Failed to connect to HA API: {e}") from e
 
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    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(
@@ -48,7 +48,7 @@ def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | E
                 raise PowerMeterError(f"Voltage sensor {self._voltage_entity_id} unavailable")
 
             voltage_value = float(voltage_state.state)
-            return ExtendedPowerMeasurementResult(power_value, voltage_value, last_updated)
+            return PowerMeasurementResult(power_value, voltage_value, last_updated)
 
         return PowerMeasurementResult(power_value, last_updated)
 

diff --git a/utils/measure/measure/powermeter/kasa.py b/utils/measure/measure/powermeter/kasa.py
@@ -7,14 +7,14 @@
 from kasa import SmartPlug
 
 from measure.powermeter.errors import UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+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, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    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

diff --git a/utils/measure/measure/powermeter/manual.py b/utils/measure/measure/powermeter/manual.py
@@ -3,17 +3,17 @@
 import time
 from typing import Any
 
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class ManualPowerMeter(PowerMeter):
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    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 ExtendedPowerMeasurementResult(
+            return PowerMeasurementResult(
                 power=float(power),
                 voltage=float(voltage),
                 updated=time.time(),

diff --git a/utils/measure/measure/powermeter/mystrom.py b/utils/measure/measure/powermeter/mystrom.py
@@ -6,14 +6,14 @@
 import requests
 
 from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+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, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    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

diff --git a/utils/measure/measure/powermeter/ocr.py b/utils/measure/measure/powermeter/ocr.py
@@ -4,7 +4,7 @@
 from typing import Any
 
 from measure.powermeter.errors import UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class OcrPowerMeter(PowerMeter):
@@ -17,7 +17,7 @@ def __init__(self) -> None:
         self.file = open(filepath, "rb")  # noqa: SIM115
         super().__init__()
 
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    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.")

diff --git a/utils/measure/measure/powermeter/powermeter.py b/utils/measure/measure/powermeter/powermeter.py
@@ -8,7 +8,7 @@
 
 class PowerMeter(ABC):
     @abstractmethod
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
         """Get a power measurement from the meter. Optionally include voltage readings."""
 
     @abstractmethod
@@ -27,9 +27,4 @@ def process_answers(self, answers: dict[str, Any]) -> None:
 class PowerMeasurementResult(NamedTuple):
     power: float
     updated: float
-
-
-class ExtendedPowerMeasurementResult(NamedTuple):
-    power: float
-    voltage: float
-    updated: float
+    voltage: Optional[float] = None
diff --git a/utils/measure/measure/powermeter/shelly.py b/utils/measure/measure/powermeter/shelly.py
@@ -8,7 +8,7 @@
 import requests
 
 from measure.powermeter.errors import ApiConnectionError, UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 _LOGGER = logging.getLogger("measure")
 
@@ -42,7 +42,9 @@ def __init__(self, ip_address: str, timeout: int) -> None:
     def endpoint(self) -> str:
         return self._endpoint
 
-    def parse_json(self, json: dict) -> PowerMeasurementResult:
+    def parse_json(self, json: dict, include_voltage: bool = False) -> PowerMeasurementResult:
+        if include_voltage:
+            return PowerMeasurementResult(power=float(json["apower"]), voltage=float(json["voltage"]), updated=time.time())
         return PowerMeasurementResult(float(json["apower"]), time.time())
 
     def check_gen2_plus_endpoints(self) -> None:
@@ -76,13 +78,6 @@ def _check_endpoint_availability(self, endpoint: str) -> bool:
 
         return True
 
-    def parse_json_with_voltage(self, json: dict) -> ExtendedPowerMeasurementResult:
-        return ExtendedPowerMeasurementResult(
-            power=float(json["apower"]),
-            voltage=float(json["voltage"]),
-            updated=time.time(),
-        )
-
 
 class ShellyPowerMeter(PowerMeter):
     def __init__(self, shelly_ip: str, timeout: int = 5) -> None:
@@ -95,7 +90,7 @@ def __init__(self, shelly_ip: str, timeout: int = 5) -> None:
             self.api = ShellyApiGen2Plus(self.ip_address, self.timeout)
             self.api.check_gen2_plus_endpoints()
 
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
         """Get a new power reading from the Shelly device. Optionally include voltage."""
         try:
             r = requests.get(
@@ -112,7 +107,7 @@ def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | E
             if isinstance(self.api, ShellyApiGen1):
                 raise UnsupportedFeatureError("Voltage measurement is not supported on Shelly Gen1 devices")
             if isinstance(self.api, ShellyApiGen2Plus):
-                return self.api.parse_json_with_voltage(json)
+                return self.api.parse_json(json, include_voltage=True)
 
         return self.api.parse_json(json)
 

diff --git a/utils/measure/measure/powermeter/tasmota.py b/utils/measure/measure/powermeter/tasmota.py
@@ -6,14 +6,14 @@
 import requests
 
 from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 
 class TasmotaPowerMeter(PowerMeter):
     def __init__(self, device_ip: str) -> None:
         self._device_ip = device_ip
 
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
         """Get a new power reading from the Tasmota device. Optionally include voltage (FIXME: not yet implemented)."""
         if include_voltage:
             # FIXME: Not yet implemented

diff --git a/utils/measure/measure/powermeter/tuya.py b/utils/measure/measure/powermeter/tuya.py
@@ -6,7 +6,7 @@
 import tuyapower
 
 from measure.powermeter.errors import PowerMeterError, UnsupportedFeatureError
-from measure.powermeter.powermeter import ExtendedPowerMeasurementResult, PowerMeasurementResult, PowerMeter
+from measure.powermeter.powermeter import PowerMeasurementResult, PowerMeter
 
 STATUS_OK = "OK"
 
@@ -24,7 +24,7 @@ def __init__(
         self._device_key = device_key
         self._device_version = device_version
 
-    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult | ExtendedPowerMeasurementResult:
+    def get_power(self, include_voltage: bool = False) -> PowerMeasurementResult:
         """Get a new power reading from the Tuya device. Optionally include voltage (FIXME: not yet implemented)."""
         if include_voltage:
             # FIXME: Not yet implemented