Configure ruff linter

pyotgw/commandprocessor.py

@@ -3,9 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+from asyncio.queues import QueueFull
 import logging
 import re
-from asyncio.queues import QueueFull
 from typing import TYPE_CHECKING
 
 from . import vars as v

pyotgw/connection.py

@@ -7,10 +7,10 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from functools import partial
+import logging
 from typing import TYPE_CHECKING, Literal
 
 import serial
@@ -181,7 +181,7 @@ async def _attempt_connect(self) -> tuple[asyncio.Transport, asyncio.Protocol]:
                     )
                     self._error = err
 
-            except asyncio.TimeoutError as err:
+            except TimeoutError as err:
                 if not isinstance(err, type(self._error)):
                     _LOGGER.error(
                         "The serial device on %s is not responding. "

pyotgw/messageprocessor.py

@@ -7,8 +7,7 @@
 import re
 from typing import TYPE_CHECKING
 
-from . import messages as m
-from . import vars as v
+from . import messages as m, vars as v
 from .types import (
     OpenThermCommand,
     OpenThermDataSource,

pyotgw/poll_task.py

@@ -4,16 +4,53 @@
 
 import asyncio
 from collections.abc import Callable
+from enum import StrEnum
 import logging
 from typing import TYPE_CHECKING
 
+from . import vars as v
+
 if TYPE_CHECKING:
     from .pyotgw import OpenThermGateway
 from .types import OpenThermDataSource, OpenThermReport
 
 _LOGGER = logging.getLogger(__name__)
 
 
+class OpenThermPollTaskName(StrEnum):
+    """Poll task names."""
+
+    GPIO_STATE = "gpio_state"
+
+
+def get_all_poll_tasks(gateway: OpenThermGateway):
+    """Get all poll tasks for a gateway."""
+    return {
+        OpenThermPollTaskName.GPIO_STATE: OpenThermPollTask(
+            OpenThermPollTaskName.GPIO_STATE,
+            gateway,
+            OpenThermReport.GPIO_STATES,
+            {
+                OpenThermDataSource.GATEWAY: {
+                    v.OTGW_GPIO_A_STATE: 0,
+                    v.OTGW_GPIO_B_STATE: 0,
+                },
+            },
+            (
+                lambda: 0
+                in (
+                    gateway.status.status[OpenThermDataSource.GATEWAY].get(
+                        v.OTGW_GPIO_A
+                    ),
+                    gateway.status.status[OpenThermDataSource.GATEWAY].get(
+                        v.OTGW_GPIO_B
+                    ),
+                )
+            ),
+        )
+    }
+
+
 class OpenThermPollTask:
     """
     Describes a task that polls the gateway for certain reports.

pyotgw/protocol.py

@@ -3,9 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Awaitable, Callable
 import logging
 import re
-from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING
 
 from .commandprocessor import CommandProcessor

pyotgw/pyotgw.py

@@ -3,14 +3,14 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 from collections.abc import Awaitable, Callable
 from datetime import datetime
-from typing import TYPE_CHECKING, Final, Literal
+import logging
+from typing import TYPE_CHECKING, Literal
 
 from . import vars as v
 from .connection import ConnectionManager
-from .poll_task import OpenThermPollTask
+from .poll_task import OpenThermPollTaskName, get_all_poll_tasks
 from .reports import convert_report_response_to_status_update
 from .status import StatusManager
 from .types import (
@@ -26,39 +26,13 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-GPIO_POLL_TASK_NAME: Final = "gpio"
-
-
 class OpenThermGateway:  # pylint: disable=too-many-public-methods
     """Main OpenThermGateway object abstraction"""
 
     def __init__(self) -> None:
         """Create an OpenThermGateway object."""
         self._transport = None
-        self._poll_tasks = {
-            GPIO_POLL_TASK_NAME: OpenThermPollTask(
-                GPIO_POLL_TASK_NAME,
-                self,
-                OpenThermReport.GPIO_STATES,
-                {
-                    OpenThermDataSource.GATEWAY: {
-                        v.OTGW_GPIO_A_STATE: 0,
-                        v.OTGW_GPIO_B_STATE: 0,
-                    },
-                },
-                (
-                    lambda: 0
-                    in (
-                        self.status.status[OpenThermDataSource.GATEWAY].get(
-                            v.OTGW_GPIO_A
-                        ),
-                        self.status.status[OpenThermDataSource.GATEWAY].get(
-                            v.OTGW_GPIO_B
-                        ),
-                    )
-                ),
-            )
-        }
+        self._poll_tasks = get_all_poll_tasks(self)
         self._protocol = None
         self._skip_init = False
         self.status = StatusManager()
@@ -207,7 +181,7 @@ async def set_outside_temp(
 
     async def set_clock(
         self,
-        date: datetime = datetime.now(),
+        date: datetime | None = None,
         timeout: asyncio.Timeout = v.OTGW_DEFAULT_TIMEOUT,
     ) -> str | None:
         """
@@ -221,6 +195,8 @@ async def set_clock(
         This method is a coroutine
         """
         cmd = OpenThermCommand.SET_CLOCK
+        if date is None:
+            date = datetime.now()
         value = f"{date.strftime('%H:%M')}/{date.isoweekday()}"
         return await self._wait_for_cmd(cmd, value, timeout)
 
@@ -343,7 +319,8 @@ async def get_report(
         report_type: OpenThermReport,
         timeout: asyncio.Timeout = v.OTGW_DEFAULT_TIMEOUT,
     ) -> dict[OpenThermDataSource, dict] | None:
-        """Get the report, update status dict accordingly. Return updated status dict."""
+        """Get the report, update status dict accordingly.
+        Return updated status dict."""
         ret = await self._wait_for_cmd(OpenThermCommand.REPORT, report_type, timeout)
         if (
             ret is None
@@ -519,7 +496,9 @@ async def set_gpio_mode(
             var = getattr(v, f"OTGW_GPIO_{gpio_id}")
             status_otgw[var] = ret
             self.status.submit_partial_update(OpenThermDataSource.GATEWAY, status_otgw)
-            await self._poll_tasks[GPIO_POLL_TASK_NAME].start_or_stop_as_needed()
+            await self._poll_tasks[
+                OpenThermPollTaskName.GPIO_STATE
+            ].start_or_stop_as_needed()
             return ret
 
     async def set_setback_temp(
@@ -865,7 +844,7 @@ async def _wait_for_cmd(
                 self._protocol.command_processor.issue_cmd(cmd, value),
                 timeout,
             )
-        except asyncio.TimeoutError:
+        except TimeoutError:
             _LOGGER.error("Timed out waiting for command: %s, value: %s.", cmd, value)
             return
         except (RuntimeError, SyntaxError, ValueError) as exc:

pyotgw/status.py

@@ -3,9 +3,9 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 from collections.abc import Awaitable, Callable
 from copy import deepcopy
+import logging
 
 from . import vars as v
 from .types import OpenThermDataSource

pyproject.toml

@@ -1,6 +1,16 @@
 [tool.pytest.ini_options]
 asyncio_default_fixture_loop_scope = "function"
 
+[tool.ruff.lint]
+select = [
+    "E",      # pycodestyle
+    "F",      # Pyflakes
+    "UP",     # pyupgrade
+    "B",      # flake8-bugbear
+    "SIM",    # flake8-simplify
+    "I",      # isort
+]
+
 [tool.ruff.lint.isort]
 force-sort-within-sections = true
 known-first-party = [

tests/data.py

@@ -2,8 +2,8 @@
 
 from types import SimpleNamespace
 
-import pyotgw.vars as v
 from pyotgw.types import OpenThermDataSource, OpenThermMessageID, OpenThermMessageType
+import pyotgw.vars as v
 
 _report_responses_51 = {
     v.OTGW_REPORT_ABOUT: "A=OpenTherm Gateway 5.1",

tests/helpers.py

@@ -32,13 +32,23 @@ async def _wait():
 
 
 def respond_to_reports(
-    cmds: list[OpenThermReport] = [], responses: list[str] = []
+    cmds: list[OpenThermReport] | None = None, responses: list[str] | None = None
 ) -> Callable[[OpenThermCommand, str, float | None], str]:
     """
     Respond to PR= commands with test values.
     Override response values by specifying cmds and responses in order.
     """
 
+    if len(cmds) != len(responses):
+        raise ValueError(
+            "There should be an equal amount of provided cmds and responses"
+        )
+
+    if cmds is None:
+        cmds = []
+    if responses is None:
+        responses = []
+
     default_responses = {
         OpenThermReport.ABOUT: "A=OpenTherm Gateway 5.8",
         OpenThermReport.BUILD: "B=17:52 12-03-2023",
@@ -58,11 +68,6 @@ def respond_to_reports(
         OpenThermReport.DHW: "W=1",
     }
 
-    if len(cmds) != len(responses):
-        raise ValueError(
-            "There should be an equal amount of provided cmds and responses"
-        )
-
     for cmd, response in zip(cmds, responses):
         if cmd not in default_responses:
             raise ValueError(f"Command {cmd} not found in default responses.")

tests/test_commandprocessor.py

@@ -149,7 +149,8 @@ async def test_issue_cmd(caplog, pygw_proto):
         (
             "pyotgw.commandprocessor",
             logging.WARNING,
-            f"Command {OpenThermCommand.CONTROL_SETPOINT_2} failed with InvalidCommand, retrying...",
+            f"Command {OpenThermCommand.CONTROL_SETPOINT_2} failed with InvalidCommand,"
+            " retrying...",
         ),
     ]
     caplog.clear()
@@ -185,12 +186,14 @@ async def test_issue_cmd(caplog, pygw_proto):
         (
             "pyotgw.commandprocessor",
             logging.WARNING,
-            f"Command {OpenThermCommand.CONTROL_HEATING_2} failed with Error 03, retrying...",
+            f"Command {OpenThermCommand.CONTROL_HEATING_2} failed with Error 03,"
+            " retrying...",
         ),
         (
             "pyotgw.commandprocessor",
             logging.WARNING,
-            f"Command {OpenThermCommand.CONTROL_HEATING_2} failed with {OpenThermCommand.CONTROL_HEATING_2}: BV, retrying...",
+            f"Command {OpenThermCommand.CONTROL_HEATING_2} failed with"
+            " {OpenThermCommand.CONTROL_HEATING_2}: BV, retrying...",
         ),
     ]
 

tests/test_messages.py

@@ -1,7 +1,7 @@
 """Tests for pyotgw/messages.py"""
 
-import pyotgw.messages as m
 from pyotgw.messageprocessor import MessageProcessor
+import pyotgw.messages as m
 
 
 def test_message_registry():

tests/test_poll_task.py

@@ -1,18 +1,19 @@
 """Tests for pyotgw/poll_tasks.py"""
 
+from unittest.mock import AsyncMock, call
+
 import pytest
-from unittest.mock import call, AsyncMock
 
-from pyotgw.poll_task import OpenThermPollTask
-from pyotgw.pyotgw import OpenThermGateway, GPIO_POLL_TASK_NAME
+from pyotgw.poll_task import OpenThermPollTask, OpenThermPollTaskName
+from pyotgw.pyotgw import OpenThermGateway
 from pyotgw.types import OpenThermDataSource, OpenThermReport
 import pyotgw.vars as v
 
 from .helpers import called_x_times
 
 TASK_TEST_PARAMETERS = ("task_name",)
 TASK_TEST_VALUES = [
-    (GPIO_POLL_TASK_NAME,),
+    (OpenThermPollTaskName.GPIO_STATE,),
 ]
 
 
@@ -31,7 +32,7 @@ async def test_init(pygw: OpenThermGateway, task_name: str) -> None:
 @pytest.mark.asyncio
 async def test_gpio_start_stop(pygw: OpenThermGateway) -> None:
     """Test task.start() and task.stop()"""
-    task = pygw._poll_tasks[GPIO_POLL_TASK_NAME]
+    task = pygw._poll_tasks[OpenThermPollTaskName.GPIO_STATE]
     assert not task.is_running
     pygw.status.submit_partial_update(OpenThermDataSource.GATEWAY, {v.OTGW_GPIO_A: 0})
     task.start()
@@ -43,7 +44,7 @@ async def test_gpio_start_stop(pygw: OpenThermGateway) -> None:
 @pytest.mark.asyncio
 async def test_gpio_start_or_stop_as_needed(pygw: OpenThermGateway) -> None:
     """Test task.start_or_stop_as_needed()"""
-    task = pygw._poll_tasks[GPIO_POLL_TASK_NAME]
+    task = pygw._poll_tasks[OpenThermPollTaskName.GPIO_STATE]
     assert task.is_running is False
     await task.start_or_stop_as_needed()
     assert task.is_running is False
@@ -59,7 +60,7 @@ async def test_gpio_start_or_stop_as_needed(pygw: OpenThermGateway) -> None:
 
 def test_gpio_should_run(pygw: OpenThermGateway) -> None:
     """Test task.should_run()"""
-    task = pygw._poll_tasks[GPIO_POLL_TASK_NAME]
+    task = pygw._poll_tasks[OpenThermPollTaskName.GPIO_STATE]
     assert task.should_run is False
     pygw.status.submit_partial_update(OpenThermDataSource.GATEWAY, {v.OTGW_GPIO_A: 0})
     assert task.should_run is True
@@ -68,7 +69,7 @@ def test_gpio_should_run(pygw: OpenThermGateway) -> None:
 @pytest.mark.asyncio
 async def test_gpio_is_running(pygw: OpenThermGateway) -> None:
     """Test task.should_run()"""
-    task = pygw._poll_tasks[GPIO_POLL_TASK_NAME]
+    task = pygw._poll_tasks[OpenThermPollTaskName.GPIO_STATE]
     assert task.is_running is False
     pygw.status.submit_partial_update(OpenThermDataSource.GATEWAY, {v.OTGW_GPIO_A: 0})
     task.start()
@@ -79,7 +80,7 @@ async def test_gpio_is_running(pygw: OpenThermGateway) -> None:
 async def test_gpio_polling_routing(pygw: OpenThermGateway) -> None:
     """Test task._polling_routing()"""
     pygw.get_report = AsyncMock()
-    task = pygw._poll_tasks[GPIO_POLL_TASK_NAME]
+    task = pygw._poll_tasks[OpenThermPollTaskName.GPIO_STATE]
     task._interval = 0.01
     pygw.status.submit_partial_update(OpenThermDataSource.GATEWAY, {v.OTGW_GPIO_A: 0})
     task.start()