Add speaker diarization capabilities for long audios. (4+ hours) 
State: closed | Pull request: yes
Labels: name: ASR

name: Speaker Tasks

# What does this PR do ?

This PR enables speaker diarization on long audio files (even +4 hours) without running out of memory. Only one long audio file is accepted for the moment, because this is the use case I needed, but it can easily be made to accept multiple audios. (It was tested mainly on CPU, so some arguments regarding CUDA were set as "False", just to be safe)

**Collection**:  collections/asr/models, collections/asr/parts/utils

# Changelog 
The logic behind this method is that, if the same speaker would be in different audios, his speaker representations should be the same/very similar, and if those 2 different audios would be merged, he would be detected as the same speaker. That is why this method will imply 2 steps, 1st where we split a long audio in smaller audios and perform diarization, and the second where we extract just a few speaker representations from each speaker detected in the smaller audios and merge them in order to perform a global diarization over all audios, but without using all speaker representations. ( just a few embeddings from each speaker already detected in the 1st step)


The following figure presents the process for enabling long audio diarization through "2 step clusterization. The steps are the following:
- We split a long input audio (greater than 45 minutes) in equally sized ordered audio parts ( for example a 4 hour audio => 8 x 30 min audio parts). If audio < 45 minutes : normal diarization is performed.
- We perform speaker diarization on each audio part and obtain speaker labels for each individual audio part. Furthermore, for each audio part, we save the speaker embeddings for each speaker inside a dictionary called *split_audios".
- After all audio parts were diarized and the speaker embeddings for each part were saved. We enter the second step in the process called "reclustering". From each audio part, we take the first *E* number of speaker embeddings from each speaker(we know at what timestamp a speaker is based on the labels) and merge them with other speaker embeddings extracted from other speakers, from other audio parts. ( e.g. : if we have 2 speakers detected in audio part 1; 3 speakers in audio part 2 and 4 speakers in audio part 3; if we take *E*=100, we will have a merged embeddings matrix of (2+3+4)*100= *a matrix of [900 x 192]* ) In this way, we already know which embeddings we should take, based on the speakers detected in the 1st step, so we do not need to take ALL embeddings in a e.g: 4 hours audio, but just a few from each speaker.
- After merging the embeddings ( from each speaker ,from each audio), we perform clusterization again. (*speaker_utils.py/perform_reclustering()*)  We obtained a set of final labels for each row in the merged embeddings matrix. To be noted, each label corresponds to a certain timestamp, so we can map the new detected labels over the merged embeddings matrix to the old labels detected over each individual audio part.
- We will now map the new labels to the old labels using functions found in /asr/parts/utils/speaker_utils.py: label_mapping(), label_conversion(), order_labels(). For each old label, we take the majority new label, we also order the labels and map them to all the old labels. ( e.g.: if for a speaker we only took 100 embeddings, but he had 2000 embeddings, we have to map the majority *new label* to all 2000 embeddings.)
- Lastly, after mapping all the labels, as the audio inputs are ordered cronologically, all we have to do is merge the final output labels rttms into a single output labels rttm.

TO BE NOTED: 
- One final step we actually implement is extracting multiple *E* embeddings from each speaker and then taking the majority label over all *E* scales using *speaker_utils.py/majority_label()* function. In *conf/inference/diar_infer_meeting.yaml*, we created a *diarizer.clustering.parameters.embs_count_recluster* parameter that is a list of the number of embeddings to be extracted from each speaker, across every audio part. We do this in order to have a more robust result, as we observed across several experiments, that the DER results differ depending on the number of speaker embeddings extracted. This way, we take the majority label detected for each speaker embedding extracted/timestamp over multiple *E* scales. (it is a similar concept to extracting multiple window lengths)
- If a speaker doesn't have enough speaker embeddings in an audio part (e.g.: the speaker has only 210 speaker embeddings and we want to take *E*=300), we take his maximum number of speaker embeddings (e.g.: 210).

![2step drawio](https://github.com/NVIDIA/NeMo/assets/53811655/efdc9873-a055-479f-8557-3f651013930a)

This PR also adds the ability to automatically create a manifest file based on the input audio. Additional argument to the manifest file: "split". If audio is bigger than 45 minutes and smaller than 1 hour we will split it in 2 equal audio parts. If audio is bigger than 1 hour, we will split it in maximum 45 minutes parts.

Changes/additional files: 

- /examples/speaker_tasks/diarization/clustering_diarizer/long_offline_diar_infer.py
- /examples/speaker_tasks/diarization/conf/inference/* (NOTE: added the diarizer.clustering.parameters.embs_count_recluster parameter)
- /nemo/collections/asr/models/long_clustering_diarizer.py
- /nemo/collections/asr/parts/utils/speaker_utils.py: 
     - Changed perform_clustering() function, by adding split_audios dictionary where we save information about speaker labels, speaker embeddings, etc. for each audio part. PROBLEM: we now return all_reference, all_hypothesis, split_audios => we need to change the normal *clustering_diarizer.py* file, in order to not get an error, by adding "all_reference,all_hypothesis, _".
     - Added perform_reclustering function that uses split_audios info from previous step as input
     - Added the label_mapping(), label_conversion(), order_labels(), majority_label(), recluster_write() functions for the post processing of final labels obtain.
- /nemo/collections/asr/parts/utils/offline_clustering.py: 
     - added getRepeatedEmbeddings function. Everytime we extract embeddings at a scale (e.g: [0.5,1.5,3]s), we already save a certain number of embeddings, BUT these embeddings must target SAME TIME INTERVALS. (e.g.: if we take 100 x 0.5s segments = 50s, then we will only take ~30 segments of 3s. We will repeat the 30 segments until we get 100 segments, similar to normal clustering process. We just do this before creating the Affinity Matrix, not after.)
     - reclusteringMultiScaleCosAffinityMatrix() function, using the repeated embeddings
     - forward_reinfer() function, very similar to the foward_infer(), just used in the reclustering step 

# Usage
Inside /examples/speaker_tasks/diarization/clustering_diarizer/long_offline_diar_infer.py:

```python
@hydra_runner(config_path="../conf/inference", config_name="diar_infer_meeting.yaml")
def main(cfg):
    logging.info(f'Hydra config: {OmegaConf.to_yaml(cfg)}')
    
    cfg.diarizer.speaker_embeddings.model_path = "nemo_models/titanet-l.nemo"
    cfg.diarizer.vad.model_path = "nemo_models/vad_multilingual_marblenet.nemo"
    cfg.diarizer.speaker_embeddings.parameters.save_embeddings=False
    

    input_path = '/home/gabi/NOSFE_Gojira_Podcast_16khz_mono.wav'
    num_speakers = None 

    sd_model = LongClusteringDiarizer(cfg=cfg)
    sd_model.diarize(input_path= input_path,
                     num_speakers = num_speakers)


if __name__ == '__main__':    
    main()
```

Simply change *input_path* with the path for the audio you want. Change *num_speakers* to None, if number of speakers is unknown, or to an integer if number of speakers is known.


# Before your PR is "Ready for review"
**Pre checks**:
- [ x ] Make sure you read and followed [Contributor guidelines](https://github.com/NVIDIA/NeMo/blob/main/CONTRIBUTING.md)
- [ ] Did you write any new necessary tests?
- [ ] Did you add or update any necessary documentation?
- [ ] Does the PR affect components that are optional to install? (Ex: Numba, Pynini, Apex etc)
  - [ x ] Reviewer: Does the PR have correct import guards for all optional libraries?
  
**PR Type**:
- [ x ] New Feature
- [ ] Bugfix
- [ ] Documentation

If you haven't finished some of the above items you can still open "Draft" PR.


## Who can review?

Anyone in the community is free to review the PR once the checks have passed.  
[Contributor guidelines](https://github.com/NVIDIA/NeMo/blob/main/CONTRIBUTING.md) contains specific people who can review PRs to various areas.

I kindly ask: @titu1994, @redoctopus, @jbalam-nv, or @okuchaiev to review this PR and ask any questions that could help to clarify my implementation.

# Additional Information

I would kindly ask for further testing from the NeMo team to see if the solution is consistent and has the expected results. Then we can further clarify comments, and reformat functions, as they may not be the most efficiently written. I have tested the method on multiple smaller audios (1 hour), and for 2 speaker audios, it gets the expected result that you would obtain by doing normal diarization. However, for multiple speakers, the results slightly changes( sometimes the DER is improved, sometimes not). My main goal using this method is to be able to actually diarize long audios (mainly containing 2 speakers, but more speakers proved to work very well too, in my opinion), without facing computational barriers. 

The reclustering step, on a i7-12700 with 20 cores took less than 1 second, using 10 embeddings scales (100-500 with step of 50).