From a16c4366c13730074120debf09d31686a3bc22a3 Mon Sep 17 00:00:00 2001 From: Glenn Waters Date: Wed, 21 Aug 2024 20:20:13 -0400 Subject: [PATCH] Add more ruff checks across codebase. (#81) * Fix some formatting to use f strings. * ruff auto-fixes. * More ruff found fixes. * Ruff auto-fixed errors. * Add UP ruff checks. * Last ruff check added. * Formatting. --- bin/cmdr.py | 35 ++++++++++++++++++----------------- elkm1_lib/connection.py | 6 +++--- elkm1_lib/discovery.py | 17 ++++++++--------- elkm1_lib/elements.py | 9 +++++---- elkm1_lib/keypads.py | 7 +++---- elkm1_lib/message.py | 4 +--- elkm1_lib/panel.py | 4 ++-- elkm1_lib/thermostats.py | 2 +- pyproject.toml | 4 ++-- 9 files changed, 43 insertions(+), 45 deletions(-) diff --git a/bin/cmdr.py b/bin/cmdr.py index 844ac0d..f9363ff 100644 --- a/bin/cmdr.py +++ b/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, - "{} [subcommand]".format(cmd), - "Displays internal state of {}".format(element), + f"{cmd} [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() diff --git a/elkm1_lib/connection.py b/elkm1_lib/connection.py index 83354ed..257ed55 100644 --- a/elkm1_lib/connection.py +++ b/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)") diff --git a/elkm1_lib/discovery.py b/elkm1_lib/discovery.py index aa79def..77b5038 100644 --- a/elkm1_lib/discovery.py +++ b/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) diff --git a/elkm1_lib/elements.py b/elkm1_lib/elements.py index 0a3b39d..4058622 100644 --- a/elkm1_lib/elements.py +++ b/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 diff --git a/elkm1_lib/keypads.py b/elkm1_lib/keypads.py index 9653835..f2e179b 100644 --- a/elkm1_lib/keypads.py +++ b/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) diff --git a/elkm1_lib/message.py b/elkm1_lib/message.py index 3c7888b..c523d49 100644 --- a/elkm1_lib/message.py +++ b/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} diff --git a/elkm1_lib/panel.py b/elkm1_lib/panel.py index 045eab5..aae1a87 100644 --- a/elkm1_lib/panel.py +++ b/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() diff --git a/elkm1_lib/thermostats.py b/elkm1_lib/thermostats.py index 9e91a1d..efb6f3c 100644 --- a/elkm1_lib/thermostats.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 27a0c21..7d3ec8c 100644 --- a/pyproject.toml +++ b/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 ]