STT-Assistant/src/stt_daemon/events.py

"""NDJSON event envelope and codec shared by the STT daemon and its subscribers."""

import json
import time

PROTOCOL_VERSION = 1

TRANSCRIPT = "transcript"
SEGMENT_DONE = "segment_done"
SPEECH_START = "speech_start"
SPEECH_STOP = "speech_stop"
HEARTBEAT = "heartbeat"
STATUS = "status"
RING_END = "ring_end"

ALL_TYPES = frozenset(
    {
        TRANSCRIPT,
        SEGMENT_DONE,
        SPEECH_START,
        SPEECH_STOP,
        HEARTBEAT,
        STATUS,
        RING_END,
    }
)

_REQUIRED_FIELDS = ("v", "type", "seq", "ts")


class DecodeError(ValueError):
    """Raised when a line cannot be decoded into a valid event envelope."""


def make_event(type_: str, seq: int, ts: float | None = None, **fields) -> dict:
    """Build an event envelope, adding any extra keyword arguments as fields."""
    event = {
        "v": PROTOCOL_VERSION,
        "type": type_,
        "seq": int(seq),
        "ts": time.time() if ts is None else float(ts),
    }
    event.update(fields)
    return event


def encode(event: dict) -> bytes:
    """Encode an event as one newline-terminated UTF-8 NDJSON line."""
    return (json.dumps(event, ensure_ascii=False) + "\n").encode("utf-8")


def decode(line: bytes | bytearray | str) -> dict:
    """Decode one NDJSON line into an event dict, validating the envelope."""
    if isinstance(line, (bytes, bytearray)):
        try:
            text = bytes(line).decode("utf-8")
        except UnicodeDecodeError as e:
            raise DecodeError(f"invalid utf-8: {e}") from e
    elif isinstance(line, str):
        text = line
    else:
        raise DecodeError(f"expected bytes or str, got {type(line).__name__}")

    try:
        event = json.loads(text)
    except json.JSONDecodeError as e:
        raise DecodeError(f"invalid json: {e}") from e

    if not isinstance(event, dict):
        raise DecodeError("event must be a JSON object")

    for field in _REQUIRED_FIELDS:
        if field not in event:
            raise DecodeError(f"missing required field: {field}")

    return event