app.py

import numpy as np
import pandas as pd
import torch
from transformers import HubertModel
import torchaudio
from scipy.stats import zscore
from librosa.sequence import dtw as lib_dtw
import gradio as gr
import spaces
from itertools import combinations
import os

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):
    """
    Compute the Dynamic Time Warping distance between two sequences.
    sq1: numpy array of shape (t1, d)
    sq2: numpy array of shape (t2, d)
    """
    return lib_dtw(sq1.transpose(), sq2.transpose())[0][-1, -1]


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


def calc_distance(df_subset, speaker1, speaker2, cols):
    features_speaker1 = df_subset[df_subset['speaker'] == speaker1][cols].to_numpy()
    features_speaker2 = df_subset[df_subset['speaker'] == speaker2][cols].to_numpy()
    features_speaker1, features_speaker2 = mut_normalize_sequences(features_speaker1, features_speaker2, True)
    
    distance = librosa_dtw(features_speaker1, features_speaker2)
    distance = distance / (len(features_speaker1) + len(features_speaker2))
    return distance
    

# Model's label rate is 0.02 seconds. To not overflow the plot, time is shown every 5 samples (0.1 seconds).
# To change that, change "time_frame" below.
time_frame = 5

# @spaces.GPU(duration=120)
def grMeasureDistance(wav_paths, map_file):
	map_df = pd.read_csv(map_file)
	#for index, row in map_df.iterrows():
    #	gr.Info(row['File1'].astype(str))	

	if wav_paths is None:
		gr.Warning("Please upload some sound files!")
		return None
	seed = 31415
	# Load wav files
	expected_sr = 16000
	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]
	# Not batched to know the actual seqence shape
	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
	data_subset, df_subset, hubert_feature_columns = create_df(features, speaker_len, names)

	# Evaluate Distance of all speaker pairs
	distances_list = []
	#wav_pairs = list(combinations(names, 2))
	wav_pairs = []
			
	for index, row in map_df.iterrows():
		#file1_index = find_substring_index(names, row['S1'])
		#file2_index = find_substring_index(names, row['S2'])
		file1_index = find_exactstring_index(names, row['S1'])
		file2_index = find_exactstring_index(names, row['S2'])
		
		if(file1_index != -1 and file2_index != -1):
			wav_pairs.append((names[file1_index], names[file2_index]))

	#print(len(wav_pairs))
	for wav_pair in wav_pairs:
		S1 = wav_pair[0]
		S2 = wav_pair[1]
		#print("*** " + S1 + " *** " + S2 + " ***")
		
		# FULL DIMENSIONALITY
		distance = calc_distance(df_subset, S1, S2, hubert_feature_columns)
		distances_list.append([os.path.basename(S1), os.path.basename(S2), distance])
	return distances_list

def find_substring_index(string_list, substring):
    for index, string in enumerate(string_list):
        if substring in string:
            return index
    return -1

def find_exactstring_index(string_list, substring):
    for index, string in enumerate(string_list):
        if substring == os.path.basename(string):
            return index
    return -1

#csv export function
def export_csv(d):
	if(len(d.iloc[0,0])>0):
		d.to_csv("output.csv")
		return gr.File(value="output.csv", visible=True)

def clearInterface():
	return gr.File(interactive=False, visible=False), gr.Dataframe(value=None)

#main GradIO interface
with gr.Blocks() as demo:
	gr.Markdown(
	"""
	# PS3-PDM: Perceptual Similarity Space for Speech-Pairwise Distance Matrix
	## Project
	- Perceptual Similarity Space for Speech
	- Supported by the National Science Foundation (DRL 2219843) and Binational Science Foundation (2022618)
	
	## Description
	Takes a set of utterance files (.wav format) and a two column .csv *map file*. Generates pair-wise distances of the corresponding trajectories in HuBERT embedding spaces. Methods are based on Kim et al. (2025) and Chernyak et al. (2024). We report distances for embeddings in the original embedding space of transformer layer 12, without any form of dimensionality reduction.
	 """)
 
	with gr.Accordion("Click for more details", open=False):
		gr.Markdown(
		"""
		## Project team
		- [Matt Goldrick](https://faculty.wcas.northwestern.edu/matt-goldrick/)
		- [Yossi Keshet](https://keshet.net.technion.ac.il/)
		- [Ann Bradlow](https://faculty.wcas.northwestern.edu/ann-bradlow/)
		- [Seung-Eun Kim](https://seungeun-kim.github.io/)
		- [Roni Chernyak](https://bronichern.github.io/)
		- [Chun Liang Chan](https://staff.wcas.northwestern.edu/clc500/)
 
		## Requirements
		- All speech files must be in a single channel .wav format. (Note: It is recommended to normalize the loudness of the files.)
		- Stereo or multi channel audio files should be reduced to a single channel before processing. A [Praat](https://www.fon.hum.uva.nl/praat/) script that extracts a single channel from a directory of .wav files is available [here](https://huggingface.co/spaces/MLSpeech/perceptual-similarity/resolve/main/extractSingleChannel.praat).
		- All speech files that are being compared must contain productions of the identical linguistic content (i.e., same words in same order). 
		- For example, the files may contain productions of a given sentence by different talkers, or by a single talker under different conditions. 
		- Note that while the utility will return distance values for files with different content the interpretation of these values is meaningless.
 
		## Usage
		- Upload wav files.
		- Upload csv *map file* that contains two columns with the headers "S1" and "S2". 
		
		| S1              | S2              |
		| --------------- | --------------- |
		| my_sentence_1_1 | my_sentence_1_2 |
		| my_sentence_2_1 | my_sentence_2_2 |
		| etc...          | etc...          |
		
		- Example csv map file available [here](https://huggingface.co/spaces/MLSpeech/perceptual-similarity/resolve/main/example.csv)
		- Each cell should contain the name of a wav file that was uploaded **without the ".wav" extension**
		- Distances will be measured by comparing the files in the "S1" column to the files in the "S2" column		
		- Click 'run' to get distances.
		- Output (download in .csv format) consists of a table with 4 columns (index, S1, S2, distance)
 
		## Capacity limits
		- Processing time is approximately 7 times the duration of the input audio files. For example, a minute of audio can take up to 7 minutes to process. If processing is taking longer than expected, please refresh the page and reupload your files.
		- Ocassionally the app may fail when uploading a large number of files in a single session. Consider running in smaller batches if possible.
		- Networks with slower upload speeds may experience reduced performance.
 
		## References
		- Kim, S-E, Chernyak, B. R., Keshet, J., Goldrick, M., & Bradlow, A. R. (2025).  Predicting relative intelligibility from inter-talker distances in a perceptual similarity space for speech.  Psychonomic Bulletin and Review. https://doi.org/10.3758/s13423-025-02652-2
		- [for full-dimensional data and analysis of Kim et al. (2025), [see this OSF](https://doi.org/10.17605/osf.io/v5tru) repository] Kim, S.-E., Goldrick, M., & Bradlow, A. R. (2025). Predicting relative talker intelligibility using HuBERT perceptual similarity space distances (full-dimension). https://doi.org/10.17605/osf.io/v5tru
		- Chernyak, B. R., Bradlow, A. R., Keshet, J., & Goldrick, M., & (2024).  A perceptual similarity space for speech based on self-supervised speech representations.  Journal of the Acoustical Society of America, 155(6), 3915-3929.  

		"""
		)
	with gr.Row():
		inputFiles = gr.File(label="wav files", file_count="multiple", file_types=[".wav"])
		mapFile = gr.File(label="map file", file_count="single", file_types=[".csv", ".txt"])		
		with gr.Column():
			runbtn = gr.Button("Run")
			csv = gr.File(interactive=False, visible=False)
			dataframe = gr.Dataframe(headers=["S1", "S2", "distance"], visible=True, row_count=[1, 'dynamic'])


	runbtn.click(fn=grMeasureDistance, inputs=[inputFiles, mapFile], outputs=dataframe)
	dataframe.change(export_csv, inputs=dataframe, outputs=csv)
	
	inputFiles.change(fn=clearInterface, inputs=None, outputs=[csv, dataframe])

if __name__ == "__main__":
	demo.launch(ssr_mode=False)