diff --git a/README.md b/README.md index aed0cc8..919935f 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ input from microphone and pre-recorded audio files. - [Speaker Diarization](#speaker-diarization) - [Batch Inference](#batch-inference) - [Raw PCM Input](#raw-pcm-input) +- [Streaming Client (Manual Audio Chunking)](#streaming-client-manual-audio-chunking) - [Browser Extensions](#browser-extensions) - [Whisper Live Server in Docker](#whisper-live-server-in-docker) - [Troubleshooting](#troubleshooting) @@ -263,6 +264,63 @@ python3 run_server.py --port 9090 --backend faster_whisper --raw_pcm_input ``` Audio is automatically normalized to float32 range [-1.0, 1.0]. Clients can also set `audio_format` in the initial websocket options to `float32` (default), `int16`, or `uint8`. +## Streaming Client (manual audio streaming from any source) + +`StreamingTranscriptionClient` lets you push raw PCM audio bytes from any source — a live microphone capture loop, a network stream, an audio pipeline — and receive transcripts via callbacks as speech is detected. Unlike `TranscriptionClient`, it does not manage audio capture internally; you control when and how audio is fed. + +A runnable example that reads from an audio file and streams the chunks is at [`examples/manual_audio_chunking.py`](examples/manual_audio_chunking.py): + +```bash +python examples/manual_audio_chunking.py --file assets/jfk.flac +``` + +Example usage: + +```python +from whisper_live.client import StreamingTranscriptionClient + +client = StreamingTranscriptionClient( + "localhost", 9090, + lang="en", + model="small", + on_session_started=lambda: print("Server ready"), + on_partial_transcript=lambda text, segs: print(f"… {text}", end="\r"), + on_committed_transcript=lambda text, segs: print(f"✓ {text}"), + on_error=lambda e: print(f"Error: {e}"), + on_close=lambda: print("Closed"), +) + +with client: + for chunk in my_audio_source: # any cadence, any chunk size + client.send(chunk, pcm_format="int16") +``` + +Audio must be **mono, 16 kHz PCM**. Two formats are accepted: + +| `pcm_format` | Description | +|---|---| +| `"int16"` (default for raw microphone data) | 16-bit signed integers, normalized internally | +| `"float32"` | 32-bit floats in `[-1, 1]`, passed through directly | + +NumPy arrays can be sent with `send_array()`: + +```python +import numpy as np +samples = np.frombuffer(raw_bytes, dtype=np.int16) +client.send_array(samples) +``` + +**Callbacks** + +| Callback | Signature | When fired | +|---|---|---| +| `on_session_started` | `() -> None` | Server handshake complete, ready to receive audio | +| `on_partial_transcript` | `(text, segments) -> None` | In-progress segment updated | +| `on_committed_transcript` | `(text, segments) -> None` | Segment finalized | +| `on_translation` | `(text, segments) -> None` | Translated segment ready (requires `enable_translation=True`) | +| `on_error` | `(error) -> None` | WebSocket error | +| `on_close` | `() -> None` | Connection closed | + ## Browser Extensions - Run the server with your desired backend as shown [here](https://github.com/collabora/WhisperLive?tab=readme-ov-file#running-the-server). - Transcribe audio directly from your browser using our Chrome or Firefox extensions. Refer to [Audio-Transcription-Chrome](https://github.com/collabora/whisper-live/tree/main/Audio-Transcription-Chrome#readme) and https://github.com/collabora/WhisperLive/blob/main/TensorRT_whisper.md diff --git a/examples/manual_audio_chunking.py b/examples/manual_audio_chunking.py new file mode 100644 index 0000000..790853b --- /dev/null +++ b/examples/manual_audio_chunking.py @@ -0,0 +1,78 @@ +""" +Manual audio chunking example for WhisperLive. + +Streams an audio file to a running WhisperLive server in real-time sized chunks, +printing partial transcripts when speech is detected and committed transcripts +when each segment is finalized. + +Usage: + python examples/manual_audio_chunking.py --file assets/jfk.flac +""" + +import argparse +import os +import sys +import time +import wave + +try: + from whisper_live.client import StreamingTranscriptionClient + from whisper_live.utils import resample +except ImportError: # just in case whisper_live isn't installed. + sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + print("[INFO] whisper_live not installed or the current version does not have StreamingTranscriptionClient. Will attempt to import from local source.") + from whisper_live.client import StreamingTranscriptionClient + from whisper_live.utils import resample + +SAMPLE_RATE = 16000 + + +def stream_audio_file(path: str, client: StreamingTranscriptionClient, chunk_ms: int = 50) -> None: + """Read an audio file, resample to 16 kHz mono if needed, and pace chunks in real time.""" + resampled_path = resample(path) + try: + with wave.open(resampled_path, "rb") as wf: + frames_per_chunk = SAMPLE_RATE * chunk_ms // 1000 + chunk_duration = frames_per_chunk / SAMPLE_RATE + while chunk := wf.readframes(frames_per_chunk): + client.send(chunk, pcm_format="int16") + time.sleep(chunk_duration) + finally: + os.remove(resampled_path) + + +def main(): + parser = argparse.ArgumentParser(description="Stream an audio file to WhisperLive.") + parser.add_argument("--file", "-f", required=True, help="Audio file to transcribe (any format supported by ffmpeg).") + parser.add_argument("--server", "-s", default="localhost") + parser.add_argument("--port", "-p", type=int, default=9090) + parser.add_argument("--model", "-m", default="small") + parser.add_argument("--lang", "-l", default="en") + parser.add_argument("--chunk_ms", type=int, default=50, help="Chunk size in ms.") + args = parser.parse_args() + + client = StreamingTranscriptionClient( + args.server, args.port, + lang=args.lang, + model=args.model, + on_session_started=lambda: print("[INFO] Server ready.\n"), + on_partial_transcript=lambda text, _: print(f"\r… {text:<80}", end="", flush=True), + on_committed_transcript=lambda text, _: print(f"\r✓ {text:<80}"), + on_error=lambda e: print(f"\n[ERROR] {e}"), + on_close=lambda: print("\n[INFO] Connection closed."), + ) + + with client: + print(f"[INFO] Streaming {args.file} in {args.chunk_ms} ms chunks.") + stream_audio_file(args.file, client, chunk_ms=args.chunk_ms) + + print("\n[INFO] Final transcript:") + for seg in client.transcript: + print(f" [{float(seg['start']):.2f}s → {float(seg['end']):.2f}s] {seg['text'].strip()}") + seg = client.last_partial + if seg: + print(f" [{float(seg['start']):.2f}s → {float(seg['end']):.2f}s] {seg['text'].strip()} (partial)") + + +if __name__ == "__main__": + main() diff --git a/tests/test_streaming_client.py b/tests/test_streaming_client.py new file mode 100644 index 0000000..d986467 --- /dev/null +++ b/tests/test_streaming_client.py @@ -0,0 +1,210 @@ +import json +import time +import unittest +from unittest.mock import patch, MagicMock + +import numpy as np + +from whisper_live.client import Client, StreamingTranscriptionClient + + +class StreamingClientTestCase(unittest.TestCase): + @patch('whisper_live.client.websocket.WebSocketApp') + def setUp(self, mock_websocket): + self.mock_websocket = mock_websocket + self.mock_ws_app = mock_websocket.return_value + self.mock_ws_app.send = MagicMock() + + self.committed = [] + self.partials = [] + self.session_started = [] + + self.client = StreamingTranscriptionClient( + host='localhost', + port=9090, + lang="en", + on_session_started=lambda: self.session_started.append(True), + on_committed_transcript=lambda text, segs: self.committed.append((text, segs)), + on_partial_transcript=lambda text, segs: self.partials.append((text, segs)), + ) + self._inner = self.client._client + + def tearDown(self): + self._inner.close_websocket() + self.mock_websocket.stop() + + def _server_ready(self, backend="faster_whisper"): + self._inner.on_message(self.mock_ws_app, json.dumps({ + "uid": self._inner.uid, + "message": "SERVER_READY", + "backend": backend, + })) + + def _send_segments(self, segments): + self._inner.on_message(self.mock_ws_app, json.dumps({ + "uid": self._inner.uid, + "segments": segments, + })) + + +class TestPcmFormatConversion(StreamingClientTestCase): + def test_int16_is_normalized_to_float32(self): + self._server_ready() + raw = np.array([0, 16384, -32768], dtype=np.int16).tobytes() + with patch.object(self._inner, 'send_packet_to_server') as mock_send: + self.client.send(raw, pcm_format="int16") + sent = np.frombuffer(mock_send.call_args[0][0], dtype=np.float32) + np.testing.assert_allclose(sent, [0.0, 0.5, -1.0], atol=1e-4) + + def test_float32_passes_through(self): + self._server_ready() + raw = np.array([0.1, -0.2], dtype=np.float32).tobytes() + with patch.object(self._inner, 'send_packet_to_server') as mock_send: + self.client.send(raw, pcm_format="float32") + self.assertEqual(mock_send.call_args[0][0], raw) + + def test_default_format_is_int16(self): + self._server_ready() + raw = np.array([32767], dtype=np.int16).tobytes() + with patch.object(self._inner, 'send_packet_to_server') as mock_send: + self.client.send(raw) + sent = np.frombuffer(mock_send.call_args[0][0], dtype=np.float32) + self.assertAlmostEqual(float(sent[0]), 32767 / 32768.0, places=4) + + def test_unsupported_format_raises(self): + self._server_ready() + with self.assertRaises(ValueError): + self.client.send(b"\x00\x00", pcm_format="int8") + + def test_send_after_close_raises(self): + self.client._closed = True + with self.assertRaises(RuntimeError): + self.client.send(b"\x00\x00", pcm_format="int16") + + def test_send_array_normalizes_integers(self): + with patch.object(self._inner, 'send_packet_to_server') as mock_send: + self.client.send_array(np.array([0, 16384, -32768], dtype=np.int16)) + sent = np.frombuffer(mock_send.call_args[0][0], dtype=np.float32) + np.testing.assert_allclose(sent, [0.0, 0.5, -1.0], atol=1e-4) + + +class TestTranscriptDispatch(StreamingClientTestCase): + def test_partial_then_committed(self): + self._server_ready() + self._send_segments([{"start": 0, "end": 1, "text": "hello", "completed": False}]) + self.assertEqual(len(self.partials), 1) + self.assertEqual(self.partials[0][0], "hello") + self.assertEqual(len(self.committed), 0) + + self._send_segments([{"start": 0, "end": 1, "text": "hello world", "completed": True}]) + self.assertEqual(len(self.committed), 1) + self.assertEqual(self.committed[0][0], "hello world") + self.assertEqual(len(self.client.transcript), 1) + + def test_committed_deduplicated(self): + self._server_ready() + seg = {"start": 0, "end": 1, "text": "hi", "completed": True} + self._send_segments([seg]) + self._send_segments([seg]) + self.assertEqual(len(self.committed), 1) + self.assertEqual(len(self.client.transcript), 1) + + def test_committed_backend_agnostic(self): + """Committed dispatch must work for non-faster_whisper backends.""" + self._server_ready(backend="tensorrt") + self._send_segments([{"start": 0, "end": 1, "text": "trt seg", "completed": True}]) + self.assertEqual(len(self.committed), 1) + self.assertEqual(len(self.client.transcript), 1) + + def test_last_partial_alias(self): + self._server_ready() + self._send_segments([{"start": 0, "end": 1, "text": "pending", "completed": False}]) + self.assertIsNotNone(self.client.last_partial) + self.assertIs(self.client.last_partial, self.client.last_segment) + + +class TestConnectLifecycle(StreamingClientTestCase): + def test_connect_returns_after_ready(self): + self._server_ready() + self.assertIs(self.client.connect(), self.client) + self.assertEqual(len(self.session_started), 1) + + def test_connect_times_out(self): + self.client._ready_timeout = 0.1 + with self.assertRaises(TimeoutError): + self.client.connect() + + def test_connect_raises_on_server_error(self): + self._inner.on_message(self.mock_ws_app, json.dumps({ + "uid": self._inner.uid, + "status": "ERROR", + "message": "boom", + })) + with self.assertRaises(RuntimeError): + self.client.connect() + + def test_connect_raises_when_server_full(self): + self._inner.on_message(self.mock_ws_app, json.dumps({ + "uid": self._inner.uid, + "status": "WAIT", + "message": 5, + })) + with self.assertRaises(RuntimeError): + self.client.connect() + + def test_close_sends_end_of_audio(self): + self._server_ready() + self._inner.recording = False # pretend server already closed + with patch.object(self._inner, 'send_packet_to_server') as mock_send, \ + patch.object(self._inner, 'close_websocket') as mock_close: + self.client.close() + mock_send.assert_called_once_with(Client.END_OF_AUDIO.encode("utf-8")) + mock_close.assert_called_once() + + def test_close_waits_for_server_then_times_out(self): + self._server_ready() + self.assertTrue(self._inner.recording) # server still "processing" + start = time.time() + with patch.object(self._inner, 'send_packet_to_server'), \ + patch.object(self._inner, 'close_websocket') as mock_close: + self.client.close(timeout=0.2) + self.assertGreaterEqual(time.time() - start, 0.2) + mock_close.assert_called_once() + + def test_close_returns_early_when_server_closes(self): + self._server_ready() + + def close_soon(_msg): + self._inner.recording = False + + with patch.object(self._inner, 'send_packet_to_server', side_effect=close_soon), \ + patch.object(self._inner, 'close_websocket') as mock_close: + start = time.time() + self.client.close(timeout=10.0) + self.assertLess(time.time() - start, 1.0) + mock_close.assert_called_once() + + +class TestErrorHandling(StreamingClientTestCase): + def test_close_frame_not_reported_as_error(self): + """A normal CLOSE control frame (opcode 8) must not fire on_error.""" + self._server_ready() + errors = [] + self.client._client._on_error_hook = errors.append + close_frame = MagicMock() + close_frame.opcode = 8 + self._inner.on_error(self.mock_ws_app, close_frame) + self.assertEqual(errors, []) + self.assertFalse(self._inner.server_error) + + def test_real_error_still_reported(self): + self._server_ready() + errors = [] + self.client._client._on_error_hook = errors.append + self._inner.on_error(self.mock_ws_app, RuntimeError("boom")) + self.assertEqual(len(errors), 1) + self.assertTrue(self._inner.server_error) + + +if __name__ == '__main__': + unittest.main() diff --git a/whisper_live/client.py b/whisper_live/client.py index 6f2b674..759e2e7 100644 --- a/whisper_live/client.py +++ b/whisper_live/client.py @@ -11,6 +11,7 @@ import websocket import uuid import time import av +from typing import Callable, Literal, Optional import whisper_live.utils as utils @@ -911,3 +912,230 @@ class TranscriptionClient(TranscriptionTeeClient): output_recording_filename=output_recording_filename, mute_audio_playback=mute_audio_playback ) + + +PcmFormat = Literal["float32", "int16"] + + +class _HookedClient(Client): + """Client subclass that exposes lifecycle callbacks not available on the base class.""" + + def __init__(self, *args, on_session_started=None, on_error_hook=None, on_close_hook=None, **kwargs): + self._on_session_started = on_session_started + self._on_error_hook = on_error_hook + self._on_close_hook = on_close_hook + super().__init__(*args, **kwargs) + + def on_message(self, ws, message): + was_recording = self.recording + super().on_message(ws, message) + if not was_recording and self.recording and self._on_session_started: + self._on_session_started() + + def on_error(self, ws, error): + # websocket-client surfaces the server's CLOSE control frame (opcode 8) + # through on_error during shutdown; a normal close is not an error. + if getattr(error, "opcode", None) == 8: + return + if self._on_error_hook: + self._on_error_hook(error) + super().on_error(ws, error) + + def on_close(self, ws, close_status_code, close_msg): + if self._on_close_hook: + self._on_close_hook() + super().on_close(ws, close_status_code, close_msg) + + +class StreamingTranscriptionClient: + """Feed raw PCM audio in chunks; receive partial and committed transcripts via callbacks. + + Args: + host: WhisperLive server hostname. + port: WhisperLive server port. + lang: Language code (e.g. ``"en"``). ``None`` enables auto-detection. + model: Whisper model size (``"tiny"``, ``"base"``, ``"small"``, ``"medium"``, ``"large"``). + use_vad: Enable server-side voice activity detection. + use_wss: Use ``wss://`` instead of ``ws://``. + send_last_n_segments: How many recent segments the server echoes per update. + no_speech_thresh: Segments with no-speech probability above this are discarded. + clip_audio: Drop audio with no valid segments. + same_output_threshold: Repeated identical outputs before a segment is committed. + enable_translation: Enable post-transcription translation. + target_language: Target language for translation (e.g. ``"fr"``). + ready_timeout: Seconds to wait for ``SERVER_READY`` before raising ``TimeoutError``. + on_session_started: Called once when the server is ready to receive audio. + on_partial_transcript: Called on each in-progress segment update with ``(text, segments)``. + on_committed_transcript: Called for each finalized segment with ``(text, segments)``. + on_translation: Called for each translated segment with ``(text, segments)``. + on_error: Called on WebSocket errors with the exception. + on_close: Called when the connection closes. + + Example:: + + client = StreamingTranscriptionClient( + "localhost", 9090, + lang="en", + on_partial_transcript=lambda text, _: print(f"… {text}", end="\\r"), + on_committed_transcript=lambda text, _: print(f"✓ {text}"), + ) + with client: + for chunk in my_audio_source: + client.send(chunk, pcm_format="int16") + """ + + def __init__( + self, + host: str, + port: int, + *, + lang: Optional[str] = None, + model: str = "small", + use_vad: bool = True, + use_wss: bool = False, + send_last_n_segments: int = 10, + no_speech_thresh: float = 0.45, + clip_audio: bool = False, + same_output_threshold: int = 10, + enable_translation: bool = False, + target_language: str = "fr", + ready_timeout: float = 30.0, + on_session_started: Optional[Callable[[], None]] = None, + on_partial_transcript: Optional[Callable[[str, list], None]] = None, + on_committed_transcript: Optional[Callable[[str, list], None]] = None, + on_translation: Optional[Callable[[str, list], None]] = None, + on_error: Optional[Callable[[Exception], None]] = None, + on_close: Optional[Callable[[], None]] = None, + ): + self._on_partial_transcript = on_partial_transcript + self._on_committed_transcript = on_committed_transcript + self._ready_timeout = ready_timeout + self._closed = False + self._transcript = [] + self._committed_keys = set() + + self._client = _HookedClient( + host=host, + port=port, + lang=lang, + model=model, + use_vad=use_vad, + use_wss=use_wss, + log_transcription=False, + send_last_n_segments=send_last_n_segments, + no_speech_thresh=no_speech_thresh, + clip_audio=clip_audio, + same_output_threshold=same_output_threshold, + enable_translation=enable_translation, + target_language=target_language, + transcription_callback=self._dispatch_transcript, + translation_callback=on_translation, + on_session_started=on_session_started, + on_error_hook=on_error, + on_close_hook=on_close, + ) + + def _dispatch_transcript(self, text: str, segments: list) -> None: + for seg in segments: + if not seg.get("completed", False): + continue + key = (seg.get("start"), seg.get("end"), seg.get("text")) + if key in self._committed_keys: + continue + self._committed_keys.add(key) + self._transcript.append(seg) + if self._on_committed_transcript: + self._on_committed_transcript(seg["text"].strip(), [seg]) + + last = segments[-1] if segments else None + if last and not last.get("completed", False) and self._on_partial_transcript: + self._on_partial_transcript(last["text"].strip(), [last]) + + def connect(self) -> "StreamingTranscriptionClient": + """Block until the server is ready. Returns self for use as a context manager.""" + deadline = time.time() + self._ready_timeout + while not self._client.recording: + if self._client.server_error: + raise RuntimeError(getattr(self._client, "error_message", "Server reported an error.")) + if self._client.waiting: + raise RuntimeError("Server is full.") + if time.time() > deadline: + raise TimeoutError("Timed out waiting for server ready.") + time.sleep(0.05) + return self + + def send(self, audio_bytes: bytes, pcm_format: PcmFormat = "int16") -> None: + """Send one PCM chunk. Any chunk size is fine; must be mono 16 kHz. + + Args: + audio_bytes: Raw PCM payload. + pcm_format: ``"int16"`` is normalized to float32; ``"float32"`` passes through. + """ + if self._closed: + raise RuntimeError("Client is already closed.") + if not audio_bytes: + return + if pcm_format == "float32": + payload = audio_bytes + elif pcm_format == "int16": + samples = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32) / 32768.0 + payload = samples.tobytes() + else: + raise ValueError(f"Unsupported pcm_format: {pcm_format!r}") + self._client.send_packet_to_server(payload) + + def send_array(self, samples: np.ndarray) -> None: + """Send a numpy array (any numeric dtype, mono, 16 kHz). + + Args: + samples: 1-D numpy array of audio samples. + """ + if samples.ndim != 1: + raise ValueError("Expected mono (1-D) array.") + if np.issubdtype(samples.dtype, np.integer): + info = np.iinfo(samples.dtype) + samples = samples.astype(np.float32) / max(abs(info.min), info.max) + elif samples.dtype != np.float32: + samples = samples.astype(np.float32) + self._client.send_packet_to_server(samples.tobytes()) + + @property + def transcript(self) -> list: + """All committed segments received so far.""" + return self._transcript + + @property + def last_partial(self) -> Optional[dict]: + """The most recent in-progress segment, or ``None`` if none pending.""" + return self._client.last_segment + + # Alias for ``last_partial``; kept for readability at call sites. + last_segment = last_partial + + def close(self, timeout: float = 15.0) -> None: + """Signal end-of-stream, wait for the server to finish, then close. + + After ``END_OF_AUDIO`` the server transcribes any buffered audio, sends + the final committed segment, and closes the connection. Waiting for that + server-initiated close keeps the last segment from being dropped. + + Args: + timeout: Maximum seconds to wait for the server to close before + forcing the connection shut. + """ + if self._closed: + return + self._closed = True + try: + self._client.send_packet_to_server(Client.END_OF_AUDIO.encode("utf-8")) + deadline = time.time() + timeout + while self._client.recording and time.time() < deadline: + time.sleep(0.05) + finally: + self._client.close_websocket() + + def __enter__(self) -> "StreamingTranscriptionClient": + return self.connect() + + def __exit__(self, exc_type, exc, tb) -> None: + self.close()