Diarization

def assign_speakers(segments: list[Segment], turns: list[Turn]) -> None:
    """Label each segment with the speaker whose turns overlap it most; leave it unchanged if none do.

    Args:
        segments: Segments to label in place.
        turns: Speaker turns from :func:`diarize`.
    """
    for seg in segments:
        overlap: dict[str, float] = {}
        for start, end, label in turns:
            ov = min(seg.end, end) - max(seg.start, start)
            if ov > 0:
                overlap[label] = overlap.get(label, 0.0) + ov
        if overlap:
            seg.speaker = max(overlap, key=overlap.__getitem__)

def diarize(audio: np.ndarray, cfg: dict) -> list[Turn]:
    """Find speaker turns.

    Args:
        audio: 16 kHz mono float32 samples.
        cfg: The ``[diarization]`` config section.

    Returns:
        ``(start, end, label)`` turns, labelled ``speaker 1``, ``speaker 2``, ... in order of first appearance.
    """
    import torch

    out = _pipeline(cfg)({"waveform": torch.from_numpy(np.ascontiguousarray(audio))[None], "sample_rate": SR})
    annotation = getattr(out, "speaker_diarization", out)  # pyannote 4 wraps the annotation
    raw = [(float(t.start), float(t.end), str(label)) for t, _, label in annotation.itertracks(yield_label=True)]
    names: dict[str, str] = {}
    for _, _, label in sorted(raw):
        names.setdefault(label, f"speaker {len(names) + 1}")
    return [(s, e, names[label]) for s, e, label in raw]