diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..01692ce --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +whisper_env +__pycache__ +*.srt +*.wav diff --git a/client_oldapi.py b/client_oldapi.py new file mode 100644 index 0000000..d4e8cff --- /dev/null +++ b/client_oldapi.py @@ -0,0 +1,25 @@ +import sys +from whisper_live.client import TranscriptionClient + +if len(sys.argv) < 2: + print("Usage: python transcribe_file.py ") + sys.exit(1) + +audio_file = sys.argv[1] + +client = TranscriptionClient( + "localhost", + 9090, + lang="en", + translate=False, + model="small", # also support hf_model => `Systran/faster-whisper-small` + use_vad=False, + save_output_recording=True, # Only used for microphone input, False by Default + output_recording_filename="./output_recording.wav", # Only used for microphone input + mute_audio_playback=False, # Only used for file input, False by Default + enable_translation=True, + target_language="hi", +) + +# Transcribe the offline audio file +client(audio_file) diff --git a/run_server.py b/run_server.py index c021bac..d48eb70 100644 --- a/run_server.py +++ b/run_server.py @@ -90,6 +90,12 @@ if __name__ == "__main__": help='Expect raw PCM int16 audio from clients instead of float32. ' 'Audio will be normalized to float32 range [-1.0, 1.0].' ) + parser.add_argument( + '--metrics_port', + type=int, + default=0, + help='Port for Prometheus /metrics endpoint. 0 = disabled (default). Requires prometheus_client.' + ) args = parser.parse_args() if args.backend == "tensorrt": @@ -120,4 +126,5 @@ if __name__ == "__main__": batch_max_size=args.batch_max_size, batch_window_ms=args.batch_window_ms, raw_pcm_input=args.raw_pcm_input, + metrics_port=args.metrics_port, ) \ No newline at end of file diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..af73502 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,137 @@ +import unittest +from unittest.mock import patch, MagicMock + +from whisper_live import metrics as wl_metrics + +_skip_no_prometheus = unittest.skipUnless( + wl_metrics.is_available(), "prometheus_client not installed" +) + + +class TestMetricsAvailability(unittest.TestCase): + def test_is_available_returns_bool(self): + self.assertIsInstance(wl_metrics.is_available(), bool) + + +@_skip_no_prometheus +class TestTrackConnectionOpened(unittest.TestCase): + def test_increments_total_and_active(self): + total_before = wl_metrics.CONNECTIONS_TOTAL._value.get() + active_before = wl_metrics.CONNECTIONS_ACTIVE._value.get() + wl_metrics.track_connection_opened() + self.assertEqual(wl_metrics.CONNECTIONS_TOTAL._value.get(), total_before + 1) + self.assertEqual(wl_metrics.CONNECTIONS_ACTIVE._value.get(), active_before + 1) + + +@_skip_no_prometheus +class TestTrackConnectionClosed(unittest.TestCase): + def test_decrements_active(self): + wl_metrics.track_connection_opened() + active_before = wl_metrics.CONNECTIONS_ACTIVE._value.get() + wl_metrics.track_connection_closed() + self.assertEqual(wl_metrics.CONNECTIONS_ACTIVE._value.get(), active_before - 1) + + +@_skip_no_prometheus +class TestTrackConnectionRejected(unittest.TestCase): + def test_rejected_full(self): + before = wl_metrics.CONNECTIONS_REJECTED.labels(reason="full")._value.get() + wl_metrics.track_connection_rejected(reason="full") + self.assertEqual(wl_metrics.CONNECTIONS_REJECTED.labels(reason="full")._value.get(), before + 1) + + def test_rejected_auth(self): + before = wl_metrics.CONNECTIONS_REJECTED.labels(reason="auth")._value.get() + wl_metrics.track_connection_rejected(reason="auth") + self.assertEqual(wl_metrics.CONNECTIONS_REJECTED.labels(reason="auth")._value.get(), before + 1) + + +@_skip_no_prometheus +class TestTrackTranscriptionLatency(unittest.TestCase): + def test_observe_records_value(self): + count_before = wl_metrics.TRANSCRIPTION_LATENCY._sum.get() + wl_metrics.track_transcription_latency(0.5) + self.assertAlmostEqual(wl_metrics.TRANSCRIPTION_LATENCY._sum.get(), count_before + 0.5, places=3) + + +@_skip_no_prometheus +class TestTrackAudioProcessed(unittest.TestCase): + def test_increments_by_duration(self): + before = wl_metrics.AUDIO_PROCESSED._value.get() + wl_metrics.track_audio_processed(3.5) + self.assertAlmostEqual(wl_metrics.AUDIO_PROCESSED._value.get(), before + 3.5, places=3) + + +@_skip_no_prometheus +class TestTrackSegmentEmitted(unittest.TestCase): + def test_completed_true(self): + before = wl_metrics.SEGMENTS_EMITTED.labels(completed="true")._value.get() + wl_metrics.track_segment_emitted(completed=True) + self.assertEqual(wl_metrics.SEGMENTS_EMITTED.labels(completed="true")._value.get(), before + 1) + + def test_completed_false(self): + before = wl_metrics.SEGMENTS_EMITTED.labels(completed="false")._value.get() + wl_metrics.track_segment_emitted(completed=False) + self.assertEqual(wl_metrics.SEGMENTS_EMITTED.labels(completed="false")._value.get(), before + 1) + + +@_skip_no_prometheus +class TestTrackRestRequest(unittest.TestCase): + def test_tracks_200(self): + before = wl_metrics.REST_REQUESTS.labels(endpoint="transcriptions", status="200")._value.get() + wl_metrics.track_rest_request(endpoint="transcriptions", status=200) + self.assertEqual(wl_metrics.REST_REQUESTS.labels(endpoint="transcriptions", status="200")._value.get(), before + 1) + + def test_tracks_500(self): + before = wl_metrics.REST_REQUESTS.labels(endpoint="transcriptions", status="500")._value.get() + wl_metrics.track_rest_request(endpoint="transcriptions", status=500) + self.assertEqual(wl_metrics.REST_REQUESTS.labels(endpoint="transcriptions", status="500")._value.get(), before + 1) + + +@_skip_no_prometheus +class TestTrackError(unittest.TestCase): + def test_tracks_transcription_error(self): + before = wl_metrics.ERRORS.labels(type="transcription")._value.get() + wl_metrics.track_error("transcription") + self.assertEqual(wl_metrics.ERRORS.labels(type="transcription")._value.get(), before + 1) + + def test_tracks_rest_error(self): + before = wl_metrics.ERRORS.labels(type="rest_transcription")._value.get() + wl_metrics.track_error("rest_transcription") + self.assertEqual(wl_metrics.ERRORS.labels(type="rest_transcription")._value.get(), before + 1) + + +class TestStartMetricsServer(unittest.TestCase): + @patch("whisper_live.metrics.start_http_server") + def test_starts_on_given_port(self, mock_start): + wl_metrics.start_metrics_server(9999) + mock_start.assert_called_once_with(9999) + + @patch("whisper_live.metrics.start_http_server", side_effect=OSError("port in use")) + def test_logs_error_on_failure(self, mock_start): + with self.assertLogs(level="ERROR") as cm: + wl_metrics.start_metrics_server(9999) + self.assertTrue(any("Failed to start" in msg for msg in cm.output)) + + +class TestNoOpWhenUnavailable(unittest.TestCase): + """Verify helper functions are no-ops when _AVAILABLE is False.""" + + def test_all_helpers_are_noop(self): + original = wl_metrics._AVAILABLE + try: + wl_metrics._AVAILABLE = False + # None of these should raise + wl_metrics.track_connection_opened() + wl_metrics.track_connection_closed() + wl_metrics.track_connection_rejected("full") + wl_metrics.track_transcription_latency(1.0) + wl_metrics.track_audio_processed(1.0) + wl_metrics.track_segment_emitted() + wl_metrics.track_rest_request() + wl_metrics.track_error() + finally: + wl_metrics._AVAILABLE = original + + +if __name__ == "__main__": + unittest.main() diff --git a/whisper_live/backend/base.py b/whisper_live/backend/base.py index 855e310..78182c8 100644 --- a/whisper_live/backend/base.py +++ b/whisper_live/backend/base.py @@ -5,6 +5,8 @@ import time import queue import numpy as np +from whisper_live import metrics as wl_metrics + class ServeClientBase(object): RATE = 16000 @@ -92,16 +94,20 @@ class ServeClientBase(object): continue try: input_sample = input_bytes.copy() + t0 = time.time() result = self.transcribe_audio(input_sample) if result is None or self.language is None: self.timestamp_offset += duration time.sleep(0.25) # wait for voice activity, result is None when no voice activity continue + wl_metrics.track_transcription_latency(time.time() - t0) + wl_metrics.track_audio_processed(duration) self.handle_transcription_output(result, duration) except Exception as e: logging.error(f"[ERROR]: Failed to transcribe audio chunk: {e}") + wl_metrics.track_error("transcription") time.sleep(0.01) def transcribe_audio(self): @@ -247,6 +253,8 @@ class ServeClientBase(object): "segments": segments, }) ) + for seg in segments: + wl_metrics.track_segment_emitted(completed=seg.get("completed", False)) except Exception as e: logging.error(f"[ERROR]: Sending data to client: {e}") diff --git a/whisper_live/metrics.py b/whisper_live/metrics.py new file mode 100644 index 0000000..00a4f9c --- /dev/null +++ b/whisper_live/metrics.py @@ -0,0 +1,122 @@ +""" +Prometheus metrics for WhisperLive server. + +Exposes a /metrics HTTP endpoint on a configurable port for Prometheus scraping. +All metrics are optional — the server works fine without prometheus_client installed. +""" + +import logging +import threading + +try: + from prometheus_client import ( + Counter, + Gauge, + Histogram, + start_http_server, + ) + + CONNECTIONS_TOTAL = Counter( + "whisperlive_connections_total", + "Total WebSocket connections accepted", + ) + CONNECTIONS_ACTIVE = Gauge( + "whisperlive_connections_active", + "Currently active WebSocket connections", + ) + CONNECTIONS_REJECTED = Counter( + "whisperlive_connections_rejected_total", + "Connections rejected (server full or auth failure)", + ["reason"], + ) + TRANSCRIPTION_LATENCY = Histogram( + "whisperlive_transcription_latency_seconds", + "Time to transcribe a single audio chunk", + buckets=(0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0), + ) + AUDIO_PROCESSED = Counter( + "whisperlive_audio_processed_seconds_total", + "Total seconds of audio processed", + ) + SEGMENTS_EMITTED = Counter( + "whisperlive_segments_emitted_total", + "Total transcription segments sent to clients", + ["completed"], + ) + REST_REQUESTS = Counter( + "whisperlive_rest_requests_total", + "Total REST API requests", + ["endpoint", "status"], + ) + ERRORS = Counter( + "whisperlive_errors_total", + "Total errors by type", + ["type"], + ) + + _AVAILABLE = True + +except ImportError: + _AVAILABLE = False + + +def is_available(): + """Check if prometheus_client is installed.""" + return _AVAILABLE + + +def start_metrics_server(port=9091): + """Start the Prometheus metrics HTTP server on the given port. + + Args: + port (int): Port to serve /metrics on. Default 9091. + """ + if not _AVAILABLE: + logging.warning("prometheus_client not installed; metrics endpoint disabled") + return + try: + start_http_server(port) + logging.info(f"Prometheus metrics available at http://0.0.0.0:{port}/metrics") + except Exception as e: + logging.error(f"Failed to start metrics server: {e}") + + +def track_connection_opened(): + if _AVAILABLE: + CONNECTIONS_TOTAL.inc() + CONNECTIONS_ACTIVE.inc() + + +def track_connection_closed(): + if _AVAILABLE: + CONNECTIONS_ACTIVE.dec() + + +def track_connection_rejected(reason="full"): + if _AVAILABLE: + CONNECTIONS_REJECTED.labels(reason=reason).inc() + + +def track_transcription_latency(seconds): + if _AVAILABLE: + TRANSCRIPTION_LATENCY.observe(seconds) + + +def track_audio_processed(seconds): + if _AVAILABLE: + AUDIO_PROCESSED.inc(seconds) + + +def track_segment_emitted(completed=True): + if _AVAILABLE: + SEGMENTS_EMITTED.labels(completed=str(completed).lower()).inc() + + +def track_rest_request(endpoint="/v1/audio/transcriptions", status="200"): + if _AVAILABLE: + REST_REQUESTS.labels(endpoint=endpoint, status=str(status)).inc() + + +def track_error(error_type="transcription"): + if _AVAILABLE: + ERRORS.labels(type=error_type).inc() diff --git a/whisper_live/server.py b/whisper_live/server.py index 66bcd21..32d24aa 100644 --- a/whisper_live/server.py +++ b/whisper_live/server.py @@ -16,6 +16,8 @@ from faster_whisper import WhisperModel import torch from enum import Enum + +from whisper_live import metrics as wl_metrics from typing import List, Optional import numpy as np from websockets.sync.server import serve @@ -345,6 +347,7 @@ class TranscriptionServer: self.use_vad = options.get('use_vad') if self.client_manager.is_server_full(websocket, options): + wl_metrics.track_connection_rejected(reason="full") websocket.close() return False # Indicates that the connection should not continue @@ -352,6 +355,7 @@ class TranscriptionServer: self.vad_detector = VoiceActivityDetector(frame_rate=self.RATE) self.initialize_client(websocket, options, faster_whisper_custom_model_path, whisper_tensorrt_path, trt_multilingual, trt_py_session=trt_py_session) + wl_metrics.track_connection_opened() return True except json.JSONDecodeError: logging.error("Failed to decode JSON from client") @@ -430,6 +434,7 @@ class TranscriptionServer: if self.client_manager.get_client(websocket): self.cleanup(websocket) websocket.close() + wl_metrics.track_connection_closed() del websocket def run(self, @@ -450,7 +455,8 @@ class TranscriptionServer: batch_enabled=False, batch_max_size=8, batch_window_ms=50, - raw_pcm_input=False): + raw_pcm_input=False, + metrics_port: int = 0): """ Run the transcription server. @@ -507,6 +513,10 @@ class TranscriptionServer: if not BackendType.is_valid(backend): raise ValueError(f"{backend} is not a valid backend type. Choose backend from {BackendType.valid_types()}") + # Start Prometheus metrics endpoint if port is specified + if metrics_port > 0: + wl_metrics.start_metrics_server(metrics_port) + # New OpenAI-compatible REST API (toggleable via enable_rest boolean) if enable_rest: app = FastAPI(title="WhisperLive OpenAI-Compatible API") @@ -538,12 +548,14 @@ class TranscriptionServer: hotwords: Optional[str] = Form(default=None), ): if stream: + wl_metrics.track_rest_request(endpoint="transcriptions", status=400) return JSONResponse({"error": "Streaming not supported in this backend."}, status_code=400) if chunking_strategy or known_speaker_names or known_speaker_references: logging.warning("Diarization/chunking params ignored; not supported.") supported_formats = ["json", "text", "srt", "verbose_json", "vtt"] if response_format not in supported_formats: + wl_metrics.track_rest_request(endpoint="transcriptions", status=400) return JSONResponse({"error": f"Unsupported response_format. Supported: {supported_formats}"}, status_code=400) if model != "whisper-1": @@ -574,8 +586,10 @@ class TranscriptionServer: os.unlink(tmp_path) if response_format == "text": + wl_metrics.track_rest_request(endpoint="transcriptions", status=200) return PlainTextResponse(text) elif response_format == "json": + wl_metrics.track_rest_request(endpoint="transcriptions", status=200) return {"text": text} elif response_format == "verbose_json": verbose = { @@ -601,6 +615,7 @@ class TranscriptionServer: if timestamp_granularities and "word" in timestamp_granularities: seg_dict["words"] = [{"word": w.word, "start": w.start, "end": w.end, "probability": w.probability} for w in seg.words] verbose["segments"].append(seg_dict) + wl_metrics.track_rest_request(endpoint="transcriptions", status=200) return verbose elif response_format in ["srt", "vtt"]: output = [] @@ -611,8 +626,11 @@ class TranscriptionServer: output.append(f"{i}\n{start.replace('.', ',')} --> {end.replace('.', ',')}\n{seg.text.strip()}\n") else: # vtt output.append(f"{start} --> {end}\n{seg.text.strip()}\n") + wl_metrics.track_rest_request(endpoint="transcriptions", status=200) return PlainTextResponse("\n".join(output)) except Exception as e: + wl_metrics.track_rest_request(endpoint="transcriptions", status=500) + wl_metrics.track_error("rest_transcription") return JSONResponse({"error": str(e)}, status_code=500) threading.Thread(