Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions src/wokwi_client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,9 @@

import base64
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 (
read_framebuffer_png_bytes,
save_framebuffer_png,
Expand All @@ -16,7 +17,7 @@
from .control import set_control
from .event_queue import EventQueue
from .file_ops import download, 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
Expand Down Expand Up @@ -256,6 +257,18 @@ 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) -> 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]
) -> ResponseMessage:
Expand Down
10 changes: 10 additions & 0 deletions src/wokwi_client/pins.py
Original file line number Diff line number Diff line change
Expand Up @@ -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", {})
Loading