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
31 changes: 30 additions & 1 deletion src/wokwi_client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
#
# SPDX-License-Identifier: MIT

import base64
from pathlib import Path
from typing import Any, Optional, Union

Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 4 additions & 0 deletions src/wokwi_client/file_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,7 @@ 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})
Loading