random unimportant cleanups

Author: bookman-dius

docs/source/conf.py

@@ -10,16 +10,15 @@
 
 # Project information
 project = "powersensor_local"
-copyright = "2025, DiUS"
+copyright = "2025, DiUS" # noqa A001
 author = "Powersensor Team!"
 
 html_favicon = "_static/powersensor-logo.png"
 html_logo = "_static/powersensor-logo.png"
 
 # The full version, including alpha/beta/rc tags
 try:
-    from powersensor_local import __version__
-    release = __version__
+    from powersensor_local import __version__ as release
 except ImportError:
     release = "0.1.0"
 

src/powersensor_local/EventBuffer.py

@@ -0,0 +1,26 @@
+from typing import Any
+
+
+class EventBuffer:
+    def __init__(self, keep: int):
+        self._keep = keep
+        self._evs = []
+
+    def find_by_key(self, key: str, value: Any):
+        for ev in self._evs:
+            if key in ev and ev[key] == value:
+                return ev
+        return None
+
+    def append(self, ev: dict):
+        self._evs.append(ev)
+        if len(self._evs) > self._keep:
+            del self._evs[0]
+
+    def evict_older(self, key: str, value: float):
+        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/__init__.py

@@ -18,15 +18,15 @@
 Lower-level interfaces are available in the PlugListenerUdp and PlugListenerTcp
 classes, though they are not recommended for general use.
 
-Additionally a convience abstraction for translating some of the events into
+Additionally, a convenience abstraction for translating some of the events into
 a household view is available in VirtualHousehold.
 
 Quick overview:
 • PlugApi is the recommended API layer
 • PlugListenerUdp is the UDP lower-level abstraction used by PlugApi
 • PlugListenerTcp is the TCP lower-level abstraction used by PlugApi
 • PowersensorDevices is the legacy main API layer
-• LegadyDiscovery provides access to the legacy discovery mechanism
+• LegacyDiscovery provides access to the legacy discovery mechanism
 • VirtualHousehold can be used to translate events into a household view
 
 The 'plugevents' and 'rawplug' modules are helper utilities provided as
@@ -35,14 +35,13 @@
 nder the names ps-events, and offers up the events from PowersensorDevices.
 """
 __all__ = [
-    'devices',
-    'legacy_discovery',
-    'plug_api',
-    'plug_listener_tcp',
-    'plug_listener_udp',
-    'virtual_household'
+    'VirtualHousehold',
+    'PlugApi',
+    '__version__',
+    'PlugListenerTcp',
+    'PlugListenerUdp'
 ]
-__version__ = "2.0.0"
+__version__ = "2.0.1"
 from .devices import PowersensorDevices
 from .legacy_discovery import LegacyDiscovery
 from .plug_api import PlugApi

src/powersensor_local/devices.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import sys
 
 from datetime import datetime, timezone
@@ -10,7 +9,6 @@
 
 from powersensor_local.legacy_discovery import LegacyDiscovery
 from powersensor_local.plug_api import PlugApi
-from powersensor_local.xlatemsg import translate_raw_message
 
 EXPIRY_CHECK_INTERVAL_S = 30
 EXPIRY_TIMEOUT_S = 5 * 60

src/powersensor_local/legacy_discovery.py

@@ -13,6 +13,7 @@ def __init__(self, broadcast_addr = '<broadcast>'):
         """
         super().__init__()
         self._dst_addr = broadcast_addr
+        self._found = dict()
 
     async def scan(self, timeout_sec = 2.0):
         """Scans the local network for discoverable devices.

src/powersensor_local/plug_api.py

@@ -59,9 +59,8 @@ async def disconnect(self):
     async def _on_message(self, _, message):
         """Translates the raw message and emits the resulting messages, if any.
 
-        Also synthesises 'now_relaying_for' messages as needed.
+        Also synthesizes 'now_relaying_for' messages as needed.
         """
-        evs = None
         try:
             evs = translate_raw_message(message, self._mac)
         except KeyError:

src/powersensor_local/plug_listener_tcp.py

@@ -69,7 +69,7 @@ async def _close_connection(self):
 
     async def _do_connection(self, backoff = 0):
         if self._disconnecting:
-            return
+            return None
         if backoff < 9:
             backoff += 1
         try:
@@ -89,7 +89,7 @@ async def _do_connection(self, backoff = 0):
             # Handle disconnection and retry with exponential backoff
             await self._close_connection()
             if self._disconnecting:
-                return
+                return None
             await asyncio.sleep(min(5 * 60, 2**backoff * 1))
             return await self._do_connection(backoff)
 
@@ -108,10 +108,11 @@ async def _process_line(self, reader, writer):
                     pass
                 else:
                     await self.emit('message', message)
-            except (json.decoder.JSONDecodeError) as ex:
+            except json.decoder.JSONDecodeError:
                 await self.emit('malformed', data)
 
-    async def _send_subscribe(self, writer):
+    @staticmethod
+    async def _send_subscribe(writer):
         writer.write(b'subscribe(60)\n')
         await writer.drain()
 

src/powersensor_local/plug_listener_udp.py

@@ -92,7 +92,7 @@ async def _do_connection(self):
             family = socket.AF_INET,
             remote_addr = (self._ip, self._port))
         self._reconnect = loop.call_later(
-            min(5*60, 2**self._backoff + 2), self._retry)
+            min(5*60, 2**self._backoff + 2), self._retry) # noqa
 
     def _send_subscribe(self):
         if self._transport is not None:
@@ -123,7 +123,7 @@ def datagram_received(self, data, addr):
         if self._inactive is not None:
             self._inactive.cancel()
         loop = asyncio.get_running_loop()
-        self._inactive = loop.call_later(60, self._on_inactivity)
+        self._inactive = loop.call_later(60, self._on_inactivity) # noqa
 
         lines = data.decode('utf-8').splitlines()
         for line in lines:
@@ -137,7 +137,7 @@ def datagram_received(self, data, addr):
                     pass
                 else:
                     asyncio.create_task(self.emit('message', message))
-            except (json.decoder.JSONDecodeError) as ex:
+            except json.decoder.JSONDecodeError:
                 asyncio.create_task(self.emit('malformed', data))
 
     def error_received(self, exc):

src/powersensor_local/virtual_household.py

@@ -2,6 +2,9 @@
 
 import sys
 from pathlib import Path
+
+from powersensor_local.EventBuffer import EventBuffer
+
 project_root = str(Path(__file__).parents[1])
 if project_root not in sys.path:
     sys.path.append(project_root)
@@ -48,7 +51,7 @@ def same_duration(ev1: dict, ev2: dict):
     d2 = round(ev2[dur], 0)
     return d1 == d2
 
-def matching_instants(starttime_utc: int, solar_events: list, housenet_events: list) -> Optional[InstantaneousValues]:
+def matching_instants(starttime_utc: int, solar_events: EventBuffer, housenet_events: EventBuffer) -> Optional[InstantaneousValues]:
     """Attempts to match and merge solar+housenet average_power events."""
     solar = solar_events.find_by_key(KEY_START, starttime_utc)
     housenet = housenet_events.find_by_key(KEY_START, starttime_utc)
@@ -73,7 +76,7 @@ def make_instant_housenet(ev: dict) -> Optional[InstantaneousValues]:
         duration_s = round(ev[KEY_DUR_S], 0)
     )
 
-def matching_summations(starttime_utc: int, solar_events: list, housenet_events: list) -> Optional[SummationValues]:
+def matching_summations(starttime_utc: int, solar_events: EventBuffer, housenet_events: EventBuffer) -> Optional[SummationValues]:
     """Attempts to match and merge solar+housenet summation events."""
     solar = solar_events.find_by_key(KEY_START, starttime_utc)
     housenet = housenet_events.find_by_key(KEY_START, starttime_utc)
@@ -148,10 +151,10 @@ def __init__(self, with_solar: bool):
         self._expect_solar = with_solar
         self._summation = self.SummationInfo(0, 0, 0, 0)
         self._counters = self.Counters(0, 0, 0, 0, 0)
-        self._solar_instants = self.EventBuffer(31)
-        self._housenet_instants = self.EventBuffer(31)
-        self._solar_summations = self.EventBuffer(5)
-        self._housenet_summations = self.EventBuffer(5)
+        self._solar_instants = EventBuffer(31)
+        self._housenet_instants = EventBuffer(31)
+        self._solar_summations = EventBuffer(5)
+        self._housenet_summations = EventBuffer(5)
 
     async def process_average_power_event(self, ev: dict):
         """Ingests an event of type 'average_power'."""
@@ -186,7 +189,6 @@ async def process_summation_event(self, ev: dict):
                 await self._process_summations(starttime_utc)
 
     async def _process_instants(self, starttime_utc: int):
-        v = None
         if self._expect_solar:
             v = matching_instants(starttime_utc, self._solar_instants, self._housenet_instants)
         else:
@@ -216,7 +218,6 @@ async def _process_instants(self, starttime_utc: int):
             })
 
     async def _process_summations(self, starttime_utc: int):
-        v = None
         if self._expect_solar:
             v = matching_summations(starttime_utc, self._solar_summations, self._housenet_summations)
         else:
@@ -295,30 +296,6 @@ def _increment_counters(self, d: SummationDeltas):
         self._counters.from_grid += d.from_grid
         self._counters.home_use += d.home_use
 
-    class EventBuffer:
-        def __init__(self, keep: int):
-            self._keep = keep
-            self._evs = []
-
-        def find_by_key(self, key: str, value: any):
-            for ev in self._evs:
-                if key in ev and ev[key] == value:
-                  return ev
-            return None
-
-        def append(self, ev: dict):
-            self._evs.append(ev)
-            if len(self._evs) > self._keep:
-                del self._evs[0]
-
-        def evict_older(self, key: str, value: float):
-            while len(self._evs) > 0:
-                ev = self._evs[0]
-                if key in ev and ev[key] <= value:
-                    del self._evs[0]
-                else:
-                    return
-
     @dataclass
     class SummationInfo:
         solar_resettime: int