vocal_helper.growing module

Diariser le tampon entier, périodiquement : la qualité du différé, en direct.

Le raisonnement

Deux mesures se combinent mieux qu’il n’y paraît. D’une part l’identité voyage très bien par recouvrement temporel entre deux diarisations qui partagent de l’audio. D’autre part le diariseur différé coûte RTF 0,039 : une réunion d’une heure se diarise en deux minutes.

Un découpage en morceaux a besoin d’un amorçage synthétique pour transporter l’identité, parce qu’un locuteur peut être absent de la zone commune aux deux diarisations. Si la seconde passe porte sur tout l’audio depuis le début, ce cas disparaît : quiconque a parlé est dans la zone commune, qui couvre alors la quasi-totalité du signal. L’appariement devient presque exact, aucun son synthétique n’est nécessaire, et surtout la dernière passe est une diarisation du fichier entier — donc la qualité du différé, atteinte en direct.

Le coût se calcule et se mesure. Rafraîchir aux instants t, r·t, r²·t… diarise au total un petit multiple de la durée, donc le coût reste proportionnel à la séance et non à son carré : 2,6 à 3,3 fois l’audio mesuré, pour un facteur temps réel de 0,10 à 0,16.

Ce que ça donne

Sur les huit réunions d’AMI dev-slice, le résultat final est exactement celui du différé pleine fenêtre, à la troisième décimale — taux d’erreur médian 0,176, le bon nombre de locuteurs, et le meilleur accord longue portée mesuré. Pour comparaison sur le même corpus : l’étage en ligne glouton donne 0,369 avec soixante-cinq étiquettes pour quatre locuteurs, et le regroupement incrémental 0,254 avec vingt-quatre.

Ce que ça coûte

Pas la qualité, la fraîcheur : la parole postérieure à la dernière passe n’a pas encore d’étiquette faisant autorité. Un espacement géométrique rend ce retard proportionnel à la séance — 0,375 fois sa durée à ratio=1.6 — ce qui va pour vingt minutes et non pour deux heures ; max_period_s le borne au prix d’un facteur temps réel plus élevé. Et les étiquettes se révisent : c’est le prix de l’exactitude, et LabelRevision le dit à l’appelant plutôt que de le cacher.

class vocal_helper.growing.GrowingBufferDiarStage(backend='auto', policy=<factory>, offline=None, online=None, on_revision=None)[source]

Bases: object

Diarisation en direct par re-diarisation périodique du tampon entier.

L’étage accumule l’audio et, aux instants que la politique fixe, le fait diariser en entier par un moteur différé. Les étiquettes de chaque passe sont rattachées à celles de la précédente par recouvrement temporel, de sorte que le lecteur voie des identités stables ; les changements sont rendus comme des révisions explicites.

Parameters:

  • backend (str) – Moteur différé, passé à OfflineDiarStage. "auto" prend le meilleur installé.

  • policy (GrowingBufferPolicy, optional) – Le calendrier des passes.

  • offline (object, optional) – Un diariseur déjà construit, qui doit exposer diarize(pcm, sr) -> [(t0, t1, étiquette), …]. Sert aux tests et permet de brancher un autre moteur sans toucher à cette classe.

  • online (object, optional) – L’étage en ligne qui donne l’étiquette provisoire, en une demi- seconde, le temps que la première passe arrive. None ne publie rien avant la première passe.

  • on_revision (callable, optional) – Appelée pour chaque étiquette démentie par une passe. Sans elle, les corrections sont calculées et perdues.

Examples

Un diariseur de pacotille suffit à montrer le mécanisme : les étiquettes qu’il rend changent de nom à chaque passe, et l’étage les stabilise.

>>> import numpy as np
>>> class Flighty:
...     def __init__(self): self.calls = 0
...     def diarize(self, pcm, sr, **kw):
...         self.calls += 1
...         tag = f"local{self.calls}"
...         return [(0.0, len(pcm) / sr, tag)]
>>> stage = GrowingBufferDiarStage(
...     offline=Flighty(), policy=GrowingBufferPolicy(first_s=1.0, ratio=2.0)
... )
>>> segments = stage.diarize(np.zeros(16000 * 4, dtype=np.float32), 16000)
>>> [label for _, _, label in segments]
['P0']
>>> stage.speaker_labels
['P0']

Trois passes ont eu lieu, et le locuteur n’a été découvert qu’une fois.

>>> stage.passes
3
backend: str = 'auto'
diarize(pcm, sr, *, num_speakers=None, min_speakers=None, max_speakers=None)[source]

Dérouler un tampon comme un direct, et rendre l’état final.

Utile pour mesurer : la dernière passe voit tout, donc le résultat est celui du différé pleine fenêtre, obtenu en ayant payé le prix du direct. Pour un vrai flux, appeler refresh() aux instants que GrowingBufferPolicy.times() indique et lire les révisions.

Parameters:

  • pcm (NDArray[np.float32]) – Tampon mono.

  • sr (int) – Sa fréquence.

  • num_speakers (int, optional) – Transmis au moteur différé à chaque passe.

  • min_speakers (int, optional) – Transmis au moteur différé à chaque passe.

  • max_speakers (int, optional) – Transmis au moteur différé à chaque passe.

Returns:

[(t0, t1, locuteur), …], la même forme que les autres étages.

Return type:

list of tuple

property discovered_speakers: int

Combien de locuteurs ont été découverts.

offline: Any = None
on_revision: Any = None
online: Any = None
property passes: int

Combien de passes de re-diarisation ont eu lieu.

policy: GrowingBufferPolicy
refresh(pcm, sr, **hints)[source]

Rediariser tout le tampon, et rendre ce que ça change.

Parameters:

  • pcm (NDArray[np.float32]) – Tout l’audio depuis le début, pas seulement le nouveau.

  • sr (int) – Sa fréquence.

  • **hints – Transmis au moteur différé, par exemple max_speakers.

Returns:

Les passages dont l’étiquette change par rapport à ce qui avait été publié. Vide à la première passe, puisque rien n’avait été dit.

Return type:

list of LabelRevision

reset()[source]

Oublier le flux, en gardant la configuration.

Return type:

None

async run(inbox, outbox)[source]

Consommer des segments voisés, publier aussitôt, corriger ensuite.

Même contrat que l’étage en ligne, donc c’est un remplacement direct : un consommateur qui lit outbox n’a rien à changer. Ce qui est nouveau, c’est qu’une étiquette déjà publiée peut être démentie, et on_revision le dit — un affichage qui ignore les révisions montre des étiquettes dont on lui a signalé qu’elles sont fausses.

L’étage garde tout l’audio voisé, puisque chaque passe rediarise depuis le début : à 16 kHz en flottant simple cela fait 64 ko par seconde de parole, soit environ 230 Mo pour une heure de réunion. C’est le prix de l’exactitude, et il est borné par la durée de la séance.

Parameters:

  • inbox – Files d’attente de segments voisés puis diarisés, comme pour les autres étages.

  • outbox – Files d’attente de segments voisés puis diarisés, comme pour les autres étages.

Return type:

None

segments()[source]

La diarisation publiée, telle que la dernière passe l’a laissée.

Return type:

list[tuple[float, float, str]]

property speaker_labels: list[str]

Les locuteurs publiés, dans l’ordre de découverte.

class vocal_helper.growing.GrowingBufferPolicy(first_s=60.0, ratio=1.6, max_period_s=None, min_speaker_s=0.0, gap_s=0.8)[source]

Bases: object

Quand rafraîchir, et jusqu’où laisser le retard grandir.

Parameters:
first_s

Instant de la première passe. En dessous, le diariseur n’a pas de quoi travailler et rendrait un seul locuteur.

Type:

float

ratio

Facteur d’espacement. 1,6 mesure un coût de 2,6 à 3,3 fois l’audio pour un retard valant 0,375 fois la séance ; descendre vers 1,2 raccourcit le retard et monte le coût.

Type:

float

max_period_s

Attente maximale entre deux passes. None laisse le retard croître avec la séance, ce qui va pour une réunion et non pour une journée.

Type:

float, optional

min_speaker_s

Temps de parole minimal pour qu’une étiquette soit publiée. Écarte les locuteurs d’une seconde que le diariseur invente sur un bruit.

Type:

float

gap_s

Silence considéré comme une séparation de tours, pour récolter le bruit de fond qui reconstitue les creux entre segments voisés.

Type:

float

first_s: float = 60.0
gap_s: float = 0.8
max_period_s: float | None = None
min_speaker_s: float = 0.0
ratio: float = 1.6
times(duration_s)[source]

Les instants de rafraîchissement pour cette durée.

Parameters:

duration_s (float)

Return type:

list[float]

worst_staleness_s(duration_s)[source]

Le plus long délai avant qu’un passage reçoive son étiquette définitive.

Examples

>>> GrowingBufferPolicy(first_s=60.0, ratio=1.6).worst_staleness_s(1200.0)
377.48736
Parameters:

duration_s (float)

Return type:

float

vocal_helper.growing.map_by_overlap(left, right)[source]

La correspondance un-pour-un entre deux diarisations, par temps partagé.

Les noms de grappes sont arbitraires et interchangeables ; ce qui doit être conservé d’une passe à l’autre, c’est qui est qui. C’est un appariement bijectif, donc résolu par algorithme hongrois et non par un argmax par étiquette : un argmax peut envoyer deux étiquettes de gauche sur la même à droite, ce qui fusionnerait deux locuteurs sans le dire.

Parameters:

  • left (list of tuple) – Deux diarisations (t0, t1, étiquette) sur la même horloge.

  • right (list of tuple) – Deux diarisations (t0, t1, étiquette) sur la même horloge.

Returns:

{étiquette de gauche: étiquette de droite}, seulement pour les paires qui partagent effectivement du temps.

Return type:

dict of str

Examples

>>> map_by_overlap([(0.0, 2.0, "a"), (2.0, 4.0, "b")],
...                [(0.0, 2.0, "y"), (2.0, 4.0, "x")])
{'a': 'y', 'b': 'x'}

Rien de partagé, rien d’apparié — plutôt qu’un rapprochement inventé.

>>> map_by_overlap([(0.0, 1.0, "a")], [(9.0, 10.0, "z")])
{}
vocal_helper.growing.refresh_times(duration_s, *, first_s, ratio, max_period_s=None)[source]

Les instants de rafraîchissement, espacés géométriquement.

Un pas constant ferait croître le coût comme le carré de la durée, puisque chaque passe rediarise tout depuis le début. Un pas géométrique le garde proportionnel : la somme des préfixes reste un petit multiple de la durée totale. max_period_s borne l’attente au prix d’un coût plus élevé, pour une séance longue où un retard proportionnel serait inacceptable.

Examples

>>> refresh_times(100.0, first_s=10.0, ratio=2.0)
[10.0, 20.0, 40.0, 80.0, 100.0]

Plus court que la première passe : une seule, à la fin.

>>> refresh_times(8.0, first_s=10.0, ratio=2.0)
[8.0]

Avec un plafond, l’attente ne grandit plus indéfiniment.

>>> refresh_times(100.0, first_s=10.0, ratio=2.0, max_period_s=25.0)
[10.0, 20.0, 40.0, 65.0, 90.0, 100.0]
Parameters:
Return type:

list[float]