Merge pull request #244 from plugwise/pw_0_29_10

Various bugfixes 2

Author: bouwew

CHANGELOG.md

@@ -1,8 +1,16 @@
 # Changelog
 
-# v0.26.9: Hide cooling-related switch, binary_sensors when there is no cooling present
+# v0.25.10: Thermostats: more improvements
+- Anna + Elga: hide cooling_enable switch, (hardware-)switch is on Elga, not in Plugwise App
+- Adam: improve collecting regulation_mode-related data. Fix for https://github.com/plugwise/python-plugwise/issues/240
+- Anna: remove device availability, fix for https://github.com/home-assistant/core/issues/81716
+- Anna + OnOff device: fix incorrect heating-state, fix for https://github.com/home-assistant/core/issues/81839
+- Improve handling of xml-data missing, raise exception with warning. Solution for https://github.com/home-assistant/core/issues/81672
+- Improve handling of empty schedule, fix for https://github.com/plugwise/python-plugwise/issues/241
+
+# v0.25.9: Adam: hide cooling-related switch, binary_sensors when there is no cooling present
 - This fixes the unexpected appearance of new entities after the Adam 3.7.1 firmware-update
-- Properly handle an empty schedule, should fix #313
+- Properly handle an empty schedule, should fix https://github.com/plugwise/plugwise-beta/issues/313
 
 # v0.25.8: Make collection of toggle-data future-proof
 

plugwise/__init__.py

@@ -1,6 +1,6 @@
 """Plugwise module."""
 
-__version__ = "0.25.9"
+__version__ = "0.25.10"
 
 from plugwise.smile import Smile
 from plugwise.stick import Stick

plugwise/constants.py

@@ -471,7 +471,6 @@
     "electricity_consumed": UOM(POWER_WATT),
     "electricity_produced": UOM(POWER_WATT),
     "relay": UOM(NONE),
-    "regulation_mode": UOM(NONE),
 }
 
 # Heater Central related measurements

plugwise/helper.py

@@ -4,7 +4,6 @@
 from __future__ import annotations
 
 import aiohttp
-from dateutil.parser import parse
 from defusedxml import ElementTree as etree
 
 # Dict as class
@@ -46,7 +45,7 @@
     ResponseError,
     UnsupportedDeviceError,
 )
-from .helper import SmileComm, SmileHelper, update_helper
+from .helper import SmileComm, SmileHelper, _find, _findall, update_helper
 
 
 class SmileData(SmileHelper):
@@ -130,11 +129,11 @@ def get_all_devices(self) -> None:
         # Start by determining the system capabilities:
         # Find the connected heating/cooling device (heater_central), e.g. heat-pump or gas-fired heater
         if self.smile_type == "thermostat":
-            onoff_boiler: etree = self._domain_objects.find(
-                "./module/protocols/onoff_boiler"
+            onoff_boiler: etree = _find(
+                self._domain_objects, "./module/protocols/onoff_boiler"
             )
-            open_therm_boiler: etree = self._domain_objects.find(
-                "./module/protocols/open_therm_boiler"
+            open_therm_boiler: etree = _find(
+                self._domain_objects, "./module/protocols/open_therm_boiler"
             )
             self._on_off_device = onoff_boiler is not None
             self._opentherm_device = open_therm_boiler is not None
@@ -144,11 +143,12 @@ def get_all_devices(self) -> None:
             locator_2 = "./gateway/features/elga_support"
             search = self._domain_objects
             self._cooling_present = False
-            if search.find(locator_1) is not None:
+            if _find(search, locator_1) is not None:
                 self._cooling_present = True
             # Alternative method for the Anna with Elga
-            elif search.find(locator_2) is not None:
+            elif _find(search, locator_2) is not None:
                 self._cooling_present = True
+                self._elga = True
 
         # Gather all the device and initial data
         self._scan_thermostats()
@@ -267,23 +267,6 @@ def _check_availability(
                     if "P1 does not seem to be connected to a smart meter" in msg:
                         device_data["available"] = False
 
-        # Anna thermostat
-        if "modified" in device_data:
-            time_now: str | None = None
-            if (
-                time_now := self._domain_objects.find("./gateway/time").text
-            ) is not None:
-                interval = (
-                    parse(time_now) - parse(device_data["modified"])
-                ).total_seconds()
-                if interval > 0:
-                    if details["dev_class"] == "thermostat":
-                        device_data["available"] = False
-                        if interval < 90:
-                            device_data["available"] = True
-
-            device_data.pop("modified")
-
     def _get_device_data(self, dev_id: str) -> DeviceData:
         """Helper-function for _all_device_data() and async_update().
         Provide device-data, based on Location ID (= dev_id), from APPLIANCES.
@@ -368,12 +351,12 @@ def __init__(
     async def connect(self) -> bool:
         """Connect to Plugwise device and determine its name, type and version."""
         result = await self._request(DOMAIN_OBJECTS)
-        vendor_names: list[etree] = result.findall("./module/vendor_name")
-        vendor_models: list[etree] = result.findall("./module/vendor_model")
+        vendor_names: list[etree] = _findall(result, "./module/vendor_name")
+        vendor_models: list[etree] = _findall(result, "./module/vendor_model")
         # Work-around for Stretch fv 2.7.18
         if not vendor_names:
             result = await self._request(MODULES)
-            vendor_names = result.findall("./module/vendor_name")
+            vendor_names = _findall(result, "./module/vendor_name")
 
         names: list[str] = []
         models: list[str] = []
@@ -382,7 +365,7 @@ async def connect(self) -> bool:
         for model in vendor_models:
             models.append(model.text)
 
-        dsmrmain = result.find("./module/protocols/dsmrmain")
+        dsmrmain = _find(result, "./module/protocols/dsmrmain")
         if "Plugwise" not in names and dsmrmain is None:  # pragma: no cover
             LOGGER.error(
                 "Connected but expected text not returned, we got %s. Please create \
@@ -409,40 +392,42 @@ async def connect(self) -> bool:
     async def _smile_detect_legacy(self, result: etree, dsmrmain: etree) -> str:
         """Helper-function for _smile_detect()."""
         # Stretch: find the MAC of the zigbee master_controller (= Stick)
-        if network := result.find("./module/protocols/master_controller"):
-            self.smile_zigbee_mac_address = network.find("mac_address").text
+        if network := _find(result, "./module/protocols/master_controller"):
+            self.smile_zigbee_mac_address = _find(network, "mac_address").text
         # Find the active MAC in case there is an orphaned Stick
-        if zb_networks := result.findall("./network"):
+        if zb_networks := _findall(result, "./network"):
             for zb_network in zb_networks:
-                if zb_network.find("./nodes/network_router"):
-                    network = zb_network.find("./master_controller")
-                    self.smile_zigbee_mac_address = network.find("mac_address").text
+                if _find(zb_network, "./nodes/network_router"):
+                    network = _find(zb_network, "./master_controller")
+                    self.smile_zigbee_mac_address = _find(network, "mac_address").text
 
         # Assume legacy
         self._smile_legacy = True
         # Try if it is a legacy Anna, assuming appliance thermostat,
         # fake insert version assuming Anna, couldn't find another way to identify as legacy Anna
         self.smile_fw_version = "1.8.0"
         model = "smile_thermo"
-        if result.find('./appliance[type="thermostat"]') is None:
+        if _find(result, './appliance[type="thermostat"]') is None:
             # It's a P1 legacy:
             if dsmrmain is not None:
                 self._status = await self._request(STATUS)
-                self.smile_fw_version = self._status.find("./system/version").text
-                model = self._status.find("./system/product").text
-                self.smile_hostname = self._status.find("./network/hostname").text
-                self.smile_mac_address = self._status.find("./network/mac_address").text
+                self.smile_fw_version = _find(self._status, "./system/version").text
+                model = _find(self._status, "./system/product").text
+                self.smile_hostname = _find(self._status, "./network/hostname").text
+                self.smile_mac_address = _find(
+                    self._status, "./network/mac_address"
+                ).text
 
             # Or a legacy Stretch:
             elif network is not None:
                 self._system = await self._request(SYSTEM)
-                self.smile_fw_version = self._system.find("./gateway/firmware").text
-                model = self._system.find("./gateway/product").text
-                self.smile_hostname = self._system.find("./gateway/hostname").text
+                self.smile_fw_version = _find(self._system, "./gateway/firmware").text
+                model = _find(self._system, "./gateway/product").text
+                self.smile_hostname = _find(self._system, "./gateway/hostname").text
                 # If wlan0 contains data it's active, so eth0 should be checked last
                 for network in ("wlan0", "eth0"):
                     locator = f"./{network}/mac"
-                    if (net_locator := self._system.find(locator)) is not None:
+                    if (net_locator := _find(self._system, locator)) is not None:
                         self.smile_mac_address = net_locator.text
 
             else:  # pragma: no cover
@@ -460,12 +445,12 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None:
         Detect which type of Smile is connected.
         """
         model: str | None = None
-        if (gateway := result.find("./gateway")) is not None:
-            model = gateway.find("vendor_model").text
-            self.smile_fw_version = gateway.find("firmware_version").text
-            self.smile_hw_version = gateway.find("hardware_version").text
-            self.smile_hostname = gateway.find("hostname").text
-            self.smile_mac_address = gateway.find("mac_address").text
+        if (gateway := _find(result, "./gateway")) is not None:
+            model = _find(gateway, "vendor_model").text
+            self.smile_fw_version = _find(gateway, "firmware_version").text
+            self.smile_hw_version = _find(gateway, "hardware_version").text
+            self.smile_hostname = _find(gateway, "hostname").text
+            self.smile_mac_address = _find(gateway, "mac_address").text
         else:
             model = await self._smile_detect_legacy(result, dsmrmain)
 
@@ -520,11 +505,11 @@ async def _update_domain_objects(self) -> None:
 
         # If Plugwise notifications present:
         self._notifications = {}
-        for notification in self._domain_objects.findall("./notification"):
+        for notification in _findall(self._domain_objects, "./notification"):
             try:
                 msg_id = notification.attrib["id"]
-                msg_type = notification.find("type").text
-                msg = notification.find("message").text
+                msg_type = _find(notification, "type").text
+                msg = _find(notification, "message").text
                 self._notifications.update({msg_id: {msg_type: msg}})
                 LOGGER.debug("Plugwise notifications: %s", self._notifications)
             except AttributeError:  # pragma: no cover
@@ -580,8 +565,8 @@ async def _set_schedule_state_legacy(
     ) -> None:
         """Helper-function for set_schedule_state()."""
         schedule_rule_id: str | None = None
-        for rule in self._domain_objects.findall("rule"):
-            if rule.find("name").text == name:
+        for rule in _findall(self._domain_objects, "rule"):
+            if _find(rule, "name").text == name:
                 schedule_rule_id = rule.attrib["id"]
 
         if schedule_rule_id is None:
@@ -595,7 +580,7 @@ async def _set_schedule_state_legacy(
             return
 
         locator = f'.//*[@id="{schedule_rule_id}"]/template'
-        for rule in self._domain_objects.findall(locator):
+        for rule in _findall(self._domain_objects, locator):
             template_id = rule.attrib["id"]
 
         uri = f"{RULES};id={schedule_rule_id}"
@@ -643,13 +628,13 @@ async def set_schedule_state(
         )
         if self.smile_name != "Adam":
             locator = f'.//*[@id="{schedule_rule_id}"]/template'
-            template_id = self._domain_objects.find(locator).attrib["id"]
+            template_id = _find(self._domain_objects, locator).attrib["id"]
             template = f'<template id="{template_id}" />'
 
         locator = f'.//*[@id="{schedule_rule_id}"]/contexts'
-        contexts = self._domain_objects.find(locator)
+        contexts = _find(self._domain_objects, locator)
         locator = f'.//*[@id="{loc_id}"].../...'
-        if (subject := contexts.find(locator)) is None:
+        if (subject := _find(contexts, locator)) is None:
             subject = f'<context><zone><location id="{loc_id}" /></zone></context>'
             subject = etree.fromstring(subject)
 
@@ -672,7 +657,7 @@ async def set_schedule_state(
     async def _set_preset_legacy(self, preset: str) -> None:
         """Set the given Preset on the relevant Thermostat - from DOMAIN_OBJECTS."""
         locator = f'rule/directives/when/then[@icon="{preset}"].../.../...'
-        rule = self._domain_objects.find(locator)
+        rule = _find(self._domain_objects, locator)
         data = f'<rules><rule id="{rule.attrib["id"]}"><active>true</active></rule></rules>'
 
         await self._request(RULES, method="put", data=data)
@@ -688,9 +673,9 @@ async def set_preset(self, loc_id: str, preset: str) -> None:
             await self._set_preset_legacy(preset)
             return
 
-        current_location = self._locations.find(f'location[@id="{loc_id}"]')
-        location_name = current_location.find("name").text
-        location_type = current_location.find("type").text
+        current_location = _find(self._locations, f'location[@id="{loc_id}"]')
+        location_name = _find(current_location, "name").text
+        location_type = _find(current_location, "type").text
 
         uri = f"{LOCATIONS};id={loc_id}"
         data = (
@@ -732,9 +717,9 @@ async def set_number_setpoint(self, key: str, temperature: float) -> None:
         temp = str(temperature)
         thermostat_id: str | None = None
         locator = f'appliance[@id="{self._heater_id}"]/actuator_functionalities/thermostat_functionality'
-        if th_func_list := self._appliances.findall(locator):
+        if th_func_list := _findall(self._appliances, locator):
             for th_func in th_func_list:
-                if th_func.find("type").text == key:
+                if _find(th_func, "type").text == key:
                     thermostat_id = th_func.attrib["id"]
 
         if thermostat_id is None:
@@ -752,7 +737,7 @@ async def _set_groupswitch_member_state(
         """
         for member in members:
             locator = f'appliance[@id="{member}"]/{switch.actuator}/{switch.func_type}'
-            switch_id = self._appliances.find(locator).attrib["id"]
+            switch_id = _find(self._appliances, locator).attrib["id"]
             uri = f"{APPLIANCES};id={member}/{switch.device};id={switch_id}"
             if self._stretch_v2:
                 uri = f"{APPLIANCES};id={member}/{switch.device}"
@@ -791,9 +776,9 @@ async def set_switch_state(
             return await self._set_groupswitch_member_state(members, state, switch)
 
         locator = f'appliance[@id="{appl_id}"]/{switch.actuator}/{switch.func_type}'
-        found: list[etree] = self._appliances.findall(locator)
+        found: list[etree] = _findall(self._appliances, locator)
         for item in found:
-            if (sw_type := item.find("type")) is not None:
+            if (sw_type := _find(item, "type")) is not None:
                 if sw_type.text == switch.act_type:
                     switch_id = item.attrib["id"]
             else:
@@ -810,7 +795,7 @@ async def set_switch_state(
                 f'appliance[@id="{appl_id}"]/{switch.actuator}/{switch.func_type}/lock'
             )
             # Don't bother switching a relay when the corresponding lock-state is true
-            if self._appliances.find(locator).text == "true":
+            if _find(self._appliances, locator).text == "true":
                 raise PlugwiseError("Plugwise: the locked Relay was not switched.")
 
         await self._request(uri, method="put", data=data)