Import Logger, start adding typing hints

Author: bouwew

plugwise/helper.py

@@ -3,6 +3,7 @@
 """
 import asyncio
 import datetime as dt
+from typing import Any
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientSession, ClientTimeout, ServerTimeoutError
@@ -271,36 +272,36 @@ class SmileHelper:
 
     def __init__(self):
         """Set the constructor for this class."""
-        self._appl_data = None
-        self._appliances = None
-        self._cooling_present = None
-        self._cp_state = None
-        self._devices = None
-        self._domain_objects = None
-        self._heater_id = None
-        self._home_location = None
-        self._last_active = {}
-        self._loc_data = None
-        self._locations = None
-        self._modules = None
-        self._on_off_device = None
-        self._ot_device = None
-        self._outdoor_temp = None
-        self._sm_thermostat = None
-        self._thermo_locs = None
-
-        self._smile_legacy = None
-        self._stretch_v2 = None
-        self._stretch_v3 = None
-
-        self.cooling_active = None
-        self.gateway_id = None
-        self.gw_data = {}
-        self.gw_devices = {}
-
-        self.smile_name = None
-        self.smile_type = None
-        self.smile_version = []
+        self._appl_data: dict[str, Any] = {}
+        self._appliances: etree = None
+        self._cooling_present: bool = False
+        self._devices: dict[str, str] = {}
+        self._domain_objects: etree = None
+        self._heater_id: str = None
+        self._home_location: str = None
+        self._last_active: dict[str, str] = {}
+        self._loc_data: dict[str, Any] = {}
+        self._locations: etree = None
+        self._modules: etree = None
+        self._on_off_device: bool = False
+        self._opentherm_device: bool = False
+        self._outdoor_temp: float = None
+        self._is_thermostat: bool = False
+        self._multi_thermostats: bool = False
+        self._thermo_locs: dict[str, Any] = {}
+
+        self._smile_legacy: bool = False
+        self._stretch_v2: bool = False
+        self._stretch_v3: bool = False
+
+        self.cooling_active: bool = False
+        self.gateway_id: str = None
+        self.gw_data: dict[str, Any] = {}
+        self.gw_devices: dict[str, Any] = {}
+
+        self.smile_name: str = None
+        self.smile_type: str = None
+        self.smile_version: list[str] = []
 
     def _locations_legacy(self):
         """Helper-function for _all_locations().
@@ -348,7 +349,6 @@ def _locations_specials(self, loc, location):
 
     def _all_locations(self):
         """Collect all locations."""
-        self._loc_data = {}
         loc = Munch()
 
         # Legacy Anna without outdoor_temp and Stretches have no locations, create one containing all appliances
@@ -476,7 +476,7 @@ def _appliance_info_finder(self, appliance, appl):
                 return appl
 
             # Remove heater_central when no active device present
-            if not self._ot_device and not self._on_off_device:
+            if not self._opentherm_device and not self._on_off_device:
                 return None
 
             self._heater_id = appliance.attrib["id"]
@@ -531,9 +531,6 @@ def _appliance_types_finder(self, appliance, appl):
 
     def _all_appliances(self):
         """Collect all appliances with relevant info."""
-        self._appl_data = {}
-        self._cp_state = None
-
         self._all_locations()
 
         # Create a gateway for legacy Anna, P1 and Stretches
@@ -572,7 +569,7 @@ def _all_appliances(self):
         ot_fault_code = self._appliances.find(
             ".//logs/point_log[type='open_therm_oem_fault_code']"
         )
-        self._ot_device = ch_state is not None and ot_fault_code is not None
+        self._opentherm_device = ch_state is not None and ot_fault_code is not None
         self._on_off_device = ch_state is not None and ot_fault_code is None
 
         for appliance in self._appliances.findall("./appliance"):
@@ -767,7 +764,7 @@ def _get_appliance_data(self, d_id):
 
         appliance = self._appliances.find(f'.//appliance[@id="{d_id}"]')
         measurements = DEVICE_MEASUREMENTS.items()
-        if self._ot_device or self._on_off_device:
+        if self._opentherm_device or self._on_off_device:
             measurements = {
                 **DEVICE_MEASUREMENTS,
                 **HEATER_CENTRAL_MEASUREMENTS,
@@ -1203,7 +1200,7 @@ def _create_dicts_from_data(self, data, bs_dict, s_dict, sw_dict):
             for item in BINARY_SENSORS:
                 if list(item.keys())[0] == key:
                     data.pop(key)
-                    if self._ot_device or self._on_off_device:
+                    if self._opentherm_device or self._on_off_device:
                         bs_dict[key] = value
             for item in SENSORS:
                 if list(item.keys())[0] == key:

plugwise/smile.py

@@ -1,9 +1,8 @@
 """Use of this source code is governed by the MIT license found in the LICENSE file.
 Plugwise backend module for Home Assistant Core.
 """
-import logging
-
 import aiohttp
+from defusedxml import ElementTree as etree
 
 # Dict as class
 from munch import Munch
@@ -18,6 +17,7 @@
     DEFAULT_USERNAME,
     DOMAIN_OBJECTS,
     LOCATIONS,
+    LOGGER,
     MODULES,
     NOTIFICATIONS,
     PW_NOTIFICATION,
@@ -31,8 +31,6 @@
 from .exceptions import ConnectionFailedError, InvalidXMLError, UnsupportedDeviceError
 from .helper import SmileComm, SmileHelper, pw_notification_updater, update_helper
 
-_LOGGER = logging.getLogger(__name__)
-
 
 class SmileData(SmileHelper):
     """The Plugwise Smile main class."""
@@ -42,7 +40,7 @@ def _append_special(self, data, d_id, bs_dict, s_dict):
         When conditions are met, the plugwise_notification binary_sensor is appended.
         """
         if d_id == self.gateway_id:
-            if self._sm_thermostat is not None:
+            if self._is_thermostat:
                 bs_dict.update(PW_NOTIFICATION)
 
     def _all_device_data(self):
@@ -69,11 +67,13 @@ def _all_device_data(self):
 
             self.gw_devices[dev_id] = dev_and_data
 
-        self.gw_data["active_device"] = self._ot_device or self._on_off_device
+        self.gw_data["active_device"] = self._opentherm_device or self._on_off_device
         self.gw_data["cooling_present"] = self._cooling_present
         self.gw_data["gateway_id"] = self.gateway_id
         self.gw_data["heater_id"] = self._heater_id
-        self.gw_data["single_master_thermostat"] = self._sm_thermostat
+        self.gw_data["single_master_thermostat"] = (
+            self._is_thermostat and not self._multi_thermostats
+        )
         self.gw_data["smile_name"] = self.smile_name
 
     def get_all_devices(self):
@@ -221,20 +221,19 @@ def _get_device_data(self, dev_id):
         return device_data
 
     def single_master_thermostat(self):
-        """Determine if there is a single master thermostat in the setup.
-        Possible output: None, True, False.
-        """
+        """Determine if there is a single master thermostat in the setup."""
         if self.smile_type == "thermostat":
+            self._is_thermostat = True
             count = 0
             for dummy, data in self._thermo_locs.items():
                 if "master_prio" in data:
                     if data.get("master_prio") > 0:
                         count += 1
 
             if count == 1:
-                self._sm_thermostat = True
+                self._multi_thermostats = False
             if count > 1:
-                self._sm_thermostat = False
+                self._multi_thermostats = True
 
 
 class Smile(SmileComm, SmileData):
@@ -244,13 +243,13 @@ class Smile(SmileComm, SmileData):
 
     def __init__(
         self,
-        host,
-        password,
-        username=DEFAULT_USERNAME,
-        port=DEFAULT_PORT,
-        timeout=DEFAULT_TIMEOUT,
+        host: str,
+        password: str,
+        username: str = DEFAULT_USERNAME,
+        port: str = DEFAULT_PORT,
+        timeout: str = DEFAULT_TIMEOUT,
         websession: aiohttp.ClientSession = None,
-    ):
+    ) -> None:
         """Set the constructor for this class."""
         super().__init__(
             host,
@@ -262,17 +261,17 @@ def __init__(
         )
         SmileData.__init__(self)
 
-        self._notifications = {}
-        self.smile_hostname = None
+        self._notifications: dict[str, str] = {}
+        self.smile_hostname: str = "Unknown"
 
-    async def connect(self):
+    async def connect(self) -> bool:
         """Connect to Plugwise device and determine its name, type and version."""
-        names = []
+        names: list[str] = []
 
-        result = await self._request(DOMAIN_OBJECTS)
-        dsmrmain = result.find(".//module/protocols/dsmrmain")
+        result: etree = await self._request(DOMAIN_OBJECTS)
+        dsmrmain: etree | None = result.find(".//module/protocols/dsmrmain")
 
-        vendor_names = result.findall(".//module/vendor_name")
+        vendor_names: list[etree] = result.findall(".//module/vendor_name")
         if not vendor_names:
             # Work-around for Stretch fv 2.7.18
             result = await self._request(MODULES)
@@ -283,7 +282,7 @@ async def connect(self):
 
         if "Plugwise" not in names:
             if dsmrmain is None:  # pragma: no cover
-                _LOGGER.error(
+                LOGGER.error(
                     "Connected but expected text not returned, \
                               we got %s",
                     result,
@@ -298,23 +297,23 @@ async def connect(self):
 
         return True
 
-    async def _smile_detect_legacy(self, result, dsmrmain):
+    async def _smile_detect_legacy(self, result, dsmrmain) -> None:
         """Helper-function for _smile_detect()."""
-        network = result.find(".//module/protocols/master_controller")
+        network: etree = result.find(".//module/protocols/master_controller")
 
         # Assume legacy
         self._smile_legacy = True
         # Try if it is an Anna, assuming appliance thermostat
-        anna = result.find('.//appliance[type="thermostat"]')
+        anna: etree = result.find('.//appliance[type="thermostat"]')
         # Fake insert version assuming Anna
         # couldn't find another way to identify as legacy Anna
-        version = "1.8.0"
-        model = "smile_thermo"
+        version: str = "1.8.0"
+        model: str = "smile_thermo"
         if anna is None:
             # P1 legacy:
             if dsmrmain is not None:
                 try:
-                    status = await self._request(STATUS)
+                    status: etree = await self._request(STATUS)
                     version = status.find(".//system/version").text
                     model = status.find(".//system/product").text
                     self.smile_hostname = status.find(".//network/hostname").text
@@ -334,7 +333,7 @@ async def _smile_detect_legacy(self, result, dsmrmain):
                     raise ConnectionFailedError
             else:  # pragma: no cover
                 # No cornercase, just end of the line
-                _LOGGER.error("Connected but no gateway device information found")
+                LOGGER.error("Connected but no gateway device information found")
                 raise ConnectionFailedError
         return model, version
 
@@ -353,14 +352,14 @@ async def _smile_detect(self, result, dsmrmain):
 
         if model is None or version is None:  # pragma: no cover
             # Corner case check
-            _LOGGER.error("Unable to find model or version information")
+            LOGGER.error("Unable to find model or version information")
             raise UnsupportedDeviceError
 
         ver = semver.VersionInfo.parse(version)
         target_smile = f"{model}_v{ver.major}"
-        _LOGGER.debug("Plugwise identified as %s", target_smile)
+        LOGGER.debug("Plugwise identified as %s", target_smile)
         if target_smile not in SMILES:
-            _LOGGER.error(
+            LOGGER.error(
                 'Your version Smile identified as "%s" seems\
                  unsupported by our plugin, please create an issue\
                  on http://github.com/plugwise/python-plugwise!',
@@ -407,9 +406,9 @@ async def _update_domain_objects(self):
                 msg_type = notification.find("type").text
                 msg = notification.find("message").text
                 self._notifications.update({msg_id: {msg_type: msg}})
-                _LOGGER.debug("Plugwise notifications: %s", self._notifications)
+                LOGGER.debug("Plugwise notifications: %s", self._notifications)
             except AttributeError:  # pragma: no cover
-                _LOGGER.info(
+                LOGGER.info(
                     "Plugwise notification present but unable to process, manually investigate: %s",
                     f"{self._endpoint}{DOMAIN_OBJECTS}",
                 )