Smile.py: import _find and _findall and implement

Author: bouwew

plugwise/smile.py

@@ -45,7 +45,7 @@
     ResponseError,
     UnsupportedDeviceError,
 )
-from .helper import SmileComm, SmileHelper, update_helper
+from .helper import SmileComm, SmileHelper, _find, _findall, update_helper
 
 
 class SmileData(SmileHelper):
@@ -129,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
@@ -143,10 +143,10 @@ 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
 
@@ -351,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] = []
@@ -365,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 \
@@ -392,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
@@ -443,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)
 
@@ -503,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
@@ -563,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:
@@ -578,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}"
@@ -626,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)
 
@@ -655,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)
@@ -671,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 = (
@@ -715,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:
@@ -735,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}"
@@ -774,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:
@@ -793,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)