Improved type hinting.

Now running mypy with --strict.

Fixed issue in PlugListenerUdp's connection retry backoff; it wasn't
backing off in the error path correctly.

Also fixed a few more potential races of create_task/gc interactions.

Author: jmattsson

scripts/run-tests.sh

@@ -11,13 +11,14 @@ extra=
 [ -z "$@" ] && extra="--cov-fail-under=100"
 
 echo "Linting src..."
-python3 -m mypy --show-error-codes --show-column-numbers "${rootdir}/src"
+python3 -m mypy --strict --show-error-codes --show-column-numbers "${rootdir}/src"
 
 echo "Linting tests..."
-mypy "${testsdir}"
+python3 -m mypy --strict "${testsdir}"
 
 pytest \
   --cov=powersensor_local.xlatemsg \
+  --cov=powersensor_local.async_event_emitter \
   --cov-report term-missing \
   --cov-config="${testsdir}/.coveragerc" \
   --cache-clear \

src/powersensor_local/__init__.py

@@ -33,6 +33,10 @@
 debug aids, which get installed under the names ps-plugevents and ps-rawplug
 respectively. There is also the legacy 'events' debug aid which get installed
 nder the names ps-events, and offers up the events from PowersensorDevices.
+Similarly, 'zc_events' offers up events from PowersensorZeroconfDevices,
+installed as 'ps-zcevents', again, for debug use only.
+
+Type hints for emitted data are available in Event/Events/Message.
 """
 __all__ = [
     'VirtualHousehold',
@@ -43,6 +47,9 @@
     'PowersensorDevices',
     'PowersensorLegacyDevices',
     'PowersensorZeroconfDevices',
+    'Event',
+    'Events',
+    'Message',
 ]
 __version__ = "2.2.1"
 from .devices import PowersensorDevices, PowersensorLegacyDevices
@@ -52,3 +59,4 @@
 from .plug_listener_udp import PlugListenerUdp
 from .virtual_household import VirtualHousehold
 from .zeroconf_devices import PowersensorZeroconfDevices
+from .xlatemsg import Event, Events, Message

src/powersensor_local/abstract_event_handler.py

@@ -2,6 +2,7 @@
 import asyncio
 import signal
 from abc import ABC, abstractmethod
+from types import FrameType
 
 class AbstractEventHandler(ABC):
     """Base class to handle signals and the asyncio loop.
@@ -13,15 +14,15 @@ class AbstractEventHandler(ABC):
     """
     exiting: bool = False
     @abstractmethod
-    async def on_exit(self):
+    async def on_exit(self) -> None:
         """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):
+    async def _do_exit(self) -> None:
         """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.
@@ -30,7 +31,7 @@ async def _do_exit(self):
         self.exiting = True
 
     @abstractmethod
-    async def main(self):
+    async def main(self) -> None:
         """Main coroutine to be executed by the event loop.
 
         Subclasses must implement this method.  It should contain the
@@ -39,7 +40,7 @@ async def main(self):
         """
 
     # Signal handler for Ctrl+C
-    def register_sigint_handler(self):
+    def register_sigint_handler(self) -> None:
         """Register the SIGINT (Ctrl‑C) handler.
 
         This method sets :py:meth:`__handle_sigint` as the callback for
@@ -48,21 +49,22 @@ def register_sigint_handler(self):
         """
         signal.signal(signal.SIGINT, self.__handle_sigint)
 
-    def __handle_sigint(self, signum, frame):
+    def __handle_sigint(self, signum: int, frame: FrameType | None) -> None:
         """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}")
+        if frame is not None:
+            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())
+        self._exit_task = asyncio.create_task(self._do_exit())
 
-    def run(self):
+    def run(self) -> None:
         """Start the event loop and execute :py:meth:`main`.
 
         A new event loop is created, the SIGINT handler is registered,
@@ -73,7 +75,7 @@ def run(self):
         asyncio.run(self.main())
         loop.stop()
 
-    async def wait(self, seconds=1):
+    async def wait(self, seconds: int = 1) -> None:
         """Keep the event loop alive until a SIGINT is received.
 
         Parameters

src/powersensor_local/async_event_emitter.py

@@ -1,31 +1,33 @@
 """Small helper class for pub/sub functionality with async handlers."""
 import logging
-from typing import Callable
+from typing import Any, Awaitable, Callable
+
+_OpenCallback = Callable[..., Awaitable[None]]
 
 class AsyncEventEmitter:
     """Small helper class for pub/sub functionality with async handlers.
     An optional Logger can be provided, which will be used to log any
     unhandled exceptions."""
     def __init__(self, logger: logging.Logger | None = None):
-        self._listeners: dict[str,list[Callable]] = {}
+        self._listeners: dict[str,list[_OpenCallback]] = {}
         self._logger = logger
 
-    def subscribe(self, event_name: str, callback: Callable):
+    def subscribe(self, event_name: str, callback: _OpenCallback) -> None:
         """Registers an event handler for the given event key. The handler must
         be async. Duplicate registrations are ignored."""
         if self._listeners.get(event_name) is None:
             self._listeners[event_name] = []
         if not callback in self._listeners[event_name]:
             self._listeners[event_name].append(callback)
 
-    def unsubscribe(self, event_name: str, callback: Callable):
+    def unsubscribe(self, event_name: str, callback: _OpenCallback) -> None:
         """Unregisters the given event handler from the given event type."""
         if self._listeners.get(event_name) is None:
             return
         if callback in self._listeners[event_name]:
             self._listeners[event_name].remove(callback)
 
-    async def emit(self, event_name: str, *args):
+    async def emit(self, event_name: str, *args: Any) -> None:
         """Emits an event to all registered listeners for that event type.
         Additional arguments may be supplied with event as appropriate. Each
         event handler is awaited before delivering the event to the next.

src/powersensor_local/devices.py

@@ -5,9 +5,13 @@
 
 from datetime import datetime, timezone
 from enum import Enum
+from typing import Any, Callable, Coroutine
 
 from .legacy_discovery import LegacyDiscovery
 from .plug_api import PlugApi
+from .xlatemsg import Event
+
+_AsyncCallback = Callable[[Event], Coroutine[None, None, None]]
 
 EXPIRY_CHECK_INTERVAL_S = 30
 EXPIRY_TIMEOUT_S = 5 * 60
@@ -46,9 +50,17 @@ class _PowersensorDevicesBase:
 
     **device_found**
         A device has been discovered or re-discovered.
+
         Note that due to device hardware limitations, role information is NOT
         reliably available at this time, and therefore not included in this
-        message.
+        message. There are situations where role information will never become
+        available, and therefore the API can make no promises otherwise. If a
+        user requires role information, they must manage that themselves and
+        also provide a mechanism for handling the situation of a device not
+        being able to provide role information in the first place. If a device
+        does supply a role at any point, it should be considered authoritative
+        and override any user provided value.
+
         ``{ event: "device_found", device_type: "plug"|"sensor", mac: "..." }``
 
     **device_lost**
@@ -96,7 +108,7 @@ def __init__(
             library emits debug/warning/error messages via this logger.  When
             None (default) the library is completely silent.
         """
-        self._event_cb = None
+        self._event_cb: _AsyncCallback | None = None
         self._devices: dict[str, '_PowersensorDevicesBase._Device'] = {}
         self._plug_apis: dict[str, PlugApi] = {}
         self._timer: '_PowersensorDevicesBase._Timer | None' = None
@@ -107,7 +119,7 @@ def __init__(
     # Internal logging helper
     # ------------------------------------------------------------------
 
-    def _maybe_log(self, level: _LogLevel, msg: str, *args) -> None:
+    def _maybe_log(self, level: _LogLevel, msg: str, *args: Any) -> None:
         """Emit a log message if a logger was provided at construction."""
         if self._logger is None:
             return
@@ -191,19 +203,20 @@ async def _plug_lost(self, mac: str) -> None:
     # Internal event routing
     # ------------------------------------------------------------------
 
-    async def _emit_if_subscribed(self, ev: str, mac: str, obj: dict) -> None:
+    async def _emit_if_subscribed(self, ev: str, mac: str, obj: Event) -> None:
         if self._event_cb is None:
             return
         device = self._devices.get(mac)
         if device is not None and device.subscribed:
             obj['event'] = ev
             await self._event_cb(obj)
 
-    async def _reemit(self, ev: str, obj: dict[str, str]) -> None:
-        mac: str|None = obj.get('mac')
+    async def _reemit(self, ev: str, obj: Event) -> None:
+        mac = obj.get('mac')
         if mac is None:
             self._maybe_log(_LogLevel.WARNING, "Received event '%s' with no MAC address — ignoring", ev)
             return
+        mac = str(mac)
         device = self._devices.get(mac)
         if device is not None:
             device.mark_active()
@@ -264,7 +277,7 @@ def has_expired(self) -> bool:
             return delta.total_seconds() > EXPIRY_TIMEOUT_S
 
     class _Timer:
-        def __init__(self, interval_s: float, callback) -> None:
+        def __init__(self, interval_s: float, callback: Callable[[], Coroutine[Any, Any, None]]):
             self._terminate = False
             self._interval = interval_s
             self._callback = callback
@@ -300,7 +313,7 @@ def __init__(
         super().__init__(relay_now_relaying_for=relay_now_relaying_for, logger=logger)
         self._discovery = LegacyDiscovery(bcast_addr)
 
-    async def start(self, async_event_cb) -> int:
+    async def start(self, async_event_cb: _AsyncCallback) -> int:
         """Register the async event callback and scan the local network.
 
         The callback has the form::
@@ -329,7 +342,7 @@ async def rescan(self) -> None:
         """Perform a fresh scan to discover added or moved devices."""
         await self._on_scanned(await self._discovery.scan())
 
-    async def _on_scanned(self, found: list) -> None:
+    async def _on_scanned(self, found: list[dict[str,str]]) -> None:
         for device in found:
             mac = device['id']
             ip = device['ip']

src/powersensor_local/event_buffer.py

@@ -14,9 +14,9 @@ class EventBuffer:
     """
     def __init__(self, keep: int):
         self._keep = keep
-        self._evs: list = []
+        self._evs: list[Any] = []
 
-    def find_by_key(self, key: str, value: Any):
+    def find_by_key(self, key: str, value: Any) -> Any | None:
         """Return the first event that contains ``key`` with the given ``value``.
 
         Parameters
@@ -36,7 +36,7 @@ def find_by_key(self, key: str, value: Any):
                 return ev
         return None
 
-    def append(self, ev: dict):
+    def append(self, ev: dict[str, Any]) -> None:
         """Add an event to the buffer.
 
         If adding the new event would exceed ``self._keep``, the oldest event
@@ -51,7 +51,7 @@ def append(self, ev: dict):
         if len(self._evs) > self._keep:
             del self._evs[0]
 
-    def evict_older(self, key: str, value: float):
+    def evict_older(self, key: str, value: float) -> None:
         """Remove events that are older than a given timestamp.
 
         Events are considered *older* if they contain ``key`` and its value is

src/powersensor_local/events.py

@@ -8,6 +8,7 @@
 
 from powersensor_local.devices import PowersensorDevices
 from powersensor_local.abstract_event_handler import AbstractEventHandler
+from powersensor_local.xlatemsg import Event
 
 class EventLoopRunner(AbstractEventHandler):
     """Main logic wrapper."""
@@ -18,11 +19,11 @@ async def on_exit(self) -> None:
         if self.devices is not None:
             await self.devices.stop()
 
-    async def on_message(self, obj) -> None:
+    async def on_message(self, obj: Event) -> None:
         """Callback for printing received events."""
         print(obj)
         if obj['event'] == 'device_found':
-            self.devices.subscribe(obj['mac'])
+            self.devices.subscribe(str(obj['mac']))
 
     async def main(self) -> None:
         if self.devices is None:

src/powersensor_local/legacy_discovery.py

@@ -8,15 +8,15 @@
 class LegacyDiscovery(asyncio.DatagramProtocol):
     """The legacy alternative to using mDNS discovery."""
 
-    def __init__(self, broadcast_addr = '<broadcast>'):
+    def __init__(self, broadcast_addr: str = '<broadcast>'):
         """Initialises a new discovery object.
         Optionally takes a specific broadcast address to use.
         """
         super().__init__()
-        self._dst_addr = broadcast_addr
-        self._found = {}
+        self._dst_addr: str = broadcast_addr
+        self._found: dict[str,dict[str,str]] = {}
 
-    async def scan(self, timeout_sec = 2.0):
+    async def scan(self, timeout_sec: float = 2.0) -> list[dict[str,str]]:
         """Scans the local network for discoverable devices.
         Returns the list of devices found, with each device represented
         in the format:
@@ -45,11 +45,11 @@ async def scan(self, timeout_sec = 2.0):
         transport.close()
         return list(self._found.values())
 
-    def protocol_factory(self):
+    def protocol_factory(self) -> asyncio.DatagramProtocol:
         """UDP protocol factory."""
         return self
 
-    def datagram_received(self, data, addr):
+    def datagram_received(self, data, addr) -> None: # type: ignore
         try:
             response = json.loads(data.decode('utf-8'))
             ip = response['ip']