Add more ruff checks across codebase.

bin/cmdr.py

@@ -55,7 +55,7 @@ def parse_range(rng, max):
             xr = [s.strip() for s in x.split("-")]
             ids.extend(range(int(xr[0]), int(xr[1]) + 1))
         else:
-            raise ValueError('Unknown range type: "%s"' % x)
+            raise ValueError(f'Unknown range type: "{x}"')
     return ids
 
 
@@ -87,7 +87,7 @@ def get_helpers(element, clas):
                 params = " ".join(["<" + p + ">" for p in params])
                 helpers[function_name] = (
                     fn,
-                    "{} {}".format(function_name, params),
+                    f"{function_name} {params}",
                     fn.__doc__[8:],
                 )
     return helpers
@@ -120,8 +120,8 @@ def __init__(self, elk):
                 cmd = element[:-1]
             self.element_cmds[cmd] = (
                 fn,
-                "{} <range list> [subcommand]".format(cmd),
-                "Displays internal state of {}".format(element),
+                f"{cmd} <range list> [subcommand]",
+                f"Displays internal state of {element}",
                 get_helpers(element, cmd.capitalize()),
             )
 
@@ -133,7 +133,7 @@ def __call__(self, line):
         if cmd in self._quit_cmd:
             return Commander.Exit
 
-        print("#blue#{}".format(line))
+        print(f"#blue#{line}")
 
         if cmd in self._help_cmd:
             return self.help(cmd, args)
@@ -147,7 +147,7 @@ def __call__(self, line):
         if cmd in self.element_cmds:
             return self.element_cmds[cmd][0](cmd, args)
 
-        return "#error#Unknown command: {}".format(cmd)
+        return f"#error#Unknown command: {cmd}"
 
     def help(self, cmd, args):
         if len(args) == 0:
@@ -162,16 +162,17 @@ def help(self, cmd, args):
             help_for = args[0]
             if help_for in self.encode_cmds:
                 command = self.encode_cmds[help_for]
-                res = "#green#{}\n{}".format(command.help, command.docs)
+                res = f"#green#{command.help}\n{command.docs}"
 
             elif help_for in self.element_cmds:
-                res = "#green#{}\n{}".format(
-                    self.element_cmds[help_for][1], self.element_cmds[help_for][2]
+                res = (
+                    f"#green#{self.element_cmds[help_for][1]}\n"
+                    f"{self.element_cmds[help_for][2]}"
                 )
-                for k, v in self.element_cmds[help_for][3].items():
-                    res += "\nSubcommand: {}\n{}".format(v[1], v[2])
+                for _k, v in self.element_cmds[help_for][3].items():
+                    res += f"\nSubcommand: {v[1]}\n{v[2]}"
             else:
-                res = "#error#Unknown command: {}".format(help_for)
+                res = f"#error#Unknown command: {help_for}"
         return res
 
     def print_elements(self, cmd, args):
@@ -211,11 +212,11 @@ def encoder(self, cmd, args):
         self.elk.send(self.encode_cmds[cmd].function(*converted))
 
 
-class FocusMixin(object):
+class FocusMixin:
     def mouse_event(self, size, event, button, x, y, focus):
         if focus and hasattr(self, "_got_focus") and self._got_focus:
             self._got_focus()
-        return super(FocusMixin, self).mouse_event(size, event, button, x, y, focus)
+        return super().mouse_event(size, event, button, x, y, focus)
 
 
 class ListView(FocusMixin, urwid.ListBox):
@@ -227,7 +228,7 @@ def __init__(self, model, got_focus, max_size=None):
 
     def mouse_event(self, size, event, button, x, y, focus):
         direction = "up" if button == 4 else "down"
-        return super(ListView, self).keypress(size, direction)
+        return super().keypress(size, direction)
 
     def add(self, line):
         with self._lock:
@@ -281,7 +282,7 @@ class Commander(urwid.Frame):
     Commander.loop(). You can also asynchronously output messages
     with Commander.output('message')"""
 
-    class Exit(object):
+    class Exit:
         pass
 
     PALLETE = [
@@ -344,7 +345,7 @@ def on_line_entered(self, line):
                 res = self._cmd(line)
             except Exception as e:
                 traceback.print_exc()
-                self.output("Error: %s" % e, "error")
+                self.output(f"Error: {e}", "error")
                 return
             if res == Commander.Exit:
                 raise urwid.ExitMainLoop()

elkm1_lib/connection.py

@@ -62,7 +62,7 @@ async def connect(self) -> None:
                         reader, self._writer = await asyncio.open_connection(
                             host=dest, port=param, ssl=ssl_context
                         )
-            except (ValueError, OSError, asyncio.TimeoutError) as err:
+            except (TimeoutError, ValueError, OSError) as err:
                 LOG.warning(
                     "Error connecting to ElkM1 (%s). Retrying in %d seconds",
                     err,
@@ -115,7 +115,7 @@ async def await_msg_response() -> None:
             try:
                 async with asyncio_timeout(MESSAGE_RESPONSE_TIME):
                     await self._response_received.wait()
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 self._notifier.notify("timeout", {"msg_code": q_entry.response_cmd})
             self._response_received.clear()
             self._awaiting_response_command = None
@@ -183,7 +183,7 @@ async def _heartbeat_timer(self) -> None:
             try:
                 async with asyncio_timeout(HEARTBEAT_TIME):
                     await self._heartbeat_event.wait()
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 if self._paused:
                     continue
                 self.disconnect("(heartbeat timeout)")

elkm1_lib/discovery.py

@@ -9,7 +9,6 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from struct import unpack
-from typing import Optional
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -48,11 +47,11 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
         """Trigger on_response."""
         self.on_response(data, addr)
 
-    def error_received(self, exc: Optional[Exception]) -> None:
+    def error_received(self, exc: Exception | None) -> None:
         """Handle error."""
         _LOGGER.error("ELKDiscovery error: %s", exc)
 
-    def connection_lost(self, exc: Optional[Exception]) -> None:
+    def connection_lost(self, exc: Exception | None) -> None:
         """Do nothing on connection lost."""
 
 
@@ -96,16 +95,16 @@ class AIOELKDiscovery:
     def __init__(self) -> None:
         self.found_devices: list[ElkSystem] = []
 
-    def _destination_from_address(self, address: Optional[str]) -> tuple[str, int]:
+    def _destination_from_address(self, address: str | None) -> tuple[str, int]:
         if address is None:
             address = self.BROADCAST_ADDRESS
         return (address, self.DISCOVERY_PORT)
 
     def _process_response(
         self,
-        data: Optional[bytes],
+        data: bytes | None,
         from_address: tuple[str, int],
-        address: Optional[str],
+        address: str | None,
         response_list: dict[tuple[str, int], ElkSystem],
     ) -> bool:
         """Process a response.
@@ -130,7 +129,7 @@ async def _async_run_scan(
         transport: asyncio.DatagramTransport,
         destination: tuple[str, int],
         timeout: int,
-        found_all_future: "asyncio.Future[bool]",
+        found_all_future: asyncio.Future[bool],
     ) -> None:
         """Send the scans."""
         _LOGGER.debug("discover: %s => %s", destination, self.DISCOVER_MESSAGE)
@@ -145,7 +144,7 @@ async def _async_run_scan(
                 await asyncio.wait_for(
                     asyncio.shield(found_all_future), timeout=time_out
                 )
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 if time.monotonic() >= quit_time:
                     return
                 # No response, send broadcast again in cast it got lost
@@ -156,7 +155,7 @@ async def _async_run_scan(
             remain_time = quit_time - time.monotonic()
 
     async def async_scan(
-        self, timeout: int = 10, address: Optional[str] = None
+        self, timeout: int = 10, address: str | None = None
     ) -> list[ElkSystem]:
         """Discover ELK devices."""
         sock = create_udp_socket(self.DISCOVERY_PORT)

elkm1_lib/elements.py

@@ -6,8 +6,8 @@
 
 import re
 from abc import abstractmethod
-from collections.abc import Callable
-from typing import Any, Generator, Generic, Type, TypeVar
+from collections.abc import Callable, Generator
+from typing import Any, Generic, TypeVar
 
 from .connection import Connection
 from .const import TextDescription, TextDescriptions
@@ -81,7 +81,8 @@ def __str__(self) -> str:
             if not k.startswith("_") and k != "name"
         }.items()
         varstr = " ".join(
-            "%s:%s" % item  # pylint: disable=consider-using-f-string
+            # pylint: disable=consider-using-f-string
+            "%s:%s" % item  # noqa
             for item in varlist
         )
         return f"{self._index} '{self.name}' {varstr}"
@@ -105,7 +106,7 @@ def __init__(
         self,
         connection: Connection,
         notifier: Notifier,
-        class_: Type[T],
+        class_: type[T],
         max_elements: int,
     ) -> None:
         self._connection = connection

elkm1_lib/keypads.py

@@ -1,7 +1,6 @@
 """Definition of an ElkM1 Keypad."""
 
 import datetime as dt
-from typing import Optional
 
 from .connection import Connection
 from .const import FunctionKeys, KeypadKeys, Max, TextDescriptions
@@ -17,10 +16,10 @@ def __init__(self, index: int, connection: Connection, notifier: Notifier) -> No
         super().__init__(index, connection, notifier)
         self.area = -1
         self.temperature = -40
-        self.last_user_time = dt.datetime.now(dt.timezone.utc)
+        self.last_user_time = dt.datetime.now(dt.UTC)
         self.last_user = -1
         self.code = ""
-        self.last_keypress: Optional[tuple[str, int]] = None
+        self.last_keypress: tuple[str, int] | None = None
         self.last_function_key = FunctionKeys.FORCE_KF_SYNC
 
     def press_function_key(self, functionkey: FunctionKeys) -> None:
@@ -51,7 +50,7 @@ def _ic_handler(self, code: int, user: int, keypad: int) -> None:
         keypad_ = self.elements[keypad]
 
         # By setting a time this will force the IC change to always be reported
-        keypad_.setattr("last_user_time", dt.datetime.now(dt.timezone.utc), False)
+        keypad_.setattr("last_user_time", dt.datetime.now(dt.UTC), False)
 
         # If user is negative then invalid code entered
         keypad_.setattr("code", code if user < 0 else "****", False)

elkm1_lib/message.py

@@ -221,9 +221,7 @@ def ld_decode(msg: str) -> dict[str, Any]:
     log["event"] = int(msg[4:8])
     log["number"] = int(msg[8:11])
     log["index"] = int(msg[20:23])
-    log["timestamp"] = dt.datetime(
-        *log_gm_timestruct[:6], tzinfo=dt.timezone.utc
-    ).isoformat()
+    log["timestamp"] = dt.datetime(*log_gm_timestruct[:6], tzinfo=dt.UTC).isoformat()
 
     return {"area": area, "log": log}
 

elkm1_lib/panel.py

@@ -2,7 +2,7 @@
 
 import datetime as dt
 import time
-from typing import Any, Optional
+from typing import Any
 
 from .connection import Connection
 from .const import ElkRPStatus
@@ -48,7 +48,7 @@ def speak_phrase(self, phrase: int) -> None:
         """(Helper) Speak phrase."""
         self._connection.send(sp_encode(phrase))
 
-    def set_time(self, datetime: Optional[dt.datetime] = None) -> None:
+    def set_time(self, datetime: dt.datetime | None = None) -> None:
         """(Helper) Set the time given a datetime."""
         if datetime is None:
             struct_time = time.localtime()

elkm1_lib/thermostats.py

@@ -49,7 +49,7 @@ def set(
             raise ValueError("Wrong type for thermostat setting.")
         if isinstance(val, bool):
             setting = 1 if val else 0
-        elif isinstance(val, (ThermostatFan, ThermostatMode)):
+        elif isinstance(val, ThermostatFan | ThermostatMode):
             setting = val.value
         else:
             setting = val

pyproject.toml

@@ -79,8 +79,8 @@ pythonVersion = "3.11"
 select = [
     "E",    # pycodestyle
     "F",    # Pyflakes
-    # "UP", # pyupgrade
-    # "B",  # flake8-bugbear
+    "UP",   # pyupgrade
+    "B",    # flake8-bugbear
     "SIM",  # flake8-simplify
     "I",    # isort
 ]