feat: Add option to capture payloads

docs/concepts/key-concepts.rst

@@ -149,3 +149,41 @@ The ``--qemu-cli-args`` option applies to the first DUT (with the ``qemu`` servi
 ``pytest-embedded`` prints all DUT output with a timestamp. To remove the timestamp, run pytest with the ``--with-timestamp n`` option.
 
 By default, ``pytest`` swallows ``stdout``. To see the live output, run pytest with the ``-s`` option.
+
+*************
+ Echo Muting
+*************
+
+The listener process that echoes serial output to ``stdout`` supports two complementary muting mechanisms. When muted, output is still written to the log file — only the terminal echo is suppressed.
+
+Automatic Muting via ``mute_patterns``
+======================================
+
+Override the ``mute_patterns`` fixture in your ``conftest.py`` to provide ``(start, end)`` string pairs. When the listener sees *start* in the serial stream it suppresses echo until *end* is seen:
+
+.. code:: python
+
+   @pytest.fixture
+   def mute_patterns():
+       return [("<<<DUMP_START>>>", "<<<DUMP_END>>>")]
+
+This is the recommended approach because it operates inside the listener process itself, so there is no race condition between the DUT producing output and the test code reacting to it.
+
+Manual Muting via ``Dut``
+=========================
+
+For ad-hoc suppression you can call :func:`~pytest_embedded.dut.Dut.mute_echo` / :func:`~pytest_embedded.dut.Dut.unmute_echo`, or use the :func:`~pytest_embedded.dut.Dut.muted_echo` context manager:
+
+.. code:: python
+
+   def test_verbose_section(dut):
+       with dut.muted_echo():
+           dut.expect("noisy output")
+
+*****************
+ Payload Capture
+*****************
+
+:func:`~pytest_embedded.dut.Dut.capture_payload` and :func:`~pytest_embedded.dut.Dut.capture_payload_to_file` let you extract delimited data blocks from the serial output — for example Base64-encoded binary dumps, coverage data, or diagnostic payloads.
+
+Both methods pair naturally with ``mute_patterns``: register the same start/end markers so the raw dump never clutters the terminal, then call ``capture_payload`` to retrieve the data programmatically.

docs/usages/payload_capture.rst

@@ -0,0 +1,121 @@
+###############################
+ Payload Capture & Echo Muting
+###############################
+
+Embedded devices sometimes emit large blocks of data over the serial port — Base64-encoded binary blobs, coverage dumps, diagnostic payloads, etc. ``pytest-embedded`` provides tools to **capture** these blocks programmatically and **mute** the terminal echo so the console stays readable.
+
+***********************
+ Automatic Echo Muting
+***********************
+
+The listener process that echoes serial output to ``stdout`` supports pattern-based muting. Override the ``mute_patterns`` fixture in your project's ``conftest.py``:
+
+.. code:: python
+
+   # conftest.py
+   import pytest
+
+   @pytest.fixture
+   def mute_patterns():
+       return [
+           ("<<<PAYLOAD_START>>>", "<<<PAYLOAD_END>>>"),
+       ]
+
+Each ``(start, end)`` pair defines a muting region. When the listener detects *start* in the serial stream, it suppresses ``stdout`` echo until *end* is seen. Log-file writing is **never** affected — all data is always recorded to the pexpect log file.
+
+Multiple patterns can be registered simultaneously:
+
+.. code:: python
+
+   @pytest.fixture
+   def mute_patterns():
+       return [
+           ("<<<GCOV_DUMP_START>>>", "<<<GCOV_DUMP_END>>>"),
+           ("<<<DIAG_START>>>", "<<<DIAG_END>>>"),
+       ]
+
+.. note::
+
+   ``mute_patterns`` operates inside the listener process itself, so there is no race condition between the device producing output and the test code reacting to it.
+
+********************
+ Manual Echo Muting
+********************
+
+For sections that are not delimited by fixed markers, use the :func:`~pytest_embedded.dut.Dut.muted_echo` context manager:
+
+.. code:: python
+
+   def test_noisy_section(dut):
+       with dut.muted_echo():
+           dut.expect("some verbose output")
+           # stdout is silent, but the log file still records everything
+
+You can also call :func:`~pytest_embedded.dut.Dut.mute_echo` and :func:`~pytest_embedded.dut.Dut.unmute_echo` directly for fine-grained control.
+
+********************
+ Capturing Payloads
+********************
+
+:func:`~pytest_embedded.dut.Dut.capture_payload` waits for a *start* marker, collects all bytes until the *end* marker, and returns them as raw ``bytes``:
+
+.. code:: python
+
+   def test_extract_data(dut):
+       data = dut.capture_payload(
+           start="<<<DATA_START>>>",
+           end="<<<DATA_END>>>",
+           start_timeout=10,
+           timeout=60,
+       )
+       assert data is not None
+       # process `data` ...
+
+If you prefer to persist the captured data to a file (for example, to accumulate data from multiple reboots), use :func:`~pytest_embedded.dut.Dut.capture_payload_to_file`:
+
+.. code:: python
+
+   def test_multi_boot_capture(dut):
+       for boot in range(3):
+           dut.write("reboot\n")
+           ok = dut.capture_payload_to_file(
+               start="<<<DUMP_START>>>",
+               end="<<<DUMP_END>>>",
+               filepath="/tmp/dump.txt",
+               append=True,          # accumulate across boots
+               include_markers=True,  # wrap each block with markers
+           )
+           assert ok
+
+*************************
+ Combining Both Features
+*************************
+
+Registering start/end markers in ``mute_patterns`` **and** using them in ``capture_payload`` is the recommended pattern — the listener mutes the echo automatically while the test code captures the data:
+
+.. code:: python
+
+   # conftest.py
+   import pytest
+
+   DUMP_START = "<<<GCOV_DUMP_START>>>"
+   DUMP_END   = "<<<GCOV_DUMP_END>>>"
+
+   @pytest.fixture
+   def mute_patterns():
+       return [(DUMP_START, DUMP_END)]
+
+.. code:: python
+
+   # test_coverage.py
+   from conftest import DUMP_START, DUMP_END
+
+   def test_gcov(dut):
+       ok = dut.capture_payload_to_file(
+           start=DUMP_START,
+           end=DUMP_END,
+           filepath="gcov_raw.txt",
+       )
+       assert ok
+
+The console stays clean, the log file has everything, and ``gcov_raw.txt`` contains just the payload.

pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py

@@ -225,13 +225,6 @@ def _start_serial_monitoring(self):
         """Start monitoring serial output and forward to stdout and message queue."""
 
         def serial_callback(data: bytes):
-            # Write to stdout for live monitoring
-            try:
-                decoded = data.decode('utf-8', errors='replace')
-                print(decoded, end='', flush=True)
-            except Exception as e:
-                logging.debug(f'Error writing to stdout: {e}')
-
             # Write to log file if available
             try:
                 if hasattr(self, '_fw') and self._fw and not self._fw.closed:
@@ -241,7 +234,9 @@ def serial_callback(data: bytes):
             except Exception as e:
                 logging.debug(f'Error writing to log file: {e}')
 
-            # Put in message queue for expect() functionality
+            # Put in message queue for expect() functionality.
+            # The listener process handles timestamped stdout echo and
+            # respects mute_patterns / mute_event, so we don't print here.
             try:
                 if hasattr(self, '_q') and self._q:
                     self._q.put(data)

pytest-embedded/pytest_embedded/dut.py

@@ -1,5 +1,7 @@
+import contextlib
 import functools
 import logging
+import os
 import os.path
 import re
 from collections.abc import Callable
@@ -33,10 +35,12 @@ def __init__(
         pexpect_logfile: str,
         test_case_name: str,
         meta: Meta | None = None,
+        _mute_event=None,
         **kwargs,
     ) -> None:
         self._q = msg_queue
         self._meta = meta
+        self._mute_event = _mute_event
 
         self.pexpect_proc = pexpect_proc
         self.app = app
@@ -59,6 +63,139 @@ def close(self) -> None:
             self.testsuite.dump(junit_report)
             logging.info(f'Created unity output junit report: {junit_report}')
 
+    def mute_echo(self) -> None:
+        """Suppress stdout echo from the listener process. Log file writing is unaffected."""
+        if self._mute_event is not None:
+            self._mute_event.set()
+
+    def unmute_echo(self) -> None:
+        """Resume stdout echo from the listener process."""
+        if self._mute_event is not None:
+            self._mute_event.clear()
+
+    @contextlib.contextmanager
+    def muted_echo(self):
+        """Context manager that suppresses stdout echo while still logging to file."""
+        self.mute_echo()
+        try:
+            yield
+        finally:
+            self.unmute_echo()
+
+    def capture_payload(
+        self,
+        start: str,
+        end: str,
+        start_timeout: float = 10,
+        timeout: float = 60,
+        mute: bool = False,
+    ) -> bytes | None:
+        """Capture a delimited payload from the serial output.
+
+        Waits for the *start* marker, then reads until the *end* marker and
+        returns the raw bytes **between** the two markers.
+
+        Stdout echo suppression is best handled automatically by
+        registering the markers via the ``mute_patterns`` fixture (see
+        :func:`pytest_embedded.plugin.mute_patterns`).  When *mute* is
+        ``True`` the event-based manual mute is also engaged as a
+        fallback.
+
+        Args:
+            start: exact string that marks the beginning of the payload.
+            end: exact string that marks the end of the payload.
+            start_timeout: seconds to wait for *start* before giving up.
+            timeout: seconds to wait for *end* after *start* has been seen.
+            mute: additionally engage the manual mute event. (default: False)
+
+        Returns:
+            The captured bytes between the markers, or ``None`` if the start
+            marker was not seen or the end marker timed out.
+        """
+        ctx = self.muted_echo() if mute else contextlib.nullcontext()
+        with ctx:
+            try:
+                self.pexpect_proc.expect(start, timeout=start_timeout)
+            except (pexpect.EOF, pexpect.TIMEOUT):
+                logging.debug('capture_payload: start marker %r not found within %ss', start, start_timeout)
+                return None
+            except Exception:
+                logging.debug('capture_payload: unexpected error waiting for start marker', exc_info=True)
+                return None
+
+            try:
+                self.pexpect_proc.expect(end, timeout=timeout)
+                return self.pexpect_proc.before
+            except pexpect.TIMEOUT:
+                logging.warning('capture_payload: end marker %r not found within %ss', end, timeout)
+                return None
+            except Exception:
+                logging.debug('capture_payload: unexpected error waiting for end marker', exc_info=True)
+                return None
+
+    def capture_payload_to_file(
+        self,
+        start: str,
+        end: str,
+        filepath: str,
+        start_timeout: float = 10,
+        timeout: float = 60,
+        mute: bool = True,
+        append: bool = True,
+        include_markers: bool = True,
+    ) -> bool:
+        """Capture a delimited payload and write it to a file.
+
+        This is a convenience wrapper around :meth:`capture_payload` that
+        persists the result to *filepath*.
+
+        Args:
+            start: exact string that marks the beginning of the payload.
+            end: exact string that marks the end of the payload.
+            filepath: destination file path (parent directories are created
+                automatically).
+            start_timeout: seconds to wait for *start* before giving up.
+            timeout: seconds to wait for *end* after *start* has been seen.
+            mute: suppress stdout echo while capturing. (default: True)
+            append: open the file in append mode so that successive calls
+                accumulate data (e.g. one call per reboot). (default: True)
+            include_markers: wrap the written data with ``start`` / ``end``
+                lines so downstream parsers can re-identify each block.
+                (default: True)
+
+        Returns:
+            ``True`` if the payload was captured and written, ``False``
+            otherwise.
+        """
+        data = self.capture_payload(
+            start=start,
+            end=end,
+            start_timeout=start_timeout,
+            timeout=timeout,
+            mute=mute,
+        )
+        if data is None:
+            return False
+
+        if isinstance(data, bytes):
+            data = data.decode('utf-8', errors='replace')
+
+        try:
+            os.makedirs(os.path.dirname(filepath) or '.', exist_ok=True)
+            mode = 'a' if append else 'w'
+            with open(filepath, mode) as fh:
+                if include_markers:
+                    fh.write(start + '\n')
+                fh.write(data)
+                if include_markers:
+                    fh.write(end + '\n')
+            logging.info('capture_payload_to_file: payload written to %s', filepath)
+        except Exception:
+            logging.debug('capture_payload_to_file: failed to write %s', filepath, exc_info=True)
+            return False
+
+        return True
+
     def write(self, s: AnyStr) -> None:
         """
         Write to the `MessageQueue` instance