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