Merge pull request #839 from plugwise/xml-get

Optimize 3

Author: bouwew

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Ongoing
 
-- Code optimizations via PR [#837](https://github.com/plugwise/python-plugwise/pull/837), [#838](https://github.com/plugwise/python-plugwise/pull/838)
+- Code optimizations via PR [#837](https://github.com/plugwise/python-plugwise/pull/837), [#838](https://github.com/plugwise/python-plugwise/pull/838), [#839](https://github.com/plugwise/python-plugwise/pull/839)
 
 ## v1.11.0
 

plugwise/common.py

@@ -188,29 +188,28 @@ def _entity_switching_group(self, entity: GwEntityData) -> None:
             entity["switches"]["relay"] = counter != 0
             self._count += 1
 
-    def _get_groups(self) -> dict[str, GwEntityData]:
+    def _get_groups(self) -> None:
         """Helper-function for smile.py: get_all_gateway_entities().
 
         Collect switching-, pumping- or report-group info.
         """
-        groups: dict[str, GwEntityData] = {}
         # P1 and Anna don't have groups
         if self.smile.type == "power" or self.check_name(ANNA):
-            return groups
+            return
 
         for group in self._domain_objects.findall("./group"):
             members: list[str] = []
-            group_id = group.attrib["id"]
+            group_id = group.get("id")
             group_name = group.find("name").text
             group_type = group.find("type").text
             group_appliances = group.findall("appliances/appliance")
             for item in group_appliances:
                 # Check if members are not orphaned - stretch
-                if item.attrib["id"] in self.gw_entities:
-                    members.append(item.attrib["id"])
+                if item.get("id") in self.gw_entities:
+                    members.append(item.get("id"))
 
-            if group_type in GROUP_TYPES and members:
-                groups[group_id] = {
+            if group_type in GROUP_TYPES and members and group_id:
+                self.gw_entities[group_id] = {
                     "dev_class": group_type,
                     "model": "Group",
                     "name": group_name,
@@ -219,8 +218,6 @@ def _get_groups(self) -> dict[str, GwEntityData]:
                 }
                 self._count += 5
 
-        return groups
-
     def _get_lock_state(
         self, xml: etree.Element, data: GwEntityData, stretch_v2: bool = False
     ) -> None:
@@ -266,7 +263,7 @@ def _get_module_data(
             if key is not None and key not in link_tag:
                 continue
 
-            link_id = appl_search.attrib["id"]
+            link_id = appl_search.get("id")
             loc = f".//services/{link_tag}[@id='{link_id}']...."
             # Not possible to walrus for some reason...
             # xml_2: self._modules for legacy, self._domain_objects for actual

plugwise/helper.py

@@ -117,7 +117,7 @@ def _get_appliances(self) -> None:
         for appliance in self._domain_objects.findall("./appliance"):
             appl = Munch()
             appl.available = None
-            appl.entity_id = appliance.attrib["id"]
+            appl.entity_id = appliance.get("id")
             appl.location = None
             appl.name = appliance.find("name").text
             appl.model = None
@@ -139,7 +139,7 @@ def _get_appliances(self) -> None:
                 continue
 
             if (appl_loc := appliance.find("location")) is not None:
-                appl.location = appl_loc.attrib["id"]
+                appl.location = appl_loc.get("id")
             # Set location to the _home_loc_id when the appliance-location is not found,
             # except for thermostat-devices without a location, they are not active
             elif appl.pwclass not in THERMOSTAT_CLASSES:
@@ -206,7 +206,7 @@ def _get_locations(self) -> None:
         loc = Munch()
         locations = self._domain_objects.findall("./location")
         for location in locations:
-            loc.loc_id = location.attrib["id"]
+            loc.loc_id = location.get("id")
             loc.name = location.find("name").text
             loc._type = location.find("type").text
             self._loc_data[loc.loc_id] = {
@@ -269,7 +269,7 @@ def _appliance_info_finder(self, appl: Munch, appliance: etree.Element) -> Munch
 
     def _appl_gateway_info(self, appl: Munch, appliance: etree.Element) -> Munch:
         """Helper-function for _appliance_info_finder()."""
-        self._gateway_id = appliance.attrib["id"]
+        self._gateway_id = appliance.get("id")
         appl.firmware = str(self.smile.version)
         appl.hardware = self.smile.hw_version
         appl.mac = self.smile.mac_address
@@ -319,7 +319,7 @@ def _get_appliances_with_offset_functionality(self) -> list[str]:
             './/actuator_functionalities/offset_functionality[type="temperature_offset"]/offset/../../..'
         )
         for item in offset_appls:
-            therm_list.append(item.attrib["id"])
+            therm_list.append(item.get("id"))
 
         return therm_list
 
@@ -393,12 +393,12 @@ def _get_measurement_data(self, entity_id: str, entity: GwEntityData) -> None:
     def _collect_group_sensors(
         self,
         data: GwEntityData,
-        entity_id: str,
+        group_id: str,
         measurements: dict[str, UOM],
     ) -> None:
         """Collect group sensors."""
         if (
-            group := self._domain_objects.find(f'./group[@id="{entity_id}"]')
+            group := self._domain_objects.find(f'./group[@id="{group_id}"]')
         ) is not None:
             for measurement, attrs in measurements.items():
                 locator = f'.//logs/point_log[type="{measurement}"]/period/measurement'
@@ -502,7 +502,7 @@ def _get_plugwise_notifications(self) -> None:
         self._notifications = {}
         for notification in self._domain_objects.findall("./notification"):
             try:
-                msg_id = notification.attrib["id"]
+                msg_id = notification.get("id")
                 msg_type = notification.find("type").text
                 msg = notification.find("message").text
                 self._notifications[msg_id] = {msg_type: msg}
@@ -903,7 +903,7 @@ def _presets(self, loc_id: str) -> dict[str, list[float]]:
             directives = self._domain_objects.find(f'rule[@id="{rule_id}"]/directives')
             for directive in directives:
                 preset = directive.find("then").attrib
-                presets[directive.attrib["preset"]] = [
+                presets[directive.get("preset")] = [
                     float(preset["heating_setpoint"]),
                     float(preset["cooling_setpoint"]),
                 ]
@@ -920,13 +920,13 @@ def _rule_ids_by_name(self, name: str, loc_id: str) -> dict[str, dict[str, str]]
         for rule in self._domain_objects.findall(f'./rule[name="{name}"]'):
             active = rule.find("active").text
             if rule.find(locator) is not None:
-                schedule_ids[rule.attrib["id"]] = {
+                schedule_ids[rule.get("id")] = {
                     "location": loc_id,
                     "name": name,
                     "active": active,
                 }
             else:
-                schedule_ids[rule.attrib["id"]] = {
+                schedule_ids[rule.get("id")] = {
                     "location": NONE,
                     "name": name,
                     "active": active,
@@ -947,13 +947,13 @@ def _rule_ids_by_tag(self, tag: str, loc_id: str) -> dict[str, dict[str, str]]:
                 name = rule.find("name").text
                 active = rule.find("active").text
                 if rule.find(locator2) is not None:
-                    schedule_ids[rule.attrib["id"]] = {
+                    schedule_ids[rule.get("id")] = {
                         "location": loc_id,
                         "name": name,
                         "active": active,
                     }
                 else:
-                    schedule_ids[rule.attrib["id"]] = {
+                    schedule_ids[rule.get("id")] = {
                         "location": NONE,
                         "name": name,
                         "active": active,
@@ -1002,6 +1002,6 @@ def _thermostat_uri(self, loc_id: str) -> str:
         Determine the location-set_temperature uri - from LOCATIONS.
         """
         locator = f'./location[@id="{loc_id}"]/actuator_functionalities/thermostat_functionality'
-        thermostat_functionality_id = self._domain_objects.find(locator).attrib["id"]
+        thermostat_functionality_id = self._domain_objects.find(locator).get("id")
 
         return f"{LOCATIONS};id={loc_id}/thermostat;id={thermostat_functionality_id}"

plugwise/legacy/helper.py

@@ -107,7 +107,7 @@ def _all_appliances(self) -> None:
                 continue  # pragma: no cover
 
             appl.location = self._home_loc_id
-            appl.entity_id = appliance.attrib["id"]
+            appl.entity_id = appliance.get("id")
             appl.name = appliance.find("name").text
             # Extend device_class name when a Circle/Stealth is type heater_central -- Pw-Beta Issue #739
             if (
@@ -148,7 +148,7 @@ def _all_locations(self) -> None:
             return
 
         for location in locations:
-            loc.loc_id = location.attrib["id"]
+            loc.loc_id = location.get("id")
             loc.name = location.find("name").text
             loc._type = location.find("type").text
             # Filter the valid single location for P1 legacy: services not empty
@@ -393,10 +393,14 @@ def _presets(self) -> dict[str, list[float]]:
         """Helper-function for presets() - collect Presets for a legacy Anna."""
         presets: dict[str, list[float]] = {}
         for directive in self._domain_objects.findall("rule/directives/when/then"):
-            if directive is not None and directive.get("icon") is not None:
+            if (
+                directive is not None
+                and directive.get("icon") is not None
+                and directive.get("temperature") is not None
+            ):
                 # Ensure list of heating_setpoint, cooling_setpoint
-                presets[directive.attrib["icon"]] = [
-                    float(directive.attrib["temperature"]),
+                presets[directive.get("icon")] = [
+                    float(directive.get("temperature")),
                     0,
                 ]
 
@@ -411,7 +415,7 @@ def _schedules(self) -> tuple[list[str], str]:
         search = self._domain_objects
         if (result := search.find("./rule[name='Thermostat schedule']")) is not None:
             name = "Thermostat schedule"
-            rule_id = result.attrib["id"]
+            rule_id = result.get("id")
 
         log_type = "schedule_state"
         locator = f"./appliance[type='thermostat']/logs/point_log[type='{log_type}']/period/measurement"
@@ -432,5 +436,5 @@ def _schedules(self) -> tuple[list[str], str]:
     def _thermostat_uri(self) -> str:
         """Determine the location-set_temperature uri - from APPLIANCES."""
         locator = "./appliance[type='thermostat']"
-        appliance_id = self._appliances.find(locator).attrib["id"]
+        appliance_id = self._appliances.find(locator).get("id")
         return f"{APPLIANCES};id={appliance_id}/thermostat"

plugwise/legacy/smile.py

@@ -82,9 +82,7 @@ def get_all_gateway_entities(self) -> None:
         Finally, collect the data and states for each entity.
         """
         self._all_appliances()
-        if group_data := self._get_groups():
-            self.gw_entities.update(group_data)
-
+        self._get_groups()
         self._all_entity_data()
 
     async def async_update(self) -> dict[str, GwEntityData]:
@@ -154,13 +152,17 @@ async def set_offset(self, dev_id: str, offset: float) -> None:
 
     async def set_preset(self, _: str, preset: str) -> None:
         """Set the given Preset on the relevant Thermostat - from DOMAIN_OBJECTS."""
-        if (presets := self._presets()) is None:
+        if not (presets := self._presets()):
             raise PlugwiseError("Plugwise: no presets available.")  # pragma: no cover
         if preset not in list(presets):
             raise PlugwiseError("Plugwise: invalid preset.")
 
         locator = f'rule/directives/when/then[@icon="{preset}"].../.../...'
-        rule_id = self._domain_objects.find(locator).attrib["id"]
+        if (rule := self._domain_objects.find(locator)) is None:
+            raise PlugwiseError("Plugwise: no preset rule found.")  # pragma: no cover
+        if (rule_id := rule.get("id")) is None:
+            raise PlugwiseError("Plugwise: no preset id found.")  # pragma: no cover
+
         data = f"<rules><rule id='{rule_id}'><active>true</active></rule></rules>"
         await self.call_request(RULES, method="put", data=data)
 
@@ -192,7 +194,7 @@ async def set_schedule_state(
         schedule_rule_id: str | None = None
         for rule in self._domain_objects.findall("rule"):
             if rule.find("name").text == name:
-                schedule_rule_id = rule.attrib["id"]
+                schedule_rule_id = rule.get("id")
                 break
 
         if schedule_rule_id is None:
@@ -205,7 +207,7 @@ async def set_schedule_state(
             new_state = "true"
 
         locator = f'.//*[@id="{schedule_rule_id}"]/template'
-        template_id = self._domain_objects.find(locator).attrib["id"]
+        template_id = self._domain_objects.find(locator).get("id")
 
         data = (
             "<rules>"

plugwise/smile.py

@@ -113,9 +113,7 @@ def get_all_gateway_entities(self) -> None:
             )
             self._scan_thermostats()
 
-        if group_data := self._get_groups():
-            self.gw_entities.update(group_data)
-
+        self._get_groups()
         self._all_entity_data()
 
     async def async_update(self) -> dict[str, GwEntityData]:
@@ -178,7 +176,7 @@ async def set_number(
         if th_func_list := self._domain_objects.findall(locator):
             for th_func in th_func_list:
                 if th_func.find("type").text == key:
-                    thermostat_id = th_func.attrib["id"]
+                    thermostat_id = th_func.get("id")
 
         if thermostat_id is None:
             raise PlugwiseError(f"Plugwise: cannot change setpoint, {key} not found.")
@@ -358,7 +356,7 @@ async def set_schedule_state(
         )
         if self.check_name(ANNA):
             locator = f'.//*[@id="{schedule_rule_id}"]/template'
-            template_id = self._domain_objects.find(locator).attrib["id"]
+            template_id = self._domain_objects.find(locator).get("id")
             template = f'<template id="{template_id}" />'
 
         contexts = self.determine_contexts(loc_id, new_state, schedule_rule_id)
@@ -427,10 +425,10 @@ async def set_switch_state(
             # multiple types of e.g. toggle_functionality present
             if (sw_type := item.find("type")) is not None:
                 if sw_type.text == switch.act_type:
-                    switch_id = item.attrib["id"]
+                    switch_id = item.get("id")
                     break
             else:  # actuators with a single item like relay_functionality
-                switch_id = item.attrib["id"]
+                switch_id = item.get("id")
 
         uri = f"{APPLIANCES};id={appl_id}/{switch.device};id={switch_id}"
         if model == "relay":
@@ -456,7 +454,7 @@ async def _set_groupswitch_member_state(
         switched = 0
         for member in members:
             locator = f'appliance[@id="{member}"]/{switch.actuator}/{switch.func_type}'
-            switch_id = self._domain_objects.find(locator).attrib["id"]
+            switch_id = self._domain_objects.find(locator).get("id")
             uri = f"{APPLIANCES};id={member}/{switch.device};id={switch_id}"
             lock_blocked = self.gw_entities[member]["switches"].get("lock")
             # Assume Plugs under Plugwise control are not part of a group

plugwise/util.py

@@ -82,30 +82,33 @@ def check_heater_central(xml: etree.Element) -> str:
     for a system that has two heater_central appliances.
     """
     locator = "./appliance[type='heater_central']"
-    hc_count = 0
-    hc_list: list[dict[str, bool]] = []
+    heater_central_count = 0
+    heater_central_list: list[dict[str, bool]] = []
     for heater_central in xml.findall(locator):
-        hc_count += 1
-        hc_id: str = heater_central.attrib["id"]
-        has_actuators: bool = (
-            heater_central.find("actuator_functionalities/") is not None
-        )
+        if (heater_central_id := heater_central.get("id")) is None:
+            continue  # pragma: no cover
+
+        if (heater_central_name := heater_central.find("name")) is None:
+            continue  # pragma: no cover
+
+        has_actuators = heater_central.find("actuator_functionalities/") is not None
         # Filter for Plug/Circle/Stealth heater_central -- Pw-Beta Issue #739
-        if heater_central.find("name").text == "Central heating boiler":
-            hc_list.append({hc_id: has_actuators})
+        if heater_central_name.text == "Central heating boiler":
+            heater_central_list.append({heater_central_id: has_actuators})
+            heater_central_count += 1
 
-    if not hc_list:
+    if not heater_central_list:
         return NONE  # pragma: no cover
 
-    heater_central_id = list(hc_list[0].keys())[0]
-    if hc_count > 1:
-        for item in hc_list:
-            hc_id, has_actuators = next(iter(item.items()))
+    heater_id = list(heater_central_list[0].keys())[0]
+    if heater_central_count > 1:
+        for item in heater_central_list:
+            heater_central_id, has_actuators = next(iter(item.items()))
             if has_actuators:
-                heater_central_id = hc_id
+                heater_id = heater_central_id
                 break
 
-    return heater_central_id
+    return heater_id
 
 
 def check_model(name: str | None, vendor_name: str | None) -> str | None: