o oi\8@sddlZddlZddlZddlmZmZddlmZddl m Z m Z ddl m Z ddlmZmZdd lmZdd lmZdd lmZGd d d ZdS)N) load_datasetconcatenate_datasets)List) ConfigManager DatasetConfig)DatasetProcessor)worker_process AudioWorker)reader_worker_process)SpeakerEmbeddingProcessor)SpeakerEmbeddingGrouperc@sVeZdZdZddefddZddZdd Zd d Zd e fd dZ ddZ ddZ dS)PipelineManagerz,Manages the entire audio processing pipeline config.yaml config_pathcCsvt||_|j|_|j|_|j|_tj |_ t j |jjddz tjdddWdSty:YdSw)NT)exist_okspawn)force)rconfig_managerget_base_settings base_settingsget_save_settings save_settingsget_speaker_embedding_settingsspeaker_emb_settingstorchcuda device_countnum_gpusosmakedirsOUT_DIRmpset_start_method RuntimeError)selfrr'K/home/ubuntu/kanitts-2-dataset-pipeline/utils/nanocodec/pipeline_manager.py__init__s      zPipelineManager.__init__cs|jjsdS|jjs dS|jj|j}tfdd|D}|s$dStdtdtdtddtd d td td td  }|sRd}||j_td|dtddS)z Check whether the clustering speaker column name conflicts with any add_constant key in the dataset configs. If a conflict is found, prompt the user to supply an alternative column name. Nc3s|] }|vVqdSN)get_constant_columns).0dsconflict_columnr'r( 0s  z=PipelineManager._check_clustering_conflict..= ============================================================u<⚠️ WARNING: Clustering column naming conflict detected!<============================================================z& 'do_clusters: true' will produce a 'z' column (integer cluster IDs),z but 'z<' is also defined in 'add_constant' (string constant value).zC Please provide a different name for the clustering output column.<------------------------------------------------------------z4 New clustering column name [default: cluster_id]: cluster_idu( ✅ Clustering column will be named: ''============================================================= ) radd_speaker_emb do_clustersclustering_speaker_columnr get_datasetsanyprintinputstrip)r&datasets has_conflictnew_namer'r.r(_check_clustering_conflict"s@     z*PipelineManager._check_clustering_conflictcCs@|jj}|r |js dS|j}|j}g}|D]}||vr)|d|jdq|j}|j r=|j r=|j |kr=|d|D]}|j |krP|d|jdq?|rbt d|dd|dSt d t d t d t d |dt d t dt dtd}|dkrd|_t dnt dt ddS)a Warn the user early if 'group_sp_emb.do_this: true' but the group_by_column_name cannot be traced to any known source: - add_constant in any dataset config - clustering output (do_clusters: true + matching column name) - speaker_column_name of any dataset config If no source is found, prompt the user to disable grouping via interactive input. Nzadd_constant in 'r5zclustering (do_clusters: true)zspeaker_column_name in 'z#[Validation] group_sp_emb: column 'z' provided by: z, r1u6⚠️ WARNING: group_sp_emb column source not found!r2z0 'group_sp_emb.do_this: true' requires column 'zR but it will NOT be produced by add_constant, clustering, or speaker_column_name.uN It may already exist in the source dataset — proceed only if you are sure.r3z, Disable group_sp_emb for this run? [Y/n]: nFu ✅ group_sp_emb disabled.uB ⚠️ Proceeding — will fail at runtime if column is absent.r6)r group_embdo_thisgroup_by_column_namerr:r+appendnamer7r8r9speaker_column_namer<joinr=r>lower)r&rD group_colr?sourcesr-spanswerr'r'r(_check_group_emb_feasibilityQs`          z,PipelineManager._check_group_emb_feasibilitycCsJtd|j|||jdkrtdtd|jddS)z&Validate configuration and environmentu 🔍 Validating configuration...ru!❌ ERROR: No CUDA devices found!u ✅ Found z GPU(s)N)r<rvalidate_datasetsrBrPrr%r&r'r'r(validates  zPipelineManager.validatedataset_configcstddtdjtdjtdt}|jjjdjj}jj rctddtdtdt j}| | j }||_||_tdtd |jn8|r|jrtddtd tdt|jjd }|| }||_||_td td |jtjjjd tdtdtjtdjtdjjtdjjtdjjtdjjdtdjjtdfddtjD}|D]}| q| fddtjjD} gtjjD]} t} | | | _|j| _!| qfddtjjD} | D]} | qCzn| D]} | "qNtjD] } #t$j%q[|D]}|"qgtdtdjdt&j'(jjrfddt&)jjD}|rt*fd d!|D}td"t+|d#|d$d%d&Wd+SWd+SWd+St,ytd'| D]} | -q|D]}|-q| D] } | j"d(d)q|D] }|j"d(d)qtd*w),z-Process a single dataset through the pipeline r2u🎯 Processing dataset: u📝 Dataset prefix: )num_procu────────────────────────────────────────────────────────────u'🎙️ Speaker Embedding: starting...u✅ Speaker Embedding done.z+ Columns preserved through tokenization: u0📊 Group Embeddings (standalone): starting...)settingsembedding_columnu✅ Group Embeddings done.)maxsizeu" 🚀 Starting processing pipelineu💻 CUDA available: u🔥 GPU workers: u📖 Reader workers: u ⚙️ Dataset load processes: u📁 Output directory: u🗂️ Lines per file: ,u📦 Queue size: r3c sFg|]}tjt|jjjjjjjjjjj jj f dqS)targetargs) r#Processr rr"dataset_prefix gzip_level buffer_sizelines_per_file num_readers audio_codecr,i)rTqr&r'r( s z:PipelineManager.process_single_dataset..csg|] }jjj|dqS)) num_shardsindex)shardrrcre)datasetr&r'r(rhscs*g|]}tjt|jj|fdqSr[)r#r^r rrcre)rgr&shard_processorsr'r(rhsr1u 🎉 Dataset z processed successfully!csg|] }|jr|qSr') startswithr_r,f)rTr'r(rh%s  c3s*|]}tjtjjj|VqdSr*)r pathgetsizerJrr"rorRr'r(r0(s  z9PipelineManager.process_single_dataset..u📊 Generated z files for this dataset, size: i@z.2fz GBu. ⚠️ Interrupted! Terminating processes... )timeoutu🛑 All processes terminatedN).r<rHr_r rrload_dataset_num_procrrDr7r process get_datasetaudio_column_namerlget_preserved_columnspreserve_columnsrErrXgroupr#Queueqsizerr is_availablerrcr"rbrangestartrGrJputr SENTINELr rqexistslistdirsumlenKeyboardInterrupt terminate)r&rT processorrD sp_processorupdated_datasetgrouperworkerspsharded_datasetsrf shard_procreadersprfiles total_sizer')rlrTrgr&rmr(process_single_datasets                  z&PipelineManager.process_single_datasetcCstddtdtdtj|jjd}td|td|jjdddd }td t|d |jj rMtd |jj | |jj td |jj rgtd|jj |j |jj ddtdtddtdtddS)z@Assemble all processed shards into final dataset and save/uploadrUr2u0🔨 Assembling final dataset from all shards...z *.jsonl.gzu📂 Loading shards from: jsontrain no_checks)data_dir data_filessplitverification_modeu✅ Final dataset assembled: z samplesu! 💾 Saving dataset locally to: u✅ Dataset saved to disku+ ☁️ Uploading dataset to HuggingFace: T)privateu'✅ Dataset uploaded to HuggingFace Hubu%🎊 Pipeline completed successfully!N) r<r rqrJrr"rrrlocal save_to_disk hf_upload push_to_hub)r& shard_files final_datasetr'r'r(assemble_and_save_final_datasetAs0 z/PipelineManager.assemble_and_save_final_datasetcCsp||j}tdt|dt|dD]\}}td|dt|||q|tddS)z*Run the complete pipeline for all datasetsu 📋 Found z dataset(s) to processru 🔄 Processing dataset /u 👋 Pipeline finished!N)rSrr:r<r enumeraterr)r&r?idxrTr'r'r(runes   zPipelineManager.runN)r) __name__ __module__ __qualname____doc__strr)rBrPrSrrrrr'r'r'r(rs/H  $r)rmultiprocessingr#r r?rrtypingrrrrdataset_processorr audio_workerr r reader_workerr speaker_emb.processorr speaker_emb.group_embeddingsrrr'r'r'r(s