diff --git a/src/grbl/session.rs b/src/grbl/session.rs index ecb79e1..98ba312 100644 --- a/src/grbl/session.rs +++ b/src/grbl/session.rs @@ -44,6 +44,14 @@ pub trait SessionEvents: Send + Sync { fn wcs_updated(&self, offsets: &HashMap); fn config_changed(&self, key: &str, value: i64); fn command_done(&self, op_index: i64); + + /// A complete line received from the device, including status + /// reports, `ok`/`error:` acknowledgements and `$` replies. + fn raw_rx(&self, _line: &str) {} + + /// A complete line sent to the device (interactive commands and + /// realtime bytes; not bulk job streaming). + fn raw_tx(&self, _line: &str) {} } /// No-op event sink. @@ -663,9 +671,18 @@ impl SessionCore { } for resp in lines.into_iter().chain(acks) { match resp.rtype { - GrblResponseType::Ok => self.handle_ok(resp.pending), - GrblResponseType::Error => self.handle_error(&resp.text), - GrblResponseType::Line => self.handle_line(&resp.text), + GrblResponseType::Ok => { + self.events.raw_rx("ok"); + self.handle_ok(resp.pending) + } + GrblResponseType::Error => { + self.events.raw_rx(&resp.text); + self.handle_error(&resp.text) + } + GrblResponseType::Line => { + self.events.raw_rx(&resp.text); + self.handle_line(&resp.text) + } } } // Wake parked senders only after the responses were @@ -1088,6 +1105,7 @@ impl SessionCore { let text = String::from_utf8_lossy(payload).to_string(); self.flow .track_command(payload.len(), None, text.trim_end()); + self.events.raw_tx(text.trim_end()); self.transport.write(payload).await?; Ok(()) } diff --git a/src/python/events.rs b/src/python/events.rs index f7a45c4..4d43e57 100644 --- a/src/python/events.rs +++ b/src/python/events.rs @@ -65,6 +65,7 @@ impl EventQueue { struct EventDrainer { queue: Arc, callback: Py, + progress: Option>, } #[pymethods] @@ -130,6 +131,12 @@ impl PyEvents { EventDrainer { queue: self.queue.clone(), callback: callback.clone().unbind(), + progress: self + .progress_callback + .lock() + .unwrap() + .as_ref() + .map(|cb| cb.clone_ref(py)), }, )?; loop_py @@ -172,6 +179,7 @@ impl PyEvents { EventDrainer { queue: self.queue.clone(), callback, + progress: None, }, )?; if let Err(err) = loop_py @@ -271,6 +279,20 @@ impl SessionEvents for PyEvents { fn command_done(&self, op_index: i64) { self.schedule_progress(op_index); } + + 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()) + }); + } + + 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()) + }); + } } impl EventDrainer { @@ -279,7 +301,9 @@ impl EventDrainer { fn take_and_deliver(&self, py: Python<'_>) { for (name, payload) in self.queue.take() { if name == "__progress__" { - PyEvents::deliver_progress(&self.callback, py, &payload); + 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)) { diff --git a/tests/freeze-dump.txt b/tests/freeze-dump.txt new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_grbl_raw_lines.py b/tests/test_grbl_raw_lines.py new file mode 100644 index 0000000..f4e0f8b --- /dev/null +++ b/tests/test_grbl_raw_lines.py @@ -0,0 +1,60 @@ +"""Raw RX/TX line events surfaced to the Python event callback.""" + +from conftest import make_session, start_device + + +class TestRawLines: + async def test_raw_tx_reports_interactive_command(self, mock, events): + emulator, device_task = await start_device(mock) + session = make_session(mock, events) + try: + await session.connect() + await events.wait_for( + "connection_status_changed", lambda p: p[0] == "CONNECTED" + ) + events.events.clear() + await session.execute_interactive_command("$$") + await events.wait_for("raw_tx", lambda line: line == "$$") + finally: + await session.disconnect() + device_task.cancel() + + 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) + assert any( + name == "raw_rx" and line.startswith("$") + for name, line 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(">") + ) + assert "Idle" in report or "Run" in report or "Hold" in report + + async def test_raw_rx_reports_welcome_message(self, mock, events): + emulator, device_task = await start_device(mock) + session = make_session(mock, events) + try: + await session.connect() + await events.wait_for( + "raw_rx", lambda line: line.startswith("Grbl ") + ) + finally: + await session.disconnect() + device_task.cancel() + + 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") + ] + assert "$I" in raw + assert any(line.startswith("[VER:") for line in raw) + assert "ok" in raw