#!/usr/bin/env python3 """ Genesis Voice-to-Terminal Server ================================ WebSocket server that receives speech transcriptions from the browser and pipes them to the Windows clipboard (and optionally to a file/pipe). Usage: python3 voice_server.py [--port 9877] [--auto-copy] [--pipe /tmp/genesis_voice] The browser page connects via WebSocket, sends transcribed text, and this server handles clipboard injection and display. """ import asyncio import json import os import signal import subprocess import sys import time from datetime import datetime from pathlib import Path import websockets # --- Configuration --- DEFAULT_PORT = 9877 CLIP_EXE = "/mnt/c/Windows/System32/clip.exe" VOICE_LOG = Path("/mnt/e/genesis-system/scripts/whisper/voice_log.txt") PIPE_PATH = Path("/tmp/genesis_voice_pipe") # Terminal colors class C: RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" BLUE = "\033[38;5;69m" GREEN = "\033[38;5;46m" RED = "\033[38;5;196m" YELLOW = "\033[38;5;226m" CYAN = "\033[38;5;87m" GRAY = "\033[38;5;240m" WHITE = "\033[38;5;255m" BG_DARK = "\033[48;5;234m" class VoiceServer: def __init__(self, port=DEFAULT_PORT, auto_copy=True, pipe_path=None): self.port = port self.auto_copy = auto_copy self.pipe_path = pipe_path self.clients = set() self.transcript_buffer = [] self.message_count = 0 self.start_time = None def copy_to_clipboard(self, text: str) -> bool: """Copy text to Windows clipboard via clip.exe.""" try: process = subprocess.Popen( [CLIP_EXE], stdin=subprocess.PIPE, stderr=subprocess.PIPE ) # clip.exe on Windows accepts UTF-8 piped from WSL2 process.communicate(input=text.encode('utf-8')) return process.returncode == 0 except FileNotFoundError: print(f" {C.RED}clip.exe not found at {CLIP_EXE}{C.RESET}") return False except Exception as e: print(f" {C.RED}Clipboard error: {e}{C.RESET}") return False def write_to_pipe(self, text: str): """Write text to named pipe if configured.""" if not self.pipe_path: return try: pipe = Path(self.pipe_path) with open(pipe, 'a') as f: f.write(text + '\n') except Exception as e: print(f" {C.RED}Pipe error: {e}{C.RESET}") def log_transcript(self, text: str): """Append transcript to log file.""" try: with open(VOICE_LOG, 'a', encoding='utf-8') as f: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") f.write(f"[{timestamp}] {text}\n") except Exception: pass async def handle_client(self, websocket): """Handle a single WebSocket client connection.""" self.clients.add(websocket) client_id = id(websocket) % 10000 print(f"\n {C.GREEN}+ Client connected{C.RESET} {C.GRAY}(#{client_id}){C.RESET}") print(f" {C.DIM}Active connections: {len(self.clients)}{C.RESET}") try: async for message in websocket: try: data = json.loads(message) except json.JSONDecodeError: # Plain text message data = {"type": "transcript", "text": message} msg_type = data.get("type", "unknown") if msg_type == "transcript": text = data.get("text", "").strip() if not text: continue self.message_count += 1 is_final = data.get("final", True) mode = data.get("mode", "manual") # manual or auto if is_final: # Display the transcription print(f"\n {C.BLUE}{C.BOLD}VOICE{C.RESET} {C.GRAY}#{self.message_count}{C.RESET}") print(f" {C.WHITE}{text}{C.RESET}") # Auto-copy to clipboard if self.auto_copy: if self.copy_to_clipboard(text): print(f" {C.GREEN}>> Copied to clipboard{C.RESET} {C.DIM}(Ctrl+V to paste){C.RESET}") else: print(f" {C.RED}>> Clipboard copy failed{C.RESET}") # Write to pipe if configured if self.pipe_path: self.write_to_pipe(text) print(f" {C.CYAN}>> Written to pipe{C.RESET}") # Log transcript self.log_transcript(text) # Store in buffer self.transcript_buffer.append({ "text": text, "time": datetime.now().isoformat(), "count": self.message_count }) # Send acknowledgment back to browser ack = json.dumps({ "type": "ack", "count": self.message_count, "clipboard": self.auto_copy }) await websocket.send(ack) else: # Interim result - show in-progress interim_display = text[:80] + ("..." if len(text) > 80 else "") print(f"\r {C.GRAY}... {interim_display}{C.RESET}", end="", flush=True) elif msg_type == "ping": await websocket.send(json.dumps({"type": "pong"})) elif msg_type == "clear": self.transcript_buffer.clear() print(f"\n {C.YELLOW}Buffer cleared{C.RESET}") elif msg_type == "status": status = { "type": "status", "messages": self.message_count, "clients": len(self.clients), "uptime": int(time.time() - self.start_time) if self.start_time else 0, "auto_copy": self.auto_copy } await websocket.send(json.dumps(status)) except websockets.exceptions.ConnectionClosed: pass except Exception as e: print(f"\n {C.RED}Client error: {e}{C.RESET}") finally: self.clients.discard(websocket) print(f"\n {C.RED}- Client disconnected{C.RESET} {C.GRAY}(#{client_id}){C.RESET}") def print_banner(self): """Print startup banner.""" print(f""" {C.BLUE}{C.BOLD}{'='*60} GENESIS VOICE-TO-TERMINAL SERVER {'='*60}{C.RESET} {C.GREEN}Server running on:{C.RESET} ws://localhost:{self.port} {C.GREEN}Auto-copy:{C.RESET} {"ON - text goes to clipboard" if self.auto_copy else "OFF"} {C.GREEN}Log file:{C.RESET} {VOICE_LOG} {C.GREEN}Pipe:{C.RESET} {self.pipe_path or "disabled"} {C.YELLOW}{C.BOLD}HOW TO USE:{C.RESET} {C.WHITE}1. Open Chrome and go to the voice page:{C.RESET} {C.CYAN}E:\\genesis-system\\scripts\\whisper\\voice_terminal.html{C.RESET} {C.DIM}(or in Chrome: file:///E:/genesis-system/scripts/whisper/voice_terminal.html){C.RESET} {C.WHITE}2. Click the mic button (or press Space) and speak{C.RESET} {C.WHITE}3. Text is automatically copied to clipboard{C.RESET} {C.WHITE}4. Switch to terminal, press Ctrl+V to paste{C.RESET} {C.DIM}Press Ctrl+C to stop the server{C.RESET} {C.BLUE}{'='*60}{C.RESET} """) async def start(self): """Start the WebSocket server.""" self.start_time = time.time() self.print_banner() # Set up the WebSocket server stop = asyncio.Future() # Handle shutdown gracefully loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): try: loop.add_signal_handler(sig, lambda: stop.set_result(None)) except NotImplementedError: # Windows doesn't support add_signal_handler pass async with websockets.serve( self.handle_client, "0.0.0.0", # Listen on all interfaces self.port, ping_interval=30, ping_timeout=10, ): print(f" {C.GREEN}Waiting for browser connection...{C.RESET}\n") try: await stop except asyncio.CancelledError: pass print(f"\n\n {C.YELLOW}Server stopped.{C.RESET}") print(f" {C.DIM}Total messages: {self.message_count}{C.RESET}") print(f" {C.DIM}Log saved to: {VOICE_LOG}{C.RESET}\n") def main(): import argparse parser = argparse.ArgumentParser( description="Genesis Voice-to-Terminal Server", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 voice_server.py # Default: port 9877, auto-copy ON python3 voice_server.py --port 8888 # Custom port python3 voice_server.py --no-copy # Disable auto-clipboard python3 voice_server.py --pipe /tmp/voice # Also write to a file pipe """ ) parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"WebSocket port (default: {DEFAULT_PORT})") parser.add_argument("--no-copy", action="store_true", help="Disable automatic clipboard copy") parser.add_argument("--pipe", type=str, default=None, help="Path to write transcripts (file or named pipe)") args = parser.parse_args() # Verify clip.exe exists if not args.no_copy and not os.path.exists(CLIP_EXE): print(f"{C.YELLOW}Warning: clip.exe not found at {CLIP_EXE}") print(f"Clipboard auto-copy will not work.{C.RESET}") server = VoiceServer( port=args.port, auto_copy=not args.no_copy, pipe_path=args.pipe ) try: asyncio.run(server.start()) except KeyboardInterrupt: print(f"\n {C.YELLOW}Shutting down...{C.RESET}") if __name__ == "__main__": main()