diff --git a/packages/modules/devices/tasmota/tasmota/bat.py b/packages/modules/devices/tasmota/tasmota/bat.py index 870d023ab8..4cf1e2801b 100644 --- a/packages/modules/devices/tasmota/tasmota/bat.py +++ b/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) diff --git a/packages/modules/devices/tasmota/tasmota/counter.py b/packages/modules/devices/tasmota/tasmota/counter.py index 44899c6d36..6458b0be97 100644 --- a/packages/modules/devices/tasmota/tasmota/counter.py +++ b/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) diff --git a/packages/modules/devices/tasmota/tasmota/inverter.py b/packages/modules/devices/tasmota/tasmota/inverter.py index f3dffcaaca..db5f77c74b 100644 --- a/packages/modules/devices/tasmota/tasmota/inverter.py +++ b/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) diff --git a/packages/modules/devices/tasmota/tasmota/parser.py b/packages/modules/devices/tasmota/tasmota/parser.py new file mode 100644 index 0000000000..35fc123729 --- /dev/null +++ b/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') diff --git a/packages/modules/devices/tasmota/tasmota/tasmota_test.py b/packages/modules/devices/tasmota/tasmota/tasmota_test.py new file mode 100644 index 0000000000..2aa518ac5a --- /dev/null +++ b/packages/modules/devices/tasmota/tasmota/tasmota_test.py @@ -0,0 +1,148 @@ +from unittest.mock import Mock + +import pytest + +from modules.common.component_state import ConsumerState, CounterState +from modules.devices.tasmota.tasmota import bat, counter, inverter +from modules.devices.tasmota.tasmota.config import TasmotaBatSetup, TasmotaCounterSetup, TasmotaInverterSetup +from modules.devices.tasmota.tasmota.parser import TasmotaParser + + +METER_DATA = [ + ('ENERGY', {'Power': 230, 'Current': 1, 'Voltage': 230, 'Factor': 1, 'Total': 12.5}), + ('Itron', {'Power': 230, 'E_in': 12.5, 'E_out': 2.5}), + ('Itron', {'Power': 230, 'ImportActive': 12.5, 'ExportActive': 2.5}), + ('MT681', {'Watt_summe': 230, 'Total_in': 12.5, 'Total_out': 2.5}), + ('eBZ', {'Power': 230, 'E_in': 12.5, 'E_out': 2.5}), + ('MT631', {'Power': 230, 'E_in': 12.5, 'E_out': 2.5}), +] +COMPONENTS = [(counter, counter.TasmotaCounter, TasmotaCounterSetup), + (bat, bat.TasmotaBat, TasmotaBatSetup), + (inverter, inverter.TasmotaInverter, TasmotaInverterSetup)] + + +def create_component(monkeypatch, component_spec, response, filtered=False): + module, component_class, setup = component_spec + monkeypatch.setattr(module, 'get_component_value_store', Mock(return_value=Mock())) + monkeypatch.setattr(module, 'SimCounter', Mock(return_value=Mock(sim_count=Mock(return_value=(100, 200))))) + peak_filter = Mock() + peak_filter.check_values.side_effect = lambda power, imported=None, exported=None: ( + (None, None) if filtered else (imported, exported)) + monkeypatch.setattr(module, 'PeakFilter', Mock(return_value=peak_filter)) + session = Mock() + session.get.return_value.json.return_value = response + monkeypatch.setattr(module.req, 'get_http_session', Mock(return_value=session)) + component = component_class(setup(), device_id=0, ip_address='192.0.2.1', phase=2) + component.initialize() + return component + + +@pytest.mark.parametrize('meter_type, values', METER_DATA) +@pytest.mark.parametrize('component_spec', COMPONENTS) +def test_component_measurements(monkeypatch, component_spec, meter_type, values): + component = create_component(monkeypatch, component_spec, {'StatusSNS': {meter_type: values}}) + component.update() + state = component.store.set.call_args.args[0] + assert state.power == (-230 if component_spec[0] is inverter else 230) + assert state.exported == (200 if meter_type == 'ENERGY' else 2500) + if component_spec[0] is not inverter: + assert state.imported == 12500 + assert state.currents == ([0, 1, 0] if meter_type == 'ENERGY' else [0, 0, 0]) + if meter_type == 'ENERGY': + component.sim_counter.sim_count.assert_called_once_with(state.power) + else: + component.sim_counter.sim_count.assert_not_called() + + +@pytest.mark.parametrize('component_spec', COMPONENTS) +def test_filtered_export_is_not_replaced_by_simulation(monkeypatch, component_spec): + component = create_component(monkeypatch, component_spec, + {'StatusSNS': {'MT631': {'Power': 230, 'E_in': 12.5, 'E_out': 2.5}}}, + filtered=True) + component.update() + assert component.store.set.call_args.args[0].exported is None + component.sim_counter.sim_count.assert_not_called() + + +@pytest.mark.parametrize('component_spec', COMPONENTS[1:]) +def test_unused_energy_fields_are_not_required(monkeypatch, component_spec): + values = {'Power': 230, 'Current': 1} + if component_spec[0] is bat: + values['Total'] = 12.5 + component = create_component(monkeypatch, component_spec, {'StatusSNS': {'ENERGY': values}}) + component.update() + assert component.store.set.call_args.args[0].currents == [0, 1, 0] + + +@pytest.mark.parametrize('phase', [1, 2, 3]) +def test_energy_phase_assignment(phase): + data = TasmotaParser({'StatusSNS': {'ENERGY': METER_DATA[0][1]}}, phase) + for attribute, expected in [('currents', 1), ('voltages', 230), ('powers', 230), ('power_factors', 1)]: + values = getattr(data, attribute) + assert values[phase-1] == expected + assert sum(values) == (690 if attribute == 'voltages' else expected) + + +@pytest.mark.parametrize('values, expected', [ + ({}, [0, 0, 0]), + ({'36_7_0': 10}, [10, 0, 0]), + ({'36_7_0': 10, '56_7_0': 20, '76_7_0': 30}, [10, 20, 30]), +]) +def test_ebz_phase_powers(values, expected): + assert TasmotaParser({'StatusSNS': {'eBZ': values}}).powers == expected + + +def test_inverter_does_not_require_import_value(monkeypatch): + component = create_component(monkeypatch, COMPONENTS[2], + {'StatusSNS': {'MT681': {'Watt_summe': 230, 'Total_out': 2.5}}}) + component.update() + assert component.store.set.call_args.args[0].exported == 2500 + + +def test_power_only_status_zero(): + data = TasmotaParser({'Status': {'Power': 1}, 'StatusSNS': {'ENERGY': {'Power': 230}}}) + assert data.power == 230 + + +def test_unsupported_meter(): + with pytest.raises(ValueError, match='Nicht unterstützter Tasmota Zählertyp'): + TasmotaParser({'StatusSNS': {'unknown': {}}}) + + +@pytest.mark.parametrize('meter_type', ['ENERGY', 'Itron', 'MT681', 'eBZ', 'MT631']) +@pytest.mark.parametrize('attribute', [ + 'power', 'imported', 'exported', 'currents', 'voltages', 'powers', 'power_factors', +]) +def test_missing_measurements_use_counter_state_defaults(meter_type, attribute): + data = TasmotaParser({'StatusSNS': {meter_type: {}}}) + assert getattr(data, attribute) == getattr(CounterState(), attribute) + + +def test_default_lists_are_independent(): + data = TasmotaParser({'StatusSNS': {'MT631': {}}}) + values = data.voltages + values[0] = 0 + assert data.voltages == CounterState().voltages + + +def test_missing_phase_measurements_use_counter_state_defaults(): + data = TasmotaParser({'StatusSNS': {'ENERGY': {'Voltage': 240}}}, phase=2) + expected = CounterState().voltages + expected[1] = 240 + assert data.voltages == expected + + +@pytest.mark.parametrize('power, expected', [(0, False), (1, True)]) +@pytest.mark.parametrize('status_sns', [None, {}, {'ENERGY': {'Power': 230}}]) +def test_relay_state(power, expected, status_sns): + response = {'Status': {'Power': power}} + if status_sns is not None: + response['StatusSNS'] = status_sns + data = TasmotaParser(response) + assert data.state is expected + assert data.power == (230 if status_sns and 'ENERGY' in status_sns else CounterState().power) + + +def test_missing_relay_state_uses_consumer_state_default(): + data = TasmotaParser({'StatusSNS': {'ENERGY': {'Power': 230}}}) + assert data.state == ConsumerState().state