initial commit

aic_dataset.py

@@ -0,0 +1,60 @@
+import os
+from tkinter import ALL
+import gradio as gr
+from gradio_client import file
+from huggingface_hub import hf_hub_download
+from constants import DATASET_REPO, MIX_DIR, TRANS_DIR, DATASET_METADATA, HF_TOKEN
+
+
+def _get_base_filenames_from_metadata(metadata_path):
+    local_path = hf_hub_download(
+        repo_id=DATASET_REPO,
+        repo_type="dataset",
+        filename=metadata_path,
+        token=HF_TOKEN
+    )
+    with open(local_path, "r", encoding="utf-8") as f:
+        lines = f.read().splitlines()
+    base_names = []
+    for line in lines[1:]:  # skip header
+        parts = line.split(",")
+        if parts and parts[0]:
+            # Remove directory and extension
+            filename = os.path.splitext(os.path.basename(parts[0]))[0]
+            base_names.append(filename)
+    return base_names
+
+ALL_FILES = _get_base_filenames_from_metadata(DATASET_METADATA)
+
+def get_local_mix_path(file_stem: str) -> str:
+    if not file_stem:
+        return ""
+
+    mix_path = f"{MIX_DIR}/{file_stem}.wav"
+
+    # Download selected files into local cache; returns local filesystem paths
+    mix_local = hf_hub_download(
+        repo_id=DATASET_REPO, repo_type="dataset",
+        filename=mix_path, token=HF_TOKEN
+    )
+
+    return mix_local
+
+def download_transcript(file_stem: str) -> str:
+    """
+    file_stem is the base filename to be downloaded
+    """
+    if not file_stem:
+        return ""
+
+    transcript_path = f"{TRANS_DIR}/{file_stem}.txt"
+
+
+    transcript_text = ""
+    transcript_local = hf_hub_download(
+        repo_id=DATASET_REPO, repo_type="dataset",
+        filename=transcript_path, token=HF_TOKEN
+    )
+    with open(transcript_local, "r", encoding="utf-8", errors="replace") as f:
+        transcript_text = f.read()
+    return transcript_text

app.py

@@ -9,10 +9,11 @@ from constants import (
     MINUTES_KEEP,
 )
 from sdk import SDKParams, process_file_sdk
-from audio_tools import spec_image, mix_at_snr
+from audio_tools import spec_image
 import shutil
 import tempfile
-
+from aic_dataset import ALL_FILES, get_local_mix_path
+from transcribe import transcribe_file
 
 # ===============================
 # Temporary File & Cache Management
@@ -152,7 +153,7 @@ def start_processing(sample_path: str) -> tuple[Any, str, str]:
 
 
 def enable_new_input():
-    return gr.update(visible=True)
+    return gr.update(interactive=True)
 
 
 # ===============================
@@ -169,11 +170,19 @@ with gr.Blocks() as demo:
         )
     with gr.Row():
         gr.Markdown(open("intro.md").read())
+    
     with gr.Row():
-        with gr.Column():
-            audio_file = gr.Audio(
-                type="filepath", label="Input", visible=True, sources=["upload", "microphone"]
+        dataset_dropdown = gr.Dropdown(
+            choices=ALL_FILES,
+            label="Pick example from our dataset",
+            value=None
+        )
+        audio_file = gr.Audio(
+                type="filepath", label="Input", visible=True, sources=["upload"]
             )
+
+    with gr.Row():
+        with gr.Column():
             percent_slider = gr.Slider(
                 minimum=1,
                 maximum=100,
@@ -192,7 +201,10 @@ with gr.Blocks() as demo:
             noisy_image = gr.Image(label="Noisy spectrogram", format="png", type="filepath")
             enhanced_audio = gr.Audio(type="filepath", label="Enhanced audio")
             enhanced_image = gr.Image(label="Enhanced spectrogram", format="png", type="filepath")
+            transcript_box = gr.Textbox(label="Transcript", lines=8)
+            
 
+    
     btn.click(cleanup, [input_enhancement, last_audio_file, audio_file], None).then(
         start_processing,
         inputs=audio_file,
@@ -204,7 +216,10 @@ with gr.Blocks() as demo:
             percent_slider,
         ],
         outputs=[noisy_audio, noisy_image, enhanced_audio, enhanced_image],
-    ).then(enable_new_input, None, btn)
+    ).success(transcribe_file, inputs=[enhanced_audio], outputs=[transcript_box]).then(enable_new_input, None, [btn])
+
+    dataset_dropdown.change(get_local_mix_path, inputs=dataset_dropdown, outputs=[audio_file])
+
 
 cleanup_tmp(minutes_keep=0, filter=[])
 demo.launch(allowed_paths=["/tmp", "/"])

audio_tools.py

@@ -4,8 +4,6 @@ import librosa
 from PIL import Image
 import io
 import matplotlib.pyplot as plt
-import soundfile as sf
-
 
 def spec_image(
     audio_wav: str,
@@ -44,62 +42,27 @@ def spec_image(
     return Image.open(buf).convert("RGB")
 
 
-def mix_at_snr(
-    signal_path: str,
-    noise_path: str,
-    output_path: str = "output.wav",
-    snr_db: float = 10.0,
-    rng: Optional[np.random.Generator] = None,
-) -> bool:
+def compute_wer(reference: str, hypothesis: str) -> float:
     """
-    Mix noise into clean audio at a target SNR (in dB).
-
-    Args:
-        signal_wav: Path to clean/foreground audio (wav/mp3).
-        noise_wav: Path to noise audio (wav/mp3).
-        snr_db: Desired SNR in dB (signal/noise).
-        normalize: If True, peak-normalize the mixture to |x|max=1 after mixing
-                   (note: can slightly alter achieved SNR).
-        rng: Optional numpy Generator for reproducible random cropping.
-
-    Returns:
-        Bool wether clipping occurred.
+    Compute Word Error Rate (WER) between reference and hypothesis transcripts.
     """
-    clipped = False
-    rng = rng or np.random.default_rng()
-
-    sig, sr_s = librosa.load(signal_path, mono=True, sr=None)
-    noise, sr_n = librosa.load(noise_path, mono=True, sr=None)
-
-    # Resample noise if needed
-    if sr_s != sr_n:
-        noise = librosa.resample(noise, orig_sr=sr_n, target_sr=sr_s, res_type="kaiser_best")
-
-    # Match lengths:
-    L = len(sig)
-    if len(noise) < L:
-        reps = int(np.ceil(L / len(noise)))
-        noise = np.tile(noise, reps)[:L]
-    else:
-        start = rng.integers(0, len(noise) - L + 1) if len(noise) > L else 0
-        noise = noise[start : start + L]
-
-    sig_power = float(np.mean(sig**2))
-    noise_power = float(np.mean(noise**2))
-
-    if sig_power == 0.0:
-        out = noise * 0.0
-    elif noise_power == 0.0:
-        out = sig.copy()
-    else:
-        target_noise_power = sig_power / (10.0 ** (snr_db / 10.0))
-        scale = np.sqrt(target_noise_power / noise_power)
-        noise_scaled = noise * scale
-        out = sig + noise_scaled
-
-    peak = np.max(np.abs(out)) or 1.0
-    if peak > 1.0:
-        clipped = True
-        out = out / peak
-    sf.write(output_path, out, sr_s)
-    return clipped
+    ref_words = reference.split()
+    hyp_words = hypothesis.split()
+    d = np.zeros((len(ref_words) + 1, len(hyp_words) + 1), dtype=np.uint8)
+    for i in range(len(ref_words) + 1):
+        d[i][0] = i
+    for j in range(len(hyp_words) + 1):
+        d[0][j] = j
+    for i in range(1, len(ref_words) + 1):
+        for j in range(1, len(hyp_words) + 1):
+            if ref_words[i - 1] == hyp_words[j - 1]:
+                cost = 0
+            else:
+                cost = 1
+            d[i][j] = min(
+                d[i - 1][j] + 1,      # Deletion
+                d[i][j - 1] + 1,      # Insertion
+                d[i - 1][j - 1] + cost,  # Substitution
+            )
+    wer = d[len(ref_words)][len(hyp_words)] / max(len(ref_words), 1)
+    return wer

constants.py

@@ -1,48 +1,19 @@
 from typing import Final
+import os
 
-
-ENHANCEMENT_MODELS: Final = [("Quail", "QUAIL"), ("Finch", "FINCH"), ("Lark 2", "LARK_V2")]
-API_V2_URL: Final = "https://api.ai-coustics.io/v2"
 CHUNK_SIZE: Final = 1024
 TIMEOUT_FACTOR_MB: Final = 60
 BASE_TIMEOUT_SECONDS: Final = 120
 
-
 MINUTES_KEEP: Final = 60
 
-NOISES: Final = {
-    "None": "None",
-    "Noise_0": "assets/samples/extra_noise/noise0.wav",
-    "Noise_1": "assets/samples/extra_noise/noise1.wav",
-    "Noise_2": "assets/samples/extra_noise/noise2.wav",
-}
-NOISE_TYPES: Final = list(NOISES.keys())
-SNR_LEVELS: Final = ["None", "-5", "0", "10", "20"]
-EXAMPLES: Final = ["Background", "Reverb", "Distortion", "Wind", "Music"]
-EXAMPLES_DIR: Final = [
-    [
-        "assets/samples/input/Background.wav",
-        100,
-        "FINCH",
-    ],
-    [
-        "assets/samples/input/Reverb.wav",
-        100,
-        "QUAIL",
-    ],
-    [
-        "assets/samples/input/Distortion.wav",
-        100,
-        "LARK_V2",
-    ],
-    [
-        "assets/samples/input/Wind.wav",
-        100,
-        "LARK_V2",
-    ],
-    [
-        "assets/samples/input/Music.wav",
-        100,
-        "LARK_V2",
-    ],
-]
+DATASET_REPO: Final = "ai-coustics/leo_butch_voice_focus_open_source_en_vad" 
+MIX_DIR: Final = "mix"
+SPEECH_DIR: Final = "speech"
+TRANS_DIR: Final = "transcripts"
+DATASET_METADATA: Final = "metadata.csv"
+
+# Private access token from Space secrets:
+HF_TOKEN: Final = os.getenv("HF_TOKEN")  # set in HF Space Secrets
+
+DEFAULT_SR: Final = 16000

intro.md

@@ -1,10 +1,2 @@
-Welcome! This interactive demo lets you denoise and enhance speech using **ai-coustics models**. Choose from three options, each designed for different use cases:  
-
-- [**Quail**](https://ai-coustics.com/2025/07/02/meet-quail-the-most-advanced-real-time-speech-enhancement-model/): Our lightest model, created for real-time voice isolation. Try it for free or integrate via our [SDK](https://docs.ai-coustics.com/sdk/overview).  
-- [**Finch**](https://ai-coustics.com/2025/06/12/introducing-finch-2-the-new-aicoustics-model-for-studio-quality-speech/): A larger, subtractive model focused on voice isolation and noise removal while preserving the original voice.
-- [**Lark**](https://ai-coustics.com/2025/07/29/lark-2-next-generation-reconstructive-speech-enhancement/): Our most advanced, generative model for universal speech enhancement. Lark not only removes noise but also reconstructs and restores lost details in audio, keeping the speaker’s voice intact.
-
-Finch and Lark are available via our [API](https://docs.ai-coustics.com/api-reference/v2/upload-media-file). To get started, request an [API key](https://ai-coustics.com/api/). You’ll receive 120 minutes for free. For additional usage, see our [pricing details](https://ai-coustics.com/pricing/).
-
-**How to use**  
-Upload or record a noisy audio sample, optionally add background noise, and adjust the enhancement level to your preference. Select the model you want to use. Note that Quail is typically the fastest option. You can also try the example clips below to explore to preview the capabilities of each model.
+This demo showcases **Voice Focus**, our real-time speech enhancement model, combined with **speech-to-text (STT)** transcription.
+Enhance speech, view transcripts instantly, and evaluate recognition quality — either offline or live.

requirements.txt

@@ -6,4 +6,5 @@ librosa>=0.10.1,<0.11
 loguru~=0.7
 aic-sdk>=2.0.0
 dotenv
-resampy
+resampy
+whisper-normalizer

sdk.py

@@ -9,10 +9,6 @@ import soundfile as sf
 load_dotenv()
 
 
-
-
-
-
 class SDKParams:
     def __init__(self, enhancement_level: float, sdk_key: str, model_id: str = "quail-vf-1.1-l-16khz"):
         self.enhancement_level = enhancement_level

stt_streamers/__init__.py

@@ -0,0 +1,4 @@
+from .deepgram_streamer import DeepgramStreamer
+from .soniox_streamer import SonioxStreamer
+
+__all__ = ["DeepgramStreamer", "SonioxStreamer"]

stt_streamers/deepgram_streamer.py

@@ -0,0 +1,195 @@
+import json
+import os
+import threading
+import urllib.parse
+import numpy as np
+from websockets.sync.client import connect
+from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
+
+DEEPGRAM_WEBSOCKET_URL = "wss://api.deepgram.com/v1/listen"
+
+
+class DeepgramStreamer:
+    def __init__(self, fs_hz: int, stream_name: str, on_update=None) -> None:
+        api_key = os.environ.get("DEEPGRAM_API_KEY")
+        if not api_key:
+            raise RuntimeError("Missing DEEPGRAM_API_KEY.")
+
+        self.stream_name = stream_name
+        self.api_name = "Deepgram V1 Nova-3"
+        self.on_update = on_update
+        self.final_tokens: list[dict] = []
+        self.lock = threading.Lock()
+        self.finished_event = threading.Event()
+
+        # 1. Build the Deepgram URL with query parameters
+        config = self.get_config(fs_hz)
+        query_string = urllib.parse.urlencode(config)
+        url_with_params = f"{DEEPGRAM_WEBSOCKET_URL}?{query_string}"
+
+        print(f"Connecting {stream_name} to Deepgram...")
+
+        # 2. Connect with Authorization header
+        # Deepgram requires the API key in the headers
+        headers = {"Authorization": f"Token {api_key}"}
+        self.ws = connect(url_with_params, additional_headers=headers)
+
+        # 3. Start the receiving thread
+        self.thread = threading.Thread(target=self._receive_loop, daemon=True)
+        self.thread.start()
+
+    def stream_array(self, pcm: np.ndarray, fs_hz: int) -> str:
+        """
+        Streams audio chunks to Deepgram and waits for the final result.
+        """
+        chunk_size = 160  # Keeping the same chunk size as the reference
+        num_chunks = int(np.ceil(len(pcm) / chunk_size))
+        print(f"Streaming {self.stream_name} audio to Deepgram...")
+
+        for i in range(num_chunks):
+            start_idx = i * chunk_size
+            end_idx = min((i + 1) * chunk_size, len(pcm))
+            chunk = pcm[start_idx:end_idx]
+            self.process_chunk(chunk)
+
+        # Signal the end of the stream to Deepgram
+        self.close()
+
+        # Wait for the 'finished' signal (Metadata) from the receive loop
+        self.finished_event.wait()
+
+        # Ensure connection is closed and return final text
+        self._ensure_closed()
+        with self.lock:
+            return self.render_tokens(self.final_tokens, [])
+
+    def get_config(self, fs_hz: int) -> dict:
+        """
+        Returns parameters for the Deepgram V1 URL query string.
+        """
+        assert fs_hz == 16000, "Only 16 kHz audio is supported."
+
+        return {
+            "model": "nova-3",  # Recommended general model
+            "encoding": "linear16",  # Corresponds to pcm_s16le
+            "sample_rate": 16000,
+            "channels": 1,
+            "smart_format": "true",  # handling punctuation/formatting
+            "interim_results": "true",  # required for non-final updates
+            "endpointing": "500",  # ms silence to trigger finalization
+        }
+
+    def process_chunk(self, chunk: np.ndarray) -> None:
+        """
+        Converts float32 numpy array to int16 bytes and sends to WebSocket.
+        """
+        chunk = np.clip(chunk, -1.0, 1.0)
+        chunk_int16 = (chunk * 32767).astype(np.int16)
+        if len(chunk_int16) > 0:
+            try:
+                self.ws.send(chunk_int16.tobytes())
+            except Exception:
+                pass
+
+    def render_tokens(
+        self, final_tokens: list[dict], non_final_tokens: list[dict]
+    ) -> str:
+        """
+        Renders the list of token dicts into a string.
+        Matches Soniox logic: treats certain tokens as punctuation triggers.
+        """
+        text_parts = []
+        for token in final_tokens + non_final_tokens:
+            text = token["text"]
+            text_parts.append(text)
+            # Add newline if the text chunk looks like end-of-sentence punctuation
+            # Note: Deepgram 'smart_format' usually attaches punctuation to the word.
+            if text.strip() in [".", "?", "!"]:
+                text_parts.append("\n")
+        return "".join(text_parts)
+
+    def _receive_loop(self):
+        """
+        Background loop to handle incoming JSON messages from Deepgram.
+        """
+        try:
+            while True:
+                message = self.ws.recv()
+                res = json.loads(message)
+
+                # Check for metadata indicating stream end
+                if res.get("type") == "Metadata":
+                    self.finished_event.set()
+                    break
+
+                # Deepgram error handling
+                if "error" in res:
+                    print(f"Deepgram Error: {res['error']}")
+                    break
+
+                # Process Transcripts
+                # Deepgram V1 structure: result -> channel -> alternatives -> [0] -> transcript
+                if "channel" in res:
+                    is_final = res.get("is_final", False)
+                    alternatives = res["channel"].get("alternatives", [])
+
+                    if alternatives:
+                        transcript = alternatives[0].get("transcript", "")
+
+                        if transcript:
+                            # Wrap the transcript in a dict to match the
+                            # 'render_tokens' expectation of a list[dict]
+                            token_data = {
+                                "text": transcript + " ",  # Add space for readability
+                                "is_final": is_final,
+                            }
+
+                            non_final_tokens = []
+                            with self.lock:
+                                if is_final:
+                                    self.final_tokens.append(token_data)
+                                else:
+                                    non_final_tokens.append(token_data)
+
+                                current_finals = list(self.final_tokens)
+
+                            # Trigger the callback
+                            text = self.render_tokens(current_finals, non_final_tokens)
+                            if self.on_update:
+                                self.on_update(text)
+
+        except (ConnectionClosedOK, ConnectionClosedError):
+            pass
+        except Exception as e:
+            print(f"Deepgram receive loop error: {e}")
+        finally:
+            self.finished_event.set()
+
+    def close(self) -> str:
+        """
+        Sends the specific JSON message Deepgram expects to close the stream.
+        """
+        if hasattr(self, "ws"):
+            try:
+                # Deepgram V1 expects this specific JSON to close the stream
+                self.ws.send(json.dumps({"type": "CloseStream"}))
+            except Exception:
+                pass
+        return self.render_tokens(self.final_tokens, [])
+
+    def _ensure_closed(self) -> None:
+        """
+        Physical socket closure and thread cleanup.
+        """
+        if hasattr(self, "ws"):
+            try:
+                self.ws.close()
+            except Exception:
+                pass
+
+        if (
+            hasattr(self, "thread")
+            and self.thread.is_alive()
+            and self.thread != threading.current_thread()
+        ):
+            self.thread.join(timeout=1.0)

stt_streamers/soniox_streamer.py

@@ -0,0 +1,141 @@
+import json
+import os
+import threading
+import numpy as np
+from websockets.sync.client import connect
+from websockets.exceptions import ConnectionClosedOK
+
+SONIOX_WEBSOCKET_URL = "wss://stt-rt.soniox.com/transcribe-websocket"
+
+
+class SonioxStreamer:
+    def __init__(self, fs_hz: int, stream_name: str, on_update=None) -> None:
+        api_key = os.environ.get("SONIOX_API_KEY")
+        if not api_key:
+            raise RuntimeError("Missing SONIOX_API_KEY.")
+        self.stream_name = stream_name
+        self.api_name = "Soniox RT"
+        self.on_update = on_update
+        self.final_tokens: list[dict] = []
+        self.lock = threading.Lock()
+        self.finished_event = threading.Event()
+        config = self.get_config(api_key, fs_hz)
+        print(f"Connecting {stream_name} to Soniox...")
+        self.ws = connect(SONIOX_WEBSOCKET_URL)
+        self.ws.send(json.dumps(config))
+
+        self.thread = threading.Thread(target=self._receive_loop, daemon=True)
+        self.thread.start()
+
+    def stream_array(self, pcm: np.ndarray, fs_hz: int) -> str:
+        chunk_size = 160
+        num_chunks = int(np.ceil(len(pcm) / chunk_size))
+        print(f"Streaming {self.stream_name} audio to Soniox...")
+        for i in range(num_chunks):
+            start_idx = i * chunk_size
+            end_idx = min((i + 1) * chunk_size, len(pcm))
+            chunk = pcm[start_idx:end_idx]
+            self.process_chunk(chunk)
+        try:
+            self.ws.send("")
+        except Exception:
+            pass
+
+        # 3. Wait for the 'finished' message from the receive loop
+        self.finished_event.wait()
+        self.close()
+        with self.lock:
+            return self.render_tokens(self.final_tokens, [])
+
+    def get_config(self, api_key: str, fs_hz: int) -> dict:
+        config = {
+            "api_key": api_key,
+            "model": "stt-rt-v3",
+            "language_hints": ["en", "de"],
+            "language_hints_strict": True,
+            "enable_language_identification": True,
+            "enable_speaker_diarization": False,
+            "enable_endpoint_detection": True,
+        }
+        assert fs_hz == 16000, "Only 16 kHz audio is supported."
+        config["audio_format"] = "pcm_s16le"
+        config["sample_rate"] = 16000
+        config["num_channels"] = 1
+        return config
+
+    def process_chunk(self, chunk: np.ndarray) -> None:
+        chunk = np.clip(chunk, -1.0, 1.0)
+        chunk_int16 = (chunk * 32767).astype(np.int16)
+        if len(chunk_int16) > 0:
+            try:
+                self.ws.send(chunk_int16.tobytes())
+            except Exception:
+                pass
+
+    def render_tokens(
+        self, final_tokens: list[dict], non_final_tokens: list[dict]
+    ) -> str:
+        text_parts = []
+        for token in final_tokens + non_final_tokens:
+            text = token["text"]
+            text_parts.append(text)
+            if text.strip() in [".", "?", "!"]:
+                text_parts.append("\n")
+        return "".join(text_parts)
+
+    def _receive_loop(self):
+        try:
+            while True:
+                message = self.ws.recv()
+                res = json.loads(message)
+
+                if res.get("error_code") is not None:
+                    break
+
+                non_final_tokens: list[dict] = []
+
+                with self.lock:
+                    for token in res.get("tokens", []):
+                        if token.get("text"):
+                            if token.get("is_final"):
+                                self.final_tokens.append(token)
+                            else:
+                                non_final_tokens.append(token)
+
+                    current_finals = list(self.final_tokens)
+
+                text = self.render_tokens(current_finals, non_final_tokens)
+
+                if self.on_update:
+                    self.on_update(text)
+
+                if res.get("finished"):
+                    # Signal stream_array to stop waiting
+                    self.finished_event.set()
+                    break
+
+        except ConnectionClosedOK:
+            pass
+        except Exception:
+            pass
+        finally:
+            self.finished_event.set()
+
+    def close(self) -> str:
+        """Closes the connection."""
+        if hasattr(self, "ws"):
+            try:
+                self.ws.send("")
+            except Exception:
+                pass
+            self.ws.close()
+
+        if (
+            hasattr(self, "thread")
+            and self.thread.is_alive()
+            and self.thread != threading.current_thread()
+        ):
+            self.thread.join(timeout=1.0)
+
+        with self.lock:
+            return self.render_tokens(self.final_tokens, [])

transcribe.py

@@ -0,0 +1,51 @@
+from pathlib import Path
+from regex import D
+import resampy
+import soundfile as sf
+from stt_streamers import DeepgramStreamer, SonioxStreamer
+from constants import DEFAULT_SR
+
+def transcribe_file(audio_file_path: str, streamer_type: str = "deepgram", mode: str = "RAW"):
+    """
+    Transcribe an audio file using the specified STT streamer.
+
+    Args:
+        streamer_type (str): "soniox" or "deepgram"
+        audio_file_path (str): Path to WAV file
+        mode (str): Optional label for streamer instance ("RAW", "ENHANCED", etc.)
+
+    Returns:
+        str: Transcript text
+    """
+
+    # Validate file
+    audio_path = Path(audio_file_path)
+    if not audio_path.exists():
+        raise FileNotFoundError(f"Audio file not found: {audio_file_path}")
+
+    # Load audio
+    pcm, fs_hz = sf.read(audio_file_path, dtype="float32")
+    if fs_hz != DEFAULT_SR:
+        pcm = resampy.resample(pcm, fs_hz, DEFAULT_SR)
+        fs_hz = DEFAULT_SR
+        
+    # Select streamer
+    api_map = {
+        "soniox": SonioxStreamer,
+        "deepgram": DeepgramStreamer,
+    }
+
+    streamer_key = streamer_type.lower()
+    if streamer_key not in api_map:
+        raise ValueError(
+            f"Invalid streamer_type '{streamer_type}'. "
+            f"Choose from: {', '.join(api_map.keys())}"
+        )
+
+    StreamerClass = api_map[streamer_key]
+    streamer = StreamerClass(fs_hz, mode)
+
+    # Transcribe
+    transcript = streamer.stream_array(pcm, fs_hz)
+
+    return transcript

word_error_rate.py

@@ -0,0 +1,78 @@
+import numpy as np
+from whisper_normalizer.english import EnglishTextNormalizer
+
+def tokenize_text(text: str) -> list[str]:
+    """Normalize + tokenize into words."""
+    normalizer = EnglishTextNormalizer()
+    normalized = normalizer(text)
+    # simple whitespace word tokens; adjust if you need punctuation handling
+    return normalized.split()
+
+def compute_metric(y_true: str, y_pred: str, detailed: bool = False)-> float | tuple[float, float, float, float]:
+    """
+    Returns:
+        - if detailed=False: wer
+        - if detailed=True: (wer, ins_rate, del_rate, sub_rate)
+    """
+    ref = tokenize_text(y_true)
+    hyp = tokenize_text(y_pred)
+
+    total_err, ins, dels, subs = levenshtein_word_errors(ref, hyp)
+    ref_len = max(1, len(ref))
+
+    if detailed:
+        return (
+            total_err / ref_len,
+            ins / ref_len,
+            dels / ref_len,
+            subs / ref_len,
+        )
+    return total_err / ref_len
+
+
+def levenshtein_word_errors(reference: list[str], prediction: list[str]):
+    """Edit distance with insertion/deletion/substitution breakdown."""
+    n, m = len(reference), len(prediction)
+
+    # distance + counts
+    d = np.zeros((n + 1, m + 1), dtype=np.int32)
+    ins = np.zeros((n + 1, m + 1), dtype=np.int32)
+    dels = np.zeros((n + 1, m + 1), dtype=np.int32)
+    subs = np.zeros((n + 1, m + 1), dtype=np.int32)
+
+    # init borders
+    for j in range(1, m + 1):
+        d[0, j] = j
+        ins[0, j] = j
+    for i in range(1, n + 1):
+        d[i, 0] = i
+        dels[i, 0] = i
+
+    # DP
+    for i in range(1, n + 1):
+        for j in range(1, m + 1):
+            is_sub = 0 if reference[i - 1] == prediction[j - 1] else 1
+
+            # candidates: delete, insert, substitute/match
+            del_cost = d[i - 1, j] + 1
+            ins_cost = d[i, j - 1] + 1
+            sub_cost = d[i - 1, j - 1] + is_sub
+
+            # choose best (tie-breaking: prefer sub/match, then del, then ins)
+            if sub_cost <= del_cost and sub_cost <= ins_cost:
+                d[i, j] = sub_cost
+                ins[i, j] = ins[i - 1, j - 1]
+                dels[i, j] = dels[i - 1, j - 1]
+                subs[i, j] = subs[i - 1, j - 1] + is_sub
+            elif del_cost <= ins_cost:
+                d[i, j] = del_cost
+                ins[i, j] = ins[i - 1, j]
+                dels[i, j] = dels[i - 1, j] + 1
+                subs[i, j] = subs[i - 1, j]
+            else:
+                d[i, j] = ins_cost
+                ins[i, j] = ins[i, j - 1] + 1
+                dels[i, j] = dels[i, j - 1]
+                subs[i, j] = subs[i, j - 1]
+
+    return int(d[n, m]), int(ins[n, m]), int(dels[n, m]), int(subs[n, m])