clap_listener

clap_listener

Double-clap detector and optional microphone listener (welcome trigger).

Classes

ClapDetector dataclass

ClapDetector(
    cfg: ClapConfig,
    noise_floor: float = 0.0001,
    last_logged_double: float = 0.0,
    first_clap_time: float | None = None,
    first_clap_peak: float = 0.0,
    spike_armed: bool = True,
    last_miss_reason: str | None = None,
    claps_entendus: int = 0,
    dernier_clap_a: float = 0.0,
    blocs_ecoutes: int = 0,
    amorce: list[float] = list(),
)

Stateful detector: feed RMS samples, receive double-clap events.

Piece dataclass

Piece(
    niveau: float,
    dispersion: float,
    maximum: float,
    blocs_bruyants: int,
)

Ce qu'une pièce fait quand personne ne lui demande rien.

Attributes

niveau instance-attribute

niveau: float

Son niveau habituel — médiane, insensible aux évènements isolés.

dispersion instance-attribute

dispersion: float

Sa respiration, en log : de combien elle s'écarte d'ordinaire.

maximum instance-attribute

maximum: float

Le plus fort entendu, transitoire compris.

blocs_bruyants instance-attribute

blocs_bruyants: int

Combien de blocs ont dépassé six fois son niveau habituel.

haute property

haute: float

Sa portée haute ordinaire — le plafond de ce qu'elle fait seule.

Trois dispersions, pas cinq : calibré sur la pièce réelle de Carlito, dont le maximum observé vaut deux dispersions au-dessus de la médiane. Cinq donnaient 0,040 pour une pièce qui n'a jamais dépassé 0,0124 — une marge inventée devient un seuil que les claps doivent franchir pour rien.

troublee property

troublee: bool

Quelque chose est arrivé pendant que la pièce devait se taire.

Un COMPTAGE, pas un rapport. Un rapport entre le maximum et une statistique de la même fenêtre monte des deux côtés à la fois : dès que le transitoire dure quatre blocs, il définit lui-même la référence à laquelle on le compare, et la garde se tait précisément quand elle devrait parler. Un clap, attaque et réverbération, dure toujours plus que ça (démontré le 25 août 2026).

Ecoute dataclass

Ecoute(
    piece: Piece, claps: list[float], ecartes: list[float]
)

Une mesure complète : la pièce, puis ce qui s'y est ajouté.

Attributes

ecartes instance-attribute

ecartes: list[float]

Les bouffées écartées : trop faibles pour être les mêmes claps.

ClapListener

ClapListener(
    on_double_clap: Callable[[], None],
    *,
    cfg: ClapConfig | None = None,
    once: bool = True,
    device: int | None = None,
    debug: bool = False,
)

Background mic loop that invokes a callback on double clap.

Source code in src/diapason/speech/clap_listener.py

def __init__(
    self,
    on_double_clap: Callable[[], None],
    *,
    cfg: ClapConfig | None = None,
    once: bool = True,
    device: int | None = None,
    debug: bool = False,
) -> None:
    self._on_double = on_double_clap
    self._cfg = cfg or charger_reglage_claps()
    self._once = once
    self._device = device
    self._debug = debug
    self._stop = threading.Event()
    self._thread: Optional[threading.Thread] = None
    self._fired = False
    # Le fil d'écoute meurt en silence quand sounddevice manque ou que
    # le micro refuse de s'ouvrir : start() rend la main sans rien dire,
    # et l'appelant croit écouter (constaté le 25 août 2026 — le mode
    # annonçait « écoute active » alors que le fil était mort à la
    # première ligne). Ces deux signaux rendent le démarrage
    # CONSTATABLE au lieu d'être supposé.
    self._pret = threading.Event()
    self._panne: Optional[str] = None
    self._detecteur: Optional[ClapDetector] = None

Attributes

ecoute property

ecoute: bool

Le micro est-il RÉELLEMENT ouvert ? Constaté, jamais supposé.

claps_entendus property

claps_entendus: int

Combien de pics le micro a relevés — doubles ou non.

seuil property

seuil: float

Le seuil que ce fil applique VRAIMENT, à cet instant.

Le lire dans le fichier de réglage reviendrait à proclamer : un fil démarré avant une calibration garde l'ancien seuil jusqu'à ce qu'on le relance, et l'interface afficherait un chiffre auquel personne n'obéit. Zéro quand rien n'écoute — il n'y a alors pas de seuil.

fond_sonore property

fond_sonore: float

Le fond sonore que ce fil a APPRIS de la pièce où il tourne.

Comparé au seuil, il dit d'un coup d'œil si la marge est confortable ou si la pièce est montée jusqu'à frôler le déclenchement.

dernier_echec property

dernier_echec: Optional[str]

Pourquoi le dernier pic n'a pas formé un double, s'il y a lieu.

panne property

panne: Optional[str]

Pourquoi l'écoute n'a pas démarré, s'il y a une raison.

Functions:

choose_input_device

choose_input_device(
    cfg: ClapConfig,
    blocksize: int,
    *,
    override: str | None = None,
    silent_rms: float = 0.0005,
    probe_s: float = 0.5,
) -> int | None

Pick a working mic: override → default if loud → loudest input → default.

Source code in src/diapason/speech/clap_listener.py

def choose_input_device(
    cfg: ClapConfig,
    blocksize: int,
    *,
    override: str | None = None,
    silent_rms: float = 0.0005,
    probe_s: float = 0.5,
) -> int | None:
    """Pick a working mic: override → default if loud → loudest input → default."""
    import sounddevice as sd

    logger.info("Audio devices:\n%s", sd.query_devices())

    if override and override.strip():
        idx = resolve_input_device_index(override)
        peak = probe_input_max_rms(
            idx,
            blocksize,
            sample_rate=cfg.sample_rate,
            channels=cfg.channels,
            probe_s=probe_s,
        )
        logger.info(
            "Using configured mic [%d] (probe rms=%s)",
            idx,
            f"{peak:.5f}" if peak is not None else "unopenable",
        )
        return idx

    default = sd.default.device[0]
    if default is not None and default >= 0:
        peak = probe_input_max_rms(
            default,
            blocksize,
            sample_rate=cfg.sample_rate,
            channels=cfg.channels,
            probe_s=probe_s,
        )
        if peak is not None and peak >= silent_rms:
            logger.info("Using default mic [%d] (probe rms=%.5f)", default, peak)
            return int(default)
        logger.warning(
            "Default mic [%d] silent/unusable (rms=%s); scanning…",
            default,
            f"{peak:.5f}" if peak is not None else "n/a",
        )

    best_idx: int | None = None
    best_peak = -1.0
    for idx, _dev in _input_devices():
        if default is not None and idx == default:
            continue
        peak = probe_input_max_rms(
            idx,
            blocksize,
            sample_rate=cfg.sample_rate,
            channels=cfg.channels,
            probe_s=probe_s,
        )
        if peak is not None and peak > best_peak:
            best_peak = peak
            best_idx = idx

    if best_idx is not None and best_peak >= silent_rms:
        logger.info("Auto-selected mic [%d] (probe rms=%.5f)", best_idx, best_peak)
        return best_idx

    if default is not None and default >= 0:
        logger.warning("Falling back to default mic [%d]", default)
        return int(default)
    inputs = _input_devices()
    if inputs:
        logger.warning("Falling back to first input [%d]", inputs[0][0])
        return inputs[0][0]
    return None

charger_reglage_claps

charger_reglage_claps() -> ClapConfig

Le réglage de CETTE pièce — celui d'usine si rien n'est calibré.

Le min_rms calibré est conservé. Les écarts et la similarité, eux, sont ceux d'usine à chaque lecture : un claps.json du 25 août 2026 gardait max_double_gap_s=0.80 et min_double_gap_s=0.04, et deux bruits fortuits — ou l'attaque et la queue d'UN seul choc — ouvraient la caméra (29 août 2026).

Source code in src/diapason/speech/clap_listener.py

def charger_reglage_claps() -> ClapConfig:
    """Le réglage de CETTE pièce — celui d'usine si rien n'est calibré.

    Le ``min_rms`` calibré est conservé. Les écarts et la similarité, eux,
    sont ceux d'usine à chaque lecture : un ``claps.json`` du 25 août 2026
    gardait ``max_double_gap_s=0.80`` et ``min_double_gap_s=0.04``, et deux
    bruits fortuits — ou l'attaque et la queue d'UN seul choc — ouvraient
    la caméra (29 août 2026).
    """
    import json
    from dataclasses import replace

    usine = ClapConfig()
    try:
        brut = json.loads(chemin_reglage_claps().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return usine
    connus = set(ClapConfig.__dataclass_fields__)
    charges = {k: v for k, v in brut.items() if k in connus}
    # Ces champs ne se calibrent pas au micro : les laisser venir du fichier
    # réintroduisait le défaut dès qu'un vieux JSON survivait.
    for cle in (
        "min_double_gap_s",
        "max_double_gap_s",
        "similarite_min",
        "similarite_max",
        "spike_ratio",
        "retrigger_ratio",
        "cooldown_s",
    ):
        charges.pop(cle, None)
    return replace(usine, **charges)

niveau_ordinaire

niveau_ordinaire(blocs: list[float]) -> tuple[float, float]

Le niveau habituel d'un fond sonore, et sa dispersion.

La médiane, pas un quantile haut ni un maximum : un transitoire ne doit pas pouvoir déplacer la statistique qui sert à le juger. Un maximum brut cède devant un seul bloc ; le quantile 0,95 cède devant quatre. La médiane demande d'en corrompre la moitié.

Le calcul se fait sur les logarithmes, parce qu'un niveau sonore se compare en RAPPORTS et non en écarts : entre 0,005 et 0,015 il y a le même chemin qu'entre 0,05 et 0,15.

Source code in src/diapason/speech/clap_listener.py

def niveau_ordinaire(blocs: list[float]) -> tuple[float, float]:
    """Le niveau habituel d'un fond sonore, et sa dispersion.

    La médiane, pas un quantile haut ni un maximum : un transitoire ne doit
    pas pouvoir déplacer la statistique qui sert à le juger. Un maximum brut
    cède devant un seul bloc ; le quantile 0,95 cède devant quatre. La
    médiane demande d'en corrompre la moitié.

    Le calcul se fait sur les logarithmes, parce qu'un niveau sonore se
    compare en RAPPORTS et non en écarts : entre 0,005 et 0,015 il y a le
    même chemin qu'entre 0,05 et 0,15.
    """
    from math import exp, log

    if not blocs:
        return 0.0, 0.0
    logs = sorted(log(max(b, 1e-7)) for b in blocs)
    median = logs[len(logs) // 2]
    ecarts = sorted(abs(x - median) for x in logs)
    # 1,4826 : le facteur qui rend la MAD comparable à un écart-type sur une
    # loi normale, et donc lisible par qui connaît les écarts-types.
    dispersion = 1.4826 * ecarts[len(ecarts) // 2]
    return exp(median), dispersion

ecouter_la_piece

ecouter_la_piece(
    *,
    cfg: ClapConfig | None = None,
    device: int | None = None,
    duree_s: float = 2.5,
    oubli_initial_s: float = 0.6,
) -> Piece

Écouter une pièce se taire. Rien n'est enregistré : seuls des niveaux sonores sortent d'ici, jamais de son (§10).

Les premières fractions de seconde sont JETÉES : la mesure est déclenchée par un clic sur la machine qui tient le micro, et ce clic est un transitoire net.

Source code in src/diapason/speech/clap_listener.py

def ecouter_la_piece(
    *,
    cfg: ClapConfig | None = None,
    device: int | None = None,
    duree_s: float = 2.5,
    oubli_initial_s: float = 0.6,
) -> Piece:
    """Écouter une pièce se taire. Rien n'est enregistré : seuls des niveaux
    sonores sortent d'ici, jamais de son (§10).

    Les premières fractions de seconde sont JETÉES : la mesure est déclenchée
    par un clic sur la machine qui tient le micro, et ce clic est un
    transitoire net.
    """
    import sounddevice as sd

    cfg = cfg or ClapConfig()
    bloc = int(cfg.sample_rate * cfg.block_ms / 1000)
    with sd.InputStream(
        device=device,
        samplerate=cfg.sample_rate,
        channels=cfg.channels,
        dtype="float32",
        blocksize=bloc,
    ) as flux:
        fin = time.monotonic() + oubli_initial_s
        while time.monotonic() < fin:
            flux.read(bloc)
        releves: list[float] = []
        fin = time.monotonic() + duree_s
        while time.monotonic() < fin:
            donnees, _ = flux.read(bloc)
            releves.append(rms_mono(donnees))

    niveau, dispersion = niveau_ordinaire(releves)
    return Piece(
        niveau=niveau,
        dispersion=dispersion,
        maximum=max(releves) if releves else 0.0,
        blocs_bruyants=sum(1 for r in releves if r > niveau * 6.0),
    )

seuil_de_bouffee

seuil_de_bouffee(piece: Piece) -> float

À partir de quel niveau un son mérite d'être regardé.

Dix fois le fond, au minimum : un clap vaut vingt à soixante fois le niveau habituel d'une pièce. Deux fois et demie laissait entrer une touche de clavier, qui devenait ensuite « le clap le plus faible » et tirait tout le réglage vers le bas.

Source code in src/diapason/speech/clap_listener.py

def seuil_de_bouffee(piece: Piece) -> float:
    """À partir de quel niveau un son mérite d'être regardé.

    Dix fois le fond, au minimum : un clap vaut vingt à soixante fois le
    niveau habituel d'une pièce. Deux fois et demie laissait entrer une
    touche de clavier, qui devenait ensuite « le clap le plus faible » et
    tirait tout le réglage vers le bas.
    """
    return max(piece.niveau * 10.0, piece.haute * 2.0, 0.02)

ecouter_les_claps

ecouter_les_claps(
    piece: Piece,
    *,
    cfg: ClapConfig | None = None,
    device: int | None = None,
    duree_s: float = 6.0,
    oubli_initial_s: float = 0.25,
) -> Ecoute

Écouter quelqu'un claper, dans une pièce déjà mesurée.

Source code in src/diapason/speech/clap_listener.py

def ecouter_les_claps(
    piece: Piece,
    *,
    cfg: ClapConfig | None = None,
    device: int | None = None,
    duree_s: float = 6.0,
    oubli_initial_s: float = 0.25,
) -> Ecoute:
    """Écouter quelqu'un claper, dans une pièce déjà mesurée."""
    import sounddevice as sd

    cfg = cfg or ClapConfig()
    bloc = int(cfg.sample_rate * cfg.block_ms / 1000)
    with sd.InputStream(
        device=device,
        samplerate=cfg.sample_rate,
        channels=cfg.channels,
        dtype="float32",
        blocksize=bloc,
    ) as flux:
        fin = time.monotonic() + oubli_initial_s
        while time.monotonic() < fin:
            flux.read(bloc)
        releves: list[float] = []
        fin = time.monotonic() + duree_s
        while time.monotonic() < fin:
            donnees, _ = flux.read(bloc)
            releves.append(rms_mono(donnees))

    pics = _bouffees(releves, seuil_de_bouffee(piece))
    if not pics:
        return Ecoute(piece=piece, claps=[], ecartes=[])
    # Une bouffée nettement plus faible que les autres n'est pas le même
    # geste : c'est une chaise, une touche, un choc. L'écarter vaut mieux
    # que de caler tout le réglage dessus.
    milieu = sorted(pics)[len(pics) // 2]
    gardes = [p for p in pics if p >= 0.4 * milieu]
    return Ecoute(
        piece=piece,
        claps=gardes,
        ecartes=[p for p in pics if p < 0.4 * milieu],
    )

reglage_calibre

reglage_calibre(
    ecoute: Ecoute, *, base: ClapConfig | None = None
) -> ClapConfig

Un seuil posé entre DEUX mesures : la pièce, et les claps de son occupant.

Le seuil vise la moyenne géométrique — le milieu au sens de l'oreille, qui entend des rapports et non des différences.

Source code in src/diapason/speech/clap_listener.py

def reglage_calibre(
    ecoute: Ecoute,
    *,
    base: ClapConfig | None = None,
) -> ClapConfig:
    """Un seuil posé entre DEUX mesures : la pièce, et les claps de son
    occupant.

    Le seuil vise la moyenne géométrique — le milieu au sens de l'oreille,
    qui entend des rapports et non des différences.
    """
    from dataclasses import replace
    from math import sqrt

    base = base or ClapConfig()
    piece = ecoute.piece
    plancher_usine = ClapConfig().min_rms

    # Chaque refus nomme la MANŒUVRE à corriger, pas seulement le symptôme :
    # trois causes différentes donnaient le même message, et l'utilisateur
    # ne pouvait pas savoir laquelle le concernait.
    if piece.troublee:
        raise ValueError(
            f"Un bruit fort ({piece.maximum:.3f}) est arrivé pendant que "
            f"j'écoutais la pièce, qui vit d'ordinaire à {piece.niveau:.3f}. "
            "Attends que je te dise de claper, puis recommence."
        )
    if not ecoute.claps:
        raise ValueError(
            f"Aucun clap n'a été entendu — la pièce est restée à "
            f"{piece.niveau:.3f} du début à la fin. Clape plus fort, plus "
            "près du Mac, et pendant les six secondes qui suivent le signal."
        )
    if len(ecoute.claps) < 2:
        # Un seul pic ne dit pas si c'était un clap ou un choc. Deux se
        # ressemblent ; un tout seul ne se compare à rien.
        raise ValueError(
            f"Je n'ai entendu qu'un seul son fort ({ecoute.claps[0]:.3f}). "
            "Clape trois fois, bien séparément, pendant les six secondes."
        )

    # La MÉDIANE des claps retenus, jamais le plus faible : le minimum cède
    # devant un seul intrus, et c'est justement lui qui fixerait le seuil.
    ordonnes = sorted(ecoute.claps)
    reference = ordonnes[len(ordonnes) // 2]
    if reference < piece.haute * 3.0:
        raise ValueError(
            f"Tes claps ({reference:.3f}) ne se détachent pas assez du "
            f"bruit de la pièce ({piece.haute:.3f}). Clape plus fort, "
            "plus près du micro, ou dans un endroit plus calme."
        )

    seuil = sqrt(piece.haute * reference)
    # Une mesure ne doit JAMAIS rendre le détecteur plus bavard que l'usine
    # sans le dire : c'est ainsi qu'un plancher réglé pour un studio
    # imaginaire avait laissé le silence déclencher.
    # Le plancher s'appuie sur la portée HAUTE, pas sur le maximum brut :
    # un maximum cède devant un seul bloc, et un clic isolé pendant la phase
    # calme suffirait sinon à rendre le seuil inatteignable.
    seuil = max(seuil, piece.haute * 3.0, plancher_usine)
    if reference < seuil * 1.5:
        # Le plancher a rattrapé le seuil au point de dépasser les claps :
        # le poser quand même donnerait un mode qui ne s'arme jamais.
        raise ValueError(
            f"Tes claps ({reference:.3f}) sont trop faibles pour être "
            f"distingués sans risque du bruit ambiant (il faudrait au moins "
            f"{seuil * 1.5:.3f}). Rapproche-toi du Mac et recommence."
        )
    return replace(base, min_rms=round(seuil, 4))