feat: add synchronous Wokwi client and corresponding tests

Author: JakubAndrysek

.pre-commit-config.yaml

@@ -13,6 +13,7 @@ repos:
       rev: v1.16.1
       hooks:
           - id: mypy
+            exclude: ^src/wokwi_client/client_sync\.pyi$
             additional_dependencies:
                 [pydantic==2.8.0, typing-extensions, types-click, types-requests]
 

examples/hello_esp32/main.py

@@ -53,6 +53,12 @@ async def main() -> None:
 
     # Stream serial output for a few seconds
     serial_task = asyncio.create_task(client.serial_monitor_cat())
+
+    # Alternative lambda version
+    # serial_task = client.serial_monitor(
+    #     lambda line: print(line.decode("utf-8", errors="replace"), end="", flush=True)
+    # )
+
     print(f"Simulation started, waiting for {SLEEP_TIME} seconds…")
     await client.wait_until_simulation_time(SLEEP_TIME)
     serial_task.cancel()

examples/hello_esp32_sync/.gitignore

@@ -0,0 +1,3 @@
+# Ignore the firmware files, as they are downloaded from the internet
+hello_world.bin
+hello_world.elf

examples/hello_esp32_sync/__init__.py

@@ -0,0 +1,22 @@
+{
+  "version": 1,
+  "author": "Uri Shaked",
+  "editor": "wokwi",
+  "parts": [
+    {
+      "type": "wokwi-esp32-devkit-v1",
+      "id": "esp",
+      "top": 0,
+      "left": 0,
+      "attrs": { "fullBoot": "1" }
+    }
+  ],
+  "connections": [
+    ["esp:TX0", "$serialMonitor:RX", "", []],
+    ["esp:RX0", "$serialMonitor:TX", "", []]
+  ],
+  "serialMonitor": {
+    "display": "terminal"
+  },
+  "dependencies": {}
+}

examples/hello_esp32_sync/diagram.json

@@ -0,0 +1,66 @@
+# SPDX-License-Identifier: MIT
+# Copyright (C) 2025, CodeMagic LTD
+
+import os
+from pathlib import Path
+
+import requests
+
+from wokwi_client import GET_TOKEN_URL, WokwiClientSync
+
+EXAMPLE_DIR = Path(__file__).parent
+HELLO_WORLD_URL = "https://github.com/wokwi/esp-idf-hello-world/raw/refs/heads/main/bin"
+FIRMWARE_FILES = {
+    "hello_world.bin": f"{HELLO_WORLD_URL}/hello_world.bin",
+    "hello_world.elf": f"{HELLO_WORLD_URL}/hello_world.elf",
+}
+SLEEP_TIME = int(os.getenv("WOKWI_SLEEP_TIME", "10"))
+
+
+def main() -> None:
+    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}."
+        )
+
+    for filename, url in FIRMWARE_FILES.items():
+        if (EXAMPLE_DIR / filename).exists():
+            continue
+        print(f"Downloading {filename} from {url}")
+        response = requests.get(url)
+        response.raise_for_status()
+        with open(EXAMPLE_DIR / filename, "wb") as f:
+            f.write(response.content)
+
+    client = WokwiClientSync(token)
+    print(f"Wokwi client library version: {client.version}")
+
+    hello = client.connect()
+    print("Connected to Wokwi Simulator, server version:", hello["version"])
+
+    # Upload the diagram and firmware files
+    client.upload_file("diagram.json", EXAMPLE_DIR / "diagram.json")
+    client.upload_file("hello_world.bin", EXAMPLE_DIR / "hello_world.bin")
+    client.upload_file("hello_world.elf", EXAMPLE_DIR / "hello_world.elf")
+
+    # Start the simulation
+    client.start_simulation(
+        firmware="hello_world.bin",
+        elf="hello_world.elf",
+    )
+
+    # Stream serial output for a few seconds (non-blocking)
+    client.serial_monitor_cat()
+    # Alternative lambda version
+    # client.serial_monitor(lambda line: print(line.decode("utf-8", errors="replace"), end="", flush=True))
+
+    print(f"Simulation started, waiting for {SLEEP_TIME} seconds…")
+    client.wait_until_simulation_time(SLEEP_TIME)
+
+    # Disconnect from the simulator
+    client.disconnect()
+
+
+if __name__ == "__main__":
+    main()

examples/hello_esp32_sync/main.py

@@ -12,7 +12,8 @@
 
 from .__version__ import get_version
 from .client import WokwiClient
+from .client_sync import WokwiClientSync
 from .constants import GET_TOKEN_URL
 
 __version__ = get_version()
-__all__ = ["WokwiClient", "__version__", "GET_TOKEN_URL"]
+__all__ = ["WokwiClient", "WokwiClientSync", "__version__", "GET_TOKEN_URL"]

src/wokwi_client/__init__.py

@@ -2,9 +2,11 @@
 #
 # SPDX-License-Identifier: MIT
 
+import asyncio
 import base64
+import inspect
 from pathlib import Path
-from typing import Any, Optional, Union, cast
+from typing import Any, Callable, Optional, Union, cast
 
 from .__version__ import get_version
 from .constants import DEFAULT_WS_URL
@@ -48,6 +50,7 @@ def __init__(self, token: str, server: Optional[str] = None):
         self.last_pause_nanos = 0
         self._transport.add_event_listener("sim:pause", self._on_pause)
         self._pause_queue = EventQueue(self._transport, "sim:pause")
+        self._serial_monitor_tasks: set[asyncio.Task[None]] = set()
 
     async def connect(self) -> dict[str, Any]:
         """
@@ -61,7 +64,10 @@ async def connect(self) -> dict[str, Any]:
     async def disconnect(self) -> None:
         """
         Disconnect from the Wokwi simulator server.
+
+        This also stops all active serial monitors.
         """
+        self.stop_serial_monitors()
         await self._transport.close()
 
     async def upload(self, name: str, content: bytes) -> None:
@@ -188,6 +194,49 @@ async def restart_simulation(self, pause: bool = False) -> None:
         """
         await restart(self._transport, pause)
 
+    def serial_monitor(self, callback: Callable[[bytes], Any]) -> asyncio.Task[None]:
+        """
+        Start monitoring the serial output in the background and invoke `callback` for each line.
+
+        This method **does not block**: it creates and returns an asyncio.Task that runs until the
+        transport is closed or the task is cancelled. The callback may be synchronous or async.
+
+        Example:
+            task = client.serial_monitor(lambda line: print(line.decode(), end=""))
+            ... do other async work ...
+            task.cancel()
+        """
+
+        async def _runner() -> None:
+            try:
+                async for line in monitor_lines(self._transport):
+                    try:
+                        result = callback(line)
+                        if inspect.isawaitable(result):
+                            await result
+                    except Exception:
+                        # Swallow callback exceptions to keep the monitor alive.
+                        # Users can add their own error handling inside the callback.
+                        pass
+            finally:
+                # Clean up task from the set when it completes
+                self._serial_monitor_tasks.discard(task)
+
+        task = asyncio.create_task(_runner(), name="wokwi-serial-monitor")
+        self._serial_monitor_tasks.add(task)
+        return task
+
+    def stop_serial_monitors(self) -> None:
+        """
+        Stop all active serial monitor tasks.
+
+        This method cancels all tasks created by the serial_monitor method.
+        After calling this method, all active serial monitors will stop receiving data.
+        """
+        for task in self._serial_monitor_tasks.copy():
+            task.cancel()
+        self._serial_monitor_tasks.clear()
+
     async def serial_monitor_cat(self, decode_utf8: bool = True, errors: str = "replace") -> None:
         """
         Print serial monitor output to stdout as it is received from the simulation.