Make pylint happier

Author: jmattsson

src/powersensor_local/EventBuffer.py

@@ -1,38 +1,90 @@
+"""Common helper for small commandline utils."""
 import asyncio
 import signal
 from abc import ABC, abstractmethod
 
 class AbstractEventHandler(ABC):
+    """Base class to handle signals and the asyncio loop.
+
+    Subclasses must implement :py:meth:`on_exit` and :py:meth:`main`.  The
+    ``run`` method starts an event loop, registers a SIGINT handler and
+    executes the :py:meth:`main` coroutine.  The loop is stopped when a
+    SIGINT is received and the :py:meth:`on_exit` coroutine has finished.
+    """
     exiting: bool = False
     @abstractmethod
     async def on_exit(self):
-        pass
+        """Called when a SIGINT is received.
+
+        Subclasses should override this method to perform any cleanup
+        (e.g. closing connections, flushing buffers).  It is awaited before
+        the handler sets :pyattr:`exiting` to ``True``.
+        """
 
     async def _do_exit(self):
+        """Internal helper that runs ``on_exit`` and marks the handler as
+        exiting.  This coroutine is scheduled by :py:meth:`__handle_sigint`
+        when a SIGINT signal arrives.
+        """
         await self.on_exit()
         self.exiting = True
 
     @abstractmethod
     async def main(self):
-        pass
+        """Main coroutine to be executed by the event loop.
+
+        Subclasses must implement this method.  It should contain the
+        application's primary logic and can call :py:meth:`wait` to keep
+        the loop alive until a SIGINT is received.
+        """
 
     # Signal handler for Ctrl+C
     def register_sigint_handler(self):
+        """Register the SIGINT (Ctrl‑C) handler.
+
+        This method sets :py:meth:`__handle_sigint` as the callback for
+        ``signal.SIGINT``.  It should be called before :py:meth:`run` if
+        a custom handler is required.
+        """
         signal.signal(signal.SIGINT, self.__handle_sigint)
 
     def __handle_sigint(self, signum, frame):
+        """Internal SIGINT callback.
+
+        Prints diagnostic information and schedules :py:meth:`_do_exit`
+        as a task in the running event loop.  After the first SIGINT
+        the default handler is restored to allow a second Ctrl‑C to
+        terminate immediately.
+        """
         print(f"\nReceived signal: {signum}")
         print(f"Signal name: {signal.Signals(signum).name}")
         print(f"Interrupted at: {frame.f_code.co_filename}:{frame.f_lineno}")
         signal.signal(signal.SIGINT, signal.SIG_DFL)
         asyncio.create_task(self._do_exit())
 
     def run(self):
+        """Start the event loop and execute :py:meth:`main`.
+
+        A new event loop is created, the SIGINT handler is registered,
+        and :py:meth:`main` is run with ``asyncio.run``.  The loop is
+        stopped once the coroutine completes (normally or after a SIGINT).
+        """
         loop = asyncio.new_event_loop()
         asyncio.run(self.main())
         loop.stop()
 
     async def wait(self, seconds=1):
-        # Keep the event loop running until Ctrl+C is pressed
+        """Keep the event loop alive until a SIGINT is received.
+
+        Parameters
+        ----------
+        seconds : int, optional
+            Number of seconds to sleep between checks.  The default is
+            ``1`` which balances responsiveness with CPU usage.
+
+        This coroutine can be awaited by subclasses in their
+        :py:meth:`main` implementation to block until the handler
+        exits.
+        """
         while not self.exiting:
             await asyncio.sleep(seconds)

src/powersensor_local/abstract_event_handler.py

@@ -1,3 +1,4 @@
+"""Small helper class for pub/sub functionality with async handlers."""
 from typing import Callable
 
 class AsyncEventEmitter:
@@ -31,5 +32,5 @@ async def emit(self, event_name: str, *args):
         for callback in self._listeners[event_name]:
             try:
                 await callback(event_name, *args)
-            except BaseException as e:
+            except BaseException as e: # pylint: disable=W0718
                 await self.emit('exception', e)

src/powersensor_local/async_event_emitter.py

@@ -1,12 +1,14 @@
+"""Abstraction interface for unified event stream from Powersensor devices"""
 import asyncio
 import sys
 
 from datetime import datetime, timezone
 from pathlib import Path
-project_root = str(Path(__file__).parents[1])
-if project_root not in sys.path:
-    sys.path.append(project_root)
+PROJECT_ROOT = str(Path(__file__).parents[1])
+if PROJECT_ROOT not in sys.path:
+    sys.path.append(PROJECT_ROOT)
 
+# pylint: disable=C0413
 from powersensor_local.legacy_discovery import LegacyDiscovery
 from powersensor_local.plug_api import PlugApi
 
@@ -22,9 +24,9 @@ def __init__(self, bcast_addr='<broadcast>'):
         """Creates a fresh instance, without scanning for devices."""
         self._event_cb = None
         self._discovery = LegacyDiscovery(bcast_addr)
-        self._devices = dict()
+        self._devices = {}
         self._timer = None
-        self._plug_apis = dict()
+        self._plug_apis = {}
 
     async def start(self, async_event_cb):
         """Registers the async event callback function and starts the scan
@@ -81,7 +83,7 @@ async def stop(self):
         To restart the event streaming, call start() again."""
         for plug in self._plug_apis.values():
             await plug.disconnect()
-        self._plug_apis = dict()
+        self._plug_apis = {}
         self._event_cb = None
         if self._timer:
             self._timer.terminate()
@@ -175,21 +177,24 @@ def __init__(self, mac):
             self._last_active = datetime.now(timezone.utc)
 
         def mark_active(self):
+            """Updates the last activity time to prevent expiry."""
             self._last_active = datetime.now(timezone.utc)
 
         def has_expired(self):
+            """Checks whether the last activity time is past the expiry."""
             now = datetime.now(timezone.utc)
             delta = now - self._last_active
             return delta.total_seconds() > EXPIRY_TIMEOUT_S
 
-    class _Timer:
+    class _Timer: # pylint: disable=R0903
         def __init__(self, interval_s, callback):
             self._terminate = False
             self._interval = interval_s
             self._callback = callback
             self._task = asyncio.create_task(self._run())
 
         def terminate(self):
+            """Disables the timer and cancels the associated task."""
             self._terminate = True
             self._task.cancel()
 

src/powersensor_local/devices.py

@@ -0,0 +1,74 @@
+"""A simple fixed‑size buffer that stores event dictionaries."""
+from typing import Any
+
+
+class EventBuffer:
+    """A simple fixed‑size buffer that stores event dictionaries.
+
+    Parameters
+    ----------
+    keep : int
+        The maximum number of events to retain in the buffer. When a new event
+        is appended and this limit would be exceeded, the oldest event (the
+        one at index 0) is removed.
+    """
+    def __init__(self, keep: int):
+        self._keep = keep
+        self._evs = []
+
+    def find_by_key(self, key: str, value: Any):
+        """Return the first event that contains ``key`` with the given ``value``.
+
+        Parameters
+        ----------
+        key : str
+            The dictionary key to search for.
+        value : Any
+            The value that the key must match.
+
+        Returns
+        -------
+        dict | None
+            The matching event dictionary, or ``None`` if no match is found.
+        """
+        for ev in self._evs:
+            if key in ev and ev[key] == value:
+                return ev
+        return None
+
+    def append(self, ev: dict):
+        """Add an event to the buffer.
+
+        If adding the new event would exceed ``self._keep``, the oldest event
+        is removed to keep the buffer size bounded.
+
+        Parameters
+        ----------
+        ev : dict
+            The event dictionary to append.
+        """
+        self._evs.append(ev)
+        if len(self._evs) > self._keep:
+            del self._evs[0]
+
+    def evict_older(self, key: str, value: float):
+        """Remove events that are older than a given timestamp.
+
+        Events are considered *older* if they contain ``key`` and its value is
+        less than or equal to the provided ``value``. Eviction stops as soon
+        as an event that does not satisfy this condition is encountered (the
+        buffer is ordered by insertion time).
+
+        Parameters
+        ----------
+        key : str
+            The timestamp key to inspect in each event.
+        value : float
+            The cutoff timestamp; events with timestamps <= this value are removed.
+        """
+        while len(self._evs) > 0:
+            ev = self._evs[0]
+            if key in ev and ev[key] <= value:
+                del self._evs[0]
+            else:
+                return

src/powersensor_local/event_buffer.py

@@ -7,14 +7,16 @@
 import sys
 from pathlib import Path
 
-project_root = str(Path(__file__).parents[1])
-if project_root not in sys.path:
-    sys.path.append(project_root)
+PROJECT_ROOT = str(Path(__file__).parents[1])
+if PROJECT_ROOT not in sys.path:
+    sys.path.append(PROJECT_ROOT)
 
+# pylint: disable=C0413
 from powersensor_local.devices import PowersensorDevices
 from powersensor_local.abstract_event_handler import AbstractEventHandler
 
 class EventLoopRunner(AbstractEventHandler):
+    """Main logic wrapper."""
     def __init__(self):
         self.devices: typing.Union[PowersensorDevices, None] = PowersensorDevices()
 
@@ -23,6 +25,7 @@ async def on_exit(self):
             await self.devices.stop()
 
     async def on_message(self, obj):
+        """Callback for printing received events."""
         print(obj)
         if obj['event'] == 'device_found':
             self.devices.subscribe(obj['mac'])
@@ -40,6 +43,7 @@ async def main(self):
         await self.wait()
 
 def app():
+    """Application entry point."""
     EventLoopRunner().run()
 
 if __name__ == "__main__":

src/powersensor_local/events.py

@@ -1,3 +1,4 @@
+"""The legacy alternative to using mDNS discovery."""
 import asyncio
 import json
 import socket
@@ -13,7 +14,7 @@ def __init__(self, broadcast_addr = '<broadcast>'):
         """
         super().__init__()
         self._dst_addr = broadcast_addr
-        self._found = dict()
+        self._found = {}
 
     async def scan(self, timeout_sec = 2.0):
         """Scans the local network for discoverable devices.
@@ -25,7 +26,7 @@ async def scan(self, timeout_sec = 2.0):
           "id": "aabbccddeeff",
         }
         """
-        self._found = dict()
+        self._found = {}
 
         loop = asyncio.get_running_loop()
         transport, _ = await loop.create_datagram_endpoint(
@@ -45,6 +46,7 @@ async def scan(self, timeout_sec = 2.0):
         return list(self._found.values())
 
     def protocol_factory(self):
+        """UDP protocol factory."""
         return self
 
     def datagram_received(self, data, addr):