Emit raw RX/TX line events from the GRBL session

src/grbl/session.rs

@@ -44,6 +44,14 @@ pub trait SessionEvents: Send + Sync {
     fn wcs_updated(&self, offsets: &HashMap<String, (f64, f64, f64)>);
     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(())
     }

src/python/events.rs

@@ -65,6 +65,7 @@ impl EventQueue {
 struct EventDrainer {
     queue: Arc<EventQueue>,
     callback: Py<PyAny>,
+    progress: Option<Py<PyAny>>,
 }
 
 #[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)) {

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