Italian ASR app bundle only

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.py[cod]
+.Python
+*.so
+.venv/
+venv/
+.env

README.md

@@ -0,0 +1,43 @@
+---
+title: Italian Speech To Text
+emoji: 🏃
+colorFrom: yellow
+colorTo: pink
+sdk: gradio
+sdk_version: 6.12.0
+app_file: app.py
+pinned: false
+short_description: Italian ASR + English translation (Whisper + Marian). Optional local fine-tuned model in models/whisper_finetuned_it/.
+---
+
+# Italian → Italian + English
+
+Speak **Italian**; get **Italian transcription** and **English** translation.
+
+- Default: `openai/whisper-small` + `Helsinki-NLP/opus-mt-it-en`
+- Optional: copy your fine-tuned Whisper into `models/whisper_finetuned_it/` and set `ASR_REALTIME_MODE=finetuned` in Space variables.
+
+## Add your model
+
+1. Copy training outputs into `models/whisper_finetuned_it/` (`config.json`, tokenizer files, weights).
+2. Use **Git LFS** for large weight files when pushing this Space.
+3. In Space **Settings → Repository variables**: `ASR_REALTIME_MODE` = `finetuned`.
+
+## Optional environment variables
+
+| Variable | Default | Meaning |
+|----------|---------|---------|
+| `ASR_WHISPER_MODEL` | `openai/whisper-small` | HF Whisper id if not using local finetuned |
+| `ASR_REALTIME_MODE` | `quality` | `quality` = hub Whisper; `finetuned` = load `models/whisper_finetuned_it` |
+| `ASR_WHISPER_FINETUNED_DIR` | (see above) | Override path to finetuned folder |
+| `ASR_TRANSLATE` | `1` | Set `0` to disable English translation |
+| `ASR_MIN_RMS` | `0.005` | Silence gate |
+
+## Local test
+
+```bash
+cd Italian-Speech-to-Text
+python app.py
+```
+
+Configuration reference: https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,67 @@
+"""
+Hugging Face Space: speak Italian → **Italian transcription + English translation**.
+
+Whisper (Italian) + Marian IT→EN. Run locally: ``python app.py``
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+os.environ.setdefault("ASR_REALTIME_MODE", "quality")
+os.environ.setdefault("ASR_WHISPER_LANGUAGE", "italian")
+os.environ.setdefault("ASR_TRANSLATE", "1")
+
+import numpy as np
+import gradio as gr
+
+from italian_en_pipeline import ItalianEnglishPipeline
+
+_SPACE_ROOT = Path(__file__).resolve().parent
+
+_pipeline: ItalianEnglishPipeline | None = None
+
+
+def _get_pipeline() -> ItalianEnglishPipeline:
+    global _pipeline
+    if _pipeline is None:
+        _pipeline = ItalianEnglishPipeline(project_root=str(_SPACE_ROOT))
+    return _pipeline
+
+
+def transcribe(audio: tuple[int, np.ndarray] | None) -> tuple[str, str]:
+    """Gradio Audio (numpy) → (italian, english)."""
+    if audio is None:
+        return "", ""
+    sr, data = audio
+    if data is None or len(data) == 0:
+        return "", ""
+    x = np.asarray(data, dtype=np.float32)
+    if x.ndim > 1:
+        x = x.mean(axis=-1)
+    floats = x.reshape(-1).tolist()
+    return _get_pipeline().transcribe_chunk(floats, int(sr))
+
+
+with gr.Blocks(title="Italian speech → Italian + English") as demo:
+    gr.Markdown(
+        "### Italian → Italian + English\n"
+        "Speak or upload **Italian** audio. Output is **recognized Italian** and **English** translation "
+        "(Whisper + Marian). Optional fine-tuned Whisper in `models/whisper_finetuned_it/`."
+    )
+    audio_in = gr.Audio(
+        sources=["microphone", "upload"],
+        type="numpy",
+        label="Audio (Italian)",
+    )
+    run_btn = gr.Button("Transcribe", variant="primary")
+    out_it = gr.Textbox(label="Italian (ASR)", lines=4)
+    out_en = gr.Textbox(label="English (translation)", lines=4)
+
+    run_btn.click(fn=transcribe, inputs=[audio_in], outputs=[out_it, out_en])
+
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", "7860"))
+    demo.launch(server_name="0.0.0.0", server_port=port)

italian_en_pipeline.py

@@ -0,0 +1,277 @@
+"""
+Single pipeline: **Italian speech → Italian text + English translation** (Whisper + Marian).
+
+Standalone copy for Hugging Face Space (same folder as ``app.py``). ``project_root`` is this
+directory; optional fine-tuned Whisper lives under ``models/whisper_finetuned_it/``.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from collections import Counter
+from typing import Optional
+
+import numpy as np
+import torch
+
+import editdistance
+
+from whisper_asr import WhisperASR
+
+SAMPLE_RATE = 16_000
+
+
+def _pick_device() -> torch.device:
+    if torch.cuda.is_available():
+        return torch.device("cuda")
+    if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
+        return torch.device("mps")
+    return torch.device("cpu")
+
+
+_HF_ONLY_MODES = frozenset({"quality", "pretrained", "base", "hf", "huggingface"})
+
+
+def _default_realtime_mode(default_finetuned: str) -> str:
+    explicit = (os.environ.get("ASR_REALTIME_MODE") or "").strip().lower()
+    if explicit:
+        return explicit or "quality"
+    custom = (os.environ.get("ASR_WHISPER_FINETUNED_DIR") or "").strip() or default_finetuned
+    cfg = os.path.join(custom, "config.json")
+    vocab = os.path.join(custom, "vocab.json")
+    if os.path.isdir(custom) and os.path.isfile(cfg) and os.path.isfile(vocab):
+        return "finetuned"
+    return "quality"
+
+
+class ItalianEnglishPipeline:
+    """Whisper (Italian ASR) + Marian MT (IT→EN). One call returns both strings."""
+
+    def __init__(self, project_root: Optional[str] = None) -> None:
+        # Space root = directory containing this module (same as app.py).
+        _here = os.path.dirname(os.path.abspath(__file__))
+        self.project_root = project_root or _here
+        self.default_finetuned = os.path.join(self.project_root, "models", "whisper_finetuned_it")
+        self.device = _pick_device()
+
+        self.realtime_mode = _default_realtime_mode(self.default_finetuned)
+        self.whisper_model_name = (os.environ.get("ASR_WHISPER_MODEL") or "").strip() or "openai/whisper-small"
+        self.max_new = max(32, int(os.environ.get("ASR_WHISPER_MAX_NEW_TOKENS", "448")))
+        self.num_beams = max(1, int(os.environ.get("ASR_WHISPER_NUM_BEAMS", "5")))
+        self.min_rms = float(os.environ.get("ASR_MIN_RMS", "0.005"))
+
+        tr = (os.environ.get("ASR_TRANSLATE") or "1").strip().lower()
+        self.translate_enabled = tr not in ("0", "false", "no", "off")
+        self.translate_model_id = (os.environ.get("ASR_TRANSLATE_MODEL") or "Helsinki-NLP/opus-mt-it-en").strip()
+
+        self.lexicon: set[str] = self._load_lexicon_from_project()
+        self._whisper: Optional[WhisperASR] = None
+        self._whisper_failed = False
+        self._translator = None
+        self._translator_failed = False
+
+    def _load_lexicon_from_project(self) -> set[str]:
+        env_dir = (os.environ.get("ASR_WHISPER_FINETUNED_DIR") or "").strip()
+        for base in (env_dir or None, self.default_finetuned):
+            if not base:
+                continue
+            p = os.path.join(base, "lexicon.txt")
+            if os.path.isfile(p):
+                words: set[str] = set()
+                with open(p, "r", encoding="utf-8") as f:
+                    for line in f:
+                        w = line.strip().upper()
+                        if w:
+                            words.add(w)
+                return words
+        return set()
+
+    def resolve_whisper_model_id(self) -> str:
+        if self.realtime_mode in _HF_ONLY_MODES:
+            return self.whisper_model_name
+        custom = (os.environ.get("ASR_WHISPER_FINETUNED_DIR") or "").strip() or self.default_finetuned
+        if os.path.isdir(custom) and os.path.isfile(os.path.join(custom, "config.json")):
+            return custom
+        print(
+            f"[WARN] No fine-tuned Whisper at {custom}; falling back to {self.whisper_model_name}. "
+            "Use ASR_REALTIME_MODE=quality for pretrained-only."
+        )
+        return self.whisper_model_name
+
+    def _whisper_pipe(self) -> Optional[WhisperASR]:
+        if self._whisper_failed:
+            return None
+        if self._whisper is None:
+            try:
+                mid = self.resolve_whisper_model_id()
+                self._whisper = WhisperASR(mid, self.device)
+                self._whisper.load()
+            except Exception as e:
+                print(f"[WARN] Could not load Whisper ({e}).")
+                self._whisper_failed = True
+                return None
+        return self._whisper
+
+    def translate_it_to_en(self, text: str) -> str:
+        if self._translator_failed or not self.translate_enabled:
+            return ""
+        t = (text or "").strip()
+        if not t:
+            return ""
+        if self._translator is None:
+            try:
+                from transformers import pipeline
+
+                self._translator = pipeline(
+                    "translation",
+                    model=self.translate_model_id,
+                    device=-1,
+                )
+                print(f"[Pipeline] Translation loaded: {self.translate_model_id}")
+            except Exception as e:
+                self._translator_failed = True
+                print(f"[WARN] Translation pipeline failed ({e})")
+                return ""
+        try:
+            out = self._translator(t, max_length=512, clean_up_tokenization_spaces=True)
+            if isinstance(out, list) and out:
+                return (out[0].get("translation_text") or "").strip()
+            return ""
+        except Exception as e:
+            print(f"[WARN] Translation failed: {e}")
+            return ""
+
+    @staticmethod
+    def _likely_repetition_hallucination(text: str) -> bool:
+        words = [w for w in re.split(r"\s+", text.strip()) if w]
+        if len(words) < 12:
+            return False
+        stripped = [re.sub(r"^[^\w]+|[^\w]+$", "", w, flags=re.UNICODE).lower() for w in words]
+        stripped = [w for w in stripped if w]
+        if len(stripped) < 12:
+            return False
+        c = Counter(stripped)
+        return c.most_common(1)[0][1] / len(stripped) >= 0.55
+
+    def _correct_word(self, word: str, max_dist: int = 2) -> str:
+        if not self.lexicon or not word:
+            return word
+        w = word.upper()
+        if w in self.lexicon:
+            return word
+        if len(w) < 3:
+            return word
+        best = w
+        best_d = max_dist + 1
+        for cand in self.lexicon:
+            if abs(len(cand) - len(w)) > 2:
+                continue
+            if len(w) >= 5 and (not cand or cand[0] != w[0]):
+                continue
+            d = editdistance.eval(w, cand)
+            if d < best_d:
+                best_d = d
+                best = cand
+                if d == 0:
+                    break
+        if best_d <= max_dist:
+            return best
+        return word
+
+    def _correct_text(self, text: str) -> str:
+        if not text or not self.lexicon:
+            return text
+        words = text.split()
+        corrected = [self._correct_word(w, max_dist=1) for w in words]
+        merged: list[str] = []
+        i = 0
+        while i < len(corrected):
+            if i + 1 < len(corrected):
+                w1 = corrected[i]
+                w2 = corrected[i + 1]
+                joined = (w1 + w2).upper()
+                if joined in self.lexicon:
+                    merged.append(joined)
+                    i += 2
+                    continue
+            merged.append(corrected[i])
+            i += 1
+        return " ".join(merged)
+
+    def _whisper_transcribe(self, audio_16k: np.ndarray) -> str:
+        pipe = self._whisper_pipe()
+        if pipe is None:
+            return ""
+        if audio_16k.ndim != 1:
+            audio_16k = audio_16k.reshape(-1)
+        raw = pipe.transcribe(
+            audio_16k,
+            SAMPLE_RATE,
+            max_new_tokens=self.max_new,
+            num_beams=self.num_beams,
+        )
+        return (raw or "").strip()
+
+    def _preprocess_chunk(self, audio_float32: list[float], sample_rate: int) -> Optional[np.ndarray]:
+        import torchaudio
+
+        arr = np.array(audio_float32, dtype=np.float32)
+        if len(arr) < 800:
+            return None
+        if sample_rate != SAMPLE_RATE:
+            t = torch.from_numpy(arr).float().unsqueeze(0).unsqueeze(0)
+            t = torchaudio.functional.resample(t, sample_rate, SAMPLE_RATE)
+            arr = t.squeeze().numpy()
+        arr = (
+            torchaudio.functional.highpass_biquad(
+                torch.from_numpy(arr).float().unsqueeze(0), SAMPLE_RATE, 80.0
+            )
+            .squeeze(0)
+            .numpy()
+        )
+        rms = float(np.sqrt(np.mean(arr**2))) if arr.size > 0 else 0.0
+        if rms < self.min_rms:
+            return None
+        return arr
+
+    def transcribe_chunk(self, audio_float32: list[float], sample_rate: int) -> tuple[str, str]:
+        """
+        Returns ``(italian_text, english_text)``. Empty strings if silence / rejected.
+        """
+        arr = self._preprocess_chunk(audio_float32, sample_rate)
+        if arr is None:
+            return "", ""
+
+        raw = self._whisper_transcribe(arr)
+        text = (raw or "").strip()
+        if not text:
+            return "", ""
+        if self._likely_repetition_hallucination(text):
+            return "", ""
+
+        en = self.translate_it_to_en(text)
+        if self.lexicon:
+            text = self._correct_text(text.upper())
+        return text, en
+
+    def format_remote_response(self, italian_raw: str) -> tuple[str, str]:
+        """After a remote ASR returns Italian text: apply lexicon + translation."""
+        text = (italian_raw or "").strip()
+        if not isinstance(text, str):
+            text = str(text)
+        raw_for_tr = text
+        if text and self.lexicon:
+            text = self._correct_text(text.upper())
+        en = self.translate_it_to_en(raw_for_tr) if raw_for_tr else ""
+        return text, en
+
+
+_pipeline_singleton: Optional[ItalianEnglishPipeline] = None
+
+
+def get_pipeline(project_root: Optional[str] = None) -> ItalianEnglishPipeline:
+    global _pipeline_singleton
+    if _pipeline_singleton is None:
+        _pipeline_singleton = ItalianEnglishPipeline(project_root=project_root)
+    return _pipeline_singleton

models/whisper_finetuned_it/README.md

@@ -0,0 +1,17 @@
+# Fine-tuned Whisper (optional)
+
+Place your Italian fine-tuned Whisper checkpoint **here** (this folder), with at least:
+
+- `config.json`
+- `vocab.json` (or tokenizer files your training run produced)
+- Model weights (e.g. `pytorch_model.bin` or safetensors)
+
+Optional: `lexicon.txt` — one uppercase word per line for light post-correction.
+
+When this folder looks like a valid Hugging Face `from_pretrained` directory, set in the Space **Settings → Variables**:
+
+- `ASR_REALTIME_MODE=finetuned`
+
+If the folder is empty or invalid, the app falls back to `ASR_WHISPER_MODEL` (default `openai/whisper-small`) with `ASR_REALTIME_MODE=quality`.
+
+**Large files:** use [Git LFS](https://git-lfs.com) on the Space repo before pushing weights.

requirements.txt

@@ -0,0 +1,8 @@
+gradio>=4.44.0
+torch>=2.4.0
+torchaudio>=2.4.0
+transformers>=4.46.0
+accelerate>=1.0.0
+numpy>=2.1.0
+sentencepiece>=0.1.99
+editdistance>=0.8.0

whisper_asr.py

@@ -0,0 +1,149 @@
+"""
+Speech-to-text using **OpenAI Whisper** weights on Hugging Face (open weights, free to use).
+
+Models: https://huggingface.co/openai — e.g. ``openai/whisper-tiny``, ``whisper-base``, ``whisper-small``.
+
+Language for generation is set by the ``language`` argument or ``ASR_WHISPER_LANGUAGE``
+(e.g. ``italian``, ``english``); Whisper expects the full language name in English.
+"""
+
+from __future__ import annotations
+
+import math
+import os
+from typing import Optional
+
+import numpy as np
+import torch
+import torchaudio
+from transformers import WhisperForConditionalGeneration, WhisperProcessor
+
+SAMPLE_RATE = 16_000
+
+
+def pick_device() -> torch.device:
+    if torch.cuda.is_available():
+        return torch.device("cuda")
+    if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
+        return torch.device("mps")
+    return torch.device("cpu")
+
+
+class WhisperASR:
+    """Lazy-loadable Whisper STT (beam decode when supported)."""
+
+    def __init__(
+        self,
+        model_name: str,
+        device: Optional[torch.device] = None,
+        *,
+        language: Optional[str] = None,
+    ):
+        self.model_name = model_name
+        self.device = device or pick_device()
+        raw = (language or os.environ.get("ASR_WHISPER_LANGUAGE") or "italian").strip().lower()
+        self.language = raw or "italian"
+        self._model: Optional[WhisperForConditionalGeneration] = None
+        self._processor: Optional[WhisperProcessor] = None
+
+    def load(self) -> None:
+        if self._model is not None:
+            return
+        self._processor = WhisperProcessor.from_pretrained(self.model_name)
+        self._model = WhisperForConditionalGeneration.from_pretrained(self.model_name).to(
+            self.device
+        )
+        self._model.eval()
+        print(f"[WhisperASR] Loaded {self.model_name} on {self.device}")
+
+    @property
+    def model(self) -> WhisperForConditionalGeneration:
+        self.load()
+        assert self._model is not None
+        return self._model
+
+    @property
+    def processor(self) -> WhisperProcessor:
+        self.load()
+        assert self._processor is not None
+        return self._processor
+
+    def _cap_max_new_tokens(self, requested: int) -> int:
+        """Whisper caps total decoder length at ``max_target_positions``; prefix tokens count too."""
+        lim = int(getattr(self.model.config, "max_target_positions", 448))
+        # Room for language/task start tokens (transformers warns if max_new_tokens == lim exactly).
+        margin = 24
+        return max(1, min(requested, lim - margin))
+
+    @torch.no_grad()
+    def transcribe(
+        self,
+        waveform: np.ndarray,
+        sample_rate: int,
+        *,
+        max_new_tokens: int = 440,
+        num_beams: int = 5,
+        no_speech_threshold: float | None = None,
+        compression_ratio_threshold: float | None = None,
+        logprob_threshold: float | None = None,
+    ) -> str:
+        """``waveform`` mono float32; any sample rate (resampled to 16 kHz)."""
+        w = np.asarray(waveform, dtype=np.float32).reshape(-1)
+        if w.size == 0:
+            return ""
+        if not math.isfinite(float(np.max(np.abs(w)))):
+            return ""
+        sr = int(sample_rate)
+        t = torch.from_numpy(w).unsqueeze(0)
+        if sr != SAMPLE_RATE:
+            t = torchaudio.functional.resample(t, sr, SAMPLE_RATE)
+        audio_16k = t.squeeze(0).numpy()
+
+        inputs = self.processor(audio_16k, sampling_rate=SAMPLE_RATE, return_tensors="pt")
+        input_features = inputs["input_features"].to(self.device)
+        attention_mask = inputs.get("attention_mask")
+        if attention_mask is not None:
+            attention_mask = attention_mask.to(self.device)
+
+        cap = self._cap_max_new_tokens(max_new_tokens)
+
+        # Whisper-specific thresholds reduce silence/noise hallucinations (HF transformers).
+        # Defaults match OpenAI-ish behavior; stricter than "unset" when backends ignore None.
+        if no_speech_threshold is None:
+            no_speech_threshold = float(os.environ.get("ASR_NO_SPEECH_THRESHOLD", "0.65"))
+        if compression_ratio_threshold is None:
+            compression_ratio_threshold = float(os.environ.get("ASR_COMPRESSION_RATIO_THRESHOLD", "2.0"))
+        if logprob_threshold is None:
+            logprob_threshold = float(os.environ.get("ASR_LOGPROB_THRESHOLD", "-1.0"))
+
+        gen_common: dict = {
+            "max_new_tokens": cap,
+            "num_beams": num_beams,
+            "do_sample": False,
+            "language": self.language,
+            "task": "transcribe",
+            "no_speech_threshold": no_speech_threshold,
+            "compression_ratio_threshold": compression_ratio_threshold,
+            "logprob_threshold": logprob_threshold,
+        }
+        if attention_mask is not None:
+            gen_common["attention_mask"] = attention_mask
+
+        try:
+            ids = self.model.generate(input_features, **gen_common)
+        except TypeError:
+            gen_common.pop("no_speech_threshold", None)
+            gen_common.pop("compression_ratio_threshold", None)
+            gen_common.pop("logprob_threshold", None)
+            try:
+                ids = self.model.generate(input_features, **gen_common)
+            except TypeError:
+                ids = self.model.generate(
+                    input_features,
+                    max_new_tokens=cap,
+                    num_beams=num_beams,
+                    do_sample=False,
+                )
+
+        text = self.processor.batch_decode(ids, skip_special_tokens=True)[0]
+        return (text or "").strip()