Merge pull request #519 from plugwise/improve-6

Again more improvements

Author: bouwew

CHANGELOG.md

@@ -1,8 +1,8 @@
 # Changelog
 
-## Ongoing
+## V0.37.1
 
-- General improving of code
+- Further optimization / deduplication of the refactored code.
 
 ## v0.37.0
 

plugwise/common.py

@@ -4,17 +4,23 @@
 """
 from __future__ import annotations
 
+from typing import cast
+
 from plugwise.constants import (
     ANNA,
+    SPECIAL_PLUG_TYPES,
     SWITCH_GROUP_TYPES,
+    ApplianceType,
     DeviceData,
     ModelData,
     SensorType,
 )
 from plugwise.util import (
+    check_alternative_location,
     check_heater_central,
     check_model,
     get_vendor_name,
+    power_data_local_format,
     return_valid,
 )
 
@@ -103,6 +109,103 @@ def _appl_thermostat_info(self, appl: Munch, xml_1: etree, xml_2: etree = None)
 
         return appl
 
+    def _collect_power_values(self, data: DeviceData, loc: Munch, tariff: str, legacy: bool = False) -> None:
+        """Something."""
+        for loc.peak_select in ("nl_peak", "nl_offpeak"):
+            loc.locator = (
+                f'./{loc.log_type}[type="{loc.measurement}"]/period/'
+                f'measurement[@{tariff}="{loc.peak_select}"]'
+            )
+            if legacy:
+                loc.locator = (
+                    f"./{loc.meas_list[0]}_{loc.log_type}/measurement"
+                    f'[@directionality="{loc.meas_list[1]}"][@{tariff}="{loc.peak_select}"]'
+                )
+
+            loc = self._power_data_peak_value(loc, legacy)
+            if not loc.found:
+                continue
+
+            data = self._power_data_energy_diff(
+                loc.measurement, loc.net_string, loc.f_val, data
+            )
+            key = cast(SensorType, loc.key_string)
+            data["sensors"][key] = loc.f_val
+
+    def _power_data_peak_value(self, loc: Munch, legacy: bool) -> Munch:
+        """Helper-function for _power_data_from_location() and _power_data_from_modules()."""
+        loc.found = True
+        if loc.logs.find(loc.locator) is None:
+            loc = check_alternative_location(loc, legacy)
+            if not loc.found:
+                return loc
+
+        if (peak := loc.peak_select.split("_")[1]) == "offpeak":
+            peak = "off_peak"
+        log_found = loc.log_type.split("_")[0]
+        loc.key_string = f"{loc.measurement}_{peak}_{log_found}"
+        if "gas" in loc.measurement or loc.log_type == "point_meter":
+            loc.key_string = f"{loc.measurement}_{log_found}"
+        # Only for P1 Actual -------------------#
+        if "phase" in loc.measurement:
+            loc.key_string = f"{loc.measurement}"
+        # --------------------------------------#
+        loc.net_string = f"net_electricity_{log_found}"
+        val = loc.logs.find(loc.locator).text
+        loc.f_val = power_data_local_format(loc.attrs, loc.key_string, val)
+
+        return loc
+
+    def _power_data_energy_diff(
+        self,
+        measurement: str,
+        net_string: SensorType,
+        f_val: float | int,
+        direct_data: DeviceData,
+    ) -> DeviceData:
+        """Calculate differential energy."""
+        if (
+            "electricity" in measurement
+            and "phase" not in measurement
+            and "interval" not in net_string
+        ):
+            diff = 1
+            if "produced" in measurement:
+                diff = -1
+            if net_string not in direct_data["sensors"]:
+                tmp_val: float | int = 0
+            else:
+                tmp_val = direct_data["sensors"][net_string]
+
+            if isinstance(f_val, int):
+                tmp_val += f_val * diff
+            else:
+                tmp_val += float(f_val * diff)
+                tmp_val = float(f"{round(tmp_val, 3):.3f}")
+
+            direct_data["sensors"][net_string] = tmp_val
+
+        return direct_data
+
+    def _create_gw_devices(self, appl: Munch) -> None:
+        """Helper-function for creating/updating gw_devices."""
+        self.gw_devices[appl.dev_id] = {"dev_class": appl.pwclass}
+        self._count += 1
+        for key, value in {
+            "firmware": appl.firmware,
+            "hardware": appl.hardware,
+            "location": appl.location,
+            "mac_address": appl.mac,
+            "model": appl.model,
+            "name": appl.name,
+            "zigbee_mac_address": appl.zigbee_mac,
+            "vendor": appl.vendor_name,
+        }.items():
+            if value is not None or key == "location":
+                appl_key = cast(ApplianceType, key)
+                self.gw_devices[appl.dev_id][appl_key] = value
+                self._count += 1
+
     def _device_data_switching_group(
         self, device: DeviceData, data: DeviceData
     ) -> None:
@@ -154,6 +257,22 @@ def _get_group_switches(self) -> dict[str, DeviceData]:
 
         return switch_groups
 
+    def _get_lock_state(self, xml: etree, data: DeviceData, stretch_v2: bool = False) -> None:
+        """Helper-function for _get_measurement_data().
+
+        Adam & Stretches: obtain the relay-switch lock state.
+        """
+        actuator = "actuator_functionalities"
+        func_type = "relay_functionality"
+        if stretch_v2:
+            actuator = "actuators"
+            func_type = "relay"
+        if xml.find("type").text not in SPECIAL_PLUG_TYPES:
+            locator = f"./{actuator}/{func_type}/lock"
+            if (found := xml.find(locator)) is not None:
+                data["switches"]["lock"] = found.text == "true"
+                self._count += 1
+
     def _get_module_data(
         self,
         xml_1: etree,
@@ -208,34 +327,3 @@ def _get_zigbee_data(self, module: etree, model_data: ModelData, legacy: bool) -
         elif (zb_node := module.find("./protocols/zig_bee_node")) is not None:
                 model_data["zigbee_mac_address"] = zb_node.find("mac_address").text
                 model_data["reachable"] = zb_node.find("reachable").text == "true"
-
-    def _power_data_energy_diff(
-        self,
-        measurement: str,
-        net_string: SensorType,
-        f_val: float | int,
-        direct_data: DeviceData,
-    ) -> DeviceData:
-        """Calculate differential energy."""
-        if (
-            "electricity" in measurement
-            and "phase" not in measurement
-            and "interval" not in net_string
-        ):
-            diff = 1
-            if "produced" in measurement:
-                diff = -1
-            if net_string not in direct_data["sensors"]:
-                tmp_val: float | int = 0
-            else:
-                tmp_val = direct_data["sensors"][net_string]
-
-            if isinstance(f_val, int):
-                tmp_val += f_val * diff
-            else:
-                tmp_val += float(f_val * diff)
-                tmp_val = float(f"{round(tmp_val, 3):.3f}")
-
-            direct_data["sensors"][net_string] = tmp_val
-
-        return direct_data

plugwise/helper.py

@@ -27,11 +27,9 @@
     LOCATIONS,
     LOGGER,
     NONE,
-    OBSOLETE_MEASUREMENTS,
     OFF,
     P1_MEASUREMENTS,
     SENSORS,
-    SPECIAL_PLUG_TYPES,
     SPECIALS,
     SWITCHES,
     TEMP_CELSIUS,
@@ -41,7 +39,6 @@
     ActuatorData,
     ActuatorDataType,
     ActuatorType,
-    ApplianceType,
     BinarySensorType,
     DeviceData,
     GatewayData,
@@ -61,7 +58,7 @@
     check_model,
     escape_illegal_xml_characters,
     format_measure,
-    power_data_local_format,
+    skip_obsolete_measurements,
 )
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
@@ -294,22 +291,7 @@ def _all_appliances(self) -> None:
             if appl.pwclass in THERMOSTAT_CLASSES and appl.location is None:
                 continue
 
-            self.gw_devices[appl.dev_id] = {"dev_class": appl.pwclass}
-            self._count += 1
-            for key, value in {
-                "firmware": appl.firmware,
-                "hardware": appl.hardware,
-                "location": appl.location,
-                "mac_address": appl.mac,
-                "model": appl.model,
-                "name": appl.name,
-                "zigbee_mac_address": appl.zigbee_mac,
-                "vendor": appl.vendor_name,
-            }.items():
-                if value is not None or key == "location":
-                    appl_key = cast(ApplianceType, key)
-                    self.gw_devices[appl.dev_id][appl_key] = value
-                    self._count += 1
+            self._create_gw_devices(appl)
 
         # For P1 collect the connected SmartMeter info
         if self.smile_type == "power":
@@ -356,23 +338,7 @@ def _p1_smartmeter_info_finder(self, appl: Munch) -> None:
         location = self._domain_objects.find(f'./location[@id="{loc_id}"]')
         appl = self._energy_device_info_finder(appl, location)
 
-        self.gw_devices[appl.dev_id] = {"dev_class": appl.pwclass}
-        self._count += 1
-
-        for key, value in {
-            "firmware": appl.firmware,
-            "hardware": appl.hardware,
-            "location": appl.location,
-            "mac_address": appl.mac,
-            "model": appl.model,
-            "name": appl.name,
-            "zigbee_mac_address": appl.zigbee_mac,
-            "vendor": appl.vendor_name,
-        }.items():
-            if value is not None or key == "location":
-                p1_key = cast(ApplianceType, key)
-                self.gw_devices[appl.dev_id][p1_key] = value
-                self._count += 1
+        self._create_gw_devices(appl)
 
     def _appliance_info_finder(self, appl: Munch, appliance: etree) -> Munch:
         """Collect device info (Smile/Stretch, Thermostats, OpenTherm/On-Off): firmware, model and vendor name."""
@@ -591,68 +557,17 @@ def _power_data_from_location(self, loc_id: str) -> DeviceData:
         direct_data: DeviceData = {"sensors": {}}
         loc = Munch()
         log_list: list[str] = ["point_log", "cumulative_log", "interval_log"]
-        peak_list: list[str] = ["nl_peak", "nl_offpeak"]
         t_string = "tariff"
 
         search = self._domain_objects
         loc.logs = search.find(f'./location[@id="{loc_id}"]/logs')
         for loc.measurement, loc.attrs in P1_MEASUREMENTS.items():
             for loc.log_type in log_list:
-                for loc.peak_select in peak_list:
-                    loc.locator = (
-                        f'./{loc.log_type}[type="{loc.measurement}"]/period/'
-                        f'measurement[@{t_string}="{loc.peak_select}"]'
-                    )
-                    loc = self._power_data_peak_value(loc)
-                    if not loc.found:
-                        continue
-
-                    direct_data = self._power_data_energy_diff(
-                        loc.measurement, loc.net_string, loc.f_val, direct_data
-                    )
-                    key = cast(SensorType, loc.key_string)
-                    direct_data["sensors"][key] = loc.f_val
+                self._collect_power_values(direct_data, loc, t_string)
 
         self._count += len(direct_data["sensors"])
         return direct_data
 
-    def _power_data_peak_value(self, loc: Munch) -> Munch:
-        """Helper-function for _power_data_from_location() and _power_data_from_modules()."""
-        loc.found = True
-        # If locator not found look for P1 gas_consumed or phase data (without tariff)
-        if loc.logs.find(loc.locator) is None:
-            if "log" in loc.log_type and (
-                "gas" in loc.measurement or "phase" in loc.measurement
-            ):
-                # Avoid double processing by skipping one peak-list option
-                if loc.peak_select == "nl_offpeak":
-                    loc.found = False
-                    return loc
-
-                loc.locator = (
-                    f'./{loc.log_type}[type="{loc.measurement}"]/period/measurement'
-                )
-                if loc.logs.find(loc.locator) is None:
-                    loc.found = False
-                    return loc
-            else:
-                loc.found = False  # pragma: no cover
-                return loc  # pragma: no cover
-
-        if (peak := loc.peak_select.split("_")[1]) == "offpeak":
-            peak = "off_peak"
-        log_found = loc.log_type.split("_")[0]
-        loc.key_string = f"{loc.measurement}_{peak}_{log_found}"
-        if "gas" in loc.measurement or loc.log_type == "point_meter":
-            loc.key_string = f"{loc.measurement}_{log_found}"
-        if "phase" in loc.measurement:
-            loc.key_string = f"{loc.measurement}"
-        loc.net_string = f"net_electricity_{log_found}"
-        val = loc.logs.find(loc.locator).text
-        loc.f_val = power_data_local_format(loc.attrs, loc.key_string, val)
-
-        return loc
-
     def _appliance_measurements(
         self,
         appliance: etree,
@@ -663,20 +578,8 @@ def _appliance_measurements(
         for measurement, attrs in measurements.items():
             p_locator = f'.//logs/point_log[type="{measurement}"]/period/measurement'
             if (appl_p_loc := appliance.find(p_locator)) is not None:
-                # Skip known obsolete measurements
-                updated_date_locator = (
-                    f'.//logs/point_log[type="{measurement}"]/updated_date'
-                )
-                if (
-                    measurement in OBSOLETE_MEASUREMENTS
-                    and (updated_date_key := appliance.find(updated_date_locator))
-                    is not None
-                ):
-                    updated_date = updated_date_key.text.split("T")[0]
-                    date_1 = dt.datetime.strptime(updated_date, "%Y-%m-%d")
-                    date_2 = dt.datetime.now()
-                    if int((date_2 - date_1).days) > 7:
-                        continue
+                if skip_obsolete_measurements(appliance, measurement):
+                    continue
 
                 if new_name := getattr(attrs, ATTR_NAME, None):
                     measurement = new_name
@@ -719,19 +622,6 @@ def _appliance_measurements(
         # Don't count the above top-level dicts, only the remaining single items
         self._count += len(data) - 3
 
-    def _get_lock_state(self, xml: etree, data: DeviceData) -> None:
-        """Helper-function for _get_measurement_data().
-
-        Adam & Stretches: obtain the relay-switch lock state.
-        """
-        actuator = "actuator_functionalities"
-        func_type = "relay_functionality"
-        if xml.find("type").text not in SPECIAL_PLUG_TYPES:
-            locator = f"./{actuator}/{func_type}/lock"
-            if (found := xml.find(locator)) is not None:
-                data["switches"]["lock"] = found.text == "true"
-                self._count += 1
-
     def _get_toggle_state(
         self, xml: etree, toggle: str, name: ToggleNameType, data: DeviceData
     ) -> None: