Assistant vocal local ESP32 : étapes 1 à 15, et 15b écrite

État du projet au premier commit : - firmware ESP32 : micro I2S, lecture anti-gigue, MQTT (état retenu, testament), mains libres, domotique de démonstration, chien de garde, présence du serveur ; - serveur Python : sessions, VAD, faster-whisper, pyannote, identification, droits par profil, domotique, mot de réveil, Piper ; - broker Mosquitto : configuration, comptes et ACL (non activés en développement) ; - étape 15b écrite, en attente d'essai : agent MQTT pour machines Linux. Étapes 1 à 15 validées. 203 tests, pyright sans erreur. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 00:11:57 +02:00 · 2026-09-30 17:40:55 +02:00
commit d20976fea2
108 changed files with 15022 additions and 0 deletions
@@ -0,0 +1,164 @@
+"""Identification du locuteur : empreinte vocale comparée aux profils enrôlés (étape 9).
+
+Une empreinte est un vecteur de 256 nombres qui résume le timbre d'une voix.
+Deux empreintes de la même personne pointent à peu près dans la même direction :
+on les compare par similarité cosinus (1 = même direction, 0 = sans rapport).
+Au-dessus du seuil, on donne le nom du profil le plus proche ; sinon « Inconnu ».
+
+C'est une estimation, jamais une preuve : une voix enregistrée, imitée ou
+synthétisée peut tromper le système. Ne jamais s'en servir comme sécurité.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import re
+import unicodedata
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+
+import numpy as np
+
+from voice_server.settings import SpeakerSettings
+
+logger = logging.getLogger(__name__)
+
+UNKNOWN = "Inconnu"
+
+
+@dataclass(frozen=True)
+class VoiceProfile:
+    """Profil vocal d'une personne : la moyenne de ses empreintes d'enrôlement."""
+
+    name: str
+    embedding: np.ndarray   # normalisé (norme 1)
+    samples: int            # nombre de sessions moyennées
+    model: str              # modèle d'empreintes : un profil ne se compare qu'au même modèle
+    created: str
+
+
+@dataclass(frozen=True)
+class Identification:
+    """Verdict pour une voix."""
+
+    name: str | None        # None : personne n'atteint le seuil
+    score: float            # similarité avec le profil le plus proche (NaN : pas d'empreinte)
+    closest: str | None     # profil le plus proche, même sous le seuil (diagnostic)
+
+    @property
+    def label(self) -> str:
+        """Le nom, ou « Inconnu »."""
+        return self.name or UNKNOWN
+
+    def describe(self) -> str:
+        """« Denis (0.74) », « Inconnu (0.21, proche de Denis) », « Inconnu (trop peu de parole) »."""
+        if math.isnan(self.score):
+            return f"{UNKNOWN} (trop peu de parole)"
+        if self.name is not None:
+            return f"{self.name} ({self.score:.2f})"
+        if self.closest is not None:
+            return f"{UNKNOWN} ({self.score:.2f}, proche de {self.closest})"
+        return f"{UNKNOWN} (aucun profil)"
+
+
+def normalized(vector: np.ndarray) -> np.ndarray:
+    """Vecteur ramené à la norme 1 : le cosinus devient un simple produit scalaire."""
+    norm = float(np.linalg.norm(vector))
+    return vector / norm if norm > 0 and math.isfinite(norm) else np.full_like(vector, np.nan)
+
+
+def profile_path(profiles_dir: Path, name: str) -> Path:
+    """Fichier du profil : « Élodie Martin » → profiles/elodie-martin.json."""
+    ascii_name = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode()
+    slug = re.sub(r"[^a-z0-9]+", "-", ascii_name.lower()).strip("-")
+    if not slug:
+        raise ValueError(f"nom de profil inutilisable : {name!r}")
+    return profiles_dir / f"{slug}.json"
+
+
+def save_profile(profiles_dir: Path, name: str, embeddings: list[np.ndarray], model: str) -> VoiceProfile:
+    """Moyenne les empreintes d'enrôlement et écrit le profil (remplace l'ancien)."""
+    mean = normalized(np.mean([normalized(e) for e in embeddings], axis=0))
+    profile = VoiceProfile(name=name, embedding=mean, samples=len(embeddings), model=model,
+                           created=datetime.now().isoformat(timespec="seconds"))
+    path = profile_path(profiles_dir, name)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(".tmp")
+    temporary.write_text(json.dumps({
+        "name": profile.name, "model": profile.model, "samples": profile.samples,
+        "created": profile.created, "embedding": [round(float(v), 6) for v in profile.embedding],
+    }, ensure_ascii=False, indent=1), encoding="utf-8")
+    temporary.replace(path)  # écriture atomique : le serveur ne lit jamais un fichier à moitié écrit
+    return profile
+
+
+def load_profiles(profiles_dir: Path, model: str) -> list[VoiceProfile]:
+    """Lit les profils ; ceux d'un autre modèle d'empreintes sont ignorés (incomparables)."""
+    profiles: list[VoiceProfile] = []
+    for path in sorted(profiles_dir.glob("*.json")):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            profile = VoiceProfile(name=str(data["name"]), embedding=normalized(np.asarray(data["embedding"], dtype=np.float32)),
+                                   samples=int(data["samples"]), model=str(data["model"]), created=str(data["created"]))
+        except (OSError, ValueError, KeyError) as error:
+            logger.warning("profil illisible %s : %s", path.name, error)
+            continue
+        if profile.model != model:
+            logger.warning("profil %s ignore : cree avec %s, pas %s (re-enrolez)", path.name, profile.model, model)
+            continue
+        profiles.append(profile)
+    return profiles
+
+
+class SpeakerIdentifier:
+    """Donne un nom à une empreinte, d'après les profils du dossier.
+
+    Le dossier est relu dès qu'il change : un enrôlement fait pendant que le
+    serveur tourne est pris en compte sans le redémarrer.
+    """
+
+    def __init__(self, settings: SpeakerSettings, model: str) -> None:
+        """
+        Args:
+            settings: dossier des profils et seuil.
+            model: nom du modèle d'empreintes utilisé par la diarisation.
+        """
+        self._settings = settings
+        self._model = model
+        self._stamp: float | None = None
+        self._profiles: list[VoiceProfile] = []
+        self._reload_if_changed()
+
+    @property
+    def profiles(self) -> list[VoiceProfile]:
+        """Profils chargés."""
+        self._reload_if_changed()
+        return self._profiles
+
+    def identify(self, embedding: np.ndarray) -> Identification:
+        """Profil le plus proche, s'il dépasse le seuil."""
+        self._reload_if_changed()
+        vector = normalized(np.asarray(embedding, dtype=np.float32))
+        if not np.all(np.isfinite(vector)):
+            return Identification(name=None, score=math.nan, closest=None)
+        if not self._profiles:
+            return Identification(name=None, score=0.0, closest=None)
+        scores = [float(profile.embedding @ vector) for profile in self._profiles]
+        best = int(np.argmax(scores))
+        closest = self._profiles[best].name
+        if scores[best] >= self._settings.similarity_threshold:
+            return Identification(name=closest, score=scores[best], closest=closest)
+        return Identification(name=None, score=scores[best], closest=closest)
+
+    def _reload_if_changed(self) -> None:
+        directory = self._settings.profiles_dir
+        stamp = directory.stat().st_mtime if directory.is_dir() else None
+        if stamp == self._stamp:
+            return
+        self._stamp = stamp
+        self._profiles = load_profiles(directory, self._model) if stamp is not None else []
+        names = ", ".join(f"{p.name} ({p.samples} sessions)" for p in self._profiles) or "aucun"
+        logger.info("profils vocaux : %s", names)