diff --git a/README.md b/README.md index b23aaff..15e8255 100644 --- a/README.md +++ b/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")) diff --git a/python/raydriver/cli/main.py b/python/raydriver/cli/main.py index 8b2964b..ea6cae6 100644 --- a/python/raydriver/cli/main.py +++ b/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: diff --git a/python/raydriver/grbl/__init__.pyi b/python/raydriver/grbl/__init__.pyi index 4267dcb..afee7b8 100644 --- a/python/raydriver/grbl/__init__.pyi +++ b/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: ... + + ... + diff --git a/src/python/events.rs b/src/python/events.rs index 4d43e57..8e92602 100644 --- a/src/python/events.rs +++ b/src/python/events.rs @@ -3,22 +3,64 @@ //! Implements [`SessionEvents`] by marshaling each event onto the //! caller's asyncio loop. Events are queued on the Rust side and a //! single `call_soon_threadsafe(drain)` signal per batch wakes the -//! loop, which invokes `event_callback(name, payload)` for every -//! queued event on the loop thread. Batching keeps the loop's -//! self-pipe write rate at one signal per batch regardless of event -//! volume. +//! loop, which materializes one [`SessionEvent`] per queued event and +//! invokes `event_callback(event)` for it on the loop thread. +//! Batching keeps the loop's self-pipe write rate at one signal per +//! batch regardless of event volume. use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use pyo3::prelude::*; +use pyo3_stub_gen::derive::gen_stub_pyclass_complex_enum; use crate::grbl::session::SessionEvents; -use crate::grbl::types::{DeviceState, TransportStatus}; +use crate::grbl::types::{DeviceState as CoreDeviceState, TransportStatus}; -/// One queued event: the event name and its payload object. -type QueuedEvent = (String, Py); +use crate::python::grbl::types::DeviceState as PyDeviceState; + +/// One typed session event, delivered to `event_callback`. +#[gen_stub_pyclass_complex_enum] +#[pyclass(module = "raydriver.grbl", skip_from_py_object)] +#[derive(Clone)] +pub enum SessionEvent { + StateChanged { + state: PyDeviceState, + }, + ConnectionStatusChanged { + status: String, + message: Option, + }, + CommandStatusChanged { + status: String, + message: Option, + }, + JobFinished {}, + ProbeStatusChanged { + message: String, + }, + WcsUpdated { + offsets: HashMap, + }, + ConfigChanged { + key: String, + value: i64, + }, + RawRx { + line: String, + }, + RawTx { + line: String, + }, +} + +/// One queued event: session events go to the event callback, a +/// progress report goes to the progress callback. +enum QueuedEvent { + Session(SessionEvent), + Progress(i64), +} pub(crate) struct PyEvents { event_callback: Option>, @@ -60,11 +102,13 @@ impl EventQueue { /// The loop-side drain callable: invoked via /// `loop.call_soon_threadsafe(drain)`; runs on the event loop -/// thread and delivers every queued event. +/// thread and delivers every queued event. Carries both callbacks +/// so each queued event routes to its own callback no matter which +/// schedule call drained the batch. #[pyclass(module = "raydriver.grbl")] struct EventDrainer { queue: Arc, - callback: Py, + events: Option>, progress: Option>, } @@ -100,20 +144,16 @@ impl PyEvents { *self.progress_callback.lock().unwrap() = cb; } - /// Queue `(name, payload)` for delivery and schedule the loop + /// Queue a session event for delivery and schedule the loop /// drain if this caller flipped the scheduled flag. The GIL is - /// attached only for payload construction and the single - /// `call_soon_threadsafe` signal. - fn schedule(&self, name: &str, payload: F) - where - F: FnOnce(Python<'_>) -> PyResult>, - { + /// attached only for the single `call_soon_threadsafe` signal; + /// the event is materialized into a Python object at drain time. + fn schedule(&self, event: SessionEvent) { if self.event_callback.is_none() { return; } let result: PyResult<()> = Python::attach(|py| { - let payload = payload(py)?; - if self.queue.push((name.to_string(), payload)) { + if self.queue.push(QueuedEvent::Session(event)) { let loop_guard = self.loop_.lock().unwrap(); let Some(loop_) = loop_guard.as_ref() else { self.queue.unschedule(); @@ -130,7 +170,7 @@ impl PyEvents { py, EventDrainer { queue: self.queue.clone(), - callback: callback.clone().unbind(), + events: Some(callback.clone().unbind()), progress: self .progress_callback .lock() @@ -147,7 +187,7 @@ impl PyEvents { }); if let Err(err) = result { self.queue.unschedule(); - log::warn!("failed to schedule session event {name}: {err}"); + log::warn!("failed to schedule session event: {err}"); } } @@ -169,17 +209,16 @@ impl PyEvents { }; let loop_py = loop_.clone_ref(py); drop(loop_guard); - let payload = { - let obj = op_index.into_pyobject(py)?; - obj.unbind().into_any() - }; - if self.queue.push(("__progress__".to_string(), payload)) { + if self.queue.push(QueuedEvent::Progress(op_index)) { let drainer = Py::new( py, EventDrainer { queue: self.queue.clone(), - callback, - progress: None, + events: self + .event_callback + .as_ref() + .map(|cb| cb.clone_ref(py)), + progress: Some(callback), }, )?; if let Err(err) = loop_py @@ -197,29 +236,12 @@ impl PyEvents { log::warn!("failed to schedule command_done: {err}"); } } - - /// Deliver one queued progress report (called on the loop - /// thread by the drainer). - fn deliver_progress( - callback: &Py, - py: Python<'_>, - payload: &Py, - ) { - if let Err(err) = callback.call1(py, (payload,)) { - log::warn!("progress callback failed: {err}"); - } - } } impl SessionEvents for PyEvents { - fn state_changed(&self, state: &DeviceState) { - let state = state.clone(); - self.schedule("state_changed", move |py| { - let obj = Py::new( - py, - crate::python::grbl::types::DeviceState::new(state), - )?; - Ok(obj.into_any()) + fn state_changed(&self, state: &CoreDeviceState) { + self.schedule(SessionEvent::StateChanged { + state: PyDeviceState::new(state.clone()), }); } @@ -228,51 +250,39 @@ impl SessionEvents for PyEvents { status: TransportStatus, message: Option<&str>, ) { - let status = status.name().to_string(); - let message = message.map(str::to_string); - self.schedule("connection_status_changed", move |py| { - Ok((status.clone(), message.clone()) - .into_pyobject(py)? - .unbind() - .into_any()) + self.schedule(SessionEvent::ConnectionStatusChanged { + status: status.name().to_string(), + message: message.map(str::to_string), }); } fn command_status(&self, status: TransportStatus, message: Option<&str>) { - let status = status.name().to_string(); - let message = message.map(str::to_string); - self.schedule("command_status_changed", move |py| { - Ok((status.clone(), message.clone()) - .into_pyobject(py)? - .unbind() - .into_any()) + self.schedule(SessionEvent::CommandStatusChanged { + status: status.name().to_string(), + message: message.map(str::to_string), }); } fn job_finished(&self) { - self.schedule("job_finished", move |py| { - Ok(super::pyfuture::none_obj(py)) - }); + self.schedule(SessionEvent::JobFinished {}); } fn probe_status(&self, message: &str) { - let message = message.to_string(); - self.schedule("probe_status_changed", move |py| { - Ok(message.clone().into_pyobject(py)?.unbind().into_any()) + self.schedule(SessionEvent::ProbeStatusChanged { + message: message.to_string(), }); } fn wcs_updated(&self, offsets: &HashMap) { - let offsets = offsets.clone(); - self.schedule("wcs_updated", move |py| { - Ok(offsets.clone().into_pyobject(py)?.unbind().into_any()) + self.schedule(SessionEvent::WcsUpdated { + offsets: offsets.clone(), }); } fn config_changed(&self, key: &str, value: i64) { - let key = key.to_string(); - self.schedule("config_changed", move |py| { - Ok((key.clone(), value).into_pyobject(py)?.unbind().into_any()) + self.schedule(SessionEvent::ConfigChanged { + key: key.to_string(), + value, }); } @@ -281,33 +291,49 @@ impl SessionEvents for PyEvents { } fn raw_rx(&self, line: &str) { - let line = line.to_string(); - self.schedule("raw_rx", move |py| { - Ok(line.clone().into_pyobject(py)?.unbind().into_any()) + self.schedule(SessionEvent::RawRx { + line: line.to_string(), }); } fn raw_tx(&self, line: &str) { - let line = line.to_string(); - self.schedule("raw_tx", move |py| { - Ok(line.clone().into_pyobject(py)?.unbind().into_any()) + self.schedule(SessionEvent::RawTx { + line: line.to_string(), }); } } impl EventDrainer { - /// Progress reports use a sentinel event name and route to the - /// progress callback instead of the event callback. + /// Deliver each queued event to its own callback: session + /// events to the event callback, progress reports to the + /// progress callback. fn take_and_deliver(&self, py: Python<'_>) { - for (name, payload) in self.queue.take() { - if name == "__progress__" { - let progress_cb = - self.progress.as_ref().unwrap_or(&self.callback); - PyEvents::deliver_progress(progress_cb, py, &payload); - } else { - let callback = self.callback.bind(py); - if let Err(err) = callback.call1((name, payload)) { - log::warn!("event callback failed: {err}"); + for event in self.queue.take() { + match event { + QueuedEvent::Session(event) => { + let Some(callback) = self.events.as_ref() else { + continue; + }; + let obj = match event.into_pyobject(py) { + Ok(obj) => obj.into_any().unbind(), + Err(err) => { + log::warn!( + "failed to materialize session event: {err}" + ); + continue; + } + }; + if let Err(err) = callback.bind(py).call1((obj,)) { + log::warn!("event callback failed: {err}"); + } + } + QueuedEvent::Progress(op_index) => { + let Some(callback) = self.progress.as_ref() else { + continue; + }; + if let Err(err) = callback.call1(py, (op_index,)) { + log::warn!("progress callback failed: {err}"); + } } } } diff --git a/src/python/grbl/mod.rs b/src/python/grbl/mod.rs index 4d69ac0..64e8ad7 100644 --- a/src/python/grbl/mod.rs +++ b/src/python/grbl/mod.rs @@ -25,10 +25,14 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { let grbl_mod = PyModule::new(py, "grbl")?; grbl_mod.setattr("__doc__", MODULE_DOC)?; - grbl_mod.add("__all__", vec!["GrblSession", "MockTransport", "types"])?; + grbl_mod.add( + "__all__", + vec!["GrblSession", "MockTransport", "SessionEvent", "types"], + )?; grbl_mod.add_class::()?; grbl_mod.add_class::()?; + grbl_mod.add_class::()?; let types_mod = PyModule::new(py, "types")?; types::register(&types_mod)?; diff --git a/src/python/grbl/session.rs b/src/python/grbl/session.rs index 6ab713e..82c402b 100644 --- a/src/python/grbl/session.rs +++ b/src/python/grbl/session.rs @@ -171,8 +171,8 @@ fn parse_dialect(dict: Option<&Bound<'_, PyDict>>) -> PyResult { /// 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. #[gen_stub_pyclass] #[pyclass(module = "raydriver.grbl")] pub struct GrblSession { diff --git a/src/python/grbl/types.rs b/src/python/grbl/types.rs index cb52358..d750509 100644 --- a/src/python/grbl/types.rs +++ b/src/python/grbl/types.rs @@ -1,6 +1,11 @@ //! Python-visible value types mirroring Rayforge's //! `machine.driver.driver` module. +// `from_py_object` generates a FromPyObject impl whose .clone() is +// flagged as clone_on_copy for Copy types; the lint cannot be +// targeted at the generated impl, only at this module. +#![allow(clippy::clone_on_copy)] + use pyo3::prelude::*; use pyo3_stub_gen::derive::{ gen_stub_pyclass, gen_stub_pyclass_enum, gen_stub_pymethods, diff --git a/tests/conftest.py b/tests/conftest.py index 06042b9..335e288 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,7 +8,7 @@ import pytest from raydriver.emulator import GrblEmulator -from raydriver.grbl import GrblSession, MockTransport +from raydriver.grbl import GrblSession, MockTransport, SessionEvent def _arm_freeze_watchdog(): @@ -65,24 +65,25 @@ class EventRecorder: def __init__(self): self.events = [] - def callback(self, name, payload=None): - self.events.append((name, payload)) + def callback(self, event): + self.events.append(event) def names(self): - return [name for name, _ in self.events] + return [type(event).__name__ for event in self.events] - async def wait_for(self, name, pred=None, timeout=5.0): - """Wait until an event *name* (matching *pred*) was recorded.""" + async def wait_for(self, kind, pred=None, timeout=5.0): + """Wait until an event of *kind* (a SessionEvent variant + class) matching *pred* was recorded.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: - for event_name, payload in self.events: - if event_name != name: + for event in self.events: + if not isinstance(event, kind): continue - if pred is None or pred(payload): - return payload + if pred is None or pred(event): + return event await asyncio.sleep(0.01) raise AssertionError( - f"event {name!r} not observed within {timeout:.1f}s; " + f"event {kind!r} not observed within {timeout:.1f}s; " f"got {self.names()}" ) @@ -136,7 +137,8 @@ async def rig(mock, events): session = make_session(mock, events) await session.connect() await events.wait_for( - "connection_status_changed", lambda p: p[0] == "CONNECTED" + SessionEvent.ConnectionStatusChanged, + lambda event: event.status == "CONNECTED", ) mock.clear_sent() events.events.clear() diff --git a/tests/test_grbl_raw_lines.py b/tests/test_grbl_raw_lines.py index f4e0f8b..2444a9d 100644 --- a/tests/test_grbl_raw_lines.py +++ b/tests/test_grbl_raw_lines.py @@ -2,6 +2,8 @@ from conftest import make_session, start_device +from raydriver.grbl import SessionEvent + class TestRawLines: async def test_raw_tx_reports_interactive_command(self, mock, events): @@ -10,11 +12,12 @@ async def test_raw_tx_reports_interactive_command(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", lambda p: p[0] == "CONNECTED" + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", ) events.events.clear() await session.execute_interactive_command("$$") - await events.wait_for("raw_tx", lambda line: line == "$$") + await events.wait_for(SessionEvent.RawTx, lambda e: e.line == "$$") finally: await session.disconnect() device_task.cancel() @@ -22,18 +25,20 @@ async def test_raw_tx_reports_interactive_command(self, mock, events): async def test_raw_rx_reports_settings_reply(self, rig): session, mock, emulator, events = rig await session.execute_interactive_command("$$") - await events.wait_for("raw_rx", lambda line: "ok" == line) + await events.wait_for(SessionEvent.RawRx, lambda e: "ok" == e.line) assert any( - name == "raw_rx" and line.startswith("$") - for name, line in events.events + isinstance(e, SessionEvent.RawRx) and e.line.startswith("$") + for e in events.events ) async def test_raw_rx_reports_status_report(self, rig): session, mock, emulator, events = rig - report = await events.wait_for( - "raw_rx", lambda line: line.startswith("<") and line.endswith(">") + event = await events.wait_for( + SessionEvent.RawRx, + lambda e: e.line.startswith("<") and e.line.endswith(">"), ) - assert "Idle" in report or "Run" in report or "Hold" in report + line = event.line + assert "Idle" in line or "Run" in line or "Hold" in line async def test_raw_rx_reports_welcome_message(self, mock, events): emulator, device_task = await start_device(mock) @@ -41,7 +46,7 @@ async def test_raw_rx_reports_welcome_message(self, mock, events): try: await session.connect() await events.wait_for( - "raw_rx", lambda line: line.startswith("Grbl ") + SessionEvent.RawRx, lambda e: e.line.startswith("Grbl ") ) finally: await session.disconnect() @@ -51,9 +56,9 @@ async def test_raw_lines_carry_full_conversation(self, rig): session, mock, emulator, events = rig await session.execute_interactive_command("$I") raw = [ - line - for name, line in events.events - if name in ("raw_rx", "raw_tx") + e.line + for e in events.events + if isinstance(e, (SessionEvent.RawRx, SessionEvent.RawTx)) ] assert "$I" in raw assert any(line.startswith("[VER:") for line in raw) diff --git a/tests/test_grbl_session.py b/tests/test_grbl_session.py index b787cd6..605caa0 100644 --- a/tests/test_grbl_session.py +++ b/tests/test_grbl_session.py @@ -6,6 +6,7 @@ import pytest from conftest import make_session, sent_bytes, start_device +from raydriver.grbl import SessionEvent from raydriver.grbl.types import DeviceStatus @@ -15,11 +16,11 @@ async def test_handshake_and_build_info(self, mock, events): session = make_session(mock, events) try: await session.connect() - payload = await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + event = await events.wait_for( + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", ) - assert payload[1] is None + assert event.message is None # The connection loop queried $I, whose OPT line # advertised the emulated 128-byte RX buffer. assert b"$I\n" in sent_bytes(mock) @@ -33,14 +34,14 @@ async def test_handshake_timeout_on_phantom_port(self, mock, events): session = make_session(mock, events, {"handshake_timeout": 0.15}) try: await session.connect() - payload = await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "ERROR", + event = await events.wait_for( + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "ERROR", ) - assert "No response" in payload[1] + assert "No response" in event.message await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "SLEEPING", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "SLEEPING", ) finally: await session.disconnect() @@ -50,14 +51,14 @@ async def test_reconnect_after_recovery(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "ERROR", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "ERROR", ) # The device appears (cable plugged back in). emulator, device_task = await start_device(mock) await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", timeout=8.0, ) device_task.cancel() @@ -84,8 +85,8 @@ async def test_rx_buffer_size_override_wins(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", ) # OPT and Bf advertise 128, but the override sticks. await asyncio.sleep(0.15) @@ -97,12 +98,20 @@ async def test_rx_buffer_size_override_wins(self, mock, events): async def test_fragmented_status_report_ignored(self, rig): session, mock, emulator, events = rig states_before = len( - [e for e in events.events if e[0] == "state_changed"] + [ + e + for e in events.events + if isinstance(e, SessionEvent.StateChanged) + ] ) mock.push(b" 20.0, + event = await events.wait_for( + SessionEvent.StateChanged, + lambda e: ( + e.state.machine_pos[0] is not None + and e.state.machine_pos[0] > 20.0 + ), ) # The report arrived in inches; the driver reports mm. - assert payload.machine_pos[0] == pytest.approx(25.4, abs=0.1) + assert event.state.machine_pos[0] == pytest.approx(25.4, abs=0.1) diff --git a/tests/test_grbl_streaming.py b/tests/test_grbl_streaming.py index 7a60070..c19040d 100644 --- a/tests/test_grbl_streaming.py +++ b/tests/test_grbl_streaming.py @@ -12,6 +12,7 @@ ) from raydriver.emulator import WELCOME +from raydriver.grbl import SessionEvent from raydriver.grbl.types import DeviceStatus @@ -29,7 +30,7 @@ class TestRunBasics: async def test_simple_job_completes_and_moves(self, rig): session, mock, emulator, events = rig await session.run("G1 X10 F1000\nG1 Y5 F1000\n") - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) await asyncio.sleep(0.1) assert session.buffer_count == 0 assert session.pending_commands == [] @@ -49,7 +50,7 @@ async def test_comments_and_empty_lines_skipped(self, rig): await session.run( "G1 X1 F1000 ; comment\n\n(comment only)\nG1 X2 F1000\n" ) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) sent = sent_bytes(mock) assert b"comment" not in sent assert b"G1 X1 F1000\n" in sent @@ -60,11 +61,12 @@ async def test_job_status_reflected_as_run(self, rig): emulator.speed_factor = 50 # ~0.5 s of motion job = asyncio.ensure_future(session.run(gcode(30))) await events.wait_for( - "state_changed", lambda p: p.status.name == "RUN" + SessionEvent.StateChanged, + lambda e: e.state.status.name == "RUN", ) emulator.speed_factor = 100000 await asyncio.wait_for(job, timeout=10) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) async def test_long_job_never_overflows_rx_buffer(self, rig): session, mock, emulator, events = rig @@ -73,7 +75,7 @@ async def test_long_job_never_overflows_rx_buffer(self, rig): # must never see an RX overflow from the character-counting # sender. await asyncio.wait_for(session.run(gcode(60)), timeout=30) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) await asyncio.sleep(0.1) assert emulator.rx_overrun is False assert session.buffer_count == 0 @@ -87,7 +89,7 @@ async def test_error_halts_stream(self, rig): lines[5] = "G999" job = asyncio.ensure_future(session.run("\n".join(lines))) await asyncio.wait_for(job, timeout=10) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) await asyncio.sleep(0.1) sent = sent_bytes(mock) assert sent.count(b"G1 X") < 40 @@ -95,7 +97,11 @@ async def test_error_halts_stream(self, rig): assert b"\x18" in sent # Soft reset reboots the emulator: welcome banner again. assert emulator.tx_log.count(WELCOME) >= 2 - states = [p for n, p in events.events if n == "state_changed"] + states = [ + e.state + for e in events.events + if isinstance(e, SessionEvent.StateChanged) + ] assert any(s.error is not None and s.error.code == 20 for s in states) # run() resolves normally even on failure, so the terminal # error must be readable from the device state afterwards. @@ -117,7 +123,7 @@ async def test_error_halts_stream(self, rig): # A subsequent job start clears the error again. job = asyncio.ensure_future(session.run(gcode(5))) await asyncio.wait_for(job, timeout=10) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) assert session.state.error is None async def test_soft_limit_alarm_halts_stream(self, rig): @@ -129,10 +135,14 @@ async def test_soft_limit_alarm_halts_stream(self, rig): lines[3] = "G1 X10000 F1000" job = asyncio.ensure_future(session.run("\n".join(lines))) await asyncio.wait_for(job, timeout=10) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) await asyncio.sleep(0.1) assert sent_bytes(mock).count(b"G1 X") < 40 - states = [p for n, p in events.events if n == "state_changed"] + states = [ + e.state + for e in events.events + if isinstance(e, SessionEvent.StateChanged) + ] assert any( s.error is not None and s.error.title == "Soft Limit" for s in states @@ -146,8 +156,8 @@ async def test_stall_retries_while_device_alive(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", ) mock.clear_sent() # Slow motion: the planner stays full and acks stop, @@ -161,7 +171,7 @@ async def test_stall_retries_while_device_alive(self, mock, events): # Speed the machine up: the job drains and completes. emulator.speed_factor = 100000 await asyncio.wait_for(job, timeout=30) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) assert emulator.rx_overrun is False finally: await session.disconnect() @@ -173,13 +183,13 @@ async def test_stall_aborts_when_device_dies(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", ) mock.clear_sent() emulator.silent = True # the cable is gone await asyncio.wait_for(session.run(slow_gcode(40)), timeout=40) - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) assert sent_bytes(mock).count(b"G1 X") < 40 # No protocol error was seen, so run() must surface the # job-level failure ("device stopped responding") through @@ -211,7 +221,9 @@ async def test_cancel_mid_job_sends_reset_and_safety_shutdown(self, rig): assert b"M9\n" in sent # Soft reset mid-motion reboots Grbl into the alarm lock. assert emulator.tx_log.count(WELCOME) >= 1 - finished = [e for e in events.events if e[0] == "job_finished"] + finished = [ + e for e in events.events if isinstance(e, SessionEvent.JobFinished) + ] assert len(finished) == 1 async def test_cancel_without_job(self, rig): @@ -275,7 +287,7 @@ async def test_mixed_raw(self, rig): session, mock, emulator, events = rig mock.clear_sent() await session.run_raw("G1 X1 F1000\n!\nG1 X2 F1000\n") - await events.wait_for("job_finished") + await events.wait_for(SessionEvent.JobFinished) sent = sent_bytes(mock) assert b"G1 X1 F1000\n" in sent assert b"!" in sent diff --git a/tests/test_grbl_stress.py b/tests/test_grbl_stress.py index 6d44bc3..128ab82 100644 --- a/tests/test_grbl_stress.py +++ b/tests/test_grbl_stress.py @@ -5,6 +5,8 @@ from conftest import make_session, start_device +from raydriver.grbl import SessionEvent + def gcode(n_lines): return "\n".join(f"G1 X{i % 50} F2000" for i in range(n_lines)) @@ -21,8 +23,8 @@ async def test_fragmented_output_long_job(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", timeout=5.0, ) done = [] @@ -49,8 +51,8 @@ async def test_polling_during_job_stays_stable(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", timeout=5.0, ) n = 200 @@ -71,8 +73,8 @@ async def test_1000_line_job(self, mock, events): try: await session.connect() await events.wait_for( - "connection_status_changed", - lambda p: p[0] == "CONNECTED", + SessionEvent.ConnectionStatusChanged, + lambda e: e.status == "CONNECTED", timeout=5.0, ) done = []