Save updated files

Author: bouwew

plugwise/__init__.py

@@ -2,6 +2,7 @@
 
 Plugwise backend module for Home Assistant Core.
 """
+
 from __future__ import annotations
 
 from plugwise.constants import (
@@ -63,7 +64,7 @@ def __init__(
             self._timeout,
             self._username,
             self._websession,
-         )
+        )
 
         self._cooling_present = False
         self._elga = False
@@ -125,52 +126,56 @@ async def connect(self) -> Version | None:
         # Determine smile specifics
         await self._smile_detect(result, dsmrmain)
 
-        self._smile_api = SmileAPI(
-            self._host,
-            self._password,
-            self._request,
-            self._websession,
-            self._cooling_present,
-            self._elga,
-            self._is_thermostat,
-            self._last_active,
-            self._loc_data,
-            self._on_off_device,
-            self._opentherm_device,
-            self._schedule_old_states,
-            self.gateway_id,
-            self.smile_fw_version,
-            self.smile_hostname,
-            self.smile_hw_version,
-            self.smile_mac_address,
-            self.smile_model,
-            self.smile_model_id,
-            self.smile_name,
-            self.smile_type,
-            self.smile_version,
-            self._port,
-            self._username,
-         ) if not self.smile_legacy else SmileLegacyAPI(
-            self._host,
-            self._password,
-            self._request,
-            self._websession,
-            self._is_thermostat,
-            self._loc_data,
-            self._on_off_device,
-            self._opentherm_device,
-            self._stretch_v2,
-            self._target_smile,
-            self.smile_fw_version,
-            self.smile_hostname,
-            self.smile_hw_version,
-            self.smile_mac_address,
-            self.smile_model,
-            self.smile_name,
-            self.smile_type,
-            self.smile_zigbee_mac_address,
-            self._port,
-            self._username,
+        self._smile_api = (
+            SmileAPI(
+                self._host,
+                self._password,
+                self._request,
+                self._websession,
+                self._cooling_present,
+                self._elga,
+                self._is_thermostat,
+                self._last_active,
+                self._loc_data,
+                self._on_off_device,
+                self._opentherm_device,
+                self._schedule_old_states,
+                self.gateway_id,
+                self.smile_fw_version,
+                self.smile_hostname,
+                self.smile_hw_version,
+                self.smile_mac_address,
+                self.smile_model,
+                self.smile_model_id,
+                self.smile_name,
+                self.smile_type,
+                self.smile_version,
+                self._port,
+                self._username,
+            )
+            if not self.smile_legacy
+            else SmileLegacyAPI(
+                self._host,
+                self._password,
+                self._request,
+                self._websession,
+                self._is_thermostat,
+                self._loc_data,
+                self._on_off_device,
+                self._opentherm_device,
+                self._stretch_v2,
+                self._target_smile,
+                self.smile_fw_version,
+                self.smile_hostname,
+                self.smile_hw_version,
+                self.smile_mac_address,
+                self.smile_model,
+                self.smile_name,
+                self.smile_type,
+                self.smile_zigbee_mac_address,
+                self._port,
+                self._username,
+            )
         )
 
         # Update all endpoints on first connect
@@ -203,7 +208,7 @@ async def _smile_detect(self, result: etree, dsmrmain: etree) -> None:
             )
             raise UnsupportedDeviceError
 
-        version_major= str(self.smile_fw_version.major)
+        version_major = str(self.smile_fw_version.major)
         self._target_smile = f"{model}_v{version_major}"
         LOGGER.debug("Plugwise identified as %s", self._target_smile)
         if self._target_smile not in SMILES:
@@ -318,9 +323,9 @@ async def async_update(self) -> PlugwiseData:
 
         return data
 
-########################################################################################################
-###  API Set and HA Service-related Functions                                                        ###
-########################################################################################################
+    ########################################################################################################
+    ###  API Set and HA Service-related Functions                                                        ###
+    ########################################################################################################
 
     async def set_select(
         self,
@@ -333,7 +338,9 @@ async def set_select(
         try:
             await self._smile_api.set_select(key, loc_id, option, state)
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to set select option '{option}': {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to set select option '{option}': {str(exc)}"
+            ) from exc
 
     async def set_schedule_state(
         self,
@@ -345,8 +352,9 @@ async def set_schedule_state(
         try:
             await self._smile_api.set_schedule_state(loc_id, state, name)
         except ConnectionFailedError as exc:  # pragma no cover
-            raise ConnectionFailedError(f"Failed to set schedule state: {str(exc)}") from exc  # pragma no cover
-
+            raise ConnectionFailedError(
+                f"Failed to set schedule state: {str(exc)}"
+            ) from exc  # pragma no cover
 
     async def set_preset(self, loc_id: str, preset: str) -> None:
         """Set the given Preset on the relevant Thermostat."""
@@ -360,7 +368,9 @@ async def set_temperature(self, loc_id: str, items: dict[str, float]) -> None:
         try:
             await self._smile_api.set_temperature(loc_id, items)
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to set temperature: {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to set temperature: {str(exc)}"
+            ) from exc
 
     async def set_number(
         self,
@@ -372,14 +382,18 @@ async def set_number(
         try:
             await self._smile_api.set_number(dev_id, key, temperature)
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to set number '{key}': {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to set number '{key}': {str(exc)}"
+            ) from exc
 
     async def set_temperature_offset(self, dev_id: str, offset: float) -> None:
         """Set the Temperature offset for thermostats that support this feature."""
         try:  # pragma no cover
             await self._smile_api.set_offset(dev_id, offset)  # pragma: no cover
         except ConnectionFailedError as exc:  # pragma no cover
-            raise ConnectionFailedError(f"Failed to set temperature offset: {str(exc)}") from exc  # pragma no cover
+            raise ConnectionFailedError(
+                f"Failed to set temperature offset: {str(exc)}"
+            ) from exc  # pragma no cover
 
     async def set_switch_state(
         self, appl_id: str, members: list[str] | None, model: str, state: str
@@ -388,39 +402,51 @@ async def set_switch_state(
         try:
             await self._smile_api.set_switch_state(appl_id, members, model, state)
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to set switch state: {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to set switch state: {str(exc)}"
+            ) from exc
 
     async def set_gateway_mode(self, mode: str) -> None:
         """Set the gateway mode."""
         try:  # pragma no cover
             await self._smile_api.set_gateway_mode(mode)  # pragma: no cover
         except ConnectionFailedError as exc:  # pragma no cover
-            raise ConnectionFailedError(f"Failed to set gateway mode: {str(exc)}") from exc  # pragma no cover
+            raise ConnectionFailedError(
+                f"Failed to set gateway mode: {str(exc)}"
+            ) from exc  # pragma no cover
 
     async def set_regulation_mode(self, mode: str) -> None:
         """Set the heating regulation mode."""
         try:  # pragma no cover
             await self._smile_api.set_regulation_mode(mode)  # pragma: no cover
         except ConnectionFailedError as exc:  # pragma no cover
-            raise ConnectionFailedError(f"Failed to set regulation mode: {str(exc)}") from exc  # pragma no cover
+            raise ConnectionFailedError(
+                f"Failed to set regulation mode: {str(exc)}"
+            ) from exc  # pragma no cover
 
     async def set_dhw_mode(self, mode: str) -> None:
         """Set the domestic hot water heating regulation mode."""
         try:  # pragma no cover
             await self._smile_api.set_dhw_mode(mode)  # pragma: no cover
         except ConnectionFailedError as exc:  # pragma no cover
-            raise ConnectionFailedError(f"Failed to set dhw mode: {str(exc)}") from exc  # pragma no cover
+            raise ConnectionFailedError(
+                f"Failed to set dhw mode: {str(exc)}"
+            ) from exc  # pragma no cover
 
     async def delete_notification(self) -> None:
         """Delete the active Plugwise Notification."""
         try:
             await self._smile_api.delete_notification()
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to delete notification: {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to delete notification: {str(exc)}"
+            ) from exc
 
     async def reboot_gateway(self) -> None:
         """Reboot the Plugwise Gateway."""
         try:
             await self._smile_api.reboot_gateway()
         except ConnectionFailedError as exc:
-            raise ConnectionFailedError(f"Failed to reboot gateway: {str(exc)}") from exc
+            raise ConnectionFailedError(
+                f"Failed to reboot gateway: {str(exc)}"
+            ) from exc

plugwise/common.py

@@ -2,6 +2,7 @@
 
 Plugwise Smile protocol helpers.
 """
+
 from __future__ import annotations
 
 from typing import cast
@@ -83,21 +84,23 @@ def _appl_heater_central_info(
         appl.hardware = module_data["hardware_version"]
         appl.model_id = module_data["vendor_model"] if not legacy else None
         appl.model = (
-            "Generic heater/cooler"
-            if self._cooling_present
-            else "Generic heater"
+            "Generic heater/cooler" if self._cooling_present else "Generic heater"
         )
 
         return appl
 
-    def _appl_thermostat_info(self, appl: Munch, xml_1: etree, xml_2: etree = None) -> Munch:
+    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"
         xml_2 = return_valid(xml_2, self._domain_objects)
         module_data = self._get_module_data(xml_1, locator, xml_2)
         appl.vendor_name = module_data["vendor_name"]
         appl.model = module_data["vendor_model"]
-        if appl.model != "ThermoTouch":  # model_id for Anna not present as stand-alone device
+        if (
+            appl.model != "ThermoTouch"
+        ):  # model_id for Anna not present as stand-alone device
             appl.model_id = appl.model
             appl.model = check_model(appl.model, appl.vendor_name)
 
@@ -108,7 +111,9 @@ def _appl_thermostat_info(self, appl: Munch, xml_1: etree, xml_2: etree = None)
 
         return appl
 
-    def _collect_power_values(self, data: GwEntityData, loc: Munch, tariff: str, legacy: bool = False) -> None:
+    def _collect_power_values(
+        self, data: GwEntityData, loc: Munch, tariff: str, legacy: bool = False
+    ) -> None:
         """Something."""
         for loc.peak_select in ("nl_peak", "nl_offpeak"):
             loc.locator = (
@@ -220,9 +225,7 @@ def _create_gw_entities(self, appl: Munch) -> None:
                 self.gw_entities[appl.entity_id][appl_key] = value
                 self._count += 1
 
-    def _entity_switching_group(
-        self, entity: GwEntityData, data: GwEntityData
-    ) -> None:
+    def _entity_switching_group(self, entity: GwEntityData, data: GwEntityData) -> None:
         """Helper-function for _get_device_zone_data().
 
         Determine switching group device data.
@@ -268,7 +271,9 @@ def _get_group_switches(self) -> dict[str, GwEntityData]:
 
         return switch_groups
 
-    def _get_lock_state(self, xml: etree, data: GwEntityData, stretch_v2: bool = False) -> None:
+    def _get_lock_state(
+        self, xml: etree, data: GwEntityData, stretch_v2: bool = False
+    ) -> None:
         """Helper-function for _get_measurement_data().
 
         Adam & Stretches: obtain the relay-switch lock state.
@@ -323,7 +328,9 @@ def _get_module_data(
 
         return module_data
 
-    def _get_zigbee_data(self, module: etree, module_data: ModuleData, legacy: bool) -> None:
+    def _get_zigbee_data(
+        self, module: etree, module_data: ModuleData, legacy: bool
+    ) -> None:
         """Helper-function for _get_module_data()."""
         if legacy:
             # Stretches
@@ -334,5 +341,5 @@ def _get_zigbee_data(self, module: etree, module_data: ModuleData, legacy: bool)
                 module_data["zigbee_mac_address"] = coord.find("mac_address").text
         # Adam
         elif (zb_node := module.find("./protocols/zig_bee_node")) is not None:
-                module_data["zigbee_mac_address"] = zb_node.find("mac_address").text
-                module_data["reachable"] = zb_node.find("reachable").text == "true"
+            module_data["zigbee_mac_address"] = zb_node.find("mac_address").text
+            module_data["reachable"] = zb_node.find("reachable").text == "true"

plugwise/constants.py

@@ -1,4 +1,5 @@
 """Plugwise Smile constants."""
+
 from __future__ import annotations
 
 from collections import namedtuple
@@ -167,9 +168,7 @@
     "intended_boiler_state": DATA(
         "heating_state", NONE
     ),  # Legacy Anna: shows when heating is active, we don't show dhw_state, cannot be determined reliably
-    "flame_state": UOM(
-        NONE
-    ),  # Also present when there is a single gas-heater
+    "flame_state": UOM(NONE),  # Also present when there is a single gas-heater
     "intended_boiler_temperature": UOM(
         TEMP_CELSIUS
     ),  # Non-zero when heating, zero when dhw-heating
@@ -585,4 +584,3 @@ class PlugwiseData:
 
     devices: dict[str, GwEntityData]
     gateway: GatewayData
-