From dab255f8a4082fb32b84a229263349bc9d23897e Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 10:48:34 +0200 Subject: [PATCH 1/4] feat: add framebuffer command helpers for reading and saving PNGs --- src/wokwi_client/client.py | 27 ++++++++++ src/wokwi_client/framebuffer.py | 93 +++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+) create mode 100644 src/wokwi_client/framebuffer.py diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index a1feed6..87fc315 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -5,6 +5,13 @@ from pathlib import Path from typing import Any, Optional, Union +from wokwi_client.framebuffer import ( + compare_framebuffer_png, + framebuffer_png_bytes, + framebuffer_read, + save_framebuffer_png, +) + from .__version__ import get_version from .constants import DEFAULT_WS_URL from .control import set_control @@ -233,3 +240,23 @@ async def set_control( value: Control value to set (float). """ return await set_control(self._transport, part=part, control=control, value=value) + + async def framebuffer_read(self, id: str) -> ResponseMessage: + """Read the current framebuffer for the given device id.""" + return await framebuffer_read(self._transport, id=id) + + async def framebuffer_png_bytes(self, id: str) -> bytes: + """Return the current framebuffer as PNG bytes.""" + return await framebuffer_png_bytes(self._transport, id=id) + + async def save_framebuffer_png(self, id: str, path: Path, overwrite: bool = True) -> Path: + """Save the current framebuffer as a PNG file.""" + return await save_framebuffer_png(self._transport, id=id, path=path, overwrite=overwrite) + + async def compare_framebuffer_png( + self, id: str, reference: Path, save_mismatch: Optional[Path] = None + ) -> bool: + """Compare the current framebuffer with a reference PNG file.""" + return await compare_framebuffer_png( + self._transport, id=id, reference=reference, save_mismatch=save_mismatch + ) diff --git a/src/wokwi_client/framebuffer.py b/src/wokwi_client/framebuffer.py new file mode 100644 index 0000000..ef3ae53 --- /dev/null +++ b/src/wokwi_client/framebuffer.py @@ -0,0 +1,93 @@ +"""Framebuffer command helpers. + +Provides utilities to interact with devices exposing a framebuffer (e.g. LCD +modules) via the `framebuffer:read` command. + +Exposed helpers: +* framebuffer_read -> raw response (contains base64 PNG at result.png) +* framebuffer_png_bytes -> decoded PNG bytes +* save_framebuffer_png -> save PNG to disk +* compare_framebuffer_png -> compare current framebuffer against reference +""" + +# SPDX-FileCopyrightText: 2025-present CodeMagic LTD +# +# SPDX-License-Identifier: MIT + +from __future__ import annotations + +import base64 +from pathlib import Path + +from .exceptions import WokwiError +from .protocol_types import ResponseMessage +from .transport import Transport + +__all__ = [ + "framebuffer_read", + "framebuffer_png_bytes", + "save_framebuffer_png", + "compare_framebuffer_png", +] + + +async def framebuffer_read(transport: Transport, *, id: str) -> ResponseMessage: + """Issue `framebuffer:read` for the given device id and return raw response.""" + return await transport.request("framebuffer:read", {"id": id}) + + +def _extract_png_b64(resp: ResponseMessage) -> str: + result = resp.get("result", {}) + png_b64 = result.get("png") + if not isinstance(png_b64, str): # pragma: no cover - defensive + raise WokwiError("Malformed framebuffer:read response: missing 'png' base64 string") + return png_b64 + + +async def framebuffer_png_bytes(transport: Transport, *, id: str) -> bytes: + """Return decoded PNG bytes for the framebuffer of device `id`.""" + resp = await framebuffer_read(transport, id=id) + return base64.b64decode(_extract_png_b64(resp)) + + +async def save_framebuffer_png( + transport: Transport, *, id: str, path: Path, overwrite: bool = True +) -> Path: + """Save the framebuffer PNG to `path` and return the path. + + Args: + transport: Active transport. + id: Device id (e.g. "lcd1"). + path: Destination file path. + overwrite: Overwrite existing file (default True). If False and file + exists, raises WokwiError. + """ + if path.exists() and not overwrite: + raise WokwiError(f"File already exists and overwrite=False: {path}") + data = await framebuffer_png_bytes(transport, id=id) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "wb") as f: + f.write(data) + return path + + +async def compare_framebuffer_png( + transport: Transport, *, id: str, reference: Path, save_mismatch: Path | None = None +) -> bool: + """Compare the current framebuffer PNG with a reference file. + + Performs a byte-for-byte comparison. If different and `save_mismatch` is + provided, writes the current framebuffer PNG there. + + Returns True if identical, False otherwise. + """ + if not reference.exists(): + raise WokwiError(f"Reference image does not exist: {reference}") + current = await framebuffer_png_bytes(transport, id=id) + ref_bytes = reference.read_bytes() + if current == ref_bytes: + return True + if save_mismatch: + save_mismatch.parent.mkdir(parents=True, exist_ok=True) + save_mismatch.write_bytes(current) + return False From 321b1036bb32f6c65368034653109d68c409c7d8 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 11:00:26 +0200 Subject: [PATCH 2/4] feat: add download and download_file methods to WokwiClient and file_ops --- src/wokwi_client/client.py | 24 +++++++++++++++++++++++- src/wokwi_client/file_ops.py | 13 +++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index 87fc315..78ed94c 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -16,7 +16,7 @@ from .constants import DEFAULT_WS_URL from .control import set_control from .event_queue import EventQueue -from .file_ops import upload, upload_file +from .file_ops import download, download_file, upload, upload_file from .pins import pin_listen, pin_read from .protocol_types import EventMessage, ResponseMessage from .serial import monitor_lines, write_serial @@ -93,6 +93,28 @@ async def upload_file( """ return await upload_file(self._transport, filename, local_path) + async def download(self, name: str) -> ResponseMessage: + """ + Download a file from the simulator. + + Args: + name: The name of the file to download. + + Returns: + The response message from the server. + """ + return await download(self._transport, name) + + async def download_file(self, name: str, local_path: Optional[Path] = None) -> None: + """ + Download a file from the simulator and save it to a local path. + + Args: + name: The name of the file to download. + local_path: The local path to save the downloaded file. If not provided, uses the name as the path. + """ + await download_file(self._transport, name, local_path) + async def start_simulation( self, firmware: str, diff --git a/src/wokwi_client/file_ops.py b/src/wokwi_client/file_ops.py index 032bfd9..0907509 100644 --- a/src/wokwi_client/file_ops.py +++ b/src/wokwi_client/file_ops.py @@ -22,3 +22,16 @@ async def upload_file( 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()) + + +async def download(transport: Transport, name: str) -> ResponseMessage: + return await transport.request("file:download", {"name": name}) + + +async def download_file(transport: Transport, name: str, local_path: Optional[Path] = None) -> None: + if local_path is None: + local_path = Path(name) + + result = await download(transport, name) + with open(local_path, "wb") as f: + f.write(base64.b64decode(result["result"]["binary"])) From 8a158b9deefe08421ebc4550b211be1f06e9522d Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 11:10:44 +0200 Subject: [PATCH 3/4] feat: add gpio_list method to retrieve all GPIO pins in WokwiClient --- src/wokwi_client/client.py | 6 +++++- src/wokwi_client/pins.py | 10 ++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index 78ed94c..d313572 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -17,7 +17,7 @@ from .control import set_control from .event_queue import EventQueue from .file_ops import download, download_file, upload, upload_file -from .pins import pin_listen, pin_read +from .pins import gpio_list, pin_listen, pin_read from .protocol_types import EventMessage, ResponseMessage from .serial import monitor_lines, write_serial from .simulation import pause, restart, resume, start @@ -251,6 +251,10 @@ async def listen_pin(self, part: str, pin: str, listen: bool = True) -> Response """ return await pin_listen(self._transport, part=part, pin=pin, listen=listen) + async def gpio_list(self) -> ResponseMessage: + """Get a list of all GPIO pins available in the simulation.""" + return await gpio_list(self._transport) + async def set_control( self, part: str, control: str, value: Union[int, bool, float] ) -> ResponseMessage: diff --git a/src/wokwi_client/pins.py b/src/wokwi_client/pins.py index 82879ad..87d96a9 100644 --- a/src/wokwi_client/pins.py +++ b/src/wokwi_client/pins.py @@ -43,3 +43,13 @@ async def pin_listen( """ return await transport.request("pin:listen", {"part": part, "pin": pin, "listen": listen}) + + +async def gpio_list(transport: Transport) -> ResponseMessage: + """List all GPIO pins and their current states. + + Args: + transport: The active Transport instance. + """ + + return await transport.request("gpio:list", {}) From 296b502faa38f778598dae2445263308b7a54802 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 14:53:54 +0200 Subject: [PATCH 4/4] feat: update gpio_list method to return a list of GPIO pins and handle errors --- src/wokwi_client/client.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index d313572..3a2b05c 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -3,8 +3,9 @@ # SPDX-License-Identifier: MIT from pathlib import Path -from typing import Any, Optional, Union +from typing import Any, Optional, Union, cast +from wokwi_client.exceptions import ProtocolError from wokwi_client.framebuffer import ( compare_framebuffer_png, framebuffer_png_bytes, @@ -251,9 +252,17 @@ async def listen_pin(self, part: str, pin: str, listen: bool = True) -> Response """ return await pin_listen(self._transport, part=part, pin=pin, listen=listen) - async def gpio_list(self) -> ResponseMessage: - """Get a list of all GPIO pins available in the simulation.""" - return await gpio_list(self._transport) + async def gpio_list(self) -> list[str]: + """Get a list of all GPIO pins available in the simulation. + + Returns: + list[str]: Example: ["esp32:GPIO0", "esp32:GPIO1", ...] + """ + resp = await gpio_list(self._transport) + pins_val: Any = resp.get("result", {}).get("pins") + if not isinstance(pins_val, list) or not all(isinstance(p, str) for p in pins_val): + raise ProtocolError("Malformed gpio:list response: expected result.pins: list[str]") + return cast(list[str], pins_val) async def set_control( self, part: str, control: str, value: Union[int, bool, float]