Initial app commit

app.py

@@ -0,0 +1,142 @@
+import numpy as np
+import pandas as pd
+import torch
+import torchaudio
+from librosa.sequence import dtw as lib_dtw
+from scipy.stats import zscore
+import gradio as gr
+from transformers import HubertModel
+
+time_frame = 1
+expected_sr = 16000
+
+def calculateDistances(snd1, snd2):
+	# Load wav files and resample if neeeded
+	wav_paths = [
+	#"audio/KEI_EF08_EN038.wav",
+	#"audio/KEI_KF04_EN038.wav"
+		snd1 , snd2
+	]
+
+	wavs = []
+	for wav_path in wav_paths:
+		wav, sr = torchaudio.load(wav_path)
+		if sr != expected_sr:
+			print(f"Sampling rate of {wav_path} is not {expected_sr} -> Resampling the file")
+			resampler = torchaudio.transforms.Resample(orig_freq=sr, new_freq=expected_sr)
+			wav = resampler(wav)
+			wav.squeeze()
+		wavs.append(wav)
+		
+	# Generate Features
+	device_name = "cuda" if torch.cuda.is_available() else "cpu"
+	device = torch.device(device_name)
+	print(f'Running on {device_name}')
+
+	
+	model = HubertModel.from_pretrained("facebook/hubert-base-ls960")
+	
+	features = None
+	speaker_len = []
+	layer = 12
+	names = [f.rsplit(".", 1)[0] for f in wav_paths]
+	
+	for wav in wavs:
+		wav_features = model(wav, return_dict=True, output_hidden_states=True).hidden_states[
+			layer].squeeze().detach().numpy()
+		features = wav_features if features is None else np.concatenate([features, wav_features], axis=0)
+		speaker_len.append(wav_features.shape[0])
+		
+	
+	#Create & Fill a dataframe with the details - full dimensionality
+	data_subset, df_subset, hubert_feature_columns = create_df(features, speaker_len, names)
+	
+	speaker1 = names[0]
+	speaker2 = names[1]	
+	#print(speaker1)
+	#print(speaker2)
+			
+	#using full dimensionality hubert_feature_columns
+	features_speaker1 = df_subset[df_subset['speaker'] == speaker1][hubert_feature_columns].to_numpy()
+	features_speaker2 = df_subset[df_subset['speaker'] == speaker2][hubert_feature_columns].to_numpy()
+	features_speaker1, features_speaker2 = mut_normalize_sequences(features_speaker1, features_speaker2, True)
+	
+	distances = librosa_dtw(features_speaker1, features_speaker2)
+	return(distances)
+
+def mut_normalize_sequences(sq1, sq2, normalize: bool):
+	"""
+	Normalize the sequences together by z-scoring each dimension.
+	sq1: numpy array of shape (t1, d)
+	sq2: numpy array of shape (t2, d)
+	normalize: if True, normalize the sequences together
+	"""
+	if normalize:
+		sq1 = np.copy(sq1)
+		sq2 = np.copy(sq2)
+		len_sq1 = sq1.shape[0]
+
+		arr = np.concatenate((sq1, sq2), axis=0)
+		for dim in range(sq1.shape[1]):
+			arr[:, dim] = zscore(arr[:, dim])
+		sq1 = arr[:len_sq1, :]
+		sq2 = arr[len_sq1:, :]
+	return sq1, sq2
+
+
+def librosa_dtw(sq1, sq2):
+	D, wp = lib_dtw(sq1.transpose(), sq2.transpose(), backtrack=True)
+	#out_wp_cols = ["EF08_EN038 index", "frame_range","frame_range_ms", "KF04_EN038 index", 'frame_range',"frame_range_ms", "cost"]
+	out_wp = []
+	samples_out = []
+	for i, j in wp[::-1]:
+		match_cost = D[i, j]/(len(sq1)+len(sq2))
+		#print(match_cost)
+		
+		samples_per_chunk = 0.02 * expected_sr
+		out_wp.append({"spk1_index":i, "spk1_frame_start":(i*samples_per_chunk), "spk1_frame_end":(i*samples_per_chunk+samples_per_chunk), "spk1_time_start":i*0.02, "spk1_time_end":i*0.02+0.02 , "spk2_index":j, "spk2_frame_start":j*samples_per_chunk, "spk2_frame_end":j*samples_per_chunk+samples_per_chunk, "spk2_time_start":j*0.02, "spk2_time_end":j*0.02+0.02 ,"cost":match_cost})
+	
+	return out_wp
+	
+
+def time_txt(time, time_frame=5):
+	if time % time_frame == 0:
+		return f"{round(time * 0.02, 2)}"
+	return ""
+
+def create_df(feats, speaker_len, names):
+	cols = [f"val {i}" for i in range(feats.shape[1])]
+	df = pd.DataFrame(feats, columns=cols)
+	df['idx'] = df.index
+	time_index = {i: speaker_len[i] for i in range(len(speaker_len))}
+	com_time_index = {i: sum(speaker_len[:i]) for i in range(len(speaker_len))}
+	df_speaker_count = pd.Series(time_index)
+	df_speaker_count = df_speaker_count.reindex(df_speaker_count.index.repeat(df_speaker_count.to_numpy())).rename_axis('speaker_id').reset_index()
+	df['speaker_id'] = df_speaker_count['speaker_id']
+	df['speaker_len'] = df['speaker_id'].apply(lambda row: speaker_len[row])
+	df['com_sum'] = df['speaker_id'].apply(lambda i: com_time_index[i])
+	df['speaker'] = df['speaker_id'].apply(lambda i: names[i])
+	df['time'] = df['idx'] - df['com_sum']
+	df['time_txt'] = df[['time', 'speaker_len']].apply(lambda row: time_txt(row['time'], time_frame), axis=1)
+	assert len(df.loc[df['speaker'] == -1]) == 0
+	assert len(df_speaker_count) == len(df)
+	df_subset = df.copy()
+	data_subset = df_subset[cols].values
+	return data_subset, df_subset, cols	
+
+
+#main GradIO interface
+with gr.Blocks() as demo:
+	sound1 = gr.Audio(sources=["microphone", "upload"], type="filepath")
+	sound2 = gr.Audio(sources=["microphone", "upload"], type="filepath")
+	runbtn = gr.Button("Run")
+	json = gr.JSON()
+
+	runbtn.click(fn=calculateDistances, inputs=[sound1, sound2], outputs=json)
+
+
+
+# need to wrap function calls in this. See https://github.com/huggingface/transformers/pull/34966#issuecomment-2538598145
+if __name__ == '__main__':
+	demo.launch(ssr_mode=False)
+	#calculateDistances()

requirements.txt

@@ -0,0 +1,8 @@
+numpy==1.26.4
+pandas
+torch
+torchaudio
+librosa
+Spicy
+gradio
+transformers