Tasmota: use parser for meter types

packages/modules/devices/tasmota/tasmota/bat.py

@@ -12,6 +12,7 @@
 from modules.common.component_state import BatState
 from modules.common.utils.peak_filter import PeakFilter
 from modules.common.component_type import ComponentType
+from modules.devices.tasmota.tasmota.parser import TasmotaParser
 
 log = logging.getLogger(__name__)
 
@@ -40,47 +41,18 @@ def update(self):
         url = "http://" + self.__ip_address + "/cm?cmnd=Status%208"
         response = req.get_http_session().get(url, timeout=5).json()
 
-        currents = None
-        if 'ENERGY' in response['StatusSNS']:
-            currents = [0.0, 0.0, 0.0]
-
-            power = float(response['StatusSNS']['ENERGY']['Power'])
-            currents[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Current'])
-            imported = float(response['StatusSNS']['ENERGY']['Total']*1000)
-            imported, _ = self.peak_filter.check_values(power, imported, None)
+        data = TasmotaParser(response, self.__phase)
+        power = data.power
+        measured_exported = None if data.meter_type == 'ENERGY' else data.exported
+        imported, exported = self.peak_filter.check_values(power, data.imported, measured_exported)
+        if data.meter_type == 'ENERGY':
             _, exported = self.sim_counter.sim_count(power)
-        elif 'Itron' in response['StatusSNS']:
-            power = float(response['StatusSNS']['Itron']['Power'])
-            if 'E_in' in response['StatusSNS']['Itron']:
-                imported = float(response['StatusSNS']['Itron']['E_in']*1000)
-                exported = float(response['StatusSNS']['Itron']['E_out']*1000)
-            else:
-                imported = float(response['StatusSNS']['Itron']['ImportActive']*1000)
-                exported = float(response['StatusSNS']['Itron']['ExportActive']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        elif 'MT681' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT681']['Watt_summe'])
-            imported = float(response['StatusSNS']['MT681']['Total_in']*1000)
-            exported = float(response['StatusSNS']['MT681']['Total_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        elif 'eBZ' in response['StatusSNS']:
-            power = float(response['StatusSNS']['eBZ']['Power'])
-            imported = float(response['StatusSNS']['eBZ']['E_in']*1000)
-            exported = float(response['StatusSNS']['eBZ']['E_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        elif 'MT631' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT631']['Power'])
-            imported = float(response['StatusSNS']['MT631']['E_in']*1000)
-            exported = float(response['StatusSNS']['MT631']['E_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        else:
-            raise ValueError("Nicht unterstützter Tasmota Zählertyp. Bitte an den Support wenden.")
 
         bat_state = BatState(
             power=power,
             imported=imported,
             exported=exported,
-            currents=currents
+            currents=data.currents
         )
 
         self.store.set(bat_state)

packages/modules/devices/tasmota/tasmota/counter.py

@@ -12,6 +12,7 @@
 from modules.common.component_state import CounterState
 from modules.common.utils.peak_filter import PeakFilter
 from modules.common.component_type import ComponentType
+from modules.devices.tasmota.tasmota.parser import TasmotaParser
 
 log = logging.getLogger(__name__)
 
@@ -40,68 +41,21 @@ def update(self):
         url = "http://" + self.__ip_address + "/cm?cmnd=Status%208"
         response = req.get_http_session().get(url, timeout=5).json()
 
-        voltages = None
-        currents = None
-        powers = None
-        power_factors = None
-        if 'ENERGY' in response['StatusSNS']:
-            voltages = [0.0, 0.0, 0.0]
-            powers = [0.0, 0.0, 0.0]
-            currents = [0.0, 0.0, 0.0]
-            power_factors = [0.0, 0.0, 0.0]
-
-            power = float(response['StatusSNS']['ENERGY']['Power'])
-            voltages[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Voltage'])
-            powers[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Power'])
-            currents[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Current'])
-            power_factors[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Factor'])
-            imported = float(response['StatusSNS']['ENERGY']['Total']*1000)
-            imported, _ = self.peak_filter.check_values(power, imported, None)
+        data = TasmotaParser(response, self.__phase)
+        power = data.power
+        measured_exported = None if data.meter_type == 'ENERGY' else data.exported
+        imported, exported = self.peak_filter.check_values(power, data.imported, measured_exported)
+        if data.meter_type == 'ENERGY':
             _, exported = self.sim_counter.sim_count(power)
-        elif 'Itron' in response['StatusSNS']:
-            power = float(response['StatusSNS']['Itron']['Power'])
-            if 'E_in' in response['StatusSNS']['Itron']:
-                imported = float(response['StatusSNS']['Itron']['E_in']*1000)
-                exported = float(response['StatusSNS']['Itron']['E_out']*1000)
-            else:
-                imported = float(response['StatusSNS']['Itron']['ImportActive']*1000)
-                exported = float(response['StatusSNS']['Itron']['ExportActive']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        elif 'MT681' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT681']['Watt_summe'])
-            imported = float(response['StatusSNS']['MT681']['Total_in']*1000)
-            exported = float(response['StatusSNS']['MT681']['Total_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        elif 'eBZ' in response['StatusSNS']:
-            powers = [0.0, 0.0, 0.0]
-            power = float(response['StatusSNS']['eBZ']['Power'])
-            imported = float(response['StatusSNS']['eBZ']['E_in']*1000)
-            exported = float(response['StatusSNS']['eBZ']['E_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-            if (
-                '36_7_0' in response['StatusSNS']['eBZ'] and
-                '56_7_0' in response['StatusSNS']['eBZ'] and
-                '76_7_0' in response['StatusSNS']['eBZ']
-            ):
-                powers = [float(response['StatusSNS']['eBZ']['36_7_0']),
-                          float(response['StatusSNS']['eBZ']['56_7_0']),
-                          float(response['StatusSNS']['eBZ']['76_7_0'])]
-        elif 'MT631' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT631']['Power'])
-            imported = float(response['StatusSNS']['MT631']['E_in']*1000)
-            exported = float(response['StatusSNS']['MT631']['E_out']*1000)
-            imported, exported = self.peak_filter.check_values(power, imported, exported)
-        else:
-            raise ValueError("Nicht unterstützter Tasmota Zählertyp. Bitte an den Support wenden.")
 
         counter_state = CounterState(
             power=power,
             imported=imported,
             exported=exported,
-            voltages=voltages,
-            currents=currents,
-            powers=powers,
-            power_factors=power_factors
+            voltages=data.voltages,
+            currents=data.currents,
+            powers=data.powers,
+            power_factors=data.power_factors
         )
 
         self.store.set(counter_state)

packages/modules/devices/tasmota/tasmota/inverter.py

@@ -12,6 +12,7 @@
 from modules.common.component_state import InverterState
 from modules.common.utils.peak_filter import PeakFilter
 from modules.common.component_type import ComponentType
+from modules.devices.tasmota.tasmota.parser import TasmotaParser
 
 log = logging.getLogger(__name__)
 
@@ -40,39 +41,19 @@ def update(self):
         url = "http://" + self.__ip_address + "/cm?cmnd=Status%208"
         response = req.get_http_session().get(url, timeout=5).json()
 
-        currents = None
-        if 'ENERGY' in response['StatusSNS']:
-            currents = [0.0, 0.0, 0.0]
-            power = float(response['StatusSNS']['ENERGY']['Power']) * -1
-            currents[self.__phase-1] = float(response['StatusSNS']['ENERGY']['Current'])
+        data = TasmotaParser(response, self.__phase)
+        power = data.power * -1
+        exported = data.exported
+        if data.meter_type == 'ENERGY':
             self.peak_filter.check_values(power)
             _, exported = self.sim_counter.sim_count(power)
-        elif 'Itron' in response['StatusSNS']:
-            power = float(response['StatusSNS']['Itron']['Power']) * -1
-            if 'E_in' in response['StatusSNS']['Itron']:
-                exported = float(response['StatusSNS']['Itron']['E_out']*1000)
-            else:
-                exported = float(response['StatusSNS']['Itron']['ExportActive']*1000)
-            _, exported = self.peak_filter.check_values(power, None, exported)
-        elif 'MT681' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT681']['Watt_summe']) * -1
-            exported = float(response['StatusSNS']['MT681']['Total_out']*1000)
-            _, exported = self.peak_filter.check_values(power, None, exported)
-        elif 'eBZ' in response['StatusSNS']:
-            power = float(response['StatusSNS']['eBZ']['Power']) * -1
-            exported = float(response['StatusSNS']['eBZ']['E_out']*1000)
-            _, exported = self.peak_filter.check_values(power, None, exported)
-        elif 'MT631' in response['StatusSNS']:
-            power = float(response['StatusSNS']['MT631']['Power']) * -1
-            exported = float(response['StatusSNS']['MT631']['E_out']*1000)
-            _, exported = self.peak_filter.check_values(power, None, exported)
         else:
-            raise ValueError("Nicht unterstützter Tasmota Zählertyp. Bitte an den Support wenden.")
+            _, exported = self.peak_filter.check_values(power, None, exported)
 
         inverter_state = InverterState(
             power=power,
             exported=exported,
-            currents=currents
+            currents=data.currents
         )
 
         self.store.set(inverter_state)

packages/modules/devices/tasmota/tasmota/parser.py

@@ -0,0 +1,88 @@
+from typing import Dict, List, Optional
+
+from modules.common.component_state import ConsumerState, CounterState
+
+
+class TasmotaParser:
+    """Read only the requested measurements from a Status 8 or Status 0 response."""
+
+    def __init__(self, response: Dict, phase: int = 1) -> None:
+        self.phase = phase
+        self.relay_status = response.get('Status', {})
+        self.meter_type: Optional[str] = None
+        self.meter: Dict = {}
+        status = response.get('StatusSNS', {})
+        for meter_type in ('ENERGY', 'Itron', 'MT681', 'eBZ', 'MT631'):
+            if meter_type in status:
+                self.meter_type = meter_type
+                self.meter = status[meter_type]
+                break
+        else:
+            if 'Power' in self.relay_status:
+                return
+            raise ValueError("Nicht unterstützter Tasmota Zählertyp. Bitte an den Support wenden.")
+
+    def _phase_values(self, key: str, attribute: str) -> List[float]:
+        values = getattr(CounterState(), attribute)
+        if self.meter_type == 'ENERGY' and self.meter.get(key) is not None:
+            values[self.phase-1] = float(self.meter[key])
+        return values
+
+    @property
+    def currents(self) -> List[float]:
+        return self._phase_values('Current', 'currents')
+
+    @property
+    def exported(self) -> float:
+        if self.meter_type == 'ENERGY':
+            return CounterState().exported
+        if self.meter_type == 'MT681':
+            key = 'Total_out'
+        elif self.meter_type == 'Itron' and 'E_in' not in self.meter:
+            key = 'ExportActive'
+        else:
+            key = 'E_out'
+        return float(self.meter[key]) * 1000 if self.meter.get(key) is not None else CounterState().exported
+
+    @property
+    def imported(self) -> float:
+        if self.meter_type == 'ENERGY':
+            key = 'Total'
+        elif self.meter_type == 'MT681':
+            key = 'Total_in'
+        elif self.meter_type == 'Itron' and 'E_in' not in self.meter:
+            key = 'ImportActive'
+        else:
+            key = 'E_in'
+        return float(self.meter[key]) * 1000 if self.meter.get(key) is not None else CounterState().imported
+
+    @property
+    def power(self) -> float:
+        key = 'Watt_summe' if self.meter_type == 'MT681' else 'Power'
+        return float(self.meter[key]) if self.meter.get(key) is not None else CounterState().power
+
+    @property
+    def power_factors(self) -> List[float]:
+        return self._phase_values('Factor', 'power_factors')
+
+    @property
+    def powers(self) -> List[float]:
+        if self.meter_type == 'ENERGY':
+            return self._phase_values('Power', 'powers')
+        values = CounterState().powers
+        if self.meter_type == 'eBZ':
+            keys = ('36_7_0', '56_7_0', '76_7_0')
+            for index, key in enumerate(keys):
+                if self.meter.get(key) is not None:
+                    values[index] = float(self.meter[key])
+        return values
+
+    @property
+    def state(self) -> bool:
+        if self.relay_status.get('Power') is not None:
+            return bool(self.relay_status['Power'])
+        return ConsumerState().state
+
+    @property
+    def voltages(self) -> List[float]:
+        return self._phase_values('Voltage', 'voltages')