Merge remote-tracking branch 'openWB/master' into changes_snaptec_repo

packages/modules/devices/kostal_piko_old/config.py

@@ -0,0 +1,40 @@
+from typing import Optional
+from helpermodules.auto_str import auto_str
+from modules.common.component_setup import ComponentSetup
+
+
+@auto_str
+class KostalPikoOldConfiguration:
+    def __init__(self, ip_address: Optional[str] = None, user: Optional[str] = None, password: Optional[str] = None):
+        self.ip_address = ip_address
+        self.user = user
+        self.password = password
+
+
+@auto_str
+class KostalPikoOld:
+    def __init__(self,
+                 name: str = "Kostal Piko (alte Generation)",
+                 type: str = "kostal_piko_old",
+                 id: int = 0,
+                 configuration: KostalPikoOldConfiguration = None) -> None:
+        self.name = name
+        self.type = type
+        self.id = id
+        self.configuration = configuration or KostalPikoOldConfiguration()
+
+
+@auto_str
+class KostalPikoOldInverterConfiguration:
+    def __init__(self):
+        pass
+
+
+@auto_str
+class KostalPikoOldInverterSetup(ComponentSetup[KostalPikoOldInverterConfiguration]):
+    def __init__(self,
+                 name: str = "Kostal Piko (alte Generation) Wechselrichter",
+                 type: str = "inverter",
+                 id: int = 0,
+                 configuration: KostalPikoOldInverterConfiguration = None) -> None:
+        super().__init__(name, type, id, configuration or KostalPikoOldInverterConfiguration())

packages/modules/devices/kostal_piko_old/device.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+import logging
+from typing import Iterable, Optional, List
+
+from helpermodules.cli import run_using_positional_cli_args
+from modules.common import req
+from modules.common.abstract_device import DeviceDescriptor
+from modules.common.configurable_device import ConfigurableDevice, ComponentFactoryByType, MultiComponentUpdater
+from modules.devices.kostal_piko_old import inverter
+from modules.devices.kostal_piko_old.config import KostalPikoOld, KostalPikoOldConfiguration, KostalPikoOldInverterSetup
+from modules.devices.kostal_piko_old.inverter import KostalPikoOldInverter
+
+log = logging.getLogger(__name__)
+
+
+def create_device(device_config: KostalPikoOld):
+    def create_inverter_component(component_config: KostalPikoOldInverterSetup):
+        return KostalPikoOldInverter(component_config)
+
+    def update_components(components: Iterable[KostalPikoOldInverter]):
+        response = req.get_http_session().get(device_config.configuration.ip_address, verify=False, auth=(
+            device_config.configuration.user, device_config.configuration.password), timeout=5).text
+        for component in components:
+            component.update(response)
+
+    return ConfigurableDevice(
+        device_config=device_config,
+        component_factory=ComponentFactoryByType(
+            inverter=create_inverter_component,
+        ),
+        component_updater=MultiComponentUpdater(update_components)
+    )
+
+
+COMPONENT_TYPE_TO_MODULE = {
+    "inverter": inverter
+}
+
+
+def read_legacy(component_type: str, ip_address: str, user: str, password: str, num: Optional[int]) -> None:
+    device_config = KostalPikoOld(
+        configuration=KostalPikoOldConfiguration(ip_address=ip_address, user=user, password=password))
+    dev = create_device(device_config)
+    if component_type in COMPONENT_TYPE_TO_MODULE:
+        component_config = COMPONENT_TYPE_TO_MODULE[component_type].component_descriptor.configuration_factory()
+    else:
+        raise Exception(
+            "illegal component type " + component_type + ". Allowed values: " +
+            ','.join(COMPONENT_TYPE_TO_MODULE.keys())
+        )
+    component_config.id = num
+    dev.add_component(component_config)
+
+    log.debug('KostalPikoOld IP-Adresse: ' + ip_address)
+    log.debug('KostalPikoOld user: ' + user)
+    log.debug('KostalPikoOld Passwort: ' + password)
+
+    dev.update()
+
+
+def main(argv: List[str]):
+    run_using_positional_cli_args(read_legacy, argv)
+
+
+device_descriptor = DeviceDescriptor(configuration_factory=KostalPikoOld)

packages/modules/devices/kostal_piko_old/inverter.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+import logging
+import re
+
+from modules.common.component_state import InverterState
+from modules.common.component_type import ComponentDescriptor
+from modules.common.fault_state import ComponentInfo
+from modules.common.store import get_inverter_value_store
+from modules.devices.kostal_piko_old.config import KostalPikoOldInverterSetup
+
+log = logging.getLogger(__name__)
+
+
+class KostalPikoOldInverter:
+    def __init__(self, component_config: KostalPikoOldInverterSetup) -> None:
+        self.component_config = component_config
+        self.store = get_inverter_value_store(self.component_config.id)
+        self.component_info = ComponentInfo.from_component_config(self.component_config)
+
+    def update(self, response) -> None:
+        # power may be a string "xxx" when the inverter is offline, so we cannot match as a number
+        # state is just for debugging currently known states:
+        # - Aus
+        # - Leerlauf
+        result = re.search(
+            r"aktuell</td>\s*<td[^>]*>\s*([^<]+).*"
+            r"Gesamtenergie</td>\s*<td[^>]*>\s*(\d+).*"
+            r"Status</td>\s*<td[^>]*>\s*([^<]+)",
+            response,
+            re.DOTALL
+        )
+        if result is None:
+            raise Exception("Given HTML does not match the expected regular expression. Ignoring.")
+        log.debug("Inverter data: state=%s, power=%s, exported=%s" %
+                  (result.group(3), result.group(1), result.group(2)))
+        try:
+            power = -int(result.group(1))
+        except ValueError:
+            log.info("Inverter power is not a number! Inverter may be offline. Setting power to 0 W.")
+            power = 0
+        inverter_state = InverterState(
+            exported=int(result.group(2)) * 1000,
+            power=power
+        )
+        self.store.set(inverter_state)
+
+
+component_descriptor = ComponentDescriptor(configuration_factory=KostalPikoOldInverterSetup)

packages/modules/devices/kostal_piko_old/inverter_test.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+from modules.common.component_state import InverterState
+
+from modules.devices.kostal_piko_old import inverter
+from modules.devices.kostal_piko_old.config import KostalPikoOldInverterSetup
+
+
+@pytest.mark.parametrize("sample_file_name, expected_inverter_state",
+                         [pytest.param("sample.html", InverterState(power=-50, exported=73288000), id="Inverter on"),
+                          pytest.param("sample_off.html", InverterState(
+                              power=0, exported=42906000), id="Inverter off")]
+                         )
+def test_parse_html(sample_file_name, expected_inverter_state, monkeypatch):
+    # setup
+    sample = (Path(__file__).parent / sample_file_name).read_text()
+    mock_inverter_value_store = Mock()
+    monkeypatch.setattr(inverter, 'get_inverter_value_store', Mock(return_value=mock_inverter_value_store))
+    inv = inverter.KostalPikoOldInverter(KostalPikoOldInverterSetup())
+
+    # execution
+    inv.update(sample)
+
+    # evaluation
+    assert mock_inverter_value_store.set.call_count == 1
+    assert vars(mock_inverter_value_store.set.call_args[0][0]) == vars(expected_inverter_state)