Breakout skip_obsolete_measurements() to util.py

Author: bouwew

plugwise/helper.py

@@ -27,7 +27,6 @@
     LOCATIONS,
     LOGGER,
     NONE,
-    OBSOLETE_MEASUREMENTS,
     OFF,
     P1_MEASUREMENTS,
     SENSORS,
@@ -55,7 +54,12 @@
     InvalidXMLError,
     ResponseError,
 )
-from plugwise.util import check_model, escape_illegal_xml_characters, format_measure
+from plugwise.util import (
+    check_model,
+    escape_illegal_xml_characters,
+    format_measure,
+    skip_obsolete_measurements,
+)
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientError, ClientResponse, ClientSession, ClientTimeout
@@ -574,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

plugwise/legacy/helper.py

@@ -4,7 +4,6 @@
 """
 from __future__ import annotations
 
-import datetime as dt
 from typing import cast
 
 from plugwise.common import SmileCommon
@@ -23,7 +22,6 @@
     HEATER_CENTRAL_MEASUREMENTS,
     LIMITS,
     NONE,
-    OBSOLETE_MEASUREMENTS,
     P1_LEGACY_MEASUREMENTS,
     SENSORS,
     SPECIALS,
@@ -43,7 +41,7 @@
     SwitchType,
     ThermoLoc,
 )
-from plugwise.util import format_measure, version_to_model
+from plugwise.util import format_measure, skip_obsolete_measurements, version_to_model
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from defusedxml import ElementTree as etree
@@ -336,20 +334,8 @@ def _appliance_measurements(
                 if measurement == "domestic_hot_water_state":
                     continue
 
-                # 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  # pragma: no cover
+                if skip_obsolete_measurements(appliance, measurement):
+                    continue  # pragma: no cover
 
                 if new_name := getattr(attrs, ATTR_NAME, None):
                     measurement = new_name

plugwise/util.py

@@ -1,13 +1,15 @@
 """Plugwise protocol helpers."""
 from __future__ import annotations
 
+import datetime as dt
 import re
 
 from plugwise.constants import (
     ATTR_UNIT_OF_MEASUREMENT,
     ELECTRIC_POTENTIAL_VOLT,
     ENERGY_KILO_WATT_HOUR,
     HW_MODELS,
+    OBSOLETE_MEASUREMENTS,
     PERCENTAGE,
     POWER_WATT,
     SPECIAL_FORMAT,
@@ -128,6 +130,20 @@ def return_valid(value: etree | None, default: etree) -> etree:
     return value if value is not None else default
 
 
+def skip_obsolete_measurements(xml: etree, measurement: str) -> bool:
+    """Skipping known obsolete measurements."""
+    locator = f".//logs/point_log[type='{measurement}']/updated_date"
+    if (
+        measurement in OBSOLETE_MEASUREMENTS
+        and (updated_date_key := xml.find(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()
+        return int((date_2 - date_1).days) > 7
+
+
 # NOTE: this function version_to_model is shared between Smile and USB
 def version_to_model(version: str | None) -> str | None:
     """Translate hardware_version to device type."""