Flash NLP : pipeline de transcription audio et analyse NLP en Python

Publié le 10 mars 2026 — Par l'équipe DEV-AI

Publié en mars 2026 — temps de lecture : 20 min — par l'équipe DEV-AI

Sur DEV-AI, notre outil de transcription audio repose sur Whisper pour convertir vos fichiers audio en texte. Mais que se passe-t-il quand on veut aller plus loin — capturer des flux audio en continu, en extraire des événements structurés et déclencher des alertes automatiques ?

C'est exactement ce que réalise Flash NLP, un projet open-source Python de Tatiana T. On l'analyse ici en intégralité : architecture, choix techniques, code commenté, tests.

Architecture globale

Le pipeline est découpé en quatre couches indépendantes et testables :

Packaging et dépendances

Le projet est un package Python installable via pyproject.toml (PEP 517/518). Python 3.11+ est requis pour zoneinfo (stdlib) et les annotations de type modernes :

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "flash-nlp"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "faster-whisper>=1.0.3",
    "numpy>=1.26",
    "scipy>=1.11",
    "sounddevice>=0.5",
    "requests>=2.31",
]

[project.optional-dependencies]
gui = ["PySide6>=6.7"]

[tool.setuptools.packages.find]
where = ["src"]

pip install -e ".[gui]"

Couche 1 — Acquisition de flux audio

Polling conditionnel HTTP

Le module fetcher.py utilise les headers HTTP conditionnels pour ne re-télécharger que si le contenu a changé. Un pattern fondamental dans tout système de polling :

headers = {}
if st.get("etag"):
    headers["If-None-Match"] = st["etag"]
if st.get("lm"):
    headers["If-Modified-Since"] = st["lm"]

r = requests.get(url, headers=headers, timeout=30)

if r.status_code == 304:
    continue  # contenu inchangé, on skip

ETag identifie une version côté serveur. If-Modified-Since déclenche un 304 Not Modified si rien n'a changé. Ce pattern réduit drastiquement la bande passante dans un contexte de polling fréquent.

Déduplification par hash MD5

Seconde ligne de défense si le serveur ne supporte pas les headers conditionnels :

def dedupe_by_md5(day_dir: Path, new_md5: str) -> bool:
    existing = sorted(day_dir.glob("flash_*.mp3"))
    latest = existing[-1] if existing else None
    if latest is not None and latest.exists():
        with latest.open("rb") as lf:
            if md5_bytes(lf.read()) == new_md5:
                return True
    return False

Seul le dernier fichier (tri alphabétique du timestamp) est comparé — suffisant car les doublons sont toujours consécutifs. Un filtre de taille minimum rejette les réponses invalides :

if len(content) < 10_000:
    continue  # moins de 10 Ko = contenu invalide

Organisation de l'archive

Les fichiers sont archivés en hiérarchie date/zone/ avec un nom encodant le timestamp :

data/flash_audio_archive/
├── index.csv
├── state.json
└── 2026-01-23/
    ├── nord/
    │   └── flash_nord_20260123_164916.mp3
    ├── sud/
    │   └── flash_sud_20260123_164917.mp3
    └── ouest/
        └── flash_ouest_20260123_164918.mp3

Chaque fichier est loggé dans index.csv avec deux timestamps : local (Europe/Paris) et UTC, pour éviter toute ambiguïté :

datetime_local;datetime_utc;zone;filename;filesize;md5
2026-01-23 16:49:16+0100;2026-01-23 15:49:16;nord;2026-01-23/nord/flash_nord_20260123_164916.mp3;1237850;5ad694...

Rotation automatique des fichiers

def rotate(path: Path, keep_days: int, tz) -> None:
    if keep_days <= 0:
        return
    limit = now_local(tz) - dt.timedelta(days=keep_days)
    for f in path.glob("**/*.mp3"):
        parts = f.stem.rsplit("_", 2)
        stamp = parts[-2] + "_" + parts[-1]
        d = parse_stamp_to_dt(stamp, tz)
        if d < limit:
            f.unlink(missing_ok=True)

Le parsing du timestamp depuis le nom de fichier supporte deux formats : avec et sans offset timezone (%Y%m%d_%H%M%z et %Y%m%d_%H%M).

Couche 2 — Transcription avec faster-whisper

faster-whisper est une réimplémentation de Whisper basée sur CTranslate2, significativement plus rapide que l'originale d'OpenAI, avec support natif de la quantification int8. C'est le même moteur que notre outil /upload.

Conversion audio avec ffmpeg

Whisper attend du WAV mono 16 kHz PCM 16-bit. La conversion est déléguée à ffmpeg via subprocess — ce qui supporte nativement mp3, m4a, aac, flac, ogg, webm et tous les formats gérés par ffmpeg :

cmd = [
    which_ffmpeg(),
    "-y",                  # overwrite sans confirmation
    "-i", src_path,
    "-vn",                 # pas de piste vidéo
    "-ac", "1",            # mono
    "-ar", "16000",        # 16 kHz
    "-c:a", "pcm_s16le",   # PCM 16-bit little-endian
    dst_path,
]
p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if p.returncode != 0:
    raise RuntimeError(f"ffmpeg conversion échouée:\n{p.stderr}")

La fonction rms() mesure le niveau RMS d'un signal pour détecter les silences avant transcription :

def rms(x: np.ndarray) -> float:
    if x.size == 0:
        return 0.0
    return float(np.sqrt(np.mean(x.astype(np.float32) ** 2)))

WhisperService — cache de modèle

La classe WhisperService implémente un cache — le modèle n'est rechargé que si les paramètres changent :

def load(self, model_name: str, device: str = "cpu", compute_type: Optional[str] = None):
    if compute_type is None:
        compute_type = "int8" if device == "cpu" else "float16"

    # Cache : ne recharge que si les paramètres changent
    if (
        self._model is not None
        and self._model_name == model_name
        and self._device == device
        and self._compute_type == compute_type
    ):
        return

    self._model = WhisperModel(model_name, device=device, compute_type=compute_type)

Sur CPU, compute_type="int8" réduit la mémoire et accélère l'inférence. Sur GPU, "float16" est utilisé automatiquement.

VAD — Voice Activity Detection

Le filtre VAD intégré à faster-whisper ignore les segments silencieux. C'est un garde-fou essentiel : Whisper hallucine du texte sur les silences prolongés.

segments, info = self._model.transcribe(
    wav_path,
    language=language,
    beam_size=beam_size,
    vad_filter=True,
    vad_parameters={"min_silence_duration_ms": 350},
)

Deux modes de sortie

Mode simple — retourne un tuple (texte, langue, probabilité) :

return " ".join(texts), info.language, float(info.language_probability)

Mode segmenté — retourne un dict structuré avec les timestamps de chaque segment, utile pour localiser précisément un événement ou générer des sous-titres :

return {
    "language": info.language,
    "language_probability": float(info.language_probability),
    "duration": float(info.duration),
    "segments": [{"start": s.start, "end": s.end, "text": text} for s in segs],
    "text": " ".join(texts),
}

Quel modèle choisir ?

Couche 3 — Analyse NLP par expressions régulières

Pourquoi du NLP symbolique et pas un modèle entraîné ?

Le choix est délibéré : pour un domaine métier à vocabulaire contraint (ici le trafic routier), les expressions régulières offrent une précision prévisible, une latence nulle, aucun GPU requis et un code facilement auditable. Un NER entraîné serait over-engineering pour ce cas d'usage.

Taxonomie des événements

_SEVERITY: dict[str, str] = {
    "accident":        "high",
    "fermeture":       "high",
    "bouchon":         "medium",
    "animal":          "medium",
    "intemperies":     "medium",
    "ralentissement":  "low",
    "travaux":         "low",
    "vehicule_panne":  "low",
}

Patterns de détection

Chaque type est détecté par une regex compilée couvrant les variantes orthographiques et morphologiques françaises. Points techniques clés :

• \b : word boundary, évite les faux positifs dans les mots composés
• (?:é|er)? : groupe non-capturant pour les variantes de conjugaison
• [cç] : classe de caractères pour les accents alternants
• re.IGNORECASE : essentiel — les transcriptions Whisper peuvent être en majuscules

"accident": re.compile(
    r'\b(accident|collision|accrochage|carambolage|heurt(?:é|er)?|percuté|renversé)\b',
    re.IGNORECASE,
),
"intemperies": re.compile(
    r'\b(neige|verglas|brouillard|pluie\s+vergla[cç]ante|givr[eé]|black.ice|'
    r'alerte\s+orange|alerte\s+rouge|conditions\s+hivernales|chaine|pneus?\s+hiver)\b',
    re.IGNORECASE,
),

Extraction contextuelle (±80 caractères)

_CONTEXT_RADIUS = 80

def _extract_context(text: str, match_start: int, match_end: int) -> str:
    start = max(0, match_start - _CONTEXT_RADIUS)
    end = min(len(text), match_end + _CONTEXT_RADIUS)
    snippet = text[start:end].replace("\n", " ").strip()
    if start > 0:
        snippet = "..." + snippet
    if end < len(text):
        snippet = snippet + "..."
    return snippet

Extraction de routes, directions et délais

_ROUTE_RE = re.compile(
    r'\b(A\d+|N\d+|D\d+|RN\d+|RD\d+|p[eé]riph[eé]rique|rocade)\b',
    re.IGNORECASE,
)
_DIRECTION_RE = re.compile(
    r'\b(sens\s+\w+(?:\s+\w+)?|direction\s+\w+(?:\s+\w+)?|vers\s+\w+(?:\s+\w+)?|'
    r'entre\s+\w+\s+et\s+\w+)\b',
    re.IGNORECASE,
)
_DELAY_RE = re.compile(
    r'(\d+)\s*(?:minute|min|km\s+de\s+bouchon|kilomètre)',
    re.IGNORECASE,
)

Modèle de données

@dataclass
class TrafficEvent:
    type: str           # "accident", "bouchon", etc.
    severity: str       # "high", "medium", "low"
    routes: List[str]   # ["A6", "N7"]
    direction: str      # "sens Paris"
    location_hint: str  # extrait ±80 chars autour du match
    zone: str           # "nord", "sud", "ouest"
    timestamp: str      # "20260123_1649"
    source_file: str    # chemin relatif du MP3
    delay_hint: str     # "20 minutes"

    def as_dict(self) -> dict: ...

Fonction principale d'extraction

def extract_events(text: str, zone: str, source_file: str, timestamp: str) -> List[TrafficEvent]:
    events = []
    routes = _extract_routes(text)
    direction = _extract_direction(text)
    delay = _extract_delay(text)

    for event_type, pattern in _PATTERNS.items():
        match = pattern.search(text)
        if not match:
            continue
        hint = _extract_context(text, match.start(), match.end())
        events.append(TrafficEvent(
            type=event_type,
            severity=_SEVERITY[event_type],
            routes=routes,
            direction=direction,
            location_hint=hint,
            zone=zone,
            timestamp=timestamp,
            source_file=source_file,
            delay_hint=delay,
        ))

    return events

Routes, direction et délai sont extraits une seule fois en global sur le texte, puis partagés entre tous les événements — correct car un flash audio porte généralement sur une zone géographique cohérente.

Couche 4 — Notification multi-canaux

Console

def notify_console(event: TrafficEvent) -> None:
    icon = _SEVERITY_ICON.get(event.severity, "⚪")
    print(
        f"{icon} [{event.timestamp}] {event.zone.upper():5s} | "
        f"{event.type.upper():18s} | {routes}{direction}{delay}\n"
        f"   {event.location_hint}",
        flush=True,
    )

Notification macOS native (osascript)

def notify_macos(event: TrafficEvent) -> None:
    script = f'display notification "{body_safe}" with title "{title_safe}"'
    subprocess.run(["osascript", "-e", script], check=False,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

L'échappement des guillemets dans le corps AppleScript est géré manuellement pour éviter l'injection de commandes. La fonction est silencieuse si osascript est absent (Linux, CI).

Webhook HTTP

def notify_webhook(event: TrafficEvent, url: str) -> None:
    payload = event.as_dict()
    payload["alerted_at"] = datetime.now(timezone.utc).isoformat()
    _requests.post(url, json=payload, timeout=5)

Compatible avec n'importe quel endpoint HTTP : Slack, Discord, n8n, Make, etc.

Log JSONL

def log_to_file(event: TrafficEvent, alerts_dir: Path) -> None:
    line = event.as_dict()
    line["alerted_at"] = datetime.now(timezone.utc).isoformat()
    with (alerts_dir / "alerts.jsonl").open("a", encoding="utf-8") as f:
        f.write(json.dumps(line, ensure_ascii=False) + "\n")

Le format JSONL (une ligne = un objet JSON) est optimal pour les logs : append-only, parsable avec jq, pandas ou DuckDB sans charger tout le fichier.

Dispatch centralisé

def dispatch(event, alerts_dir, macos=False, webhook_url=None) -> None:
    notify_console(event)
    log_to_file(event, alerts_dir)
    if macos:
        notify_macos(event)
    if webhook_url:
        notify_webhook(event, webhook_url)

Stratégie de tests

Le projet dispose de 5 fichiers de tests couvrant chaque couche indépendamment avec pytest et pytest-mock.

Test de l'extracteur NLP

_TEXT_ACCIDENT = (
    "Autoroute Info, bonjour. Un accident sur l'A6 sens Paris au niveau du km 34, "
    "deux véhicules impliqués, circulation très ralentie. Comptez 20 minutes de retard."
)

def test_detects_accident():
    events = extract_events(_TEXT_ACCIDENT, zone="nord", source_file="f.mp3", timestamp="20260122_1000")
    assert "accident" in [e.type for e in events]

def test_no_false_positive_neutral_text():
    events = extract_events(
        "Le réseau autoroutier est fluide sur l'ensemble du territoire.",
        zone="nord", source_file="f.mp3", timestamp="20260122_1000"
    )
    assert events == []

Test du fetcher avec mock HTTP

def test_fetch_once_conditional_deduplicates(tmp_path, mocker):
    content = b"\xff" * 15_000
    mocker.patch(
        "flash_nlp.acquisition.fetcher.requests.get",
        return_value=_make_response(200, content),
    )
    cond_state = {}
    added1 = fetch_once_conditional(tmp_path, keep_days=30, tz=TZ_PARIS, cond_state=cond_state)
    added2 = fetch_once_conditional(tmp_path, keep_days=30, tz=TZ_PARIS, cond_state=cond_state)
    assert added1 == 3  # 3 zones → 3 fichiers
    assert added2 == 0  # doublon MD5 → rien sauvegardé

Test de WhisperService avec mock modèle

def test_transcribe_wav_passes_correct_args(mocker):
    _, mock_instance = _mock_model(mocker)
    svc = WhisperService()
    svc.load("small")
    svc.transcribe_wav("audio.wav", language="fr", beam_size=3)

    mock_instance.transcribe.assert_called_once_with(
        "audio.wav",
        language="fr",
        beam_size=3,
        vad_filter=True,
        vad_parameters={"min_silence_duration_ms": 350},
    )

Tests des utilitaires audio

def test_rms_known_sine():
    # RMS d'un sinus d'amplitude A = A / sqrt(2)
    t = np.linspace(0, 2 * np.pi, 10000)
    x = np.sin(t).astype(np.float32)
    assert abs(rms(x) - 1.0 / np.sqrt(2)) < 1e-3

def test_save_wav_clips_overflow(tmp_path):
    audio = np.array([2.0, -3.0, 0.5], dtype=np.float32)
    save_wav(str(tmp_path / "out.wav"), audio, sr=16000)
    _, data = wav_read(str(tmp_path / "out.wav"))
    assert data[0] == 32767   # clip positif
    assert data[1] == -32767  # clip négatif

Configuration pytest

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-v --tb=short"

pip install pytest pytest-mock
pytest

Modèles TTS embarqués (Meta MMS)

Le projet embarque des modèles TTS de Meta MMS (Massively Multilingual Speech) pour l'anglais et l'ukrainien, basés sur l'architecture Vits (Variational Inference with adversarial learning for end-to-end Text-to-Speech), fine-tunée sur 1100+ langues :

models/
├── mms-tts-eng/   # TTS anglais
│   ├── config.json
│   ├── model.safetensors
│   ├── vocab.json
│   └── tokenizer_config.json
└── mms-tts-ukr/   # TTS ukrainien
    └── ...

Ces modèles permettent de synthétiser vocalement les alertes générées — par exemple pour produire une version audio traduite en ukrainien d'un bulletin trafic français.

Extension : pipeline ASR → NLP → TTS multilingue

L'architecture modulaire permet d'enchaîner les couches pour un pipeline bout-en-bout :

Application concrète : accessibilité linguistique en temps réel — diffuser des alertes trafic ou des bulletins d'information dans la langue maternelle de l'auditeur, sans intervention humaine. La partie traduction (LLM ou MarianMT) et la synthèse TTS via MMS sont les extensions naturelles du projet.

Dépendances et structure du projet

Structure du projet

.
├── pyproject.toml
├── requirements.txt
├── src/
│   └── flash_nlp/
│       ├── acquisition/
│       │   └── fetcher.py
│       ├── transcription/
│       │   ├── whisper_service.py
│       │   └── audio_utils.py
│       ├── analysis/
│       │   ├── event_extractor.py
│       │   └── notifier.py
│       └── io/
│           └── file_utils.py
├── tests/
│   ├── test_fetcher.py
│   ├── test_whisper_service.py
│   ├── test_audio_utils.py
│   ├── test_event_extractor.py
│   └── test_io.py
├── data/
│   └── flash_audio_archive/
├── models/
│   ├── mms-tts-eng/
│   └── mms-tts-ukr/
└── outputs/

Conclusion

Flash NLP démontre comment combiner plusieurs briques open-source — faster-whisper, ffmpeg, scipy, Meta MMS — en un pipeline NLP cohérent pour le traitement automatique de flux audio publics. Les choix techniques privilégient la robustesse et la maintenabilité : polling conditionnel HTTP, déduplification MD5, VAD Whisper, NLP symbolique par regex, format JSONL.

L'architecture modulaire permet d'étendre le pipeline vers la traduction automatique et la synthèse vocale multilingue, ouvrant la voie à des applications d'accessibilité linguistique en temps réel.

Pour tester la transcription audio sans code, notre outil DEV-AI /upload repose sur le même moteur faster-whisper — uploadez votre audio ou vidéo et obtenez le texte en quelques secondes.

Pour maîtriser les bases de Whisper avant d'explorer ce pipeline, consultez notre guide Whisper Python : transcription audio gratuite en français.