refactor: replace WokwiCLI with Wokwi class and update related references

Author: JakubAndrysek

docs/apis/pytest-embedded-wokwi.rst

@@ -17,7 +17,7 @@
    :undoc-members:
    :show-inheritance:
 
-.. automodule:: pytest_embedded_wokwi.wokwi_cli
+.. automodule:: pytest_embedded_wokwi.wokwi
    :members:
    :undoc-members:
    :show-inheritance:

pytest-embedded-wokwi/pyproject.toml

@@ -7,6 +7,7 @@ name = "pytest-embedded-wokwi"
 authors = [
     {name = "Fu Hanxi", email = "fuhanxi@espressif.com"},
     {name = "Uri Shaked", email = "uri@wokwi.com"},
+    {name = "Jakub Andrysek", email = "jakub.andrysek@espressif.com"},
 ]
 readme = "README.md"
 license = {file = "LICENSE"}
@@ -34,6 +35,8 @@ requires-python = ">=3.7"
 dependencies = [
     "pytest-embedded~=1.17.0a2",
     "toml~=0.10.2",
+    # Temporary workaround for Wokwi client - will be redirected to the official repo
+    "wokwi-client @ git+https://github.com/JakubAndrysek/wokwi-python-client.git@sync-wokwi-client",
 ]
 
 [project.optional-dependencies]

pytest-embedded-wokwi/pytest_embedded_wokwi/__init__.py

@@ -3,11 +3,11 @@
 WOKWI_CLI_MINIMUM_VERSION = '0.10.1'
 
 from .dut import WokwiDut  # noqa
-from .wokwi_cli import WokwiCLI  # noqa
+from .wokwi import Wokwi  # noqa
 
 __all__ = [
     'WOKWI_CLI_MINIMUM_VERSION',
-    'WokwiCLI',
+    'Wokwi',
     'WokwiDut',
 ]
 

pytest-embedded-wokwi/pytest_embedded_wokwi/dut.py

@@ -2,7 +2,7 @@
 
 from pytest_embedded.dut import Dut
 
-from .wokwi_cli import WokwiCLI
+from .wokwi import Wokwi
 
 
 class WokwiDut(Dut):
@@ -12,7 +12,7 @@ class WokwiDut(Dut):
 
     def __init__(
         self,
-        wokwi: WokwiCLI,
+        wokwi: Wokwi,
         **kwargs,
     ) -> None:
         self.wokwi = wokwi

pytest-embedded-wokwi/pytest_embedded_wokwi/wokwi.py

@@ -0,0 +1,205 @@
+import json
+import logging
+import os
+import typing as t
+
+from packaging.version import Version
+from pytest_embedded.log import DuplicateStdoutPopen, MessageQueue
+from pytest_embedded.utils import Meta
+from wokwi_client import GET_TOKEN_URL, WokwiClientSync
+
+from pytest_embedded_wokwi import WOKWI_CLI_MINIMUM_VERSION
+
+from .idf import IDFFirmwareResolver
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from pytest_embedded_idf.app import IdfApp
+
+
+target_to_board = {
+    'esp32': 'board-esp32-devkit-c-v4',
+    'esp32c3': 'board-esp32-c3-devkitm-1',
+    'esp32c6': 'board-esp32-c6-devkitc-1',
+    'esp32h2': 'board-esp32-h2-devkitm-1',
+    'esp32p4': 'board-esp32-p4-function-ev',
+    'esp32s2': 'board-esp32-s2-devkitm-1',
+    'esp32s3': 'board-esp32-s3-devkitc-1',
+}
+
+
+class Wokwi(DuplicateStdoutPopen):
+    """Synchronous Wokwi integration that inherits from DuplicateStdoutPopen.
+
+    This class provides a synchronous interface to the Wokwi simulator while maintaining
+    compatibility with pytest-embedded's logging and message queue infrastructure.
+    """
+
+    SOURCE = 'Wokwi'
+    REDIRECT_CLS = None  # We'll handle output redirection manually
+
+    def __init__(
+        self,
+        msg_queue: MessageQueue,
+        firmware_resolver: IDFFirmwareResolver,
+        wokwi_diagram: t.Optional[str] = None,
+        app: t.Optional['IdfApp'] = None,
+        _wokwi_cli_path: t.Optional[str] = None,  # ignored for compatibility
+        _wokwi_timeout: t.Optional[int] = None,  # ignored for compatibility
+        _wokwi_scenario: t.Optional[str] = None,  # ignored for compatibility
+        meta: t.Optional[Meta] = None,
+        **kwargs,
+    ):
+        self.app = app
+        self.firmware_resolver = firmware_resolver
+
+        # Get Wokwi API token
+        token = os.getenv('WOKWI_CLI_TOKEN')
+        if not token:
+            raise SystemExit(f'Set WOKWI_CLI_TOKEN in your environment. You can get it from {GET_TOKEN_URL}.')
+
+        # Initialize synchronous Wokwi client
+        self.client = WokwiClientSync(token)
+
+        # Check version compatibility
+        if Version(self.client.version) < Version(WOKWI_CLI_MINIMUM_VERSION):
+            logging.warning(
+                'Wokwi client version %s < required %s (compatibility not guaranteed)',
+                self.client.version,
+                WOKWI_CLI_MINIMUM_VERSION,
+            )
+        logging.info('Wokwi client library version: %s', self.client.version)
+
+        # Prepare diagram file if not supplied
+        if wokwi_diagram is None:
+            self.create_diagram_json()
+            wokwi_diagram = os.path.join(self.app.app_path, 'diagram.json')
+
+        # Filter out Wokwi-specific kwargs that shouldn't be passed to subprocess.Popen
+        wokwi_specific_kwargs = {'wokwi_timeout', 'wokwi_scenario', 'wokwi_diagram', 'firmware_resolver', 'app'}
+        filtered_kwargs = {k: v for k, v in kwargs.items() if k not in wokwi_specific_kwargs}
+
+        # Initialize parent class
+        super().__init__(msg_queue=msg_queue, meta=meta, **filtered_kwargs)
+
+        # Connect and start simulation
+        try:
+            firmware_path = self.firmware_resolver.resolve_firmware(app)
+            self._setup_simulation(wokwi_diagram, firmware_path, app.elf_file)
+            self._start_serial_monitoring()
+        except Exception as e:
+            self.close()
+            raise e
+
+    def _setup_simulation(self, diagram: str, firmware_path: str, elf_path: str):
+        """Set up the Wokwi simulation."""
+        hello = self.client.connect()
+        logging.info('Connected to Wokwi Simulator, server version: %s', hello.get('version', 'unknown'))
+
+        # Upload files
+        self.client.upload_file('diagram.json', diagram)
+        self.client.upload_file('pytest.bin', firmware_path)
+        self.client.upload_file('pytest.elf', elf_path)
+
+        # Start simulation
+        self.client.start_simulation(firmware='pytest.bin', elf='pytest.elf')
+
+    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:
+                    decoded = data.decode('utf-8', errors='replace')
+                    self._fw.write(decoded)
+                    self._fw.flush()
+            except Exception as e:
+                logging.debug(f'Error writing to log file: {e}')
+
+            # Put in message queue for expect() functionality
+            try:
+                if hasattr(self, '_q') and self._q:
+                    self._q.put(data)
+            except Exception as e:
+                logging.debug(f'Error putting data in message queue: {e}')
+
+        # Start monitoring in background
+        self.client.monitor_serial(serial_callback)
+
+    def write(self, s: t.Union[str, bytes]) -> None:
+        """Write data to the Wokwi serial interface."""
+        try:
+            data = s if isinstance(s, bytes) else s.encode('utf-8')
+            self.client.write_serial(data)
+            logging.debug(f'{self.SOURCE} ->: {s}')
+        except Exception as e:
+            logging.error(f'Failed to write to Wokwi serial: {e}')
+
+    def close(self):
+        """Clean up resources."""
+        try:
+            if hasattr(self, 'client') and self.client:
+                self.client.disconnect()
+        except Exception as e:
+            logging.debug(f'Error during Wokwi cleanup: {e}')
+        finally:
+            super().close()
+
+    def __del__(self):
+        """Destructor to ensure cleanup when object is garbage collected."""
+        self.close()
+
+    def terminate(self):
+        """Terminate the Wokwi connection."""
+        self.close()
+        super().terminate()
+
+    def create_diagram_json(self):
+        """Create a diagram.json file for the simulation."""
+        app = self.app
+        target_board = target_to_board[app.target]
+
+        # Check for existing diagram.json file
+        diagram_json_path = os.path.join(app.app_path, 'diagram.json')
+        if os.path.exists(diagram_json_path):
+            with open(diagram_json_path) as f:
+                json_data = json.load(f)
+            if not any(part['type'] == target_board for part in json_data['parts']):
+                logging.warning(
+                    f'diagram.json exists, no part with type "{target_board}" found. '
+                    + 'You may need to update the diagram.json file manually to match the target board.'
+                )
+            return
+
+        # Create default diagram
+        if app.target == 'esp32p4':
+            rx_pin = '38'
+            tx_pin = '37'
+        else:
+            rx_pin = 'RX'
+            tx_pin = 'TX'
+
+        diagram = {
+            'version': 1,
+            'author': 'Uri Shaked',
+            'editor': 'wokwi',
+            'parts': [{'type': target_board, 'id': 'esp'}],
+            'connections': [
+                ['esp:' + tx_pin, '$serialMonitor:RX', ''],
+                ['esp:' + rx_pin, '$serialMonitor:TX', ''],
+            ],
+        }
+
+        with open(diagram_json_path, 'w') as f:
+            json.dump(diagram, f, indent=2)
+
+    def _hard_reset(self):
+        """Fake hard_reset to maintain API consistency."""
+        raise NotImplementedError('Hard reset not supported in Wokwi simulation')