diff --git a/ffmpeg-service/app/align.py b/ffmpeg-service/app/align.py index 0662653..a33429f 100644 --- a/ffmpeg-service/app/align.py +++ b/ffmpeg-service/app/align.py @@ -116,12 +116,22 @@ def _model(): return WhisperModel(config.WHISPER_MODEL, device="cpu", compute_type=config.WHISPER_COMPUTE_TYPE) +# Silence ajouté avant la voix pour Whisper : sur une voix qui démarre +# aussitôt, il laissait souvent passer le premier mot (« Alerte »). +LEAD_SILENCE = 0.5 + + def _transcribe_sync(audio: Path) -> list[Word]: + import numpy as np + from faster_whisper.audio import decode_audio + + samples = decode_audio(str(audio), sampling_rate=16000) + padded = np.concatenate([np.zeros(int(16000 * LEAD_SILENCE), dtype=samples.dtype), samples]) # Pas de texte attendu en amorce (initial_prompt) ni d'enchaînement sur le # segment précédent : les deux font halluciner Whisper, qui répète la phrase # et masque une fin tronquée. On veut entendre ce qui est réellement dit. segments, _ = _model().transcribe( - str(audio), + padded, language="fr", word_timestamps=True, condition_on_previous_text=False, @@ -129,7 +139,7 @@ def _transcribe_sync(audio: Path) -> list[Word]: beam_size=5, ) return [ - Word(w.word.strip(), float(w.start), float(w.end)) + Word(w.word.strip(), max(0.0, w.start - LEAD_SILENCE), max(0.0, w.end - LEAD_SILENCE)) for segment in segments for w in (segment.words or []) if w.word.strip() diff --git a/ffmpeg-service/app/quality.py b/ffmpeg-service/app/quality.py index d594f19..f5944ce 100644 --- a/ffmpeg-service/app/quality.py +++ b/ffmpeg-service/app/quality.py @@ -49,8 +49,8 @@ def check_reading(expected_text: str, spoken: list[Word]) -> ReadingCheck: lettre : les petites erreurs de reconnaissance (« petit » pour « petits », « week -end » découpé) ne comptent pas comme des mots manquants, alors qu'un mot sauté, une fin tronquée ou une phrase ajoutée se voient.""" - expected = "".join(normalize(w) for w in expected_text.split()) - heard = "".join(normalize(w.text) for w in spoken) + expected = "".join(comparable_tokens(expected_text.split())) + heard = "".join(comparable_tokens([w.text for w in spoken])) if not expected: return ReadingCheck(1.0, 0.0, True) if not heard: @@ -70,13 +70,44 @@ def check_reading(expected_text: str, spoken: list[Word]) -> ReadingCheck: ) +# Unités écrites de plusieurs façons par Whisper (« 40 cm » pour « 40 centimètres ») +UNITS = { + "e", "eur", "euro", "euros", "centime", "centimes", "cm", "centimetre", "centimetres", + "mm", "millimetre", "millimetres", "m", "metre", "metres", "km", "kilometre", "kilometres", + "g", "gramme", "grammes", "kg", "kilo", "kilos", "kilogramme", "kilogrammes", + "l", "litre", "litres", "cl", "centilitre", "centilitres", "ml", "millilitre", "millilitres", + "pourcent", "pourcents", "x", "fois", +} # fmt: skip + + +def comparable_tokens(words: list[str]) -> list[str]: + """Mots normalisés, sans les nombres ni l'unité qui les suit : Whisper écrit + un prix ou une mesure à sa façon (« 11 € 99 », « 11,99 euros », « 40 cm »), + ce qui passait pour des mots sautés et relançait la génération.""" + tokens: list[str] = [] + after_number = False + for word in words: + token = normalize(word) + if not token: + continue + if any(c.isdigit() for c in token): + after_number = True + continue + if after_number and token in UNITS: + continue + after_number = False + tokens.append(token) + return tokens + + def _missing_words(expected_text: str, spoken: list[Word]) -> list[str]: """Mots attendus sans équivalent proche dans ce qui a été entendu.""" - heard = {normalize(w.text) for w in spoken} + heard = set(comparable_tokens([w.text for w in spoken])) + kept = set(comparable_tokens(expected_text.split())) missing = [] for word in expected_text.split(): token = normalize(word) - if token and not any(difflib.SequenceMatcher(a=token, b=h).ratio() >= 0.7 for h in heard): + if token in kept and not any(difflib.SequenceMatcher(a=token, b=h).ratio() >= 0.7 for h in heard): missing.append(word.strip(".,!?;:…")) return missing @@ -90,6 +121,10 @@ ACTIVE_RANGE_DB = 38.0 # Marges gardées autour de la parole LEAD_PAD = 0.08 TAIL_PAD = 0.30 +# Pause entre deux phrases : au moins 80 ms de silence, cherchée jusqu'à 1,2 s +# de la borne estimée par Whisper +MIN_PAUSE = 0.08 +PAUSE_SEARCH = 1.2 @dataclass @@ -139,6 +174,19 @@ class Envelope: active = np.nonzero(self.db[first:last] > self.threshold)[0] return (first + int(active[-1]) + 1) * FRAME if len(active) else None + def pauses(self, min_length: float = MIN_PAUSE) -> list[tuple[float, float]]: + """Silences (sous le seuil d'audibilité) d'au moins min_length secondes.""" + quiet = self.db <= self.threshold + runs, run_start = [], None + for i, is_quiet in enumerate([*quiet, False]): + if is_quiet and run_start is None: + run_start = i + elif not is_quiet and run_start is not None: + if (i - run_start) * FRAME >= min_length: + runs.append((run_start * FRAME, i * FRAME)) + run_start = None + return runs + def speech_span(self, start: float, end: float) -> tuple[float, float]: """Parole contenue dans [start, end], avec ses marges, sans en sortir.""" onset, offset = self.first_sound(start, end), self.last_sound(start, end) @@ -170,9 +218,26 @@ async def load_envelope(path: Path) -> Envelope: return envelope_from_samples(np.frombuffer(raw, dtype=np.float32), 16000) -def sentence_cut(envelope: Envelope, previous_end: float, next_start: float) -> float: - """Coupe entre deux phrases, au point le plus calme de la pause. Les bornes - Whisper sont imprécises : on cherche un peu autour.""" - low = max(0.0, min(previous_end, next_start) - 0.1) - high = max(previous_end, next_start) + 0.2 +def sentence_cut(envelope: Envelope, previous_end: float, next_start: float, after: float = 0.0) -> float: + """Coupe entre deux phrases, au cœur d'une vraie pause. + + Les bornes Whisper sont imprécises, et fausses quand un mot n'a pas été + reconnu (un prix en fin de phrase : sa place est alors estimée) ; couper + à la borne estimée tombait parfois au milieu d'une phrase. On choisit donc + la pause mesurée la plus proche de la borne estimée, les longues pauses + (fin de phrase) étant préférées aux courtes (virgule).""" + low = max(after, min(previous_end, next_start) - 0.1) + high = max(low + 0.05, max(previous_end, next_start) + 0.2) + best: tuple[float, float, float] | None = None + for start, end in envelope.pauses(): + if end <= after + 0.05 or start <= 0.0 or end >= envelope.duration: + continue # silences avant la première et après la dernière phrase exclus + distance = max(0.0, low - end, start - high) + if distance > PAUSE_SEARCH: + continue + score = distance - (end - start) + if best is None or score < best[0]: + best = (score, max(start, after), end) + if best: + return envelope.quietest(best[1], best[2]) return envelope.quietest(low, high) diff --git a/ffmpeg-service/app/srt_voice.py b/ffmpeg-service/app/srt_voice.py index fc094ee..171a2e5 100644 --- a/ffmpeg-service/app/srt_voice.py +++ b/ffmpeg-service/app/srt_voice.py @@ -103,7 +103,7 @@ def segments_from_reading(groups: list[list[Word]], envelope: quality.Envelope) marge (300 ms après le dernier son : les fins de phrases ne sont pas rognées).""" cuts = [0.0] for previous, following in zip(groups, groups[1:]): - cuts.append(quality.sentence_cut(envelope, previous[-1].end, following[0].start)) + cuts.append(quality.sentence_cut(envelope, previous[-1].end, following[0].start, after=cuts[-1])) cuts.append(envelope.duration) if any(b <= a for a, b in zip(cuts, cuts[1:])): return None diff --git a/ffmpeg-service/tests/test_quality.py b/ffmpeg-service/tests/test_quality.py index f1b6958..ebd4fef 100644 --- a/ffmpeg-service/tests/test_quality.py +++ b/ffmpeg-service/tests/test_quality.py @@ -71,3 +71,35 @@ def test_skipped_words_are_detected_and_named(): assert not check.acceptable assert "bonbons" in check.missing and "effrayante" in check.missing assert "non entendus" in check.describe() + + +def test_prices_and_units_written_differently_are_not_missing_words(): + expected = "Alerte léopard ! Le coussin de 40 centimètres à 11,99 €. Le cabas à 13,99 €." + heard = _heard("Alerte léopard ! Le coussin de 40 cm à 11 € 99. Le cabas à 13,99 euros.") + check = check_reading(expected, heard) + assert check.acceptable, check + assert check.missing == [] + + +def test_skipped_word_next_to_a_price_is_still_detected(): + expected = "Le coussin léopard à 11,99 €. Le cabas doux et chaud à 13,99 €." + heard = _heard("Le coussin à 11,99 €. Le cabas à 13,99 €.") + check = check_reading(expected, heard) + assert not check.acceptable + assert "léopard" in check.missing + + +def test_cut_uses_the_real_pause_when_word_timing_is_estimated(): + # Phrase 1 jusqu'à 2,0 s (dont un prix non reconnu), virgule à 1,0 s, + # pause de fin de phrase 2,0–2,4 s. Whisper place la fin de la phrase + # et le début de la suivante vers 2,9 s (mots interpolés). + env = _envelope([(0.3, -90), (0.7, -20), (0.1, -90), (0.9, -20), (0.4, -90), (1.5, -20), (0.3, -90)]) + cut = sentence_cut(env, previous_end=2.9, next_start=2.95) + assert 2.0 <= cut <= 2.4 + + +def test_cut_prefers_the_sentence_pause_over_a_comma(): + env = _envelope([(0.3, -90), (1.0, -20), (0.12, -90), (0.8, -20), (0.45, -90), (1.0, -20), (0.3, -90)]) + # Borne estimée entre la virgule (1,3 s) et la fin de phrase (2,22 s) + cut = sentence_cut(env, previous_end=1.7, next_start=1.75) + assert 2.22 <= cut <= 2.67