Raw interface to single plug.

Author: jmattsson

pyproject.toml

@@ -23,6 +23,7 @@ Issues = "https://github.com/DiUS/python-powersensor_local/issues"
 [project.scripts]
 ps-events = "powersensor_local.events:app"
 ps-rawfirehose = "powersensor_local.rawfirehose:app"
+ps-rawplug = "powersensor.rawplug:app"
 
 [build-system]
 requires = [ "hatchling" ]

src/powersensor_local/async_event_emitter.py

@@ -0,0 +1,28 @@
+class AsyncEventEmitter:
+    """Small helper class for pub/sub functionality with async handlers."""
+    def __init__(self):
+        self._listeners = {}
+
+    def subscribe(self, evstr, cb):
+        """Registers an event handler for the given event key. The handler must
+        be async. Duplicate registrations are ignored."""
+        if self._listeners.get(evstr) is None:
+            self._listeners[evstr] = []
+        if not cb in self._listeners[evstr]:
+            self._listeners[evstr].append(cb)
+
+    def unsubscribe(self, evstr, cb):
+        """Unregisters the given event handler from the given event type."""
+        if self._listeners.get(evstr) is None:
+            return
+        if cb in self._listeners[evstr]:
+            self._listeners[evstr].remove(cb)
+
+    async def emit(self, evstr, *args):
+        """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."""
+        if self._listeners.get(evstr) is None:
+            return
+        for cb in self._listeners[evstr]:
+            await cb(evstr, *args)

src/powersensor_local/plug_listener.py

@@ -0,0 +1,107 @@
+import asyncio
+import json
+import socket
+import time
+
+from async_event_emitter import AsyncEventEmitter
+
+class PlugListener(AsyncEventEmitter):
+    """An interface class for accessing the event stream from a single plug.
+    The following events may be emitted:
+      - ("connecting")   Whenever a connection attempt is made.
+      - ("connected")    When a connection is successful.
+      - ("disconnected") When a connection is dropped, be it intentional or not.
+      - ("message",{...}) For each event message received from the plug. The
+      plug's JSON message is decoded into a dict which is passed as the second
+      argument to the registered event handler(s). The event handlers must be
+      async.
+    """
+
+    def __init__(self, ip, port=49476):
+        """Initialises a PlugListener object, bound to the given IP address.
+        The port number may be overridden if necessary."""
+        super().__init__()
+        self._ip = ip
+        self._port = port
+        self._task = None
+        self._connection = None
+        self._disconnecting = False
+
+    def connect(self):
+        """Initiates the connection to the plug. The object will automatically
+        retry as necessary if/when it can't connect to the plug, until such
+        a time disconnect() is called."""
+        if self._task is not None:
+            raise RuntimeError("already connected/connecting")
+        self._disconnecting = False
+        self._task = asyncio.create_task(self._do_connection())
+
+    async def disconnect(self):
+        """Goes through the disconnection process towards a plug. No further
+        automatic reconnects will take place, until connect() is called."""
+        if self._task is None:
+            return
+
+        self._disconnecting = True
+
+        await self._close_connection()
+
+        if self._task is not None:
+            await self._task
+            self._task = None
+
+    async def _close_connection(self):
+        if self._connection is not None:
+            (reader, writer) = self._connection
+            self._connection = None
+
+            writer.close()
+            await writer.wait_closed()
+
+            await self.emit('disconnected')
+
+    async def _do_connection(self, backoff = 0):
+        if backoff < 9:
+            backoff += 1
+        try:
+            await self.emit('connecting')
+            reader, writer = await asyncio.open_connection(self._ip, self._port)
+            self._connection = (reader, writer)
+
+            await self._send_subscribe(writer)
+            backoff = 1
+
+            await self.emit('connected')
+
+            while not self._disconnecting:
+                await self._process_line(reader, writer)
+
+        except (ConnectionResetError, asyncio.TimeoutError):
+            # Handle disconnection and retry with exponential backoff
+            await self._close_connection()
+            if self._disconnecting:
+                return
+            await asyncio.sleep(min(5 * 60, 2**backoff * 1))
+            return await self._do_connection(backoff)
+
+    async def _process_line(self, reader, writer):
+        data = await reader.readline()
+        if data == b'':
+            raise ConnectionResetError
+        if data != b'\n': # Silently ignore empty lines
+            try:
+                message = json.loads(data.decode('utf-8'))
+                typ = message['type']
+                if typ == 'subscription':
+                    if message['subtype'] == 'warning':
+                        await self._send_subscribe(writer)
+                elif typ == 'discovery':
+                    pass
+                else:
+                    await self.emit('message', message)
+            except (json.decoder.JSONDecodeError) as ex:
+                print(f"JSON error {ex} from {data}")
+
+    async def _send_subscribe(self, writer):
+        writer.write(b'subscribe(60)\n')
+        await writer.drain()

src/powersensor_local/rawplug.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+
+"""Utility script for accessing the raw plug subscription data from a single
+network-local Powersensor devices. Intended for advanced debugging use only."""
+
+import asyncio
+import os
+import signal
+import sys
+from plug_listener import PlugListener
+
+exiting = False
+plug = None
+
+async def do_exit():
+    global exiting
+    global plug
+    if plug != None:
+        await plug.disconnect()
+        del plug
+    exiting = True
+
+async def on_evt_msg(_, msg):
+    print(msg)
+
+async def on_evt(evt):
+    print(evt)
+
+async def main():
+    if len(sys.argv) < 2:
+        print(f"Syntax: {sys.argv[0]} <ip> [port]")
+        sys.exit(1)
+
+    # Signal handler for Ctrl+C
+    def handle_sigint(signum, frame):
+        signal.signal(signal.SIGINT, signal.SIG_DFL)
+        asyncio.create_task(do_exit())
+
+    signal.signal(signal.SIGINT, handle_sigint)
+
+    global plug
+    plug = PlugListener(sys.argv[1], *sys.argv[2:2])
+    plug.subscribe('message', on_evt_msg)
+    plug.subscribe('connecting', on_evt)
+    plug.subscribe('connecting', on_evt)
+    plug.subscribe('connected', on_evt)
+    plug.subscribe('disconnected', on_evt)
+    plug.connect()
+
+    # Keep the event loop running until Ctrl+C is pressed
+    while not exiting:
+        await asyncio.sleep(1)
+
+def app():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    app()