Wyoming speech-to-text server for Home Assistant using NVIDIA Parakeet

Loads parakeet-mlx in-process (no HTTP hop) and serves it over the Wyoming
protocol. On an M4 Mac mini this transcribes typical voice commands in ~110ms
versus ~1150ms for a whisper.cpp large-v3 setup, with identical accuracy on a
ten-command benchmark.

Two behaviours matter beyond speed: silence returns an empty string rather
than whisper's "Thank you." hallucination, and there is no decoder context
carried between requests.

Notable implementation details, all covered by mutation-checked regression
tests:

- MLX streams are thread-local, so the model is loaded and evaluated on a
  single dedicated worker thread. Splitting those raises
  "There is no Stream(cpu, 1) in current thread".
- parakeet_mlx.load_audio() shells out to ffmpeg, which is unnecessary here
  since Wyoming delivers 16kHz mono PCM. The mel is built directly via
  get_logmel(), whose input must be float32 -- it views the complex STFT
  output as the input dtype, so anything narrower doubles the mel bin count.
- Wyoming's run loop has no except clause, so an exception escaping
  handle_event closes the connection without sending a Transcript and Home
  Assistant waits indefinitely. Failures are caught and returned as an empty
  transcript instead.

Defaults to parakeet-tdt-0.6b-v2 rather than the newer multilingual v3
because v2 emits digits ("21 degrees") where v3 spells numbers out, and
Home Assistant's local intent matching expects digits.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

test/conftest.py

@@ -0,0 +1,12 @@
+import numpy as np
+import pytest
+
+
+@pytest.fixture
+def pcm():
+    """Build int16 mono PCM bytes of a given sample count."""
+
+    def _pcm(n_samples: int, value: int = 8192) -> bytes:
+        return np.full(n_samples, value, dtype=np.int16).tobytes()
+
+    return _pcm

test/make-clips.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+# Generate the end-to-end test clips using macOS TTS. They are not committed
+# because they are trivially reproducible binaries.
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/clips"
+mkdir -p "$DIR"
+
+PHRASES=(
+    "turn off the kitchen lights"
+    "set the living room thermostat to twenty one degrees"
+    "what is the temperature in the bedroom"
+    "dim the hallway lights to thirty percent"
+    "is the back door locked"
+    "turn on the christmas tree in the conservatory"
+    "set a timer for twelve minutes"
+    "play radio six music in the kitchen"
+    "whats the octopus agile rate right now"
+    "close the blinds in the study and turn on the desk lamp"
+)
+
+i=1
+for phrase in "${PHRASES[@]}"; do
+    say -o "$DIR/cmd$i.aiff" "$phrase"
+    afconvert -f WAVE -d LEI16@16000 -c 1 "$DIR/cmd$i.aiff" "$DIR/cmd$i.wav"
+    rm -f "$DIR/cmd$i.aiff"
+    i=$((i + 1))
+done
+
+# 3s of digital silence -- a good model returns an empty string here rather
+# than hallucinating "Thank you." the way whisper does.
+python3 - "$DIR/silence.wav" <<'PY'
+import sys, wave
+w = wave.open(sys.argv[1], "wb")
+w.setparams((1, 2, 16000, 0, "NONE", "NONE"))
+w.writeframes(b"\x00\x00" * 16000 * 3)
+w.close()
+PY
+
+echo "Wrote $(ls "$DIR"/*.wav | wc -l | tr -d ' ') clips to $DIR"

test/test_engine.py

@@ -0,0 +1,193 @@
+"""Tests for ParakeetEngine.
+
+The model itself is mocked throughout -- these cover the audio marshalling
+and threading around it, which is where the real bugs were.
+"""
+import threading
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pytest
+
+from wyoming_parakeet.engine import MIN_SAMPLES, SAMPLE_RATE, ParakeetEngine
+
+
+def make_engine(text="turn off the kitchen lights"):
+    engine = ParakeetEngine("fake/model")
+    result = MagicMock()
+    result.text = text
+    engine.model = MagicMock()
+    engine.model.generate.return_value = [result]
+    return engine
+
+
+def test_returns_model_text(pcm):
+    engine = make_engine("set a timer for 12 minutes")
+    with patch("wyoming_parakeet.engine.get_logmel"):
+        assert engine._transcribe(pcm(SAMPLE_RATE)) == "set a timer for 12 minutes"
+
+
+def test_short_audio_short_circuits_without_touching_model(pcm):
+    """A clipped VAD flush must not reach the model at all."""
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        assert engine._transcribe(pcm(MIN_SAMPLES - 1)) == ""
+    engine.model.generate.assert_not_called()
+    get_logmel.assert_not_called()
+
+
+def test_audio_at_threshold_is_processed(pcm):
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel"):
+        assert engine._transcribe(pcm(MIN_SAMPLES)) != ""
+    engine.model.generate.assert_called_once()
+
+
+def test_empty_audio_returns_empty_string():
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel"):
+        assert engine._transcribe(b"") == ""
+
+
+def test_no_results_returns_empty_string(pcm):
+    """Silence legitimately decodes to nothing; don't IndexError on it."""
+    engine = make_engine()
+    engine.model.generate.return_value = []
+    with patch("wyoming_parakeet.engine.get_logmel"):
+        assert engine._transcribe(pcm(SAMPLE_RATE)) == ""
+
+
+def test_audio_is_float32_not_bfloat16(pcm):
+    """Regression: get_logmel views the complex STFT output as the input
+    dtype, so anything narrower than float32 silently doubles the mel bin
+    count and the downstream matmul fails."""
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        engine._transcribe(pcm(SAMPLE_RATE))
+    samples = get_logmel.call_args[0][0]
+    assert np.asarray(samples).dtype == np.float32
+
+
+def test_pcm_is_scaled_to_unit_range(pcm):
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        engine._transcribe(pcm(SAMPLE_RATE, value=16384))
+    samples = np.asarray(get_logmel.call_args[0][0])
+    assert samples.shape == (SAMPLE_RATE,)
+    assert np.allclose(samples, 0.5)
+
+
+def test_full_scale_pcm_stays_within_unit_range(pcm):
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        engine._transcribe(pcm(SAMPLE_RATE, value=-32768))
+    samples = np.asarray(get_logmel.call_args[0][0])
+    assert np.abs(samples).max() <= 1.0
+
+
+def test_preprocessor_config_is_passed_through(pcm):
+    engine = make_engine()
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        engine._transcribe(pcm(SAMPLE_RATE))
+    assert get_logmel.call_args[0][1] is engine.model.preprocessor_config
+
+
+@pytest.mark.asyncio
+async def test_load_and_inference_share_one_thread():
+    """Regression: MLX streams are thread-local, so a model loaded on one
+    thread cannot be evaluated from another -- mx.eval() raises
+    'There is no Stream(cpu, 1) in current thread'."""
+    import asyncio
+
+    engine = ParakeetEngine("fake/model")
+    threads = []
+    lock = threading.Lock()
+
+    def record():
+        with lock:
+            threads.append(threading.get_ident())
+        # Hold the worker. Sequential calls can coincidentally reuse a single
+        # thread out of a multi-worker pool, so overlap them -- a pool wider
+        # than one will hand these to different threads and fail the assert.
+        threading.Event().wait(0.05)
+
+    engine._load = record
+    engine._transcribe = lambda _pcm: (record(), "")[1]
+
+    await engine.start()
+    await asyncio.gather(*(engine.transcribe(b"") for _ in range(4)))
+
+    assert len(threads) == 5
+    assert len(set(threads)) == 1, "load and inference must share one thread"
+    assert threads[0] != threading.get_ident(), "must not run on the event loop"
+
+
+@pytest.mark.asyncio
+async def test_requests_are_serialised(pcm):
+    """A single worker means overlapping satellites queue rather than
+    racing the ANE."""
+    import asyncio
+
+    engine = ParakeetEngine("fake/model")
+    concurrent = 0
+    peak = 0
+
+    def slow(_pcm):
+        nonlocal concurrent, peak
+        concurrent += 1
+        peak = max(peak, concurrent)
+        threading.Event().wait(0.05)
+        concurrent -= 1
+        return "ok"
+
+    engine._transcribe = slow
+    await asyncio.gather(*(engine.transcribe(pcm(SAMPLE_RATE)) for _ in range(4)))
+    assert peak == 1
+
+
+def test_transcription_carries_no_state_between_calls(pcm):
+    """Regression guard for the bug class that motivated leaving whisper.cpp:
+    its server reused decoder context across requests and would return the
+    previous utterance. Each call here must stand alone."""
+    engine = make_engine()
+    audio = pcm(SAMPLE_RATE)
+
+    with patch("wyoming_parakeet.engine.get_logmel") as get_logmel:
+        engine._transcribe(pcm(SAMPLE_RATE * 2, value=4096))
+        engine._transcribe(audio)
+        first = np.asarray(get_logmel.call_args[0][0])
+
+        engine._transcribe(pcm(SAMPLE_RATE // 2, value=-2048))
+        engine._transcribe(audio)
+        second = np.asarray(get_logmel.call_args[0][0])
+
+    assert np.array_equal(first, second)
+    # generate() must be handed only the current mel, with no prompt/context.
+    assert engine.model.generate.call_args[0][0] is get_logmel.return_value
+    assert engine.model.generate.call_args.kwargs == {}
+
+
+@pytest.mark.asyncio
+async def test_start_warms_the_model():
+    """The first inference JITs Metal kernels. If warm-up is dropped, the
+    first voice command after every reboot pays that cost."""
+    engine = ParakeetEngine("fake/model")
+    model = MagicMock()
+    model.generate.return_value = []
+
+    with patch("parakeet_mlx.from_pretrained", return_value=model) as load:
+        with patch("wyoming_parakeet.engine.get_logmel"):
+            await engine.start()
+
+    load.assert_called_once_with("fake/model")
+    assert model.generate.call_count == 1, "start() should run one warm-up pass"
+
+
+@pytest.mark.asyncio
+async def test_engine_survives_a_failed_request(pcm):
+    """One bad utterance must not poison the worker for later ones."""
+    engine = make_engine("recovered")
+    with patch("wyoming_parakeet.engine.get_logmel", side_effect=[RuntimeError("boom"), MagicMock()]):
+        with pytest.raises(RuntimeError):
+            await engine.transcribe(pcm(SAMPLE_RATE))
+        assert await engine.transcribe(pcm(SAMPLE_RATE)) == "recovered"

test/test_handler.py

@@ -0,0 +1,226 @@
+"""Tests for the Wyoming event handling."""
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from wyoming.asr import Transcribe, Transcript
+from wyoming.audio import AudioChunk, AudioStop
+from wyoming.info import AsrModel, AsrProgram, Attribution, Describe, Info
+
+from wyoming_parakeet.engine import SAMPLE_RATE
+from wyoming_parakeet.handler import ParakeetEventHandler
+
+INFO = Info(
+    asr=[
+        AsrProgram(
+            name="parakeet-mlx",
+            description="test",
+            attribution=Attribution(name="t", url="http://example.invalid"),
+            installed=True,
+            version="1.0.0",
+            models=[
+                AsrModel(
+                    name="fake/model",
+                    description="test",
+                    attribution=Attribution(name="t", url="http://example.invalid"),
+                    installed=True,
+                    version=None,
+                    languages=["en"],
+                )
+            ],
+        )
+    ]
+)
+
+
+@pytest.fixture
+def handler():
+    engine = MagicMock()
+    engine.transcribe = AsyncMock(return_value="turn off the kitchen lights")
+    h = ParakeetEventHandler(INFO, MagicMock(), engine, MagicMock(), MagicMock())
+    h.write_event = AsyncMock()
+    return h
+
+
+def chunk(n_samples, rate=SAMPLE_RATE):
+    return AudioChunk(
+        rate=rate, width=2, channels=1, audio=b"\x01\x00" * n_samples
+    ).event()
+
+
+async def test_describe_returns_info(handler):
+    assert await handler.handle_event(Describe().event()) is True
+    handler.write_event.assert_awaited_once()
+    assert Info.is_type(handler.write_event.await_args[0][0].type)
+
+
+async def test_transcribe_event_is_accepted(handler):
+    assert await handler.handle_event(Transcribe(language="en").event()) is True
+    handler.write_event.assert_not_awaited()
+
+
+async def test_unknown_event_keeps_connection_open(handler):
+    from wyoming.event import Event
+
+    assert await handler.handle_event(Event(type="something-else")) is True
+
+
+async def test_chunks_accumulate_before_stop(handler):
+    for _ in range(3):
+        assert await handler.handle_event(chunk(100)) is True
+    handler.engine.transcribe.assert_not_awaited()
+    assert len(handler.audio) == 3 * 100 * 2
+
+
+async def test_stop_transcribes_accumulated_audio(handler):
+    await handler.handle_event(chunk(SAMPLE_RATE // 2))
+    await handler.handle_event(chunk(SAMPLE_RATE // 2))
+
+    assert await handler.handle_event(AudioStop().event()) is False
+
+    handler.engine.transcribe.assert_awaited_once()
+    assert len(handler.engine.transcribe.await_args[0][0]) == SAMPLE_RATE * 2
+
+
+async def test_stop_writes_transcript(handler):
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+
+    event = handler.write_event.await_args[0][0]
+    assert Transcript.is_type(event.type)
+    assert Transcript.from_event(event).text == "turn off the kitchen lights"
+
+
+async def test_empty_transcript_is_still_sent(handler):
+    """Silence must produce an empty Transcript, not a dropped response --
+    Home Assistant waits for one."""
+    handler.engine.transcribe = AsyncMock(return_value="")
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+
+    event = handler.write_event.await_args[0][0]
+    assert Transcript.is_type(event.type)
+    assert Transcript.from_event(event).text == ""
+
+
+async def test_buffer_resets_after_stop(handler):
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+    assert handler.audio == b""
+
+
+async def test_stop_with_no_audio_does_not_crash(handler):
+    assert await handler.handle_event(AudioStop().event()) is False
+    handler.engine.transcribe.assert_awaited_once_with(b"")
+
+
+async def test_resampled_input_is_converted_to_16k(handler):
+    """Satellites may send other rates; the converter must normalise them
+    before the engine sees them."""
+    await handler.handle_event(chunk(48000, rate=48000))
+    await handler.handle_event(AudioStop().event())
+
+    pcm = handler.engine.transcribe.await_args[0][0]
+    assert len(pcm) == SAMPLE_RATE * 2, "1s of 48kHz audio should become 1s at 16kHz"
+
+
+# --- failure handling -------------------------------------------------------
+# wyoming's run loop is try/finally with no except, so an exception escaping
+# handle_event closes the connection having sent nothing and Home Assistant
+# waits for a response that never arrives.
+
+
+async def test_model_failure_still_sends_a_transcript(handler):
+    handler.engine.transcribe = AsyncMock(side_effect=RuntimeError("metal exploded"))
+
+    assert await handler.handle_event(AudioStop().event()) is False
+
+    event = handler.write_event.await_args[0][0]
+    assert Transcript.is_type(event.type)
+    assert Transcript.from_event(event).text == ""
+
+
+async def test_model_failure_does_not_propagate(handler):
+    """Must not escape into wyoming's run loop."""
+    handler.engine.transcribe = AsyncMock(side_effect=RuntimeError("metal exploded"))
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())  # would raise if unhandled
+
+
+async def test_model_failure_is_logged_with_traceback(handler, caplog):
+    handler.engine.transcribe = AsyncMock(side_effect=RuntimeError("metal exploded"))
+    with caplog.at_level("ERROR"):
+        await handler.handle_event(AudioStop().event())
+    assert "metal exploded" in caplog.text
+
+
+async def test_buffer_resets_after_failure(handler):
+    handler.engine.transcribe = AsyncMock(side_effect=RuntimeError("metal exploded"))
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+    assert handler.audio == b""
+
+
+# --- isolation --------------------------------------------------------------
+# The whisper.cpp server this replaced leaked decoder context between requests
+# and would return the *previous* utterance. Guard against reintroducing any
+# shared per-request state.
+
+
+async def test_concurrent_handlers_do_not_share_audio():
+    engine = MagicMock()
+    engine.transcribe = AsyncMock(return_value="")
+    seen = []
+    engine.transcribe.side_effect = lambda pcm: seen.append(pcm) or ""
+
+    a = ParakeetEventHandler(INFO, MagicMock(), engine, MagicMock(), MagicMock())
+    b = ParakeetEventHandler(INFO, MagicMock(), engine, MagicMock(), MagicMock())
+    a.write_event = AsyncMock()
+    b.write_event = AsyncMock()
+
+    # Interleave two conversations through one shared engine.
+    await a.handle_event(chunk(100))
+    await b.handle_event(chunk(300))
+    await a.handle_event(chunk(100))
+    await a.handle_event(AudioStop().event())
+    await b.handle_event(AudioStop().event())
+
+    assert [len(p) for p in seen] == [200 * 2, 300 * 2]
+
+
+async def test_transcript_reflects_only_this_requests_audio(handler):
+    """A second utterance must not inherit the first one's text."""
+    handler.engine.transcribe = AsyncMock(side_effect=["first", "second"])
+
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+    first = Transcript.from_event(handler.write_event.await_args[0][0]).text
+
+    await handler.handle_event(chunk(SAMPLE_RATE))
+    await handler.handle_event(AudioStop().event())
+    second = Transcript.from_event(handler.write_event.await_args[0][0]).text
+
+    assert (first, second) == ("first", "second")
+
+
+# --- audio format normalisation ---------------------------------------------
+# Satellites vary; the engine must always receive 16 kHz mono 16-bit.
+
+
+async def test_stereo_input_is_downmixed(handler):
+    stereo = AudioChunk(
+        rate=SAMPLE_RATE, width=2, channels=2, audio=b"\x01\x00\x01\x00" * SAMPLE_RATE
+    ).event()
+    await handler.handle_event(stereo)
+    await handler.handle_event(AudioStop().event())
+
+    assert len(handler.engine.transcribe.await_args[0][0]) == SAMPLE_RATE * 2
+
+
+async def test_8bit_input_is_widened(handler):
+    narrow = AudioChunk(
+        rate=SAMPLE_RATE, width=1, channels=1, audio=b"\x40" * SAMPLE_RATE
+    ).event()
+    await handler.handle_event(narrow)
+    await handler.handle_event(AudioStop().event())
+
+    assert len(handler.engine.transcribe.await_args[0][0]) == SAMPLE_RATE * 2

test/test_main.py

@@ -0,0 +1,45 @@
+"""Tests for CLI wiring and the Info advertised to Home Assistant."""
+from wyoming.info import Info
+
+from wyoming_parakeet.__main__ import DEFAULT_MODEL, build_info, build_parser
+
+
+def test_default_model_is_v2_not_v3():
+    """Deliberate: v3 is newer and multilingual but spells numbers out
+    ("twenty-one degrees" vs "21 degrees"). Home Assistant's local intent
+    matching wants digits, and both pipelines run prefer_local_intents, so v3
+    would still look accurate while pushing commands onto the LLM fallback.
+    If you are changing this, re-run test/wy-test.py and check cmd2/cmd4/cmd7."""
+    assert DEFAULT_MODEL == "mlx-community/parakeet-tdt-0.6b-v2"
+
+
+def test_model_defaults_are_applied():
+    args = build_parser().parse_args(["--uri", "tcp://0.0.0.0:7892"])
+    assert args.model == DEFAULT_MODEL
+    assert args.language == "en"
+    assert args.debug is False
+
+
+def test_model_can_be_overridden():
+    args = build_parser().parse_args(
+        ["--uri", "tcp://0.0.0.0:7892", "--model", "mlx-community/other"]
+    )
+    assert args.model == "mlx-community/other"
+
+
+def test_info_advertises_the_running_model():
+    """Home Assistant's Wyoming config flow reads this; if it is malformed the
+    integration cannot be added at all."""
+    info = build_info("mlx-community/other", "en")
+    assert Info.is_type(info.event().type)
+
+    program = info.asr[0]
+    assert program.installed is True
+    assert program.models[0].name == "mlx-community/other"
+    assert program.models[0].languages == ["en"]
+
+
+def test_info_round_trips_through_an_event():
+    info = build_info(DEFAULT_MODEL, "en")
+    restored = Info.from_event(info.event())
+    assert restored.asr[0].models[0].name == DEFAULT_MODEL

test/wy-test.py

@@ -0,0 +1,30 @@
+import asyncio, os, sys, time, wave
+from wyoming.client import AsyncTcpClient
+from wyoming.asr import Transcribe, Transcript
+from wyoming.audio import AudioChunk, AudioStart, AudioStop
+from wyoming.info import Describe, Info
+
+async def once(path):
+    w = wave.open(path,"rb"); pcm = w.readframes(w.getnframes()); rate=w.getframerate()
+    async with AsyncTcpClient("127.0.0.1", int(os.environ.get("PORT", 7892))) as c:
+        await c.write_event(Transcribe(language="en").event())
+        await c.write_event(AudioStart(rate=rate, width=2, channels=1).event())
+        t=time.time()
+        for i in range(0, len(pcm), 4096):
+            await c.write_event(AudioChunk(rate=rate,width=2,channels=1,audio=pcm[i:i+4096]).event())
+        await c.write_event(AudioStop().event())
+        while True:
+            e = await c.read_event()
+            if e is None: return None, 0
+            if Transcript.is_type(e.type):
+                return Transcript.from_event(e).text, (time.time()-t)*1000
+
+async def main():
+    async with AsyncTcpClient("127.0.0.1", int(os.environ.get("PORT", 7892))) as c:
+        await c.write_event(Describe().event())
+        e = await c.read_event()
+        print("INFO ok:", Info.is_type(e.type))
+    for p in sys.argv[1:]:
+        txt, ms = await once(p)
+        print(f"{p.split('/')[-1]:14s} {ms:7.0f}ms :: {txt!r}")
+asyncio.run(main())