Josh Knapp
4d7b9d524f
- Redirect sys.stdout to stderr in Python sidecar so library print() calls don't corrupt the JSON-line IPC stream - Save real stdout fd for exclusive IPC use via init_ipc() - Skip non-JSON lines in Rust reader instead of failing with parse error - Set Tauri window background color to match dark theme (#0a0a23) - Add inline dark background on html/body to prevent white flash - Use Svelte tick() to ensure progress overlay renders before invoke - Improve ProgressOverlay with spinner, better styling, z-index 9999 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""JSON-line protocol reader/writer over stdin/stdout.
|
|
|
|
IMPORTANT: stdout is reserved exclusively for IPC messages.
|
|
At init time we save the real stdout, then redirect sys.stdout → stderr
|
|
so that any rogue print() calls from libraries don't corrupt the IPC stream.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any
|
|
|
|
from voice_to_notes.ipc.messages import IPCMessage
|
|
|
|
# Save the real stdout fd for IPC before any library can pollute it.
|
|
# Then redirect sys.stdout to stderr so library prints go to stderr.
|
|
_ipc_out: io.TextIOWrapper | None = None
|
|
|
|
|
|
def init_ipc() -> None:
|
|
"""Capture real stdout for IPC and redirect sys.stdout to stderr.
|
|
|
|
Must be called once at sidecar startup, before importing any ML libraries.
|
|
"""
|
|
global _ipc_out
|
|
if _ipc_out is not None:
|
|
return # already initialised
|
|
|
|
# Duplicate the real stdout fd so we keep it even after redirect
|
|
real_stdout_fd = os.dup(sys.stdout.fileno())
|
|
_ipc_out = io.TextIOWrapper(
|
|
io.BufferedWriter(io.FileIO(real_stdout_fd, "w")),
|
|
encoding="utf-8",
|
|
line_buffering=True,
|
|
)
|
|
|
|
# Redirect sys.stdout → stderr so print() from libraries goes to stderr
|
|
sys.stdout = sys.stderr
|
|
|
|
|
|
def _get_ipc_out() -> io.TextIOWrapper:
|
|
"""Return the IPC output stream, falling back to sys.__stdout__."""
|
|
if _ipc_out is not None:
|
|
return _ipc_out
|
|
# Fallback if init_ipc() was never called (e.g. in tests)
|
|
return sys.__stdout__
|
|
|
|
|
|
def read_message() -> IPCMessage | None:
|
|
"""Read a single JSON-line message from stdin. Returns None on EOF."""
|
|
try:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
return None # EOF
|
|
line = line.strip()
|
|
if not line:
|
|
return None
|
|
data = json.loads(line)
|
|
return IPCMessage.from_dict(data)
|
|
except json.JSONDecodeError as e:
|
|
_log(f"Invalid JSON: {e}")
|
|
return None
|
|
except Exception as e:
|
|
_log(f"Read error: {e}")
|
|
return None
|
|
|
|
|
|
def write_message(msg: IPCMessage) -> None:
|
|
"""Write a JSON-line message to the IPC channel (real stdout)."""
|
|
out = _get_ipc_out()
|
|
line = json.dumps(msg.to_dict(), separators=(",", ":"))
|
|
out.write(line + "\n")
|
|
out.flush()
|
|
|
|
|
|
def write_dict(data: dict[str, Any]) -> None:
|
|
"""Write a raw dict as a JSON-line message to the IPC channel."""
|
|
out = _get_ipc_out()
|
|
line = json.dumps(data, separators=(",", ":"))
|
|
out.write(line + "\n")
|
|
out.flush()
|
|
|
|
|
|
def _log(message: str) -> None:
|
|
"""Log to stderr (stdout is reserved for IPC)."""
|
|
print(f"[sidecar] {message}", file=sys.stderr, flush=True)
|