Refactoring with the help of Sourcery (XKNX#1157)

examples/example_disconnect.py

@@ -29,7 +29,7 @@ async def main():
     await udp_transport.connect()
     local_hpai = HPAI(*udp_transport.getsockname())
 
-    for i in range(0, 255):
+    for i in range(255):
 
         conn_state = ConnectionState(
             udp_transport, communication_channel_id=i, local_hpai=local_hpai

examples/example_powermeter_mqtt.py

@@ -15,6 +15,7 @@
 Please join XKNX on Discord (https://discord.gg/EuAQDXU) and chat with JohanElmis for
 specific questions.
 """
+
 import asyncio
 import re
 import sys
@@ -32,8 +33,8 @@
     # import pprint
 except ImportError as import_err:
     err_list = str(import_err).split(" ")
-    print("Unable to import module: " + err_list[3])
-    print("Please install the " + err_list[3] + " module for Python.")
+    print(f"Unable to import module: {err_list[3]}")
+    print(f"Please install the {err_list[3]} module for Python.")
     sys.exit()
 
 BROKER_ADDRESS = "127.0.0.1"
@@ -64,8 +65,8 @@
 async def device_updated_cb(device):
     """Do something with the updated device."""
     # print(device.name + ' ' + str(device.resolve_state()) + ' ' + device.unit_of_measurement())
-    topic = None
     value = None
+    topic = None
     if re.search("^EL-T-O_", device.name):
         metric = str(device.name)[7:]
         value = device.resolve_state()
@@ -99,12 +100,7 @@ async def device_updated_cb(device):
             )
     else:
         print(
-            "Uncatched metric: "
-            + device.name
-            + " "
-            + str(device.resolve_state())
-            + " "
-            + device.unit_of_measurement()
+            f"Uncatched metric: {device.name} {str(device.resolve_state())} {device.unit_of_measurement()}"
         )
 
     if topic and value:
@@ -124,7 +120,7 @@ async def device_updated_cb(device):
         # My latest version of the library doesn't send the value after the MQTT Topic, but a JSON structure
         # that also contains time.
 
-        print(topic + " " + str(value))
+        print(f"{topic} {str(value)}")
         # ts = int(time.time() * 1000)
         mqttc.publish(topic, value)
 

script/sensor_table_generator.py

@@ -56,7 +56,7 @@ def _update_column_width(self, index, text: str):
     def __repr__(self):
         def _format_column_ljust(index):
             content = getattr(self, index)
-            return "| " + content.ljust(Row.column_width[index] + 1)
+            return f"| {content.ljust(Row.column_width[index] + 1)}"
 
         _row = ""
         for column in COLUMN_ORDER:
@@ -94,8 +94,8 @@ def _get_dpt_number_from_docstring(self, dpt_class: DPTBase):
                 if text.startswith("DPT"):
                     return text.split()[1]
         except IndexError:
-            print("Error: Could not read docstring for: %s" % dpt_class)
-        print("Error: Could not find DPT in docstring for: %s" % dpt_class)
+            print(f"Error: Could not read docstring for: {dpt_class}")
+        print(f"Error: Could not find DPT in docstring for: {dpt_class}")
         raise ValueError
 
     def _dpt_number_sort(self, dpt_str: str) -> int:

test/devices_tests/climate_test.py

@@ -355,10 +355,7 @@ async def test_set_operation_mode_with_separate_addresses(self):
 
         await climate_mode.set_operation_mode(HVACOperationMode.COMFORT)
         assert xknx.telegrams.qsize() == 4
-        telegrams = []
-        for _ in range(4):
-            telegrams.append(xknx.telegrams.get_nowait())
-
+        telegrams = [xknx.telegrams.get_nowait() for _ in range(4)]
         test_telegrams = [
             Telegram(
                 destination_address=GroupAddress("1/2/4"),
@@ -982,9 +979,7 @@ async def test_sync_operation_mode_state(self):
         await climate_mode.sync()
         assert xknx.telegrams.qsize() == 3
 
-        telegrams = []
-        for _ in range(3):
-            telegrams.append(xknx.telegrams.get_nowait())
+        telegrams = [xknx.telegrams.get_nowait() for _ in range(3)]
         assert telegrams == [
             Telegram(GroupAddress("1/2/5"), payload=GroupValueRead()),
             Telegram(GroupAddress("1/2/6"), payload=GroupValueRead()),

test/devices_tests/light_test.py

@@ -297,10 +297,7 @@ async def test_sync(self):
         await light.sync()
         assert xknx.telegrams.qsize() == expected_telegrams
 
-        telegrams = []
-        for _ in range(expected_telegrams):
-            telegrams.append(xknx.telegrams.get_nowait())
-
+        telegrams = [xknx.telegrams.get_nowait() for _ in range(expected_telegrams)]
         test_telegrams = [
             Telegram(
                 destination_address=GroupAddress("1/2/3"), payload=GroupValueRead()

test/dpt_tests/dpt_test.py

@@ -116,18 +116,20 @@ def test_dpt_subclasses_definition_types(self):
 
     def test_dpt_subclasses_no_duplicate_value_types(self):
         """Test for duplicate value_type values in subclasses of DPTBase."""
-        value_types = []
-        for dpt in DPTBase.dpt_class_tree():
-            if dpt.value_type is not None:
-                value_types.append(dpt.value_type)
+        value_types = [
+            dpt.value_type
+            for dpt in DPTBase.dpt_class_tree()
+            if dpt.value_type is not None
+        ]
         assert len(value_types) == len(set(value_types))
 
     def test_dpt_subclasses_no_duplicate_dpt_number(self):
         """Test for duplicate value_type values in subclasses of DPTBase."""
-        dpt_tuples = []
-        for dpt in DPTBase.dpt_class_tree():
-            if dpt.dpt_main_number is not None and dpt.dpt_sub_number is not None:
-                dpt_tuples.append((dpt.dpt_main_number, dpt.dpt_sub_number))
+        dpt_tuples = [
+            (dpt.dpt_main_number, dpt.dpt_sub_number)
+            for dpt in DPTBase.dpt_class_tree()
+            if dpt.dpt_main_number is not None and dpt.dpt_sub_number is not None
+        ]
         assert len(dpt_tuples) == len(set(dpt_tuples))
 
     def test_dpt_alternative_notations(self):

test/knxip_tests/routing_indication_test.py

@@ -25,7 +25,7 @@ def test_from_knx_to_knx(self):
         raw = bytes.fromhex("06 10 05 30 00 12 29 00 bc d0 12 02 01 51 02 00 40 f0")
         knxipframe, _ = KNXIPFrame.from_knx(raw)
 
-        assert knxipframe.header.to_knx() == raw[0:6]
+        assert knxipframe.header.to_knx() == raw[:6]
         assert knxipframe.body.to_knx() == raw[6:]
         assert knxipframe.to_knx() == raw
 
@@ -49,7 +49,7 @@ def test_telegram_set(self):
             "06 10 05 30 00 14 29 00 bc d0 12 02 01 51 04 00 80 0d 17 2a"
         )
 
-        assert knxipframe.header.to_knx() == raw[0:6]
+        assert knxipframe.header.to_knx() == raw[:6]
         assert knxipframe.body.to_knx() == raw[6:]
         assert knxipframe.to_knx() == raw
 
@@ -63,7 +63,7 @@ def test_end_to_end_routing_indication(self):
         routing_indication = RoutingIndication(raw_cemi=raw_cemi)
         knxipframe2 = KNXIPFrame.init_from_body(routing_indication)
 
-        assert knxipframe2.header.to_knx() == raw[0:6]
+        assert knxipframe2.header.to_knx() == raw[:6]
         assert knxipframe2.body.to_knx() == raw[6:]
         assert knxipframe2.to_knx() == raw
 

xknx/cemi/cemi_handler.py

@@ -84,7 +84,9 @@ def telegram_received(self, telegram: Telegram) -> None:
         if isinstance(telegram.tpci, tpci.TDataGroup):
             self.xknx.telegrams.put_nowait(telegram)
             return
-        if isinstance(telegram.destination_address, IndividualAddress):
-            if telegram.destination_address != self.xknx.current_address:
-                return
+        if (
+            isinstance(telegram.destination_address, IndividualAddress)
+            and telegram.destination_address != self.xknx.current_address
+        ):
+            return
         self.xknx.management.process(telegram)

xknx/core/state_updater.py

@@ -209,9 +209,8 @@ async def connection_state_change_callback(
         if state == XknxConnectionState.CONNECTED:
             if not self.started:
                 self._start()
-        else:
-            if self.started:
-                self._stop()
+        elif self.started:
+            self._stop()
 
 
 class StateTrackerType(Enum):

xknx/devices/device.py

@@ -128,10 +128,10 @@ def get_name(self) -> str:
 
     def has_group_address(self, group_address: DeviceGroupAddress) -> bool:
         """Test if device has given group address."""
-        for remote_value in self._iter_remote_values():
-            if remote_value.has_group_address(group_address):
-                return True
-        return False
+        return any(
+            remote_value.has_group_address(group_address)
+            for remote_value in self._iter_remote_values()
+        )
 
     def __eq__(self, other: object) -> bool:
         """Compare for quality."""

xknx/devices/devices.py

@@ -61,10 +61,7 @@ def __len__(self) -> int:
 
     def __contains__(self, key: str) -> bool:
         """Return if devices with name 'key' is within devices."""
-        for device in self.__devices:
-            if device.name == key:
-                return True
-        return False
+        return any(device.name == key for device in self.__devices)
 
     def add(self, device: Device) -> None:
         """Add device to devices vector."""

xknx/devices/fan.py

@@ -171,9 +171,9 @@ def __str__(self) -> str:
         """Return object as readable string."""
 
         str_oscillation = (
-            ""
-            if not self.supports_oscillation
-            else f" oscillation={self.oscillation.group_addr_str()}"
+            f" oscillation={self.oscillation.group_addr_str()}"
+            if self.supports_oscillation
+            else ""
         )
 
         return f'<Fan name="{self.name}" speed={self.speed.group_addr_str()}{str_oscillation} />'