feat: enhance serial monitor functionality with write support and update documentation

Author: JakubAndrysek

README.md

@@ -32,7 +32,9 @@ The basic example is in the [examples/hello_esp32/main.py](examples/hello_esp32/
 - Connect to the Wokwi Simulator
 - Upload a diagram and firmware files
 - Start a simulation
-- Monitor the serial output
+- Monitor the serial output and write to them
+- Read GPIO pins
+- Control peripherals (buttons, MPU6050, etc.)
 
 You can run the example with:
 

docs/index.md

@@ -7,7 +7,7 @@ Typed, asyncio-friendly Python SDK for the **Wokwi Simulation API**.
 - Connect to the Wokwi Simulator from Python
 - Upload diagrams and firmware files
 - Start, pause, resume, and restart simulations
-- Monitor serial output asynchronously
+- Monitor serial output asynchronously and write to them
 - Control peripherals and read GPIO pins
 - Fully type-annotated and easy to use with asyncio
 

src/wokwi_client/client.py

@@ -12,7 +12,7 @@
 from .file_ops import upload, upload_file
 from .pins import pin_listen, pin_read
 from .protocol_types import EventMessage, ResponseMessage
-from .serial import monitor_lines
+from .serial import monitor_lines, write_serial
 from .simulation import pause, restart, resume, start
 from .transport import Transport
 
@@ -181,6 +181,10 @@ async def serial_monitor_cat(self) -> None:
         async for line in monitor_lines(self._transport):
             print(line.decode("utf-8"), end="", flush=True)
 
+    async def serial_write(self, data: bytes | str | list[int]) -> None:
+        """Write data to the simulation serial monitor interface."""
+        await write_serial(self._transport, data)
+
     def _on_pause(self, event: EventMessage) -> None:
         self.last_pause_nanos = int(event["nanos"])
 

src/wokwi_client/serial.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: MIT
 
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Iterable
 
 from .event_queue import EventQueue
 from .transport import Transport
@@ -14,3 +14,17 @@ async def monitor_lines(transport: Transport) -> AsyncGenerator[bytes, None]:
         while True:
             event_msg = await queue.get()
             yield bytes(event_msg["payload"]["bytes"])
+
+
+async def write_serial(transport: Transport, data: bytes | str | Iterable[int]) -> None:
+    """Write data to the serial monitor.
+
+    Accepts bytes, str (encoded as utf-8), or an iterable of integer byte values.
+    """
+    if isinstance(data, str):
+        payload = list(data.encode("utf-8"))
+    elif isinstance(data, bytes):
+        payload = list(data)
+    else:
+        payload = list(int(b) & 0xFF for b in data)
+    await transport.request("serial-monitor:write", {"bytes": payload})