feat: initial library implementation + example

Author: urish

.pre-commit-config.yaml

@@ -13,7 +13,8 @@ repos:
       rev: v1.16.1
       hooks:
           - id: mypy
-            additional_dependencies: [pydantic==2.8.0, typing-extensions, types-click]
+            additional_dependencies:
+                [pydantic==2.8.0, typing-extensions, types-click, types-requests]
 
     # Blacken-docs
     - repo: https://github.com/asottile/blacken-docs

README.md

@@ -7,15 +7,31 @@ Typed, asyncio-friendly Python SDK for the **Wokwi Simulation API**
 [![CI](https://github.com/wokwi/wokwi-python-client/actions/workflows/ci.yml/badge.svg?branch=main)](https://github.com/wokwi/wokwi-python-client/actions/workflows/ci.yml)
 [![License: MIT](https://img.shields.io/github/license/wokwi/wokwi-python-client)](LICENSE)
 
-> **TL;DR:** Run and control your Wokwi simulations from Python, synchronously **or** asynchronously, with first-class type hints and zero boilerplate.
+> **TL;DR:** Run and control your Wokwi simulations from Python with first-class type hints and zero boilerplate.
 
 ---
 
 ## Installation requirements
 
 Python ≥ 3.9
 
-An API token from https://wokwi.com/dashboard/ci
+An API token from [https://wokwi.com/dashboard/ci](https://wokwi.com/dashboard/ci).
+
+## Running the examples
+
+The basic example is in the [examples/hello_esp32/main.py](examples/hello_esp32/main.py) file. It shows how to:
+
+- Connect to the Wokwi Simulator
+- Upload a diagram and firmware files
+- Start a simulation
+- Monitor the serial output
+
+You can run the example with:
+
+```bash
+pip install -e .[dev]
+python -m examples.hello_esp32.main
+```
 
 ## License
 

examples/hello_esp32/.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/diagram.json

@@ -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/main.py

@@ -0,0 +1,64 @@
+# SPDX-License-Identifier: MIT
+# Copyright (C) 2025, CodeMagic LTD
+
+import asyncio
+import os
+from pathlib import Path
+
+import requests
+
+from wokwi_client import GET_TOKEN_URL, WokwiClient
+
+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",
+}
+
+
+async 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}."
+        )
+
+    client = WokwiClient(token)
+    print(f"Wokwi client library version: {client.version}")
+
+    hello = await client.connect()
+    print("Connected to Wokwi Simulator, server version:", hello["version"])
+
+    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)
+
+    # Upload the diagram and firmware files
+    await client.upload_file("diagram.json", EXAMPLE_DIR / "diagram.json")
+    await client.upload_file("hello_world.bin", EXAMPLE_DIR / "hello_world.bin")
+    await client.upload_file("hello_world.elf", EXAMPLE_DIR / "hello_world.elf")
+
+    # Start the simulation
+    await client.start_simulation(
+        firmware="hello_world.bin",
+        elf="hello_world.elf",
+    )
+
+    # Stream serial output for 10 seconds
+    serial_task = asyncio.create_task(client.serial_monitor_cat())
+    print("Simulation started, waiting for 10 seconds…")
+    await asyncio.sleep(10)
+    serial_task.cancel()
+
+    # Disconnect from the simulator
+    await client.disconnect()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -27,7 +27,7 @@ classifiers = [
     "Typing :: Typed",
     "License :: OSI Approved :: MIT License",
 ]
-dependencies = ["pydantic>=2.8"]
+dependencies = ["pydantic>=2.8", "websockets>=12,<13"]
 
 [project.urls]
 Documentation = "https://github.com/wokwi/wokwi-python-client#readme"
@@ -86,6 +86,7 @@ dependencies = [
     "pymdown-extensions",
     "mkdocs-material[extensions]",
     "mkdocstrings[python]",
+    "types-requests",
 ]
 
 [tool.hatch.envs.dev]

src/wokwi_client/__init__.py

@@ -1,3 +1,10 @@
 # SPDX-FileCopyrightText: 2025-present CodeMagic LTD
 #
 # SPDX-License-Identifier: MIT
+
+from .__version__ import get_version
+from .client import WokwiClient
+from .constants import GET_TOKEN_URL
+
+__version__ = get_version()
+__all__ = ["WokwiClient", "__version__", "GET_TOKEN_URL"]

src/wokwi_client/__version__.py

@@ -1,3 +1,7 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
 from importlib.metadata import PackageNotFoundError, version
 
 try:  # works for normal + editable installs

src/wokwi_client/client.py

@@ -0,0 +1,52 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+from pathlib import Path
+from typing import Any, Optional
+
+from .__version__ import get_version
+from .constants import DEFAULT_WS_URL
+from .file_ops import upload, upload_file
+from .protocol_types import ResponseMessage
+from .serial import monitor_lines
+from .simulation import pause, restart, resume, start
+from .transport import Transport
+
+
+class WokwiClient:
+    version: str
+
+    def __init__(self, token: str, server: Optional[str] = None):
+        self.version = get_version()
+        self._transport = Transport(token, server or DEFAULT_WS_URL)
+
+    async def connect(self) -> dict[str, Any]:
+        return await self._transport.connect()
+
+    async def disconnect(self) -> None:
+        await self._transport.close()
+
+    async def upload(self, name: str, content: bytes) -> ResponseMessage:
+        return await upload(self._transport, name, content)
+
+    async def upload_file(
+        self, filename: str, local_path: Optional[Path] = None
+    ) -> ResponseMessage:
+        return await upload_file(self._transport, filename, local_path)
+
+    async def start_simulation(self, **kwargs: Any) -> ResponseMessage:
+        return await start(self._transport, **kwargs)
+
+    async def pause_simulation(self) -> ResponseMessage:
+        return await pause(self._transport)
+
+    async def resume_simulation(self, pause_after: Optional[int] = None) -> ResponseMessage:
+        return await resume(self._transport, pause_after)
+
+    async def restart_simulation(self, pause: bool = False) -> ResponseMessage:
+        return await restart(self._transport, pause)
+
+    async def serial_monitor_cat(self) -> None:
+        async for line in monitor_lines(self._transport):
+            print(line.decode("utf-8"), end="", flush=True)

src/wokwi_client/constants.py

@@ -0,0 +1,13 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+DEFAULT_WS_URL = "wss://wokwi.com/api/ws/beta"
+GET_TOKEN_URL = "https://wokwi.com/dashboard/ci"
+
+MSG_TYPE_ERROR = "error"
+MSG_TYPE_RESPONSE = "response"
+MSG_TYPE_EVENT = "event"
+MSG_TYPE_COMMAND = "command"
+MSG_TYPE_HELLO = "hello"
+PROTOCOL_VERSION = 1

src/wokwi_client/exceptions.py

@@ -0,0 +1,12 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+
+class WokwiError(Exception): ...
+
+
+class ProtocolError(WokwiError): ...
+
+
+class ServerError(WokwiError): ...

src/wokwi_client/file_ops.py

@@ -0,0 +1,24 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+import base64
+from pathlib import Path
+from typing import Optional
+
+from .models import UploadParams
+from .protocol_types import ResponseMessage
+from .transport import Transport
+
+
+async def upload_file(
+    transport: Transport, filename: str, local_path: Optional[Path] = None
+) -> ResponseMessage:
+    path = Path(local_path or filename)
+    content = path.read_bytes()
+    return await upload(transport, filename, content)
+
+
+async def upload(transport: Transport, name: str, content: bytes) -> ResponseMessage:
+    params = UploadParams(name=name, binary=base64.b64encode(content).decode())
+    return await transport.request("file:upload", params.model_dump())

src/wokwi_client/models.py

@@ -0,0 +1,17 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+from pydantic import BaseModel, Field
+
+
+class UploadParams(BaseModel):
+    name: str
+    binary: str  # base64
+
+
+class SimulationParams(BaseModel):
+    firmware: str
+    elf: str
+    pause: bool = False
+    chips: list[str] = Field(default_factory=list)

src/wokwi_client/protocol_types.py

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+from typing import Any, TypedDict, Union
+
+
+class HelloMessage(TypedDict):
+    type: str  # "hello"
+    protocolVersion: int
+    appName: str
+    appVersion: str
+
+
+class CommandMessage(TypedDict, total=False):
+    type: str  # "command"
+    command: str
+    id: str
+    params: dict[str, Any]
+
+
+class ResponseMessage(TypedDict):
+    type: str  # "response"
+    command: str
+    id: str
+    result: dict[str, Any]
+    error: bool
+
+
+class ErrorResult(TypedDict):
+    code: int
+    message: str
+
+
+class EventMessage(TypedDict):
+    type: str  # "event"
+    event: str
+    payload: dict[str, Any]
+    nanos: float
+    paused: bool
+
+
+IncomingMessage = Union[HelloMessage, ResponseMessage, EventMessage]

src/wokwi_client/serial.py

@@ -0,0 +1,19 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+from collections.abc import AsyncGenerator
+from typing import cast
+
+from .protocol_types import EventMessage
+from .transport import Transport
+
+
+async def monitor_lines(transport: Transport) -> AsyncGenerator[bytes, None]:
+    await transport.request("serial-monitor:listen", {})
+    while True:
+        msg = await transport.recv()
+        if msg["type"] == "event":
+            event_msg = cast(EventMessage, msg)
+            if event_msg["event"] == "serial-monitor:data":
+                yield bytes(event_msg["payload"]["bytes"])

src/wokwi_client/simulation.py

@@ -0,0 +1,33 @@
+# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
+#
+# SPDX-License-Identifier: MIT
+
+from typing import Optional
+
+from .protocol_types import ResponseMessage
+from .transport import Transport
+
+
+async def start(
+    transport: Transport,
+    *,
+    firmware: str,
+    elf: str,
+    pause: bool = False,
+    chips: list[str] = [],
+) -> ResponseMessage:
+    return await transport.request(
+        "sim:start", {"firmware": firmware, "elf": elf, "pause": pause, "chips": chips}
+    )
+
+
+async def pause(transport: Transport) -> ResponseMessage:
+    return await transport.request("sim:pause", {})
+
+
+async def resume(transport: Transport, pause_after: Optional[int] = None) -> ResponseMessage:
+    return await transport.request("sim:resume", {"pauseAfter": pause_after})
+
+
+async def restart(transport: Transport, pause: bool = False) -> ResponseMessage:
+    return await transport.request("sim:restart", {"pause": pause})