o oi D @sBdZddlZddlZddlmZmZmZmZmZddl m Z ddl Z ddl Z ddl Z ddlmmZddlZddlmZmZmZddlmZddlmZddlmZmZe d e Gd d d Z!Gd d d Z"GdddZ#de$de!de%fddZ&GdddZ'    dde%de%dee%dee!dee%def ddZ(dS)a6 WavLM Audio Embedding Generator for HuggingFace Datasets ========================================================= Clean OOP implementation for generating speaker embeddings using the WavLM model and adding them as new columns to HuggingFace datasets with multi-GPU support. Model: Orange/Speaker-wavLM-tbr N)ListDictAnyOptionalTuple) dataclass)Dataset DatasetDict load_dataset)tqdm)EmbeddingsModelcompute_embeddingignorec@sZeZdZUdZdZeed<dZeed<dZ e ed<dZ eed <d Z eed <d Z eed <dS)EmbedderConfigz+Configuration for the WavLM audio embedder.zOrange/Speaker-wavLM-tbr model_name> target_srg>@ max_audio_sec batch_sizewavlm_embeddingembedding_column_nameTuse_multiprocessingN)__name__ __module__ __qualname____doc__rstr__annotations__rintrfloatrrrboolr#r#E/home/ubuntu/kanitts-2-dataset-pipeline/utils/speaker_emb/embedder.pyr s      rc@s^eZdZdZddedefddZdeee fd e e j fd d Z d ejd e e j fd dZdS)AudioProcessorzA Handles audio loading, preprocessing and normalization. r9@rrcCs||_||_t|||_dS)z Initialize audio processor. Args: target_sr: Target sample rate in Hz max_audio_sec: Maximum audio duration in seconds N)rrr target_samples)selfrrr#r#r$__init__1szAudioProcessor.__init__ audio_dictreturnc Csz,|d}t|tjrt|}ntj|tjd}|dkr'|j dd}| |WSt yF}zt d|WYd}~dSd}~ww)aw Process audio from HuggingFace dataset format. Note: Audio is expected to be already resampled to target_sr via dataset.cast_column('audio', Audio(sampling_rate=target_sr)) Args: audio_dict: Dictionary with 'array' and 'sampling_rate' keys Returns: Processed audio array or None if processing fails arraydtyper rdimzAudio processing error: N) isinstancenpndarraytorch from_numpyr!tensorfloat32r0mean_finalize_waveform Exceptionprint)r(r* audio_arraywaver#r#r$ process_audio=s     zAudioProcessor.process_audior=cCsf|dkrdS|jd}||jkr|d|j}|S||jkr/t|d|j|fdd}|S)z Truncate or pad waveform to target length. Args: wav: Audio waveform tensor Returns: Processed numpy array or None if empty rNconstant)numelshaper'Fpadnumpy)r(r= current_lenr#r#r$r9^s    z!AudioProcessor._finalize_waveformN)rr&)rrrrr r!r)rrrrr2r3r?r4Tensorr9r#r#r#r$r%,s  !r%c@sFeZdZdZdededefddZddZd e e d e j fd d Z d S) WavLMWorkerzX Worker class for processing dataset chunks with WavLM model on a specific GPU. rankconfigaudio_processorcCs"||_||_||_d|_d|_dS)z Initialize WavLM GPU worker. Args: rank: GPU device index config: Embedder configuration audio_processor: Audio preprocessing instance N)rJrKrLdevicemodel)r(rJrKrLr#r#r$r)zs  zWavLMWorker.__init__c Csztj|jtd|j|_Wn#ty6}ztd|jd|td|_WYd}~nd}~wwtd|jdt|j j |_ |j |j td|jddS)z0Initialize WavLM model and move to assigned GPU.zcuda:[GPU z] Failed to set device: cpuNz] Loading WavLM model...z] WavLM model ready)r4cuda set_devicerJrMr:r;WavLMEmbeddingsModelfrom_pretrainedrKrrNtoeval)r(r>r#r#r$ initializeszWavLMWorker.initialize audio_dictsr+c Cst|jj|jj}g}g}t|D]k\}}z:|d}t|tjr)t | }ntj |tj d}| dkr=|jdd}|jd|krJ|d|}||Wqty}} z!td|jd|d | |td||WYd} ~ qd} ~ wwtd d |D} g} |D]}|jd} | | krt|d| | fd d }| |qt| |j} t || }Wdn1swY|}|D] }td||<q|S)a Process a batch of audio samples with WavLM using true batching with dynamic padding. Uses dynamic padding: pads to the maximum length in the current batch (up to max_audio_sec), rather than always padding to max_audio_sec. This dramatically reduces computation for batches with shorter audio samples. Args: audio_dicts: List of audio dictionaries from dataset Returns: Numpy array of embeddings (batch_size, embedding_dim) r,r-r rr/NrOz"] Error processing audio at index z: css|]}|jdVqdS)rN)rC).0sigr#r#r$ sz,WavLMWorker.process_batch..r@rA)r rKrr enumerater1r2r3r4r5r!r6r7r0r8rCappendr:r;rJzerosmaxrDrEstackrUrMno_gradrNrPrF)r(rXmax_allowed_lengthprocessed_audios error_indicesidxr*r<rZr>batch_max_length batch_tensorsrGbatch embeddings embeddings_npr#r#r$ process_batchsF         zWavLMWorker.process_batchN)rrrrr rr%r)rWrrr2r3rlr#r#r#r$rIus rIrJrK audio_columncs(tjj}t|}|g}tfddtdjDtj ddt fdd}t j |dd } | td dd } ttD]} } || } | D]}||qd| t| qW| | jd d ||ftddt|ddS)a Worker process function for multi-GPU processing with data prefetching. Uses a background thread to load data asynchronously while GPU processes the current batch, eliminating data loading bottlenecks. Args: rank: GPU device index config: Embedder configuration dataset_shard: Subset of dataset to process audio_column: Name of audio column result_queue: Queue to store results cs$g|]}|t|jqSr#)minrrYi)rK dataset_shard num_samplesr#r$ sz&gpu_worker_process..rmaxsizec sz;zD] }|}|qWnty-}ztdd|WYd}~n d}~wwWdSWdSw)z4Background thread that loads batches asynchronously.rOz] Data loader thread error: N)putr:r;set)rirXr>)rmbatchesloading_completeprefetch_queuerJr#r$data_loader_threads   z.gpu_worker_process..data_loader_threadTtargetdaemonzGPU )totaldescpositionleave@timeoutrOz] Completed processing z samplesN)r%rrrIrWlenrangerqueueQueue threadingEventThreadstartr getrlr^updateclosejoinrwr;)rJrKrqrm result_queuerLworkerresultsr| loader_threadpbar_rXrjembr#)rmryrKrqrzrrr{rJr$gpu_worker_processs:       rc @seZdZdZdefddZ   dded ed eed ed ef d dZ ded ed e e e fddZ ded ed e e e fddZ dS) WavLMEmbedderz Main class for generating WavLM speaker embeddings on HuggingFace datasets. Supports multi-GPU parallel processing with automatic batching and progress tracking. rKcCs<||_t|j|jd|_tjrtj|_ dSd|_ dS)z Initialize the embedder. Args: config: Configuration object with model and processing parameters )rrr N) rKr%rrrLr4rQ is_available device_countnum_gpus)r(rKr#r#r$r):s $zWavLMEmbedder.__init__audioNtraindatasetrm output_path split_namer+cCstddtd|dt|dtd|jdtdd||jvr4td|d |j|jjrD|jd krD|||}n|||}| |jj |}|S) a Add embedding column to a HuggingFace dataset. Args: dataset: HuggingFace Dataset object audio_column: Name of the column containing audio data output_path: Optional path to save the dataset locally split_name: Name of the split being processed (for logging) Returns: Dataset with added embedding column  z<============================================================z Processing z split with z exampleszUsing z GPU(s)zColumn 'z+' not found in dataset. Available columns: r ) r;rr column_names ValueErrorrKr_multi_gpu_embed_single_gpu_embed add_columnr)r(rrmrrrjr#r#r$ embed_datasetJs    zWavLMEmbedder.embed_datasetc stdjj}|g}tfddtdjjDtjddfdd}t j |dd }| t d d }ttD]} } || } || |t| qH||jd d |S)z Generate embeddings using single GPU with data prefetching. Args: dataset: Dataset to process audio_column: Name of audio column Returns: List of embeddings rcs"g|] }|t|jjfqSr#)rnrKrro)rrr(r#r$rssz3WavLMEmbedder._single_gpu_embed..rtruc shzD]\}}||}|}|qWdSty3}ztd|WYd}~dSd}~ww)z2Background thread for loading data asynchronously.zData loader thread error: N)rwr:r;) start_idxend_idxrirXr>)rm batch_indicesrr{r#r$r|s   z;WavLMEmbedder._single_gpu_embed..data_loader_threadTr}zGenerating embeddings)rrrr)rIrKrLrWrrrrrrrrr rrlextendtolistrrr) r(rrmrrjr|rrrrXbatch_embeddingsr#)rmrrrrr{r(r$rts(       zWavLMEmbedder._single_gpu_embedcCs*t|}||jd|j}g}t|jD]}||}t|||}||kr1||t||qtd} | } g} tt|D]} | j t | |j || || fd} | | | qCi}tt|D] }| \} }||| <qf| D]} | qug}tt|D] } ||| qdd|DS)z Generate embeddings using multiple GPUs in parallel. Args: dataset: Dataset to process audio_column: Name of audio column Returns: List of embeddings in original order r spawn)r~argscSs$g|]}t|tjr|n|qSr#)r1r2r3r)rYrr#r#r$rssz2WavLMEmbedder._multi_gpu_embed..)rrrrnr^selectmp get_contextrProcessrrKrrrr)r(rrmrr shard_sizeshardsrprr mp_contextr processesrJp results_dictrrjall_embeddingsr#r#r$rs>      zWavLMEmbedder._multi_gpu_embed)rNr)rrrrrr)rrrrrr!rrr#r#r#r$r3s& *":rr dataset_namersplitr+c Cs|durt}td|dt||d}t|}t|trBt}|D]\}} |r2|d|nd} |j| || |d||<q%|S|j||||pJddS)a Convenience function to process a HuggingFace dataset. Args: dataset_name: Name of the dataset on HuggingFace Hub audio_column: Name of the column containing audio data output_path: Optional path to save processed dataset config: Optional custom configuration split: Specific split to process (e.g., 'train', 'test') or None for all Returns: Processed dataset with embeddings NzLoading dataset: z...)r/)rmrrr)rr;r rr1r itemsr) rrmrrKrrembedder processedr split_dataset split_outputr#r#r$process_datasets,   r)rNNN))roswarningstypingrrrrr dataclassesrrrr4torch.nn.functionalnn functionalrDrFr2datasetsrr r r torch.multiprocessingmultiprocessingr spk_wavLMr rSrfilterwarningsrr%rIr rrrrr#r#r#r$sN      IsK6