Skip to content

Commit dab255f

Browse files
committed
feat: add framebuffer command helpers for reading and saving PNGs
1 parent 5f9249a commit dab255f

File tree

2 files changed

+120
-0
lines changed

2 files changed

+120
-0
lines changed

src/wokwi_client/client.py

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

8+
from wokwi_client.framebuffer import (
9+
compare_framebuffer_png,
10+
framebuffer_png_bytes,
11+
framebuffer_read,
12+
save_framebuffer_png,
13+
)
14+
815
from .__version__ import get_version
916
from .constants import DEFAULT_WS_URL
1017
from .control import set_control
@@ -233,3 +240,23 @@ async def set_control(
233240
value: Control value to set (float).
234241
"""
235242
return await set_control(self._transport, part=part, control=control, value=value)
243+
244+
async def framebuffer_read(self, id: str) -> ResponseMessage:
245+
"""Read the current framebuffer for the given device id."""
246+
return await framebuffer_read(self._transport, id=id)
247+
248+
async def framebuffer_png_bytes(self, id: str) -> bytes:
249+
"""Return the current framebuffer as PNG bytes."""
250+
return await framebuffer_png_bytes(self._transport, id=id)
251+
252+
async def save_framebuffer_png(self, id: str, path: Path, overwrite: bool = True) -> Path:
253+
"""Save the current framebuffer as a PNG file."""
254+
return await save_framebuffer_png(self._transport, id=id, path=path, overwrite=overwrite)
255+
256+
async def compare_framebuffer_png(
257+
self, id: str, reference: Path, save_mismatch: Optional[Path] = None
258+
) -> bool:
259+
"""Compare the current framebuffer with a reference PNG file."""
260+
return await compare_framebuffer_png(
261+
self._transport, id=id, reference=reference, save_mismatch=save_mismatch
262+
)

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)