voice-to-notes/python/voice_to_notes/ipc/handlers.py

"""Message handler registry and routing."""

from __future__ import annotations

import sys
from collections.abc import Callable

from voice_to_notes.ipc.messages import IPCMessage, error_message

# Handler function type: takes a message, returns a response message
HandlerFunc = Callable[[IPCMessage], IPCMessage | None]


class HandlerRegistry:
    """Registry mapping message types to handler functions."""

    def __init__(self) -> None:
        self._handlers: dict[str, HandlerFunc] = {}

    def register(self, message_type: str, handler: HandlerFunc) -> None:
        """Register a handler for a message type."""
        self._handlers[message_type] = handler

    def handle(self, msg: IPCMessage) -> IPCMessage | None:
        """Route a message to its handler. Returns a response or error."""
        handler = self._handlers.get(msg.type)
        if handler is None:
            print(f"[sidecar] Unknown message type: {msg.type}", file=sys.stderr, flush=True)
            return error_message(msg.id, "unknown_type", f"Unknown message type: {msg.type}")
        try:
            return handler(msg)
        except Exception as e:
            print(f"[sidecar] Handler error for {msg.type}: {e}", file=sys.stderr, flush=True)
            return error_message(msg.id, "handler_error", str(e))


def ping_handler(msg: IPCMessage) -> IPCMessage:
    """Simple ping handler for testing connectivity."""
    return IPCMessage(id=msg.id, type="pong", payload={"echo": msg.payload})


def make_transcribe_handler() -> HandlerFunc:
    """Create a transcription handler with a persistent TranscribeService."""
    from voice_to_notes.services.transcribe import TranscribeService, result_to_payload

    service = TranscribeService()

    def handler(msg: IPCMessage) -> IPCMessage:
        payload = msg.payload
        result = service.transcribe(
            request_id=msg.id,
            file_path=payload["file"],
            model_name=payload.get("model", "base"),
            device=payload.get("device", "cpu"),
            compute_type=payload.get("compute_type", "int8"),
            language=payload.get("language"),
        )
        return IPCMessage(
            id=msg.id,
            type="transcribe.result",
            payload=result_to_payload(result),
        )

    return handler


def hardware_detect_handler(msg: IPCMessage) -> IPCMessage:
    """Detect hardware capabilities and return recommendations."""
    from voice_to_notes.hardware.detect import detect_hardware

    info = detect_hardware()
    return IPCMessage(
        id=msg.id,
        type="hardware.info",
        payload={
            "has_cuda": info.has_cuda,
            "cuda_device_name": info.cuda_device_name,
            "vram_mb": info.vram_mb,
            "ram_mb": info.ram_mb,
            "cpu_cores": info.cpu_cores,
            "recommended_model": info.recommended_model,
            "recommended_device": info.recommended_device,
            "recommended_compute_type": info.recommended_compute_type,
        },
    )