Function postprocess

funasr/utils/speaker_utils.py:128–175

Postprocess. Args: segments: TODO. vad_segments: TODO. labels: TODO. embeddings: TODO.

(
    segments: list, vad_segments: list, labels: np.ndarray, embeddings: np.ndarray
)

def postprocess(
    segments: list, vad_segments: list, labels: np.ndarray, embeddings: np.ndarray
) -> list:
    """Postprocess.
    
        Args:
            segments: TODO.
            vad_segments: TODO.
            labels: TODO.
            embeddings: TODO.
        """
    assert len(segments) == len(labels)
    labels = correct_labels(labels)
    distribute_res = []
    for i in range(len(segments)):
        distribute_res.append([segments[i][0], segments[i][1], labels[i]])
    # merge the same speakers chronologically
    distribute_res = merge_seque(distribute_res)

    # accquire speaker center
    spk_embs = []
    for i in range(labels.max() + 1):
        spk_emb = embeddings[labels == i].mean(0)
        spk_embs.append(spk_emb)
    spk_embs = np.stack(spk_embs)

    def is_overlapped(t1, t2):
        """Is overlapped.
        
            Args:
                t1: TODO.
                t2: TODO.
            """
        if t1 > t2 + 1e-4:
            return True
        return False

    # distribute the overlap region
    for i in range(1, len(distribute_res)):
        if is_overlapped(distribute_res[i - 1][1], distribute_res[i][0]):
            p = (distribute_res[i][0] + distribute_res[i - 1][1]) / 2
            distribute_res[i][0] = p
            distribute_res[i - 1][1] = p

    # smooth the result
    distribute_res = smooth(distribute_res)

    return distribute_res


def correct_labels(labels):