1.C. Echolalia_detection_audio-based_model.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c008b86f",
   "metadata": {},
   "source": [
    "# Audio-based model for echolalia detection"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f9b416e9",
   "metadata": {},
   "source": [
    "The following function, `audio_model_echolalia` takes as input a **multi-speaker audio file** and a **.TextGrid diarization** of that audio. The .TextGrid should contain at least one tier, named **'AC'** (Autistic Child), where the speech of the child of whom echolalic utterances will be detected is indicated in **non-empty intervals** (e.g. indicated by 'speech', 'AC'...). Additionally, the file may contain other tiers containing the speech intervals of other speakers. Please remove any non-diarization tiers. If other speakers are diarized, the parameter **'speakers_diarized'** must be set to 'all' (the default value); then, the model will compare the AC's utterances with utterances of other speakers starting at most 10 seconds before the start of the AC's utterance. If only the speech of the AC is indicated, 'speakers_diarized' must be set to 'AC', and the AC's utterances will be compared with intervals within 10 seconds before the start of the AC's utterance where the AC does not speak (i.e., where no other AC-interval occurs). The output is a **dataframe** containing the predictions ('non-echolalic' or 'echolalic') for each pair of utterances of the AC and another speaker, including the name of the other speaker's tier (if speakers_diarized='AC', then the other speaker is indicated as 'other') and the timestamps of both utterances. Furthermore, a new **.TextGrid** is created on the basis of the input .TextGrid, containing the annotations of the predicted source and echolalic intervals on two different tiers ('source' and 'echolalia'). The output path to this file must be specified in the parameter 'output_path'."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f50834ff",
   "metadata": {},
   "source": [
    "*Note*: the current trained classifier does not yet achieve satisfying results. This script will be adapted with a new version of the classifier once developed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c48b6237",
   "metadata": {},
   "outputs": [],
   "source": [
    "def audio_model_echolalia(input_audio= None, textgrid_file= None, output_path= None, speakers_diarized='all'):\n",
    "    \"\"\"Takes an audio file and its diarization in .TextGrid as input, \n",
    "    makes a prediction on the label (echolalic or not) for each suitable utterance-pair\n",
    "    and returns a dataframe containing these predictions for each pair,\n",
    "    along with an annotation of the predictions in a new TextGrid\"\"\"\n",
    "    \n",
    "    \n",
    "    # Load libraries\n",
    "    import pickle\n",
    "    from sklearn.linear_model import LogisticRegression\n",
    "    import pandas as pd\n",
    "    from pydub import AudioSegment\n",
    "    from praatio import textgrid\n",
    "    import librosa\n",
    "    import numpy as np\n",
    "    from dtw import dtw\n",
    "    import sklearn.preprocessing\n",
    "    import itertools\n",
    "    import tgt\n",
    "        \n",
    "        \n",
    "    # EXTRACT SPEECH INTERVALS\n",
    "    \n",
    "    def get_speech_intervals(tier, empty=False):\n",
    "        \"\"\"Takes a tier name as input and returns either the timestamps of the non-empty intervals ('empty= False')\n",
    "        or those of the empty intervals ('empty=True') of that tier\"\"\"\n",
    "        entries= input_tg.getTier(tier).entries\n",
    "        intervals= []\n",
    "        for entry in entries:\n",
    "            if (entry.label and empty==False) or (not entry.label and empty==True):\n",
    "                intervals.append((entry.start, entry.end))\n",
    "        return intervals\n",
    "    \n",
    "    # Get speech intervals of AC and other speaker(s)\n",
    "    input_tg= textgrid.openTextgrid(textgrid_file, includeEmptyIntervals= True)\n",
    "    \n",
    "    child_intervals= []\n",
    "    other_speaker_intervals= {}\n",
    "    \n",
    "    if speakers_diarized== 'all':\n",
    "        all_tiers= input_tg.tierNames\n",
    "        all_intervals= {}\n",
    "        for tier in all_tiers:\n",
    "            if tier== 'AC':\n",
    "                child_intervals= get_speech_intervals(tier)\n",
    "            else:\n",
    "                other_speaker_intervals[tier]= get_speech_intervals(tier)\n",
    "    \n",
    "    elif speakers_diarized== 'AC':\n",
    "        child_intervals= get_speech_intervals('AC')\n",
    "        other_speaker_intervals['other']= get_speech_intervals('AC', empty=True)\n",
    "    \n",
    "    # PREPARE THE AUDIO PROCESSING\n",
    "    \n",
    "    def audio_segment_to_np(audiosegment):\n",
    "        \"\"\"Converts an 'Audiosegment' object of the 'audiosegment' library\n",
    "        into a numpy array\"\"\"\n",
    "        samples = audiosegment.get_array_of_samples() # function of AudioSegment library\n",
    "        samples = np.array(samples)\n",
    "        samples = samples.astype(np.float32) / np.iinfo(samples.dtype).max\n",
    "        return samples\n",
    "    \n",
    "    # DEFINE FUNCTIONS FOR FEATURE EXTRACTION FROM AUDIO SIGNAL\n",
    "    \n",
    "    n_fft = 2048\n",
    "    hop_length= 512\n",
    "    \n",
    "    features= ['mfcc', 'melspectrogram', 'lpc']\n",
    "    \n",
    "    def compute_lpc_2d(y, sr, order=13, frame_length=n_fft, hop_length=hop_length):\n",
    "        # so that lpc coefficients are also computed per frame just as for mfcc and melspectrogram\n",
    "        \n",
    "        # Frame the audio signal\n",
    "        y_padded = np.pad(y, (0, frame_length - len(y) % hop_length), mode='constant')\n",
    "        frames = librosa.util.frame(y_padded, frame_length=frame_length, hop_length=hop_length).T\n",
    "        # Compute LPC for each frame\n",
    "        lpc_coeffs = [librosa.lpc(frame, order=order) for frame in frames]\n",
    "        # Stack LPC coefficients into a 2D array\n",
    "        lpc_2d = np.vstack(lpc_coeffs)\n",
    "        return lpc_2d\n",
    "    \n",
    "    def get_feature_array(feature, segment, sr):\n",
    "        if feature== 'mfcc':\n",
    "            # use 'audio_segment_to_np()' function previously defined\n",
    "            feature_vector= librosa.feature.mfcc(y= audio_segment_to_np(segment), n_mfcc=13, hop_length= hop_length, n_fft= n_fft, sr=sr).squeeze()\n",
    "        elif feature== 'melspectrogram':\n",
    "            feature_vector= librosa.feature.melspectrogram(y= audio_segment_to_np(segment), hop_length= hop_length, n_fft= n_fft, sr=sr, n_mels=32).squeeze()\n",
    "        elif feature== 'lpc':\n",
    "            feature_vector= compute_lpc_2d(y= audio_segment_to_np(segment), sr=sr).squeeze().T\n",
    "        \n",
    "        ready_feature_vector= sklearn.preprocessing.minmax_scale(feature_vector, axis=1)\n",
    "                \n",
    "        return ready_feature_vector\n",
    "\n",
    "    \n",
    "    # LOAD THE AUDIO FILE \n",
    "    \n",
    "    audio = AudioSegment.from_wav(input_audio) # Load the audio and create an 'Audiosegment' object\n",
    "    sr = librosa.load(input_audio)[1] # Get sample rate\n",
    "    \n",
    "    \n",
    "    # CREATE THE FEATURE DATAFRAME:\n",
    "    \n",
    "    # Initalize dataframe\n",
    "    df= pd.DataFrame(columns=['s2_tier', 'AC_int', 's2_int'] +[f'DTW_{feature}' for feature in features] \\\n",
    "                     + [f'DTW_{f1}_{f2}' for f1,f2 in list(itertools.combinations(features,2))]+ ['DTW_combined'])\n",
    "    row_df= 0\n",
    "    \n",
    "    \n",
    "    # Compare the utterances of the autistic child\n",
    "    # with those of all other speakers:\n",
    "    \n",
    "\n",
    "    for other_speaker in other_speaker_intervals:\n",
    "        s2_tier= other_speaker\n",
    "        \n",
    "        # Get speech intervals for this speaker:\n",
    "        s2_intervals = other_speaker_intervals[other_speaker]\n",
    "\n",
    "\n",
    "        for start, end in child_intervals: # Iterate over the child_intervals list\n",
    "\n",
    "            child_int= (start, end) # store the current timestamps in a tuple\n",
    "\n",
    "            start_child = start * 1000\n",
    "            end_child = end * 1000\n",
    "            child_segment = audio[start_child:end_child] # create a segment from the audio between start and end timestamps\n",
    "\n",
    "            \n",
    "            # Get features:\n",
    "            child_features= []\n",
    "            \n",
    "\n",
    "            for feature in features:\n",
    "                # use function 'get_feature_array()' to get features from 'Audiosegment' object\n",
    "                # Iterate through the 'features' list (= parameter of the function 'feature_df_echolalia()') \n",
    "                # to compute all the features\n",
    "                child_feature= get_feature_array(feature, child_segment, sr)\n",
    "                \n",
    "                # Store all feature arrays in a list\n",
    "                child_features.append(child_feature)\n",
    "            \n",
    "            # Concatenate different combinations of feature arrays\n",
    "            child_combinations = [np.concatenate((f1, f2)) for f1, f2 in itertools.combinations(child_features, 2)]\n",
    "            child_all_concatenated= np.concatenate([feature for feature in child_features])    \n",
    "                \n",
    "                \n",
    "                \n",
    "            # Now iterate through the speech interval dictionary of speaker 2 to compare the utterance of the child\n",
    "            # with all suitable utterances of speaker 2 (starting 10 seconds or less before the start of the child's utterance)\n",
    "\n",
    "            for s, e in s2_intervals: \n",
    "\n",
    "                start_s2 = s * 1000\n",
    "                end_s2 = e * 1000\n",
    "                \n",
    "                # if other speaker interval starts at most 10 seconds before child interval:\n",
    "                # or cut off an interval at 10 seconds before child starts in case of undiarized audio:\n",
    "\n",
    "                if 0 < start_child - start_s2 <= 10*1000\\\n",
    "                    or (0 < start_child - end_s2 < 10*1000 and speakers_diarized=='AC'): \n",
    "                    \n",
    "                    if 0 < start_child - end_s2 < 10*1000 and speakers_diarized=='AC' and not 0 < start_child - start_s2 <= 10*1000: \n",
    "                        start_s2= start_child- 10*1000\n",
    "\n",
    "                    s2_segment = audio[start_s2:min(start_child,end_s2)] # create an audiosegment\n",
    "                    \n",
    "                    s2_int= (start_s2/1000, min(start_child, end_s2)/1000)\n",
    "\n",
    "                    \n",
    "                    # Get features\n",
    "                    s2_features= []\n",
    "                                                  \n",
    "\n",
    "                    for feature in features:\n",
    "                        s2_feature= get_feature_array(feature, s2_segment, sr)\n",
    "                        # Store all feature arrays in a list\n",
    "                        s2_features.append(s2_feature)\n",
    "\n",
    "                    s2_combinations = [np.concatenate((f1, f2)) for f1, f2 in itertools.combinations(s2_features, 2)]\n",
    "                    s2_all_concatenated= np.concatenate([feature for feature in s2_features])\n",
    "                    \n",
    "                    # Get distance metrics for child and speaker 2 utterances:\n",
    "                    \n",
    "                    distance_features= []\n",
    "                    \n",
    "                    # Iterate over lists of feature arrays of both speakers and compute the normalized distance\n",
    "\n",
    "                    for child_feature, s2_feature in zip(child_features, s2_features):\n",
    "                        distance_feature= dtw(child_feature.T, s2_feature.T, distance_only=True).normalizedDistance\n",
    "                        distance_features.append(distance_feature)\n",
    "                    \n",
    "                    for child_combination, s2_combination in zip(child_combinations, s2_combinations):\n",
    "                        distance_feature= dtw(child_combination.T, s2_combination.T, distance_only=True).normalizedDistance\n",
    "                        distance_features.append(distance_feature)\n",
    "                    \n",
    "                    # dtw computation on concatenation of all feature arrays:\n",
    "                    dtw_combination= dtw(child_all_concatenated.T, s2_all_concatenated.T, distance_only=True).normalizedDistance\n",
    "                    \n",
    "                    \n",
    "                    # ADD EVERYTHING TO THE DATAFRAME\n",
    "                    df.loc[row_df] = [s2_tier, (child_int), (s2_int)] + distance_features + [dtw_combination]\n",
    "                    # Move to a new row in the output dataframe and start a new iteration of the intervals of speaker 2\n",
    "                    row_df+=1\n",
    "        \n",
    "        \n",
    "    # MAKE PREDICTIONS\n",
    "\n",
    "    # Load pretrained model and configurations\n",
    "    with open('Trained_best_classifier_echolalia.pkl', 'rb') as f:\n",
    "        model_config = pickle.load(f)\n",
    "    model= model_config['model_obj']\n",
    "    best_features= model_config['best_features']\n",
    "    threshold= model_config['threshold']\n",
    "\n",
    "    # Define function for predictions\n",
    "    def predict(model, X, threshold):\n",
    "        probs = model.predict_proba(X) \n",
    "        return (probs[:, 1] > threshold).astype(int)\n",
    "\n",
    "    # Predict outcomes\n",
    "    X= df[best_features]\n",
    "    preds= pd.DataFrame(predict(model=model, X=X, threshold=threshold), columns= ['prediction_binary'])\n",
    "    df= pd.concat([df, preds], axis=1)\n",
    "    df['prediction'] = df['prediction_binary'].apply(lambda x: 'echolalic' if x == 1 else 'non-echolalic')\n",
    "    \n",
    "    \n",
    "    # MAKE TEXTGRID\n",
    "    tg = tgt.io.read_textgrid(textgrid_file, encoding='utf-16')\n",
    "    source_tier = tgt.IntervalTier(start_time=0, name='source')\n",
    "    rep_tier = tgt.IntervalTier(start_time=0, name='echolalia')\n",
    "    \n",
    "    all_sources=[]\n",
    "    all_echoes=[]\n",
    "    \n",
    "    df_echo= df.loc[df['prediction']== 'echolalic']\n",
    "    \n",
    "    for index, row in df_echo.iterrows():\n",
    "        start_child, end_child = row['AC_int']\n",
    "        start_s2, end_s2 = row['s2_int']\n",
    "        \n",
    "        if (start_s2, end_s2) not in all_sources:\n",
    "            all_sources.append((start_s2, end_s2))\n",
    "            source_interval= tgt.Interval(start_time=float(start_s2), end_time=float(end_s2), text= 'source')\n",
    "            source_tier.add_interval(source_interval)\n",
    "        \n",
    "        if (start_child, end_child) not in all_echoes:\n",
    "            all_echoes.append((start_child, end_child))\n",
    "            rep_interval= tgt.Interval(start_time=float(start_child), end_time=float(end_child), text= 'echolalic')\n",
    "            rep_tier.add_interval(rep_interval)\n",
    "            \n",
    "    # Output the TextGrid\n",
    "    tg.add_tier(source_tier)\n",
    "    tg.add_tier(rep_tier)\n",
    "    tgt.write_to_file(tg, output_path, format='short')\n",
    "            \n",
    "        \n",
    "    return df[['s2_tier', 'AC_int', 's2_int', 'prediction']]"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}