"""Identification des locuteurs par embeddings ECAPA-TDNN + clustering incrémental. Chaque segment de parole (PCM 16 kHz mono) est projeté dans un espace de représentation de dimension 192 via ECAPA-TDNN (SpeechBrain). On maintient un profil par locuteur (moyenne mobile exponentielle des embeddings) et on attribue chaque nouveau segment au locuteur le plus proche par similarité cosinus, ou on crée un nouveau locuteur si le score est sous le seuil. """ from __future__ import annotations import struct from dataclasses import dataclass, field import numpy as np # Imports lourds (torch, speechbrain) chargés paresseusement dans load_model() # pour ne pas pénaliser le démarrage quand la diarisation est désactivée. @dataclass class _SpeakerProfile: """Profil incrémental d'un locuteur au sein d'une session.""" label: str embedding: np.ndarray # centroïde courant (moyenne mobile) count: int = 0 # nombre d'observations _ema_alpha: float = field(default=0.15, repr=False) def update(self, new_emb: np.ndarray) -> None: """Met à jour le centroïde via une moyenne mobile exponentielle. Alpha faible : le centroïde reste stable et ne dérive pas vers les variations d'un segment à l'autre (qui scinderaient un même locuteur). """ self.count += 1 if self.count == 1: self.embedding = new_emb.copy() else: self.embedding = ( self._ema_alpha * new_emb + (1 - self._ema_alpha) * self.embedding ) # Renormaliser pour que la similarité cosinus reste cohérente norm = np.linalg.norm(self.embedding) if norm > 0: self.embedding /= norm def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: """Similarité cosinus entre deux vecteurs unitaires.""" return float(np.dot(a, b)) class EmbeddingModel: """Encapsule le modèle SpeechBrain ECAPA-TDNN (singleton partagé). Chargé une seule fois au démarrage de l'application, puis réutilisé par chaque session WebSocket via des instances de SpeakerDiarizer. """ def __init__(self) -> None: import torch # noqa: F811 — import local volontaire from speechbrain.inference.speaker import EncoderClassifier self._device = "cpu" self._model = EncoderClassifier.from_hparams( source="speechbrain/spkrec-ecapa-voxceleb", savedir="/data/models/ecapa-tdnn", run_opts={"device": self._device}, ) self._torch = torch def extract(self, pcm: bytes, sample_rate: int = 16000) -> np.ndarray: """Extrait un embedding 192-d à partir d'un segment PCM 16-bit mono. Returns: np.ndarray de forme (192,), normalisé L2. Raises: ValueError: si le segment audio est trop court (< 200 ms). """ n_samples = len(pcm) // 2 min_samples = sample_rate // 5 # 200 ms minimum if n_samples < min_samples: raise ValueError( f"Segment trop court ({n_samples} samples, min {min_samples})" ) # PCM 16-bit little-endian → float32 [-1, 1] samples = struct.unpack(f"<{n_samples}h", pcm) waveform = self._torch.tensor(samples, dtype=self._torch.float32) / 32768.0 waveform = waveform.unsqueeze(0) # (1, T) with self._torch.no_grad(): embedding = self._model.encode_batch(waveform) emb = embedding.squeeze().cpu().numpy() # (192,) # Normaliser L2 norm = np.linalg.norm(emb) if norm > 0: emb /= norm return emb class SpeakerDiarizer: """Identifie le locuteur de chaque segment audio au sein d'une session. Chaque instance correspond à une réunion/session et maintient ses propres profils de locuteurs. Le modèle d'embedding est partagé (EmbeddingModel). """ # Un nouveau locuteur n'est créé que si le meilleur score est NETTEMENT # sous le seuil (zone morte) ET que le segment est assez long pour donner # une empreinte fiable. Sinon le segment est rattaché au meilleur profil. # Cela évite qu'un même locuteur soit éclaté en Locuteur 3, 4, 5... NEW_SPEAKER_MARGIN = 0.05 # écart sous le seuil pour autoriser un nouveau NEW_SPEAKER_MIN_S = 1.0 # durée mini d'un segment pour créer un locuteur STICKY_MARGIN = 0.05 # adhérence au dernier locuteur si scores proches def __init__( self, model: EmbeddingModel, threshold: float = 0.40, max_speakers: int = 8, ) -> None: self._model = model self._threshold = threshold self._max_speakers = max_speakers self._profiles: list[_SpeakerProfile] = [] self._last: _SpeakerProfile | None = None def identify(self, pcm: bytes, sample_rate: int = 16000) -> str: """Identifie le locuteur d'un segment PCM. Returns: Label du locuteur ("Locuteur 1", "Locuteur 2"...) ou chaîne vide si le segment est trop court pour être analysé. """ try: embedding = self._model.extract(pcm, sample_rate) except ValueError: return "" duration_s = (len(pcm) // 2) / sample_rate # Score de chaque profil existant scores = [(_cosine_similarity(embedding, p.embedding), p) for p in self._profiles] best_score, best_profile = max(scores, default=(-1.0, None), key=lambda x: x[0]) # Adhérence : si le dernier locuteur est presque aussi proche que le # meilleur, on le garde (réduit le clignotement entre deux profils). if self._last is not None and best_profile is not self._last: last_score = next((s for s, p in scores if p is self._last), -1.0) if last_score >= best_score - self.STICKY_MARGIN: best_score, best_profile = last_score, self._last # Rattachement à un locuteur connu if best_profile is not None and best_score >= self._threshold: best_profile.update(embedding) self._last = best_profile return best_profile.label # Création d'un nouveau locuteur : conditions strictes can_create = ( len(self._profiles) < self._max_speakers and duration_s >= self.NEW_SPEAKER_MIN_S and best_score < self._threshold - self.NEW_SPEAKER_MARGIN ) if not can_create and best_profile is not None: # Zone ambiguë ou segment court : on rattache au plus proche best_profile.update(embedding) self._last = best_profile return best_profile.label new_label = f"Locuteur {len(self._profiles) + 1}" new_profile = _SpeakerProfile(label=new_label, embedding=embedding) new_profile.update(embedding) self._profiles.append(new_profile) self._last = new_profile return new_label def reset(self) -> None: """Réinitialise les profils pour une nouvelle session.""" self._profiles.clear() self._last = None