Breakout _collect_power_values() and related

Author: bouwew

plugwise/common.py

@@ -19,6 +19,7 @@
     check_heater_central,
     check_model,
     get_vendor_name,
+    power_data_local_format,
     return_valid,
 )
 
@@ -107,6 +108,77 @@ 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
+        combination = "log" in loc.log_type and ("gas" in loc.measurement or "phase" in loc.measurement)
+        if legacy:
+             combination = "meter" in loc.log_type and ("point" in loc.log_type or "gas" in loc.measurement)
+
+        if loc.logs.find(loc.locator) is None:
+            # If locator not found look for P1 gas_consumed or phase data (without tariff)
+            # For legacy look for P1 legacy electricity_point_meter or gas_*_meter data
+            if combination:
+                # 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 legacy:
+                    loc.locator = (
+                        f"./{loc.meas_list[0]}_{loc.log_type}/"
+                        f'measurement[@directionality="{loc.meas_list[1]}"]'
+                    )
+
+                if loc.logs.find(loc.locator) is None:
+                    loc.found = False
+                    return loc
+            else:
+                loc.found = False
+                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 _device_data_switching_group(
         self, device: DeviceData, data: DeviceData
     ) -> None:

plugwise/helper.py

@@ -55,12 +55,7 @@
     InvalidXMLError,
     ResponseError,
 )
-from plugwise.util import (
-    check_model,
-    escape_illegal_xml_characters,
-    format_measure,
-    power_data_local_format,
-)
+from plugwise.util import check_model, escape_illegal_xml_characters, format_measure
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientError, ClientResponse, ClientSession, ClientTimeout
@@ -558,68 +553,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,

plugwise/legacy/helper.py

@@ -43,7 +43,7 @@
     SwitchType,
     ThermoLoc,
 )
-from plugwise.util import format_measure, power_data_local_format, version_to_model
+from plugwise.util import format_measure, version_to_model
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from defusedxml import ElementTree as etree
@@ -310,7 +310,6 @@ def _power_data_from_modules(self) -> DeviceData:
         direct_data: DeviceData = {"sensors": {}}
         loc = Munch()
         mod_list: list[str] = ["interval_meter", "cumulative_meter", "point_meter"]
-        peak_list: list[str] = ["nl_peak", "nl_offpeak"]
         t_string = "tariff_indicator"
 
         search = self._modules
@@ -319,60 +318,11 @@ def _power_data_from_modules(self) -> DeviceData:
             loc.meas_list = loc.measurement.split("_")
             for loc.logs in mod_logs:
                 for loc.log_type in mod_list:
-                    for loc.peak_select in peak_list:
-                        loc.locator = (
-                            f"./{loc.meas_list[0]}_{loc.log_type}/measurement"
-                            f'[@directionality="{loc.meas_list[1]}"][@{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, legacy=True)
 
         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 for P1 legacy electricity_point_meter or gas_*_meter data
-        if loc.logs.find(loc.locator) is None:
-            if "meter" in loc.log_type and (
-                "point" in loc.log_type or "gas" 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.meas_list[0]}_{loc.log_type}/"
-                    f'measurement[@directionality="{loc.meas_list[1]}"]'
-                )
-                if loc.logs.find(loc.locator) is None:
-                    loc.found = False
-                    return loc
-            else:
-                loc.found = False
-                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}"
-        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,