Merge pull request #516 from plugwise/improve-3

More improvements

Author: bouwew

fixtures/legacy_anna/all_data.json

@@ -1,9 +1,6 @@
 {
   "devices": {
     "0000aaaa0000aaaa0000aaaa0000aa00": {
-      "binary_sensors": {
-        "plugwise_notification": false
-      },
       "dev_class": "gateway",
       "firmware": "1.8.22",
       "location": "0000aaaa0000aaaa0000aaaa0000aa00",
@@ -67,7 +64,7 @@
     "cooling_present": false,
     "gateway_id": "0000aaaa0000aaaa0000aaaa0000aa00",
     "heater_id": "04e4cbfe7f4340f090f85ec3b9e6a950",
-    "item_count": 44,
+    "item_count": 43,
     "notifications": {},
     "smile_name": "Smile Anna"
   }

fixtures/legacy_anna_2/all_data.json

@@ -26,9 +26,6 @@
       "vendor": "Plugwise"
     },
     "be81e3f8275b4129852c4d8d550ae2eb": {
-      "binary_sensors": {
-        "plugwise_notification": false
-      },
       "dev_class": "gateway",
       "firmware": "1.8.22",
       "location": "be81e3f8275b4129852c4d8d550ae2eb",
@@ -69,7 +66,7 @@
     "cooling_present": false,
     "gateway_id": "be81e3f8275b4129852c4d8d550ae2eb",
     "heater_id": "ea5d8a7177e541b0a4b52da815166de4",
-    "item_count": 44,
+    "item_count": 43,
     "notifications": {},
     "smile_name": "Smile Anna"
   }

plugwise/__init__.py

@@ -18,12 +18,7 @@
     PlugwiseData,
     ThermoLoc,
 )
-from plugwise.exceptions import (
-    InvalidSetupError,
-    PlugwiseError,
-    ResponseError,
-    UnsupportedDeviceError,
-)
+from plugwise.exceptions import InvalidSetupError, ResponseError, UnsupportedDeviceError
 from plugwise.helper import SmileComm
 from plugwise.legacy.smile import SmileLegacyAPI
 from plugwise.smile import SmileAPI
@@ -326,58 +321,33 @@ async def set_preset(self, loc_id: str, preset: str) -> None:
 
     async def set_temperature(self, loc_id: str, items: dict[str, float]) -> None:
         """Set the given Temperature on the relevant Thermostat."""
-        try:
-            await self._smile_api.set_temperature(loc_id, items)
-        except PlugwiseError as exc:
-            raise PlugwiseError(
-                    "Plugwise: failed setting temperature: no valid input provided"
-                ) from exc
+        await self._smile_api.set_temperature(loc_id, items)
 
     async def set_number_setpoint(self, key: str, _: str, temperature: float) -> None:
         """Set the max. Boiler or DHW setpoint on the Central Heating boiler."""
-        try:
-            await self._smile_api.set_number_setpoint(key, temperature)
-        except PlugwiseError as exc:
-            raise PlugwiseError(f"Plugwise: cannot change setpoint, {key} not found.") from exc
+        await self._smile_api.set_number_setpoint(key, temperature)
 
     async def set_temperature_offset(self, _: str, dev_id: str, offset: float) -> None:
         """Set the Temperature offset for thermostats that support this feature."""
-        try:
-            await self._smile_api.set_temperature_offset(dev_id, offset)
-        except PlugwiseError as exc:
-            raise PlugwiseError(
-                "Plugwise: this device does not have temperature-offset capability."
-            ) from exc
+        await self._smile_api.set_temperature_offset(dev_id, offset)
 
     async def set_switch_state(
         self, appl_id: str, members: list[str] | None, model: str, state: str
     ) -> None:
         """Set the given State of the relevant Switch."""
-        try:
-            await self._smile_api.set_switch_state(appl_id, members, model, state)
-        except PlugwiseError as exc:
-            raise PlugwiseError("Plugwise: the locked Relay was not switched.") from exc
+        await self._smile_api.set_switch_state(appl_id, members, model, state)
 
     async def set_gateway_mode(self, mode: str) -> None:
         """Set the gateway mode."""
-        try:
-            await self._smile_api.set_gateway_mode(mode)
-        except PlugwiseError as exc:
-            raise PlugwiseError("Plugwise: invalid gateway mode.") from exc
+        await self._smile_api.set_gateway_mode(mode)
 
     async def set_regulation_mode(self, mode: str) -> None:
         """Set the heating regulation mode."""
-        try:
-            await self._smile_api.set_regulation_mode(mode)
-        except PlugwiseError as exc:
-            raise PlugwiseError("Plugwise: invalid regulation mode.") from exc
+        await self._smile_api.set_regulation_mode(mode)
 
     async def set_dhw_mode(self, mode: str) -> None:
         """Set the domestic hot water heating regulation mode."""
-        try:
-            await self._smile_api.set_dhw_mode(mode)
-        except PlugwiseError as exc:
-            raise PlugwiseError("Plugwise: invalid dhw mode.") from exc
+        await self._smile_api.set_dhw_mode(mode)
 
     async def delete_notification(self) -> None:
         """Delete the active Plugwise Notification."""

plugwise/common.py

@@ -0,0 +1,151 @@
+"""Use of this source code is governed by the MIT license found in the LICENSE file.
+
+Plugwise Smile protocol helpers.
+"""
+from __future__ import annotations
+
+from plugwise.constants import ModelData
+from plugwise.util import (
+    check_heater_central,
+    check_model,
+    get_vendor_name,
+    return_valid,
+)
+
+from defusedxml import ElementTree as etree
+from munch import Munch
+
+
+class SmileCommon:
+    """The SmileCommon class."""
+
+    def __init__(self) -> None:
+        """Init."""
+        self._appliances: etree
+        self._domain_objects: etree
+        self._cooling_present: bool
+        self._heater_id: str
+        self._on_off_device: bool
+        self._opentherm_device: bool
+        self.smile_name: str
+
+    def smile(self, name: str) -> bool:
+        """Helper-function checking the smile-name."""
+        return self.smile_name == name
+
+    def _appl_thermostat_info(self, appl: Munch, xml_1: etree, xml_2: etree = None) -> Munch:
+        """Helper-function for _appliance_info_finder()."""
+        locator = "./logs/point_log[type='thermostat']/thermostat"
+        mod_type = "thermostat"
+        xml_2 = return_valid(xml_2, self._domain_objects)
+        module_data = self._get_module_data(xml_1, locator, mod_type, xml_2)
+        appl.vendor_name = module_data["vendor_name"]
+        appl.model = check_model(module_data["vendor_model"], appl.vendor_name)
+        appl.hardware = module_data["hardware_version"]
+        appl.firmware = module_data["firmware_version"]
+        appl.zigbee_mac = module_data["zigbee_mac_address"]
+
+        return appl
+
+    def _appl_heater_central_info(
+        self,
+        appl: Munch,
+        xml_1: etree,
+        xml_2: etree = None,
+        xml_3: etree = None,
+    ) -> Munch:
+        """Helper-function for _appliance_info_finder()."""
+        # Remove heater_central when no active device present
+        if not self._opentherm_device and not self._on_off_device:
+            return None
+
+        # Find the valid heater_central
+        # xml_2 self._appliances for legacy, self._domain_objects for actual
+        xml_2 = return_valid(xml_2, self._domain_objects)
+        self._heater_id = check_heater_central(xml_2)
+
+        #  Info for On-Off device
+        if self._on_off_device:
+            appl.name = "OnOff"  # pragma: no cover
+            appl.vendor_name = None  # pragma: no cover
+            appl.model = "Unknown"  # pragma: no cover
+            return appl  # pragma: no cover
+
+        # Info for OpenTherm device
+        appl.name = "OpenTherm"
+        locator_1 = "./logs/point_log[type='flame_state']/boiler_state"
+        locator_2 = "./services/boiler_state"
+        mod_type = "boiler_state"
+        # xml_1: appliance
+        # xml_3: self._modules for legacy, self._domain_objects for actual
+        xml_3 = return_valid(xml_3, self._domain_objects)
+        module_data = self._get_module_data(xml_1, locator_1, mod_type, xml_3)
+        if not module_data["contents"]:
+            module_data = self._get_module_data(xml_1, locator_2, mod_type, xml_3)
+        appl.vendor_name = module_data["vendor_name"]
+        appl.hardware = module_data["hardware_version"]
+        appl.model = module_data["vendor_model"]
+        if appl.model is None:
+            appl.model = (
+                "Generic heater/cooler"
+                if self._cooling_present
+                else "Generic heater"
+            )
+
+        return appl
+
+    def _get_module_data(
+        self,
+        xml_1: etree,
+        locator: str,
+        mod_type: str,
+        xml_2: etree = None,
+        legacy: bool = False,
+    ) -> ModelData:
+        """Helper-function for _energy_device_info_finder() and _appliance_info_finder().
+
+        Collect requested info from MODULES.
+        """
+        model_data: ModelData = {
+            "contents": False,
+            "firmware_version": None,
+            "hardware_version": None,
+            "reachable": None,
+            "vendor_name": None,
+            "vendor_model": None,
+            "zigbee_mac_address": None,
+        }
+        # xml_1: appliance
+        if (appl_search := xml_1.find(locator)) is not None:
+            link_id = appl_search.attrib["id"]
+            loc = f".//services/{mod_type}[@id='{link_id}']...."
+            if legacy:
+                loc = f".//{mod_type}[@id='{link_id}']...."
+            # Not possible to walrus for some reason...
+            # xml_2: self._modules for legacy, self._domain_objects for actual
+            search = return_valid(xml_2, self._domain_objects)
+            module = search.find(loc)
+            if module is not None:  # pylint: disable=consider-using-assignment-expr
+                model_data["contents"] = True
+                get_vendor_name(module, model_data)
+                model_data["vendor_model"] = module.find("vendor_model").text
+                model_data["hardware_version"] = module.find("hardware_version").text
+                model_data["firmware_version"] = module.find("firmware_version").text
+                self._get_zigbee_data(module, model_data, legacy)
+
+        return model_data
+
+    def _get_zigbee_data(self, module: etree, model_data: ModelData, legacy: bool) -> None:
+        """Helper-function for _get_model_data()."""
+        if legacy:
+            # Stretches
+            if (router := module.find("./protocols/network_router")) is not None:
+                model_data["zigbee_mac_address"] = router.find("mac_address").text
+            # Also look for the Circle+/Stealth M+
+            if (coord := module.find("./protocols/network_coordinator")) is not None:
+                model_data["zigbee_mac_address"] = coord.find("mac_address").text
+        # Adam
+        elif (zb_node := module.find("./protocols/zig_bee_node")) is not None:
+                model_data["zigbee_mac_address"] = zb_node.find("mac_address").text
+                model_data["reachable"] = zb_node.find("reachable").text == "true"
+

plugwise/data.py

@@ -90,7 +90,6 @@ def _all_device_data(self) -> None:
         Collect data for each device and add to self.gw_data and self.gw_devices.
         """
         self._update_gw_devices()
-        self.device_items = self._count
         self.gw_data.update(
             {
                 "gateway_id": self.gateway_id,