feat: enhance serial monitor functionality with UTF-8 decoding options

Author: JakubAndrysek

src/wokwi_client/client.py

@@ -207,12 +207,24 @@ async def restart_simulation(self, pause: bool = False) -> ResponseMessage:
         """
         return await restart(self._transport, pause)
 
-    async def serial_monitor_cat(self) -> None:
+    async def serial_monitor_cat(self, decode_utf8: bool = True, errors: str = "replace") -> None:
         """
         Print serial monitor output to stdout as it is received from the simulation.
+
+        Args:
+            decode_utf8: Whether to decode bytes as UTF-8. If False, prints raw bytes (default: True).
+            errors: How to handle UTF-8 decoding errors. Options: 'strict', 'ignore', 'replace' (default: 'replace').
         """
         async for line in monitor_lines(self._transport):
-            print(line.decode("utf-8"), end="", flush=True)
+            if decode_utf8:
+                try:
+                    output = line.decode("utf-8", errors=errors)
+                    print(output, end="", flush=True)
+                except UnicodeDecodeError:
+                    # Fallback to raw bytes if decoding fails completely
+                    print(line, end="", flush=True)
+            else:
+                print(line, end="", flush=True)
 
     async def serial_write(self, data: typing.Union[bytes, str, list[int]]) -> None:
         """Write data to the simulation serial monitor interface."""

src/wokwi_client/client_sync.py

@@ -141,11 +141,13 @@ def restart_simulation(self, pause: bool = False) -> t.Any:
         assert self._client is not None
         return self._run_async(self._client.restart_simulation(pause))
 
-    def serial_monitor_cat(self) -> t.Any:
+    def serial_monitor_cat(self, decode_utf8: bool = True, errors: str = "replace") -> t.Any:
         if not self._connected:
             raise RuntimeError("Client not connected")
         assert self._client is not None
-        return self._run_async(self._client.serial_monitor_cat())
+        return self._run_async(
+            self._client.serial_monitor_cat(decode_utf8=decode_utf8, errors=errors)
+        )
 
     def write_serial(self, data: bytes | str | list[int]) -> t.Any:
         if not self._connected: