feat: add download and GPIO listing functionality to WokwiClient and WokwiClientSync

Author: JakubAndrysek

src/wokwi_client/client.py

@@ -10,8 +10,8 @@
 from .constants import DEFAULT_WS_URL
 from .control import set_control
 from .event_queue import EventQueue
-from .file_ops import upload, upload_file
-from .pins import pin_listen, pin_read
+from .file_ops import download, download_file, upload, upload_file
+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
@@ -88,6 +88,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,
@@ -214,6 +236,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: typing.Union[int, bool, float]
     ) -> ResponseMessage:

src/wokwi_client/client_sync.py

@@ -93,6 +93,18 @@ def upload_file(self, filename: str, local_path: Path | None = None) -> t.Any:
         assert self._client is not None
         return self._run_async(self._client.upload_file(filename, local_path))
 
+    def download(self, name: str) -> t.Any:
+        if not self._connected:
+            raise RuntimeError("Client not connected")
+        assert self._client is not None
+        return self._run_async(self._client.download(name))
+
+    def download_file(self, name: str, local_path: Path | None = None) -> t.Any:
+        if not self._connected:
+            raise RuntimeError("Client not connected")
+        assert self._client is not None
+        return self._run_async(self._client.download_file(name, local_path))
+
     def start_simulation(
         self,
         firmware: str,
@@ -153,6 +165,12 @@ def listen_pin(self, part: str, pin: str, listen: bool = True) -> t.Any:
         assert self._client is not None
         return self._run_async(self._client.listen_pin(part, pin, listen))
 
+    def gpio_list(self) -> t.Any:
+        if not self._connected:
+            raise RuntimeError("Client not connected")
+        assert self._client is not None
+        return self._run_async(self._client.gpio_list())
+
     def monitor_serial(self, callback: t.Callable[[bytes], None]) -> None:
         if not self._connected:
             raise RuntimeError("Client not connected")

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"]))

src/wokwi_client/models.py

@@ -10,6 +10,10 @@ class UploadParams(BaseModel):
     binary: str  # base64
 
 
+class DownloadParams(BaseModel):
+    binary: str  # base64
+
+
 class SimulationParams(BaseModel):
     firmware: str
     elf: str

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", {})