Further type fixes, plus py.typed marker

Author: jmattsson

src/powersensor_local/event_buffer.py

@@ -14,7 +14,7 @@ class EventBuffer:
     """
     def __init__(self, keep: int):
         self._keep = keep
-        self._evs = []
+        self._evs: list = []
 
     def find_by_key(self, key: str, value: Any):
         """Return the first event that contains ``key`` with the given ``value``.

src/powersensor_local/events.py

@@ -17,20 +17,20 @@
 
 class EventLoopRunner(AbstractEventHandler):
     """Main logic wrapper."""
-    def __init__(self):
-        self.devices: typing.Union[PowersensorDevices, None] = PowersensorDevices()
+    def __init__(self) -> None:
+        self.devices: PowersensorDevices = PowersensorDevices()
 
-    async def on_exit(self):
+    async def on_exit(self) -> None:
         if self.devices is not None:
             await self.devices.stop()
 
-    async def on_message(self, obj):
+    async def on_message(self, obj) -> None:
         """Callback for printing received events."""
         print(obj)
         if obj['event'] == 'device_found':
             self.devices.subscribe(obj['mac'])
 
-    async def main(self):
+    async def main(self) -> None:
         if self.devices is None:
             self.devices = PowersensorDevices()
 
@@ -42,7 +42,7 @@ async def main(self):
         # Keep the event loop running until Ctrl+C is pressed
         await self.wait()
 
-def app():
+def app() -> None:
     """Application entry point."""
     EventLoopRunner().run()
 

src/powersensor_local/plugevents.py

@@ -4,7 +4,6 @@
 Powersensor device. Intended for advanced debugging use only."""
 
 import sys
-from typing import Union
 from pathlib import Path
 
 PROJECT_ROOT = str(Path(__file__).parents[ 1])
@@ -15,21 +14,21 @@
 from powersensor_local.plug_api import PlugApi
 from powersensor_local.abstract_event_handler import AbstractEventHandler
 
-async def print_event_and_message(event, message):
+async def print_event_and_message(event, message) -> None:
     """Callback for printing event data."""
     print(event, message)
 
 class PlugEvents(AbstractEventHandler):
     """Main logic wrapper."""
-    def __init__(self):
-        self.plug: Union[PlugApi, None] = None
+    def __init__(self) -> None:
+        self.plug: PlugApi|None = None
 
-    async def on_exit(self):
+    async def on_exit(self) -> None:
         if self.plug is not None:
             await self.plug.disconnect()
             self.plug = None
 
-    async def main(self):
+    async def main(self) -> None:
         if len(sys.argv) < 3:
             print(f"Syntax: {sys.argv[0]} <id> <ip> [port]")
             sys.exit(1)
@@ -57,7 +56,7 @@ async def main(self):
         # Keep the event loop running until Ctrl+C is pressed
         await self.wait()
 
-def app():
+def app() -> None:
     """Application entry point."""
     PlugEvents().run()
 

src/powersensor_local/py.typed

@@ -3,7 +3,6 @@
 """Utility script for accessing the raw plug subscription data from a single
 network-local Powersensor device. Intended for advanced debugging use only."""
 
-from typing import Union
 import sys
 
 from pathlib import Path
@@ -16,29 +15,29 @@
 from powersensor_local import PlugListenerTcp,PlugListenerUdp
 from powersensor_local.abstract_event_handler import AbstractEventHandler
 
-async def print_message_ignore_event(_, message):
+async def print_message_ignore_event(_, message) -> None:
     """Callback for printing event data withou the event name."""
     print(message)
 
-async def print_event(event):
+async def print_event(event) -> None:
     """Callback for printing an event."""
     print(event)
 
 class RawPlug(AbstractEventHandler):
     """Main logic wrapper."""
-    def __init__(self, protocol=None):
-        self.plug: Union[PlugListenerTcp, PlugListenerUdp, None] = None
-        if protocol is None:
+    def __init__(self, protocol: str|None = None) -> None:
+        self.plug: PlugListenerTcp | PlugListenerUdp | None = None
+        if protocol is None or protocol == 'udp':
             self._protocol = 'udp'
         else:
             self._protocol = 'tcp'
 
-    async def on_exit(self):
+    async def on_exit(self) -> None:
         if self.plug is not None:
             await self.plug.disconnect()
             self.plug = None
 
-    async def main(self):
+    async def main(self) -> None:
         if len(sys.argv) < 2:
             print(f"Syntax: {sys.argv[0]} <ip> [port]")
             sys.exit(1)
@@ -47,25 +46,26 @@ async def main(self):
         self.register_sigint_handler()
         if len(sys.argv) >= 4:
             self._protocol = sys.argv[3]
-        plug = None
+        self.plug = None
         if self._protocol == 'udp':
-            plug = PlugListenerUdp(sys.argv[1], *sys.argv[2:3])
+            self.plug = PlugListenerUdp(sys.argv[1], *sys.argv[2:3])
         elif self._protocol == 'tcp':
-            plug = PlugListenerTcp(sys.argv[1], *sys.argv[2:3])
+            self.plug = PlugListenerTcp(sys.argv[1], *sys.argv[2:3])
         else:
             print('Unsupported protocol:', self._protocol)
-        plug.subscribe('exception', print_message_ignore_event)
-        plug.subscribe('message', print_message_ignore_event)
-        plug.subscribe('connecting', print_event)
-        plug.subscribe('connecting', print_event)
-        plug.subscribe('connected', print_event)
-        plug.subscribe('disconnected', print_event)
-        plug.connect()
+            sys.exit(1)
+        self.plug.subscribe('exception', print_message_ignore_event)
+        self.plug.subscribe('message', print_message_ignore_event)
+        self.plug.subscribe('connecting', print_event)
+        self.plug.subscribe('connecting', print_event)
+        self.plug.subscribe('connected', print_event)
+        self.plug.subscribe('disconnected', print_event)
+        self.plug.connect()
 
         # Keep the event loop running until Ctrl+C is pressed
         await self.wait()
 
-def app():
+def app() -> None:
     """Application entry point."""
     RawPlug().run()
 

src/powersensor_local/rawplug.py

@@ -7,7 +7,7 @@
 
 
 # pylint: disable=R0913,R0917
-def _pick_item(out: dict, message: dict, key: str, dstkey: str, req: bool, decis: int = None):
+def _pick_item(out: dict, message: dict, key: str, dstkey: str, req: bool, decis: int|None = None):
     val = message.get(key)
     if val is not None:
         if isinstance(val, float) and decis is not None:
@@ -23,7 +23,7 @@ def _pick_list(out: dict, message: dict, items: list):
     return out
 
 def _make_average_power_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     _pick_list(ev, message, [
         ('power', 'watts', True, 0),
@@ -32,7 +32,7 @@ def _make_average_power_event(message: dict):
     return ev
 
 def _make_average_power_components_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     _pick_list(ev, message, [
         ('current', 'apparent_current', True, 3),
@@ -43,7 +43,7 @@ def _make_average_power_components_event(message: dict):
     return ev
 
 def _make_summation_energy_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     _pick_list(ev, message, [
         ('summation', 'summation_joules', True, 0),
@@ -52,7 +52,7 @@ def _make_summation_energy_event(message: dict):
     return ev
 
 def _make_average_flow_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     # Old firmware not providing averge_flow don't account for meter scaling,
     # so the reported flow is likely wrong. To avoid displaying incorrect data
@@ -64,7 +64,7 @@ def _make_average_flow_event(message: dict):
     return ev
 
 def _make_summation_volume_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     # Old firmware not providing summation_volume don't account for meter
     # scaling, so the reported summation is likely wrong. To avoid displaying
@@ -76,7 +76,7 @@ def _make_summation_volume_event(message: dict):
     return ev
 
 def _make_uncalibrated_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     _pick_list(ev, message, [
         ('power', 'value', True, None),
@@ -85,13 +85,13 @@ def _make_uncalibrated_event(message: dict):
     return ev
 
 def _make_battery_level_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     ev['volts'] = round(float(message['batteryMicrovolt'])/1000000.0, 6)
     return ev
 
 def _make_rssi_event(message: dict):
-    ev = {}
+    ev: dict = {}
     _pick_list(ev, message, _MAC_TS_ROLE)
     _pick_list(ev, message, [
         ('duration', 'duration_s', True, 3),
@@ -247,9 +247,9 @@ def translate_raw_message(message: dict, relay_mac: str):
           - "last_rssi": The most recent RSSI value.
 
     """
-    evs = {}
-    typ = message.get('type')
-    dev = message.get('device') # plug/sensor/ble_sensor
+    evs: dict = {}
+    typ = message['type']
+    dev = message['device'] # plug/sensor/ble_sensor
 
     # Primary message type, overloaded like nothing 😅
     if typ == 'instant_power':