Clean up code.

- Remove more useless variables.
- Remove LRU cache on split byte function in schedules.

pyplumio/helpers/data_types.py

@@ -14,8 +14,7 @@ class DataType(ABC):
 
     def __init__(self, data: bytes | None = None, size: int | None = None):
         """Initialize a new data type."""
-        if size is None:
-            size = self.size
+        size = self.size if size is None else size
 
         if data is not None:
             self._data = data[0:size] if size > 0 else data

pyplumio/helpers/factory.py

@@ -11,10 +11,9 @@ def factory(class_path: str, **kwargs):
     """Return class instance from the class path."""
     try:
         module_name, class_name = class_path.rsplit(".", 1)
-        cls = getattr(
+        return getattr(
             importlib.import_module("." + module_name, "pyplumio"), class_name
-        )
-        return cls(**kwargs)
+        )(**kwargs)
     except Exception:
         _LOGGER.error("Failed to load module (%s)", class_path)
         raise

pyplumio/helpers/schedule.py

@@ -92,16 +92,16 @@ def append(self, item) -> None:
 
     def set_state(
         self,
-        state: Literal["off", "on", "day", "night"],
+        state: Literal["on", "off", "day", "night"],
         start: str = START_OF_DAY,
         end: str = END_OF_DAY,
     ) -> None:
         """Set an interval state.
 
         Can be on of the following:
-        'off', 'on', 'day' or 'night'.
+        'on', 'off', 'day' or 'night'.
         """
-        if state not in [*STATES_ON, *STATES_OFF]:
+        if state not in (STATES_ON + STATES_OFF):
             raise ValueError(f'state "{state}" is not allowed')
 
         index, stop_index = _parse_interval(start, end)

pyplumio/helpers/uid.py

@@ -15,10 +15,9 @@ def _encode_base5(data: bytes) -> str:
     """Encode bytes to a base5 encoded string."""
     key_string = "0123456789ABCDEFGHIJKLMNZPQRSTUV"
     number = int.from_bytes(data, "little")
-    output: str = ""
-    mask = (1 << 5) - 1
+    output = ""
     while number:
-        output = key_string[number & mask] + output
+        output = key_string[number & 0b00011111] + output
         number >>= 5
 
     return output

pyplumio/protocol.py

@@ -71,13 +71,13 @@ async def frame_producer(
         while self.connected.is_set():
             try:
                 if write_queue.qsize() > 0:
-                    request = await write_queue.get()
-                    await self.writer.write(request)
+                    await self.writer.write(await write_queue.get())
                     write_queue.task_done()
 
                 if (response := await self.reader.read()) is not None:
-                    device = self.setup_device_entry(response.sender)
-                    read_queue.put_nowait((device, response))
+                    read_queue.put_nowait(
+                        (self.setup_device_entry(response.sender), response)
+                    )
 
             except FrameDataError as e:
                 _LOGGER.warning("Incorrect payload: %s", e)

pyplumio/structures/__init__.py

@@ -10,9 +10,7 @@
 
 def ensure_device_data(data: EventDataType | None, *args) -> EventDataType:
     """Create or merge multiple device datasets."""
-    if data is None:
-        data = {}
-
+    data = data if data is not None else {}
     for new_data in args:
         data |= new_data
 

pyplumio/structures/regulator_data.py

@@ -58,8 +58,7 @@ def decode(
         )
 
         if schema:
-            regulator_data = data.setdefault(ATTR_REGDATA, RegulatorData())
-            regulator_data.load(
+            data.setdefault(ATTR_REGDATA, RegulatorData()).load(
                 {
                     param_id: self._unpack_regulator_data(message, data_type)
                     for param_id, data_type in schema

pyplumio/structures/schedules.py

@@ -3,7 +3,6 @@
 
 from collections.abc import Sequence
 from dataclasses import dataclass
-from functools import lru_cache
 from itertools import chain
 from typing import TYPE_CHECKING, Final
 
@@ -125,14 +124,9 @@ def collect_schedule_data(name: str, device: Device) -> EventDataType:
     }
 
 
-@lru_cache(maxsize=16)
 def _split_byte(byte: int) -> list[bool]:
     """Split single byte into an eight bits."""
-    bits = []
-    for bit in reversed(range(8)):
-        bits.append(bool(byte & (1 << bit)))
-
-    return bits
+    return [bool(byte & (1 << bit)) for bit in reversed(range(8))]
 
 
 def _join_byte(bits: Sequence[int | bool]) -> int:
@@ -168,8 +162,7 @@ class SchedulesStructure(Structure):
 
     def encode(self, data: EventDataType) -> bytearray:
         """Encode data to the bytearray message."""
-        message = bytearray()
-        message.append(1)
+        message = bytearray([1])
         try:
             message.append(SCHEDULES.index(data[ATTR_TYPE]))
             message.append(int(data[ATTR_SWITCH]))
@@ -178,13 +171,12 @@ def encode(self, data: EventDataType) -> bytearray:
         except (KeyError, ValueError) as e:
             raise FrameDataError from e
 
-        schedule_bytes = []
-        for day in list(schedule):
-            schedule_bytes += [
-                _join_byte(day[i : i + 8]) for i in range(0, len(day), 8)
-            ]
-
-        message += bytearray(schedule_bytes)
+        message += bytearray(
+            chain.from_iterable(
+                [_join_byte(day[i : i + 8]) for i in range(0, len(day), 8)]
+                for day in list(schedule)
+            )
+        )
 
         return message