whisper_web/server-py/sidecar.py

"""faster-whisper sidecar (Decisions.md D-013).

Bun spawns and supervises this process (src/server/transcribe.ts); it is the
only place server-side inference happens. The interface is JSON-lines over
stdio — one JSON object per line, no framing beyond that:

  stdout <- {"type": "ready", "version": "1.1.1"}          # startup handshake
  stdout <- {"type": "fatal", "message": "..."}             # then exit 1

  stdin  -> {"type": "transcribe", "id": "1", "model": "tiny.en",
             "audio_path": "/tmp/...", "language": "en" | null,
             "task": "transcribe" | "translate",
             "mode": "file" | "live",                 # optional, default "file"
             "prompt": "Maxwell, Voices of the Void" | null}  # optional (D-WS2a)
  stdout <- {"type": "segment", "id": "1", "start": 0.0, "end": 4.2,
             "text": " Hello world."}                       # 0..n of these
  stdout <- {"type": "done", "id": "1", "cancelled": false}
  stdout <- {"type": "error", "id": "1", "message": "..."}  # instead of done

  stdin  -> {"type": "cancel", "id": "1"}   # stop between segments; still "done"
  stdin  -> {"type": "shutdown"}            # clean exit (EOF does the same)

Audio arrives as a file of raw little-endian float32 mono PCM at 16 kHz — the
exact representation the browser already produces for its own engine — so no
container decoding happens here. Requests are processed strictly one at a
time; a reader thread keeps consuming stdin meanwhile so `cancel` can
interrupt the segment loop of the request in flight.

Decoding strategy by mode (Decisions.md D-WS5a):
  "file"  faster-whisper's BatchedInferencePipeline with its Silero VAD
          (vad_filter=True, VAD_PARAMETERS): silence is dropped, speech is cut
          into <= 30 s chunks and BATCH_SIZE chunks decode per forward pass.
          Without word-level timestamps each chunk is one segment (<= 30 s).
          Segments come out in bursts (one burst per batch), so a cancel
          takes effect at the next burst boundary; the Bun side's heartbeat
          (D-021) covers the longer quiet stretch before the first burst.
  "live"  the plain sequential WhisperModel.transcribe: server-live passes
          are short (<= 29 s, often a few seconds), so batching has nothing to
          batch and VAD only adds latency and can clip a short utterance.

The last-used model stays loaded between requests (WorkBreakdown M5 warm-up:
never reload weights per request) and is replaced — not accumulated — when a
request names a different model, bounding memory to one model.

Environment:
  FASTER_WHISPER_DEVICE         ctranslate2 device (default "auto")
  FASTER_WHISPER_COMPUTE        compute type (default "int8" — the fast,
                                low-memory CPU choice; CT2 converts on load)
  FASTER_WHISPER_DOWNLOAD_ROOT  optional model download/cache directory

Requires Python >= 3.9 and the packages in requirements.txt. This file must
keep working without them up to the import: a missing dependency is reported
as a "fatal" line, which the Bun side surfaces as "server engine unavailable"
(the anonymous/browser path never needs any of this).
"""

import json
import os
import queue
import re
import sys
import threading

SAMPLE_RATE = 16000
BATCH_SIZE = 8  # D-WS5a: VAD chunks decoded per batched forward pass
# D-WS5a: Silero VAD tuned for file mode. faster-whisper's batched default
# (threshold 0.5, 400 ms padding) dropped quiet and overlapped speech —
# deletions nearly doubled on en_alien; 0.2 / 800 ms brought WER back to the
# sequential decoder's level. min_silence_duration_ms stays at VadOptions'
# 2000 ms default (a dict replaces the pipeline's 160 ms preset).
VAD_PARAMETERS = {"threshold": 0.2, "speech_pad_ms": 800}

# "Names and terms" vocabulary hint (D-WS2a). Twins of normalizeHint /
# fitHintToBudget in src/lib/hint.ts — change both together; parity is tested
# in tests/hint-parity.test.ts. Whisper would allow 223 tokens; 120 is the
# measured cap at which small models are not garbled (D-WS2a).
PROMPT_TOKEN_BUDGET = 120
_CONTROL_CHARS = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")


def normalize_hint(raw):
    """One term per line or comma-separated -> "a, b, c"; "" means no hint."""
    lines = re.split(r"\r?\n|\r", _CONTROL_CHARS.sub("", raw))
    parts = [re.sub(r"\s+", " ", line).strip() for line in lines]
    return ", ".join(part for part in parts if part)


def _prompt_sentence(text):
    return " " + text if re.search(r"[.?!]$", text) else " " + text + "."


def fit_prompt_ids(hint, encode, budget=PROMPT_TOKEN_BUDGET):
    """Prompt token ids for `hint`, cut to `budget` tokens at a whole word.

    The prompt is a space-led sentence ending in a full stop (unless it ends
    in . ? ! already) - an unterminated list drops punctuation and loops.
    The start of the list is kept and the end dropped; a dangling ","/";" is
    removed; a first word longer than the budget is cut at the budget.
    `encode` must add no special tokens.
    """
    text = normalize_hint(hint)
    if not text:
        return []
    full = encode(_prompt_sentence(text))
    if len(full) <= budget:
        return full
    words = text.split(" ")

    def encode_prefix(count):
        return encode(_prompt_sentence(re.sub(r"[,;]+$", "", " ".join(words[:count]))))

    low, high = 0, len(words) - 1
    while low < high:
        mid = (low + high + 1) // 2
        if len(encode_prefix(mid)) <= budget:
            low = mid
        else:
            high = mid - 1
    if low == 0:
        return encode(" " + words[0])[:budget]
    return encode_prefix(low)


def emit(message):
    """Write one JSON line to stdout. Only the main thread emits."""
    sys.stdout.write(json.dumps(message) + "\n")
    sys.stdout.flush()


def main():
    try:
        import faster_whisper
        import numpy as np
    except ImportError as error:
        emit(
            {
                "type": "fatal",
                "message": "faster-whisper is not installed (%s) — "
                "see server-py/requirements.txt" % error,
            }
        )
        return 1

    device = os.environ.get("FASTER_WHISPER_DEVICE", "auto")
    compute_type = os.environ.get("FASTER_WHISPER_COMPUTE", "int8")
    download_root = os.environ.get("FASTER_WHISPER_DOWNLOAD_ROOT") or None

    requests = queue.Queue()
    cancelled = set()  # request ids; GIL-atomic add/contains is enough here
    SHUTDOWN = object()

    def read_stdin():
        """Reader thread: queue work, flag cancels, shut down on EOF."""
        for line in sys.stdin:
            line = line.strip()
            if not line:
                continue
            try:
                message = json.loads(line)
            except ValueError:
                continue  # a malformed line must not kill the sidecar
            kind = message.get("type")
            if kind == "transcribe":
                requests.put(message)
            elif kind == "cancel":
                cancelled.add(message.get("id"))
            elif kind == "shutdown":
                requests.put(SHUTDOWN)
                return
        requests.put(SHUTDOWN)  # supervisor closed the pipe / died

    threading.Thread(target=read_stdin, daemon=True).start()

    cached_name = None
    cached_model = None
    cached_pipeline = None  # BatchedInferencePipeline wrapping cached_model

    emit({"type": "ready", "version": getattr(faster_whisper, "__version__", "?")})

    while True:
        request = requests.get()
        if request is SHUTDOWN:
            return 0

        request_id = request.get("id")
        try:
            model_name = request["model"]
            audio_path = request["audio_path"]
            language = request.get("language") or None
            task = request.get("task") or "transcribe"
            mode = request.get("mode") or "file"
            prompt = request.get("prompt") or ""

            if model_name != cached_name:
                # Release both before loading the replacement: one model resident.
                cached_model = cached_pipeline = None
                cached_name = None
                cached_model = faster_whisper.WhisperModel(
                    model_name,
                    device=device,
                    compute_type=compute_type,
                    download_root=download_root,
                )
                # A thin wrapper (no weights of its own) — shares cached_model.
                cached_pipeline = faster_whisper.BatchedInferencePipeline(cached_model)
                cached_name = model_name

            audio = np.fromfile(audio_path, dtype=np.float32)
            options = {}
            prompt_ids = fit_prompt_ids(
                prompt,
                lambda text: cached_model.hf_tokenizer.encode(
                    text, add_special_tokens=False
                ).ids,
            )
            if mode == "live":
                if prompt_ids:
                    # D-WS2a: token ids, already cut to the budget by the
                    # same rule as the browser engine.
                    options["initial_prompt"] = prompt_ids
                segments, _info = cached_model.transcribe(
                    audio, language=language, task=task, **options
                )
            else:
                if prompt_ids:
                    # The batched pipeline re-encodes initial_prompt itself
                    # and only accepts text, so hand it the budget-cut ids
                    # decoded back (same tokens after re-encoding).
                    options["initial_prompt"] = cached_model.hf_tokenizer.decode(prompt_ids)
                segments, _info = cached_pipeline.transcribe(
                    audio,
                    language=language,
                    task=task,
                    vad_filter=True,
                    vad_parameters=dict(VAD_PARAMETERS),  # fresh: faster-whisper may mutate it
                    batch_size=BATCH_SIZE,
                    **options,
                )

            was_cancelled = False
            # `segments` is a generator: inference happens during iteration,
            # so segments stream out (and cancel can cut in) while later ones
            # are still being decoded — per segment in live mode, per batch of
            # BATCH_SIZE speech chunks in file mode (D-WS5a).
            for segment in segments:
                if request_id in cancelled:
                    was_cancelled = True
                    break
                emit(
                    {
                        "type": "segment",
                        "id": request_id,
                        "start": round(float(segment.start), 3),
                        "end": round(float(segment.end), 3),
                        "text": segment.text,
                    }
                )
            emit({"type": "done", "id": request_id, "cancelled": was_cancelled})
        except Exception as error:  # noqa: BLE001 — one bad request must not exit
            emit({"type": "error", "id": request_id, "message": str(error)})
        finally:
            cancelled.discard(request_id)


if __name__ == "__main__":
    sys.exit(main())