Skip to content

Commit c8d3388

Browse files
committed
feat: add framebuffer utilities for reading, saving, and comparing PNG images
1 parent 089d3a9 commit c8d3388

File tree

3 files changed

+155
-0
lines changed

3 files changed

+155
-0
lines changed

src/wokwi_client/client.py

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,13 @@
66
from pathlib import Path
77
from typing import Any, Optional
88

9+
from wokwi_client.framebuffer import (
10+
compare_framebuffer_png,
11+
framebuffer_png_bytes,
12+
framebuffer_read,
13+
save_framebuffer_png,
14+
)
15+
916
from .__version__ import get_version
1017
from .constants import DEFAULT_WS_URL
1118
from .control import set_control
@@ -251,3 +258,23 @@ async def set_control(
251258
value: Control value to set (float).
252259
"""
253260
return await set_control(self._transport, part=part, control=control, value=value)
261+
262+
async def framebuffer_read(self, id: str) -> ResponseMessage:
263+
"""Read the current framebuffer for the given device id."""
264+
return await framebuffer_read(self._transport, id=id)
265+
266+
async def framebuffer_png_bytes(self, id: str) -> bytes:
267+
"""Return the current framebuffer as PNG bytes."""
268+
return await framebuffer_png_bytes(self._transport, id=id)
269+
270+
async def save_framebuffer_png(self, id: str, path: Path, overwrite: bool = True) -> Path:
271+
"""Save the current framebuffer as a PNG file."""
272+
return await save_framebuffer_png(self._transport, id=id, path=path, overwrite=overwrite)
273+
274+
async def compare_framebuffer_png(
275+
self, id: str, reference: Path, save_mismatch: Optional[Path] = None
276+
) -> bool:
277+
"""Compare the current framebuffer with a reference PNG file."""
278+
return await compare_framebuffer_png(
279+
self._transport, id=id, reference=reference, save_mismatch=save_mismatch
280+
)

src/wokwi_client/client_sync.py

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,41 @@ def set_control(self, part: str, control: str, value: int | bool | float) -> t.A
198198
assert self._client is not None
199199
return self._run_async(self._client.set_control(part, control, value))
200200

201+
def framebuffer_read(self, id: str) -> t.Any:
202+
"""Read the current framebuffer for the given device id."""
203+
if not self._connected:
204+
raise RuntimeError("Client not connected")
205+
assert self._client is not None
206+
return self._run_async(self._client.framebuffer_read(id))
207+
208+
def framebuffer_png_bytes(self, id: str) -> bytes:
209+
"""Return the current framebuffer as PNG bytes."""
210+
if not self._connected:
211+
raise RuntimeError("Client not connected")
212+
assert self._client is not None
213+
return t.cast(bytes, self._run_async(self._client.framebuffer_png_bytes(id)))
214+
215+
def save_framebuffer_png(self, id: str, path: Path, overwrite: bool = True) -> Path:
216+
"""Save the current framebuffer as a PNG file."""
217+
if not self._connected:
218+
raise RuntimeError("Client not connected")
219+
assert self._client is not None
220+
return t.cast(
221+
Path, self._run_async(self._client.save_framebuffer_png(id, path, overwrite=overwrite))
222+
)
223+
224+
def compare_framebuffer_png(
225+
self, id: str, reference: Path, save_mismatch: Path | None = None
226+
) -> bool:
227+
"""Compare the current framebuffer with a reference PNG file."""
228+
if not self._connected:
229+
raise RuntimeError("Client not connected")
230+
assert self._client is not None
231+
return t.cast(
232+
bool,
233+
self._run_async(self._client.compare_framebuffer_png(id, reference, save_mismatch)),
234+
)
235+
201236
@property
202237
def version(self) -> str:
203238
if self._client:

src/wokwi_client/framebuffer.py

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
"""Framebuffer command helpers.
2+
3+
Provides utilities to interact with devices exposing a framebuffer (e.g. LCD
4+
modules) via the `framebuffer:read` command.
5+
6+
Exposed helpers:
7+
* framebuffer_read -> raw response (contains base64 PNG at result.png)
8+
* framebuffer_png_bytes -> decoded PNG bytes
9+
* save_framebuffer_png -> save PNG to disk
10+
* compare_framebuffer_png -> compare current framebuffer against reference
11+
"""
12+
13+
# SPDX-FileCopyrightText: 2025-present CodeMagic LTD
14+
#
15+
# SPDX-License-Identifier: MIT
16+
17+
from __future__ import annotations
18+
19+
import base64
20+
from pathlib import Path
21+
22+
from .exceptions import WokwiError
23+
from .protocol_types import ResponseMessage
24+
from .transport import Transport
25+
26+
__all__ = [
27+
"framebuffer_read",
28+
"framebuffer_png_bytes",
29+
"save_framebuffer_png",
30+
"compare_framebuffer_png",
31+
]
32+
33+
34+
async def framebuffer_read(transport: Transport, *, id: str) -> ResponseMessage:
35+
"""Issue `framebuffer:read` for the given device id and return raw response."""
36+
return await transport.request("framebuffer:read", {"id": id})
37+
38+
39+
def _extract_png_b64(resp: ResponseMessage) -> str:
40+
result = resp.get("result", {})
41+
png_b64 = result.get("png")
42+
if not isinstance(png_b64, str): # pragma: no cover - defensive
43+
raise WokwiError("Malformed framebuffer:read response: missing 'png' base64 string")
44+
return png_b64
45+
46+
47+
async def framebuffer_png_bytes(transport: Transport, *, id: str) -> bytes:
48+
"""Return decoded PNG bytes for the framebuffer of device `id`."""
49+
resp = await framebuffer_read(transport, id=id)
50+
return base64.b64decode(_extract_png_b64(resp))
51+
52+
53+
async def save_framebuffer_png(
54+
transport: Transport, *, id: str, path: Path, overwrite: bool = True
55+
) -> Path:
56+
"""Save the framebuffer PNG to `path` and return the path.
57+
58+
Args:
59+
transport: Active transport.
60+
id: Device id (e.g. "lcd1").
61+
path: Destination file path.
62+
overwrite: Overwrite existing file (default True). If False and file
63+
exists, raises WokwiError.
64+
"""
65+
if path.exists() and not overwrite:
66+
raise WokwiError(f"File already exists and overwrite=False: {path}")
67+
data = await framebuffer_png_bytes(transport, id=id)
68+
path.parent.mkdir(parents=True, exist_ok=True)
69+
with open(path, "wb") as f:
70+
f.write(data)
71+
return path
72+
73+
74+
async def compare_framebuffer_png(
75+
transport: Transport, *, id: str, reference: Path, save_mismatch: Path | None = None
76+
) -> bool:
77+
"""Compare the current framebuffer PNG with a reference file.
78+
79+
Performs a byte-for-byte comparison. If different and `save_mismatch` is
80+
provided, writes the current framebuffer PNG there.
81+
82+
Returns True if identical, False otherwise.
83+
"""
84+
if not reference.exists():
85+
raise WokwiError(f"Reference image does not exist: {reference}")
86+
current = await framebuffer_png_bytes(transport, id=id)
87+
ref_bytes = reference.read_bytes()
88+
if current == ref_bytes:
89+
return True
90+
if save_mismatch:
91+
save_mismatch.parent.mkdir(parents=True, exist_ok=True)
92+
save_mismatch.write_bytes(current)
93+
return False

0 commit comments

Comments
 (0)