Merge pull request #485 from plugwise/qual

Quality improvements (through sonar cloud)

Author: CoMPaTech

CHANGELOG.md

@@ -3,9 +3,10 @@
 ## Ongoing
 
 - Refresh adam_plus_anna_new userdata and adapt.
-- Bump actions and requirements to Python 3.12
-- Ruff as per #470 (defaulting black and isort to ruff)
-- Modularize/split testing, including separation of code and data
+- Bump actions and requirements to Python 3.12.
+- Ruff as per #470 (defaulting black and isort to ruff).
+- Modularize/split testing, including separation of code and data.
+- Improve quality as indicated by SonarCloud.
 
 ## v0.35.4
 

plugwise/__init__.py

@@ -400,6 +400,7 @@ async def _smile_detect_legacy(
         self, result: etree, dsmrmain: etree, model: str
     ) -> str:
         """Helper-function for _smile_detect()."""
+        return_model = model
         # Stretch: find the MAC of the zigbee master_controller (= Stick)
         if network := result.find("./module/protocols/master_controller"):
             self.smile_zigbee_mac_address = network.find("mac_address").text
@@ -417,7 +418,7 @@ async def _smile_detect_legacy(
         ):
             self._system = await self._request(SYSTEM)
             self.smile_fw_version = self._system.find("./gateway/firmware").text
-            model = self._system.find("./gateway/product").text
+            return_model = self._system.find("./gateway/product").text
             self.smile_hostname = self._system.find("./gateway/hostname").text
             # If wlan0 contains data it's active, so eth0 should be checked last
             for network in ("wlan0", "eth0"):
@@ -428,10 +429,9 @@ async def _smile_detect_legacy(
         elif dsmrmain is not None:
             self._status = await self._request(STATUS)
             self.smile_fw_version = self._status.find("./system/version").text
-            model = self._status.find("./system/product").text
+            return_model = self._status.find("./system/product").text
             self.smile_hostname = self._status.find("./network/hostname").text
             self.smile_mac_address = self._status.find("./network/mac_address").text
-
         else:  # pragma: no cover
             # No cornercase, just end of the line
             LOGGER.error(
@@ -441,7 +441,7 @@ async def _smile_detect_legacy(
             raise ResponseError
 
         self._smile_legacy = True
-        return model
+        return return_model
 
     async def _smile_detect(self, result: etree, dsmrmain: etree) -> None:
         """Helper-function for connect().

plugwise/constants.py

@@ -183,19 +183,25 @@
     "outdoor_temperature",
 )
 
+# Literals
+SMILE_P1 = "Smile P1"
+POWER = "power"
+STRETCH = "stretch"
+THERMOSTAT = "thermostat"
+
 # Known types of Smiles and Stretches
 SMILE = namedtuple("SMILE", "smile_type smile_name")
 SMILES: Final[dict[str, SMILE]] = {
-    "smile_v2": SMILE("power", "Smile P1"),
-    "smile_v3": SMILE("power", "Smile P1"),
-    "smile_v4": SMILE("power", "Smile P1"),
-    "smile_open_therm_v2": SMILE("thermostat", ADAM),
-    "smile_open_therm_v3": SMILE("thermostat", ADAM),
-    "smile_thermo_v1": SMILE("thermostat", ANNA),
-    "smile_thermo_v3": SMILE("thermostat", ANNA),
-    "smile_thermo_v4": SMILE("thermostat", ANNA),
-    "stretch_v2": SMILE("stretch", "Stretch"),
-    "stretch_v3": SMILE("stretch", "Stretch"),
+    "smile_v2": SMILE(POWER, SMILE_P1),
+    "smile_v3": SMILE(POWER, SMILE_P1),
+    "smile_v4": SMILE(POWER, SMILE_P1),
+    "smile_open_therm_v2": SMILE(THERMOSTAT, ADAM),
+    "smile_open_therm_v3": SMILE(THERMOSTAT, ADAM),
+    "smile_thermo_v1": SMILE(THERMOSTAT, ANNA),
+    "smile_thermo_v3": SMILE(THERMOSTAT, ANNA),
+    "smile_thermo_v4": SMILE(THERMOSTAT, ANNA),
+    "stretch_v2": SMILE(STRETCH, "Stretch"),
+    "stretch_v3": SMILE(STRETCH, "Stretch"),
 }
 REQUIRE_APPLIANCES: Final[list[str]] = [
     "smile_thermo_v1",

plugwise/helper.py

@@ -179,19 +179,22 @@ async def _request(
         """Get/put/delete data from a give URL."""
         resp: ClientResponse
         url = f"{self._endpoint}{command}"
+        use_headers = headers
 
         try:
             if method == "delete":
                 resp = await self._websession.delete(url, auth=self._auth)
             if method == "get":
                 # Work-around for Stretchv2, should not hurt the other smiles
-                headers = {"Accept-Encoding": "gzip"}
-                resp = await self._websession.get(url, headers=headers, auth=self._auth)
+                use_headers = {"Accept-Encoding": "gzip"}
+                resp = await self._websession.get(
+                    url, headers=use_headers, auth=self._auth
+                )
             if method == "put":
-                headers = {"Content-type": "text/xml"}
+                use_headers = {"Content-type": "text/xml"}
                 resp = await self._websession.put(
                     url,
-                    headers=headers,
+                    headers=use_headers,
                     data=data,
                     auth=self._auth,
                 )
@@ -314,8 +317,6 @@ def _all_locations(self) -> None:
 
             self._loc_data[loc.loc_id] = {"name": loc.name}
 
-        return
-
     def _get_module_data(
         self, appliance: etree, locator: str, mod_type: str
     ) -> ModelData:
@@ -805,15 +806,16 @@ def _appliance_measurements(
                 updated_date_locator = (
                     f'.//logs/point_log[type="{measurement}"]/updated_date'
                 )
-                if measurement in OBSOLETE_MEASUREMENTS:
-                    if (
-                        updated_date_key := appliance.find(updated_date_locator)
-                    ) is not None:
-                        updated_date = updated_date_key.text.split("T")[0]
-                        date_1 = dt.datetime.strptime(updated_date, "%Y-%m-%d")
-                        date_2 = dt.datetime.now()
-                        if int((date_2 - date_1).days) > 7:
-                            continue
+                if (
+                    measurement in OBSOLETE_MEASUREMENTS
+                    and (updated_date_key := appliance.find(updated_date_locator))
+                    is not None
+                ):
+                    updated_date = updated_date_key.text.split("T")[0]
+                    date_1 = dt.datetime.strptime(updated_date, "%Y-%m-%d")
+                    date_2 = dt.datetime.now()
+                    if int((date_2 - date_1).days) > 7:
+                        continue
 
                 if new_name := getattr(attrs, ATTR_NAME, None):
                     measurement = new_name
@@ -931,7 +933,7 @@ def _get_actuator_functionalities(
                 locator = (
                     f'.//actuator_functionalities/{functionality}[type="{item}"]/{key}'
                 )
-                if (function := xml.find(locator)) is not None:
+                if (pw_function := xml.find(locator)) is not None:
                     if key == "offset":
                         # Add limits and resolution for temperature_offset,
                         # not provided by Plugwise in the XML data
@@ -943,7 +945,7 @@ def _get_actuator_functionalities(
                         key = "setpoint"
 
                     act_key = cast(ActuatorDataType, key)
-                    temp_dict[act_key] = format_measure(function.text, TEMP_CELSIUS)
+                    temp_dict[act_key] = format_measure(pw_function.text, TEMP_CELSIUS)
                     self._count += 1
 
             if temp_dict:
@@ -1267,7 +1269,7 @@ def power_data_energy_diff(
 
         return direct_data
 
-    def _power_data_peak_value(self, direct_data: DeviceData, loc: Munch) -> Munch:
+    def _power_data_peak_value(self, loc: Munch) -> Munch:
         """Helper-function for _power_data_from_location() and _power_data_from_modules()."""
         loc.found = True
         # If locator not found look for P1 gas_consumed or phase data (without tariff)
@@ -1337,12 +1339,11 @@ def _power_data_from_location(self, loc_id: str) -> DeviceData:
         for loc.measurement, loc.attrs in P1_MEASUREMENTS.items():
             for loc.log_type in log_list:
                 for loc.peak_select in peak_list:
-                    # meter_string = ".//{}[type='{}']/"
                     loc.locator = (
                         f'./{loc.log_type}[type="{loc.measurement}"]/period/'
                         f'measurement[@{t_string}="{loc.peak_select}"]'
                     )
-                    loc = self._power_data_peak_value(direct_data, loc)
+                    loc = self._power_data_peak_value(loc)
                     if not loc.found:
                         continue
 
@@ -1377,7 +1378,7 @@ def _power_data_from_modules(self) -> DeviceData:
                             f"./{loc.meas_list[0]}_{loc.log_type}/measurement"
                             f'[@directionality="{loc.meas_list[1]}"][@{t_string}="{loc.peak_select}"]'
                         )
-                        loc = self._power_data_peak_value(direct_data, loc)
+                        loc = self._power_data_peak_value(loc)
                         if not loc.found:
                             continue
 

plugwise/util.py

@@ -27,9 +27,8 @@ def format_measure(measure: str, unit: str) -> float | int:
             result = float(measure)
     except ValueError:
         float_measure = float(measure)
-        if unit == PERCENTAGE:
-            if 0 < float_measure <= 1:
-                return int(float_measure * 100)
+        if unit == PERCENTAGE and 0 < float_measure <= 1:
+            return int(float_measure * 100)
 
         if unit == ENERGY_KILO_WATT_HOUR:
             float_measure = float_measure / 1000