pylint improvements

Author: CoMPaTech

.pre-commit-config.yaml

@@ -1,5 +1,18 @@
 repos:
   # Run manually in CI skipping the branch checks
+  - repo: https://github.com/PyCQA/pylint
+    rev: v2.12.2
+    hooks:
+      - id: pylint
+        name: pylint
+        entry: pylint
+        language: system
+        types: [python]
+        args:
+          [
+            "-rn",  # Only display messages
+            "-sn",  # Don't display the score
+          ]
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.1.0
     hooks:

plugwise/connections/__init__.py

@@ -58,6 +58,8 @@ def _reader_deamon(self):
             time.sleep(0.01)
         _LOGGER.debug("Reader daemon stopped")
 
+    #  TODO: function instead of self
+    #  pylint: disable=no-self-use
     def _read_data(self):
         """placeholder."""
         return b"0000"

plugwise/controller.py

@@ -280,11 +280,7 @@ def message_handler(self, message):
             else:
                 self._log_status_message(message)
             self.message_processor(message)
-            if (
-                message.seq_id != b"FFFF"
-                and message.seq_id != b"FFFE"
-                and message.seq_id != b"FFFD"
-            ):
+            if message.seq_id not in [b"FFFF", b"FFFE", b"FFFD"]:
                 self._post_message_action(
                     message.seq_id, None, message.__class__.__name__
                 )

plugwise/helper.py

@@ -7,7 +7,6 @@
 
 # This way of importing aiohttp is because of patch/mocking in testing (aiohttp timeouts)
 from aiohttp import BasicAuth, ClientSession, ClientTimeout, ServerTimeoutError
-from dateutil import tz
 from dateutil.parser import parse
 from defusedxml import ElementTree as etree
 from munch import Munch
@@ -17,24 +16,16 @@
 
 from .constants import (
     APPLIANCES,
-    ATTR_ICON,
-    ATTR_ID,
     ATTR_NAME,
-    ATTR_STATE,
     ATTR_TYPE,
     ATTR_UNIT_OF_MEASUREMENT,
     BINARY_SENSORS,
-    COOLING_ICON,
     DEVICE_MEASUREMENTS,
-    DOMAIN_OBJECTS,
     ENERGY_KILO_WATT_HOUR,
     ENERGY_WATT_HOUR,
     FAKE_LOC,
-    FLAME_ICON,
     HEATER_CENTRAL_MEASUREMENTS,
-    HEATING_ICON,
     HOME_MEASUREMENTS,
-    IDLE_ICON,
     LOCATIONS,
     POWER_WATT,
     SENSORS,
@@ -49,7 +40,6 @@
     ResponseError,
 )
 from .util import (
-    determine_selected,
     escape_illegal_xml_characters,
     format_measure,
     in_between,
@@ -165,9 +155,9 @@ def schemas_schedule_temp(schedules, name):
     for i in range(length):
         result_1 = schema_list[i][0]
         start = schema_list[i][1]
-        n = (i + 1) % (length - 1)
-        result_2 = schema_list[n][0]
-        end = schema_list[n][1]
+        j = (i + 1) % (length - 1)
+        result_2 = schema_list[j][0]
+        end = schema_list[j][1]
         now = dt.datetime.now().time()
         if (
             result_1 == dt.datetime.now().weekday()
@@ -326,6 +316,39 @@ async def close_connection(self):
 class SmileHelper:
     """The SmileHelper class."""
 
+    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 = []
+
     def _locations_legacy(self):
         """Helper-function for _all_locations().
         Create locations for legacy devices.
@@ -420,7 +443,8 @@ def _get_module_data(self, appliance, locator, mod_type):
         if appl_search is not None:
             link_id = appl_search.attrib["id"]
             locator = f".//{mod_type}[@id='{link_id}']...."
-            if (module := self._modules.find(locator)) is not None:
+            module = self._modules.find(locator)
+            if module is not None:
                 v_name = module.find("vendor_name").text
                 v_model = module.find("vendor_model").text
                 hw_version = module.find("hardware_version").text

plugwise/messages/responses.py

@@ -66,7 +66,7 @@ def deserialize(self, response):
         self.msg_id = response[4:8]
         self.seq_id = response[8:12]
         response = response[12:]
-        if self.format_size == MESSAGE_SMALL or self.format_size == MESSAGE_LARGE:
+        if self.format_size in [MESSAGE_SMALL, MESSAGE_LARGE]:
             self.ack_id = response[:4]
             response = response[4:]
         if self.format_size != MESSAGE_SMALL:
@@ -337,6 +337,8 @@ def __init__(self):
         self.datetime = DateTime()
         self.last_logaddr = LogAddr(0, length=8)
         self.relay_state = Int(0, length=2)
+        #  TODO: snake-style
+        #  pylint: disable=invalid-name
         self.hz = Int(0, length=2)
         self.hw_ver = String(None, length=12)
         self.fw_ver = UnixTimestamp(0)

plugwise/nodes/circle.py

@@ -685,8 +685,9 @@ def _response_energy_counters(self, message: CircleEnergyCountersResponse):
                 self._energy_last_populated_slot = _slot
 
             # Store most recent timestamp of collected pulses
-            if self._energy_last_collected_timestamp < _log_timestamp:
-                self._energy_last_collected_timestamp = _log_timestamp
+            self._energy_last_collected_timestamp = max(
+                self._energy_last_collected_timestamp, _log_timestamp
+            )
 
             # Trigger history rollover
             if (
@@ -746,11 +747,11 @@ def _response_energy_counters(self, message: CircleEnergyCountersResponse):
             )
             self._update_energy_today_now(False, _history_rollover, _midnight_rollover)
         else:
-            _LOGGER.info(
-                "_response_energy_counters for %s | self._energy_history_collecting running",
-                self.mac,
-                str(_local_midnight_timestamp),
+            logstring = (
+                "_response_energy_counters for %s | self._energy_history_collecting running %s"
+                % (self.mac, str(_local_midnight_timestamp))
             )
+            _LOGGER.info(logstring)
 
         # Cleanup energy history for more than 8 day's ago
         _8_days_ago = datetime.utcnow().replace(

plugwise/nodes/scan.py

@@ -97,10 +97,14 @@ def _process_switch_group(self, message):
                 self.mac,
             )
 
+    #  TODO: snakestyle name
+    #  pylint: disable=invalid-name
     def CalibrateLight(self, callback=None):
         """Queue request to calibration light sensitivity"""
         self._queue_request(ScanLightCalibrateRequest(self._mac), callback)
 
+    #  TODO: snakestyle name
+    #  pylint: disable=invalid-name
     def Configure_scan(
         self,
         motion_reset_timer=SCAN_MOTION_RESET_TIMER,
@@ -126,6 +130,8 @@ def Configure_scan(
             callback,
         )
 
+    #  TODO: snakestyle name
+    #  pylint: disable=invalid-name
     def SetMotionAction(self, callback=None):
         """Queue Configure Scan to signal motion"""
         # TODO:

plugwise/nodes/sed.py

@@ -75,14 +75,13 @@ def _process_awake_response(self, message):
             str(message.awake_type.value),
             self.mac,
         )
-        if (
-            message.awake_type.value == SED_AWAKE_MAINTENANCE
-            or message.awake_type.value == SED_AWAKE_FIRST
-            or message.awake_type.value == SED_AWAKE_STARTUP
-            or message.awake_type.value == SED_AWAKE_BUTTON
-        ):
-            for request in self._sed_requests:
-                (request_message, callback) = self._sed_requests[request]
+        if message.awake_type.value in [
+            SED_AWAKE_MAINTENANCE,
+            SED_AWAKE_FIRST,
+            SED_AWAKE_STARTUP,
+            SED_AWAKE_BUTTON,
+        ]:
+            for request_message, callback in self._sed_requests.items():
                 _LOGGER.info(
                     "Send queued %s message to SED node %s",
                     request_message.__class__.__name__,
@@ -136,6 +135,8 @@ def _wake_up_interval_accepted(self):
         """Callback after wake up interval is received and accepted by SED."""
         self._wake_up_interval = self._new_maintenance_interval
 
+    #  TODO: snakestyle name
+    #  pylint: disable=invalid-name
     def Configure_SED(
         self,
         stay_active=SED_STAY_ACTIVE,

plugwise/smile.py

@@ -1,14 +1,10 @@
 """Use of this source code is governed by the MIT license found in the LICENSE file.
 Plugwise backend module for Home Assistant Core.
 """
-import asyncio
-import copy
 import logging
 
 import aiohttp
 
-from defusedxml import ElementTree as etree
-
 # Dict as class
 from munch import Munch
 
@@ -74,11 +70,11 @@ def _all_device_data(self):
             self._append_special(data, dev_id, temp_bs_dict, temp_s_dict)
 
             dev_and_data.update(data)
-            if temp_bs_dict != {}:
+            if temp_bs_dict:
                 dev_and_data["binary_sensors"] = temp_bs_dict
-            if temp_s_dict != {}:
+            if temp_s_dict:
                 dev_and_data["sensors"] = temp_s_dict
-            if temp_sw_dict != {}:
+            if temp_sw_dict:
                 dev_and_data["switches"] = temp_sw_dict
 
             self.gw_devices[dev_id] = dev_and_data
@@ -468,7 +464,7 @@ async def async_update(self):
             data = self._get_device_data(dev_id)
             for key, value in list(data.items()):
                 if key in dev_dict:
-                    self.gw_devices[dev_id][key] = value
+                    dev_dict[key] = value
             if "binary_sensors" in dev_dict:
                 for key, value in list(data.items()):
                     update_helper(
@@ -531,7 +527,7 @@ async def set_schedule_state(self, loc_id, name, state):
             return await self._set_schedule_state_legacy(name, state)
 
         schema_rule = self._rule_ids_by_name(str(name), loc_id)
-        if schema_rule == {} or schema_rule is None:
+        if not schema_rule or schema_rule is None:
             return False
 
         schema_rule_id = next(iter(schema_rule))

plugwise/stick.py

@@ -610,21 +610,21 @@ def _update_loop(self):
                 if datetime.now().day != day_of_month:
                     day_of_month = datetime.now().day
                     _sync_clock = True
-                for mac in self._device_nodes:
-                    if self._device_nodes[mac]:
-                        if self._device_nodes[mac].battery_powered:
+                for mac, device in self._device_nodes.items():
+                    if device:
+                        if device.battery_powered:
                             # Check availability state of SED's
                             self._check_availability_of_seds(mac)
                         else:
                             # Do ping request for all non SED's
-                            self._device_nodes[mac].do_ping()
+                            device.do_ping()
 
-                        if self._device_nodes[mac].measures_power:
+                        if device.measures_power:
                             # Request current power usage
-                            self._device_nodes[mac].request_power_update()
+                            device.request_power_update()
                             # Sync internal clock of power measure nodes once a day
                             if _sync_clock:
-                                self._device_nodes[mac].sync_clock()
+                                device.sync_clock()
                 _sync_clock = False
 
                 # Do a single ping for undiscovered nodes once per 10 update cycles
@@ -669,8 +669,8 @@ def auto_update(self, timer=None):
             # Timer based on a minimum of 5 seconds + 1 second for each node supporting power measurement
             if not self._auto_update_manually:
                 count_nodes = 0
-                for mac in self._device_nodes:
-                    if self._device_nodes[mac].measures_power:
+                for _, node in self._device_nodes.items():
+                    if node.measures_power:
                         count_nodes += 1
                 self._auto_update_timer = 5 + (count_nodes * 1)
                 _LOGGER.info(
@@ -680,7 +680,7 @@ def auto_update(self, timer=None):
         if not self._run_update_thread:
             self._update_thread.start()
 
-    ### Helper functions ###
+    # Helper functions
     def do_callback(self, callback_type, callback_arg=None):
         """Helper to execute registered callbacks for specified callback type."""
         if callback_type in self._stick_callbacks: