Deliver typed session events to the Python callback

README.md

@@ -52,9 +52,8 @@ make check   # lint + test
 
 ```python
 import asyncio
-from raydriver.grbl import GrblSession
+from raydriver.grbl import GrblSession, SessionEvent
 
-events = []
 session = GrblSession(
     config={"port": "/dev/ttyUSB0", "baudrate": 115200},
     dialect={
@@ -64,9 +63,20 @@ session = GrblSession(
         "laser_on": "M4 S{power:.0f}",
         "laser_off": "M5",
     },
-    event_callback=lambda name, payload: events.append((name, payload)),
+    event_callback=on_event,
 )
 
+def on_event(event: SessionEvent):
+    match event:
+        case SessionEvent.StateChanged():
+            print("device state:", event.state.status.name)
+        case SessionEvent.ConnectionStatusChanged(status=status):
+            print("connection:", status)
+        case SessionEvent.RawRx(line=line):
+            print("<<", line)
+        case SessionEvent.RawTx(line=line):
+            print(">>", line)
+
 async def main():
     await session.connect()
     print(await session.execute_interactive_command("$I"))

python/raydriver/cli/main.py

@@ -20,10 +20,9 @@
 import os
 import sys
 import time
-from typing import Any
 
 import raydriver.raydriver as _native
-from raydriver.grbl import GrblSession, MockTransport
+from raydriver.grbl import GrblSession, MockTransport, SessionEvent
 from raydriver.grbl.parser import strip_gcode_comments
 from raydriver.grbl.types import DeviceState
 
@@ -160,15 +159,19 @@ async def cmd_run(args) -> int:
 
     connected = asyncio.get_running_loop().create_future()
 
-    def on_event(name: str, payload: Any = None):
-        if name == "state_changed":
-            view.on_state(payload)
-        elif name == "connection_status_changed" and not (connected.done()):
-            if payload[0] == "CONNECTED":
+    def on_event(event: SessionEvent):
+        match event:
+            case SessionEvent.StateChanged():
+                view.on_state(event.state)
+            case SessionEvent.ConnectionStatusChanged(status="CONNECTED") if (
+                not connected.done()
+            ):
                 connected.set_result(True)
-            elif payload[0] == "ERROR":
+            case SessionEvent.ConnectionStatusChanged(
+                status="ERROR", message=message
+            ) if not connected.done():
                 connected.set_exception(
-                    ConnectionError(payload[1] or "connection failed")
+                    ConnectionError(message or "connection failed")
                 )
 
     session, device_task = _open_session(args, on_event)
@@ -213,17 +216,19 @@ def on_event(name: str, payload: Any = None):
 
 
 async def cmd_status(args) -> int:
-    def on_event(name: str, payload: Any = None):
-        if name == "state_changed":
-            mpos = ",".join(
-                f"{v:8.3f}" if v is not None else "     -  "
-                for v in payload.machine_pos
-            )
-            error = f" ! {payload.error.title}" if payload.error else ""
-            print(
-                f"{time.strftime('%H:%M:%S')} {payload.status.name:8}"
-                f" MPos:{mpos}{error}"
-            )
+    def on_event(event: SessionEvent):
+        match event:
+            case SessionEvent.StateChanged():
+                state = event.state
+                mpos = ",".join(
+                    f"{v:8.3f}" if v is not None else "     -  "
+                    for v in state.machine_pos
+                )
+                error = f" ! {state.error.title}" if state.error else ""
+                print(
+                    f"{time.strftime('%H:%M:%S')} {state.status.name:8}"
+                    f" MPos:{mpos}{error}"
+                )
 
     session, device_task = _open_session(args, on_event)
     try:

python/raydriver/grbl/__init__.pyi

@@ -13,6 +13,7 @@ from . import types
 __all__ = [
     "GrblSession",
     "MockTransport",
+    "SessionEvent",
     "parser",
     "types",
 ]
@@ -23,8 +24,8 @@ class GrblSession:
     A GRBL device session: connection lifecycle, streaming and
     interactive commands, entirely in Rust.
 
-    All async methods return asyncio futures; events are delivered to
-    `event_callback(name, payload)` on the event loop thread.
+    All async methods return asyncio futures; a typed `SessionEvent`
+    is delivered to `event_callback(event)` on the event loop thread.
     """
     @property
     def state(self) -> types.DeviceState:
@@ -188,3 +189,76 @@ class MockTransport:
         """
     def __repr__(self) -> builtins.str: ...
 
+class SessionEvent:
+    r"""
+    One typed session event, delivered to `event_callback`.
+    """
+    @typing.final
+    class StateChanged(SessionEvent):
+        __match_args__ = ("state",)
+        @property
+        def state(self) -> types.DeviceState: ...
+        def __new__(cls, state: types.DeviceState) -> SessionEvent.StateChanged: ...
+
+    @typing.final
+    class ConnectionStatusChanged(SessionEvent):
+        __match_args__ = ("status", "message",)
+        @property
+        def status(self) -> builtins.str: ...
+        @property
+        def message(self) -> typing.Optional[builtins.str]: ...
+        def __new__(cls, status: builtins.str, message: typing.Optional[builtins.str]) -> SessionEvent.ConnectionStatusChanged: ...
+
+    @typing.final
+    class CommandStatusChanged(SessionEvent):
+        __match_args__ = ("status", "message",)
+        @property
+        def status(self) -> builtins.str: ...
+        @property
+        def message(self) -> typing.Optional[builtins.str]: ...
+        def __new__(cls, status: builtins.str, message: typing.Optional[builtins.str]) -> SessionEvent.CommandStatusChanged: ...
+
+    @typing.final
+    class JobFinished(SessionEvent):
+        __match_args__ = ()
+        def __new__(cls) -> SessionEvent.JobFinished: ...
+
+    @typing.final
+    class ProbeStatusChanged(SessionEvent):
+        __match_args__ = ("message",)
+        @property
+        def message(self) -> builtins.str: ...
+        def __new__(cls, message: builtins.str) -> SessionEvent.ProbeStatusChanged: ...
+
+    @typing.final
+    class WcsUpdated(SessionEvent):
+        __match_args__ = ("offsets",)
+        @property
+        def offsets(self) -> builtins.dict[builtins.str, tuple[builtins.float, builtins.float, builtins.float]]: ...
+        def __new__(cls, offsets: typing.Mapping[builtins.str, tuple[builtins.float, builtins.float, builtins.float]]) -> SessionEvent.WcsUpdated: ...
+
+    @typing.final
+    class ConfigChanged(SessionEvent):
+        __match_args__ = ("key", "value",)
+        @property
+        def key(self) -> builtins.str: ...
+        @property
+        def value(self) -> builtins.int: ...
+        def __new__(cls, key: builtins.str, value: builtins.int) -> SessionEvent.ConfigChanged: ...
+
+    @typing.final
+    class RawRx(SessionEvent):
+        __match_args__ = ("line",)
+        @property
+        def line(self) -> builtins.str: ...
+        def __new__(cls, line: builtins.str) -> SessionEvent.RawRx: ...
+
+    @typing.final
+    class RawTx(SessionEvent):
+        __match_args__ = ("line",)
+        @property
+        def line(self) -> builtins.str: ...
+        def __new__(cls, line: builtins.str) -> SessionEvent.RawTx: ...
+
+    ...
+