Merge pull request #764 from plugwise/testing

Limit amount of __init__ arguments

Author: bouwew

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Ongoing
 
-- Implement async file-open in tests
+- Implement code quality improvements as suggested by SonarCloud
 
 ## v1.7.6
 

plugwise/__init__.py

@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+from typing import cast
+
 from plugwise.constants import (
     DEFAULT_LEGACY_TIMEOUT,
     DEFAULT_PORT,
@@ -36,6 +38,7 @@
 
 import aiohttp
 from defusedxml import ElementTree as etree
+from munch import Munch
 from packaging.version import Version, parse
 
 
@@ -72,16 +75,17 @@ def __init__(
         self._smile_api: SmileAPI | SmileLegacyAPI
         self._stretch_v2 = False
         self._target_smile: str = NONE
-        self.smile_hostname: str = NONE
-        self.smile_hw_version: str | None = None
-        self.smile_legacy = False
-        self.smile_mac_address: str | None = None
-        self.smile_model: str = NONE
-        self.smile_model_id: str | None = None
-        self.smile_name: str = NONE
-        self.smile_type: str = NONE
-        self.smile_version: Version = Version("0.0.0")
-        self.smile_zigbee_mac_address: str | None = None
+        self.smile: Munch = Munch()
+        self.smile.hostname = NONE
+        self.smile.hw_version = None
+        self.smile.legacy = False
+        self.smile.mac_address = None
+        self.smile.model = NONE
+        self.smile.model_id = None
+        self.smile.name = NONE
+        self.smile.type = NONE
+        self.smile.version = Version("0.0.0")
+        self.smile.zigbee_mac_address = None
 
     @property
     def cooling_present(self) -> bool:
@@ -109,7 +113,7 @@ def reboot(self) -> bool:
 
         All non-legacy devices support gateway-rebooting.
         """
-        return not self.smile_legacy
+        return not self.smile.legacy
 
     async def connect(self) -> Version:
         """Connect to the Plugwise Gateway and determine its name, type, version, and other data."""
@@ -158,16 +162,9 @@ async def connect(self) -> Version:
                 self._opentherm_device,
                 self._request,
                 self._schedule_old_states,
-                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.smile,
             )
-            if not self.smile_legacy
+            if not self.smile.legacy
             else SmileLegacyAPI(
                 self._is_thermostat,
                 self._loc_data,
@@ -176,21 +173,14 @@ async def connect(self) -> Version:
                 self._request,
                 self._stretch_v2,
                 self._target_smile,
-                self.smile_hostname,
-                self.smile_hw_version,
-                self.smile_mac_address,
-                self.smile_model,
-                self.smile_name,
-                self.smile_type,
-                self.smile_version,
-                self.smile_zigbee_mac_address,
+                self.smile,
             )
         )
 
         # Update all endpoints on first connect
         await self._smile_api.full_xml_update()
 
-        return self.smile_version
+        return cast(Version, self.smile.version)
 
     async def _smile_detect(
         self, result: etree.Element, dsmrmain: etree.Element
@@ -203,23 +193,25 @@ async def _smile_detect(
         if (gateway := result.find("./gateway")) is not None:
             if (v_model := gateway.find("vendor_model")) is not None:
                 model = v_model.text
-            self.smile_version = parse(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
-            self.smile_model_id = gateway.find("vendor_model").text
+            self.smile.version = parse(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
+            self.smile.model_id = gateway.find("vendor_model").text
         else:
             model = await self._smile_detect_legacy(result, dsmrmain, model)
 
-        if model == "Unknown" or self.smile_version is None:  # pragma: no cover
+        if model == "Unknown" or self.smile.version == Version(
+            "0.0.0"
+        ):  # pragma: no cover
             # Corner case check
             LOGGER.error(
                 "Unable to find model or version information, please create"
                 " an issue on http://github.com/plugwise/python-plugwise"
             )
             raise UnsupportedDeviceError
 
-        version_major = str(self.smile_version.major)
+        version_major = str(self.smile.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:
@@ -230,7 +222,7 @@ async def _smile_detect(
             )
             raise UnsupportedDeviceError
 
-        if not self.smile_legacy:
+        if not self.smile.legacy:
             self._timeout = DEFAULT_TIMEOUT
 
         if self._target_smile in ("smile_open_therm_v2", "smile_thermo_v3"):
@@ -240,14 +232,14 @@ async def _smile_detect(
             )  # pragma: no cover
             raise UnsupportedDeviceError  # pragma: no cover
 
-        self.smile_model = "Gateway"
-        self.smile_name = SMILES[self._target_smile].smile_name
-        self.smile_type = SMILES[self._target_smile].smile_type
+        self.smile.model = "Gateway"
+        self.smile.name = SMILES[self._target_smile].smile_name
+        self.smile.type = SMILES[self._target_smile].smile_type
 
-        if self.smile_type == "stretch":
+        if self.smile.type == "stretch":
             self._stretch_v2 = int(version_major) == 2
 
-        if self.smile_type == "thermostat":
+        if self.smile.type == "thermostat":
             self._is_thermostat = True
             # For Adam, Anna, determine the system capabilities:
             # Find the connected heating/cooling device (heater_central),
@@ -275,36 +267,36 @@ async def _smile_detect_legacy(
         return_model = model
         # Stretch: find the MAC of the zigbee master_controller (= Stick)
         if (network := result.find("./module/protocols/master_controller")) is not None:
-            self.smile_zigbee_mac_address = network.find("mac_address").text
+            self.smile.zigbee_mac_address = network.find("mac_address").text
         # Find the active MAC in case there is an orphaned Stick
         if zb_networks := result.findall("./network"):
             for zb_network in zb_networks:
                 if zb_network.find("./nodes/network_router") is not None:
                     network = zb_network.find("./master_controller")
-                    self.smile_zigbee_mac_address = network.find("mac_address").text
+                    self.smile.zigbee_mac_address = network.find("mac_address").text
 
         # Legacy Anna or Stretch:
         if (
             result.find('./appliance[type="thermostat"]') is not None
             or network is not None
         ):
             system = await self._request(SYSTEM)
-            self.smile_version = parse(system.find("./gateway/firmware").text)
+            self.smile.version = parse(system.find("./gateway/firmware").text)
             return_model = str(system.find("./gateway/product").text)
-            self.smile_hostname = system.find("./gateway/hostname").text
+            self.smile.hostname = system.find("./gateway/hostname").text
             # If wlan0 contains data it's active, eth0 should be checked last as is preferred
             for network in ("wlan0", "eth0"):
                 locator = f"./{network}/mac"
                 if (net_locator := system.find(locator)) is not None:
-                    self.smile_mac_address = net_locator.text
+                    self.smile.mac_address = net_locator.text
 
         # P1 legacy:
         elif dsmrmain is not None:
             status = await self._request(STATUS)
-            self.smile_version = parse(status.find("./system/version").text)
+            self.smile.version = parse(status.find("./system/version").text)
             return_model = str(status.find("./system/product").text)
-            self.smile_hostname = status.find("./network/hostname").text
-            self.smile_mac_address = status.find("./network/mac_address").text
+            self.smile.hostname = status.find("./network/hostname").text
+            self.smile.mac_address = status.find("./network/mac_address").text
         else:  # pragma: no cover
             # No cornercase, just end of the line
             LOGGER.error(
@@ -313,7 +305,7 @@ async def _smile_detect_legacy(
             )
             raise ResponseError
 
-        self.smile_legacy = True
+        self.smile.legacy = True
         return return_model
 
     async def async_update(self) -> dict[str, GwEntityData]:

plugwise/common.py

@@ -55,17 +55,16 @@ def __init__(self) -> None:
         self._heater_id: str = NONE
         self._on_off_device: bool
         self.gw_entities: dict[str, GwEntityData] = {}
-        self.smile_name: str
-        self.smile_type: str
+        self.smile: Munch
 
     @property
     def heater_id(self) -> str:
         """Return the heater-id."""
         return self._heater_id
 
-    def smile(self, name: str) -> bool:
+    def check_name(self, name: str) -> bool:
         """Helper-function checking the smile-name."""
-        return self.smile_name == name
+        return bool(self.smile.name == name)
 
     def _appl_heater_central_info(
         self,
@@ -173,7 +172,7 @@ def _get_group_switches(self) -> dict[str, GwEntityData]:
         """
         switch_groups: dict[str, GwEntityData] = {}
         # P1 and Anna don't have switchgroups
-        if self.smile_type == "power" or self.smile(ANNA):
+        if self.smile.type == "power" or self.check_name(ANNA):
             return switch_groups
 
         for group in self._domain_objects.findall("./group"):

plugwise/data.py

@@ -35,7 +35,7 @@ def _all_entity_data(self) -> None:
         Collect data for each entity and add to self.gw_entities.
         """
         self._update_gw_entities()
-        if self.smile(ADAM):
+        if self.check_name(ADAM):
             self._update_zones()
             self.gw_entities.update(self._zones)
 
@@ -111,7 +111,7 @@ def _add_or_update_notifications(
         """Helper-function adding or updating the Plugwise notifications."""
         if (
             entity_id == self._gateway_id
-            and (self._is_thermostat or self.smile_type == "power")
+            and (self._is_thermostat or self.smile.type == "power")
         ) or (
             "binary_sensors" in entity
             and "plugwise_notification" in entity["binary_sensors"]
@@ -124,7 +124,7 @@ def _update_for_cooling(self, entity: GwEntityData) -> None:
         """Helper-function for adding/updating various cooling-related values."""
         # For Anna and heating + cooling, replace setpoint with setpoint_high/_low
         if (
-            self.smile(ANNA)
+            self.check_name(ANNA)
             and self._cooling_present
             and entity["dev_class"] == "thermostat"
         ):
@@ -194,11 +194,11 @@ def _get_entity_data(self, entity_id: str) -> GwEntityData:
         # Switching groups data
         self._entity_switching_group(entity, data)
         # Adam data
-        if self.smile(ADAM):
+        if self.check_name(ADAM):
             self._get_adam_data(entity, data)
 
         # Thermostat data for Anna (presets, temperatures etc)
-        if self.smile(ANNA) and entity["dev_class"] == "thermostat":
+        if self.check_name(ANNA) and entity["dev_class"] == "thermostat":
             self._climate_data(entity_id, entity, data)
             self._get_anna_control_state(data)
 

plugwise/helper.py

@@ -85,11 +85,7 @@ def __init__(self) -> None:
         self._gateway_id: str = NONE
         self._zones: dict[str, GwEntityData]
         self.gw_entities: dict[str, GwEntityData]
-        self.smile_hw_version: str | None
-        self.smile_mac_address: str | None
-        self.smile_model: str
-        self.smile_model_id: str | None
-        self.smile_version: version.Version
+        self.smile: Munch = Munch()
 
     @property
     def gateway_id(self) -> str:
@@ -160,7 +156,7 @@ def _all_appliances(self) -> None:
 
             self._create_gw_entities(appl)
 
-        if self.smile_type == "power":
+        if self.smile.type == "power":
             self._get_p1_smartmeter_info()
 
         # Sort the gw_entities
@@ -268,16 +264,16 @@ 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"]
-        appl.firmware = str(self.smile_version)
-        appl.hardware = self.smile_hw_version
-        appl.mac = self.smile_mac_address
-        appl.model = self.smile_model
-        appl.model_id = self.smile_model_id
-        appl.name = self.smile_name
+        appl.firmware = str(self.smile.version)
+        appl.hardware = self.smile.hw_version
+        appl.mac = self.smile.mac_address
+        appl.model = self.smile.model
+        appl.model_id = self.smile.model_id
+        appl.name = self.smile.name
         appl.vendor_name = "Plugwise"
 
         # Adam: collect the ZigBee MAC address of the Smile
-        if self.smile(ADAM):
+        if self.check_name(ADAM):
             if (
                 found := self._domain_objects.find(".//protocols/zig_bee_coordinator")
             ) is not None:
@@ -346,7 +342,7 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:
         # Get P1 smartmeter data from LOCATIONS
         entity = self.gw_entities[entity_id]
         # !! DON'T CHANGE below two if-lines, will break stuff !!
-        if self.smile_type == "power":
+        if self.smile.type == "power":
             if entity["dev_class"] == "smartmeter":
                 data.update(self._power_data_from_location())
 
@@ -383,7 +379,7 @@ def _get_measurement_data(self, entity_id: str) -> GwEntityData:
             data.pop("c_heating_state")
             self._count -= 1
 
-        if self._is_thermostat and self.smile(ANNA):
+        if self._is_thermostat and self.check_name(ANNA):
             self._update_anna_cooling(entity_id, data)
 
         self._cleanup_data(data)
@@ -484,7 +480,7 @@ def _get_actuator_functionalities(
                 item == "thermostat"
                 and (
                     entity["dev_class"] != "climate"
-                    if self.smile(ADAM)
+                    if self.check_name(ADAM)
                     else entity["dev_class"] != "thermostat"
                 )
             ):
@@ -539,7 +535,7 @@ def _get_actuator_mode(
 
         Collect the requested gateway mode.
         """
-        if not (self.smile(ADAM) and entity_id == self._gateway_id):
+        if not (self.check_name(ADAM) and entity_id == self._gateway_id):
             return None
 
         if (search := search_actuator_functionalities(appliance, key)) is not None:
@@ -605,10 +601,10 @@ def _process_on_off_device_c_heating_state(self, data: GwEntityData) -> None:
 
         Solution for Core issue #81839.
         """
-        if self.smile(ANNA):
+        if self.check_name(ANNA):
             data["binary_sensors"]["heating_state"] = data["c_heating_state"]
 
-        if self.smile(ADAM):
+        if self.check_name(ADAM):
             # First count when not present, then create and init to False.
             # When present init to False
             if "heating_state" not in data["binary_sensors"]:
@@ -799,8 +795,8 @@ def _control_state(self, data: GwEntityData, loc_id: str) -> str | bool:
 
         # Handle missing control_state in regulation_mode off for firmware >= 3.2.0 (issue #776)
         # In newer firmware versions, default to "off" when control_state is not present
-        if self.smile_version != version.Version("0.0.0"):
-            if self.smile_version >= version.parse("3.2.0"):
+        if self.smile.version != version.Version("0.0.0"):
+            if self.smile.version >= version.parse("3.2.0"):
                 return "off"
 
             # Older Adam firmware does not have the control_state xml-key