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 f6cc44e1c7f137e08944b06483fd85c158c919ab Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 13:45:47 +0200 Subject: [PATCH 3/4] feat: update download methods to return file content as bytes --- src/wokwi_client/client.py | 17 ++++++++++++----- src/wokwi_client/file_ops.py | 9 --------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index 78ed94c..a3b03f0 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: MIT +import base64 from pathlib import Path from typing import Any, Optional, Union @@ -16,7 +17,7 @@ from .constants import DEFAULT_WS_URL from .control import set_control from .event_queue import EventQueue -from .file_ops import download, download_file, upload, upload_file +from .file_ops import download, upload, upload_file from .pins import pin_listen, pin_read from .protocol_types import EventMessage, ResponseMessage from .serial import monitor_lines, write_serial @@ -93,7 +94,7 @@ async def upload_file( """ return await upload_file(self._transport, filename, local_path) - async def download(self, name: str) -> ResponseMessage: + async def download(self, name: str) -> bytes: """ Download a file from the simulator. @@ -101,9 +102,10 @@ async def download(self, name: str) -> ResponseMessage: name: The name of the file to download. Returns: - The response message from the server. + The downloaded file content as bytes. """ - return await download(self._transport, name) + result = await download(self._transport, name) + return base64.b64decode(result["result"]["binary"]) async def download_file(self, name: str, local_path: Optional[Path] = None) -> None: """ @@ -113,7 +115,12 @@ async def download_file(self, name: str, local_path: Optional[Path] = None) -> N 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) + if local_path is None: + local_path = Path(name) + + result = await self.download(name) + with open(local_path, "wb") as f: + f.write(result) async def start_simulation( self, diff --git a/src/wokwi_client/file_ops.py b/src/wokwi_client/file_ops.py index 0907509..f8e1def 100644 --- a/src/wokwi_client/file_ops.py +++ b/src/wokwi_client/file_ops.py @@ -26,12 +26,3 @@ async def upload(transport: Transport, name: str, content: bytes) -> ResponseMes 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 4016c5fd1fd6d814a13fd37b6ddedebcac20c897 Mon Sep 17 00:00:00 2001 From: Jakub Andrysek Date: Wed, 20 Aug 2025 15:23:25 +0200 Subject: [PATCH 4/4] feat: revert download and download_file methods to WokwiClient for file retrieval --- src/wokwi_client/client.py | 31 ++++++++++++++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/src/wokwi_client/client.py b/src/wokwi_client/client.py index 25235cf..e963a5a 100644 --- a/src/wokwi_client/client.py +++ b/src/wokwi_client/client.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: MIT +import base64 from pathlib import Path from typing import Any, Optional, Union @@ -14,7 +15,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, upload, upload_file from .pins import pin_listen, pin_read from .protocol_types import EventMessage, ResponseMessage from .serial import monitor_lines, write_serial @@ -91,6 +92,34 @@ async def upload_file( """ return await upload_file(self._transport, filename, local_path) + async def download(self, name: str) -> bytes: + """ + Download a file from the simulator. + + Args: + name: The name of the file to download. + + Returns: + The downloaded file content as bytes. + """ + result = await download(self._transport, name) + return base64.b64decode(result["result"]["binary"]) + + 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. + """ + if local_path is None: + local_path = Path(name) + + result = await self.download(name) + with open(local_path, "wb") as f: + f.write(result) + async def start_simulation( self, firmware: str,