<jupyter_start><jupyter_text>Get accoustic features of disclosures via Praat<jupyter_code>import glob
import math
import os
import re
import subprocess
import numpy as np
import pandas as pd
import parselmouth 

from parselmouth.praat import call

# Set working directory
os.chdir('/Volumes/GoogleDrive/My Drive/UCR/UCR SNL/Research Projects/SyncDisclosures/Pilot/Analysis')
# Set parameters
f0min = 75 # 75 for males, 100 for females
f0max = 500 # 300 for males, 500 for females
time_step = 1.0 # 1 second<jupyter_output><empty_output><jupyter_text>Maybe parallelize this in the future: https://medium.com/@mjschillawski/quick-and-easy-parallelization-in-python-32cb9027e490<jupyter_code># Loop through participant folders & self-disclosure recordings on Google Drive
all_accoustics = []
for subdir, dirs, files in os.walk('Data/recordings'):
    for file in files: 
        
        # Only include real participants' data and self (not partner's) disclosures
        if '/P' in os.path.join(subdir, file) and 'self' in os.path.join(subdir, file):
            
            # Convert m4a file to wav
            file_path = os.path.join(subdir, file)
            wav_file_path = file_path.replace(".m4a",".wav")
            subprocess.call(['ffmpeg', '-i', file_path, wav_file_path])
            
            # Import audio recording
            sound = parselmouth.Sound(wav_file_path)
            
            # Delete wav file to save space
            os.remove(wav_file_path)
            
            # Get participant ID
            regex = re.compile('/')
            ID = regex.split(wav_file_path)[2].replace("P","")
            
            # Get disclosure name
            regex = re.compile('_')
            disclosure = regex.split(wav_file_path)[2].replace(".wav","")
            
            print('Processing P' + ID + ', disclosure ' + disclosure)
            
            # Get audio start time and end time
            tmin = math.floor(call(sound, "Get start time"))
            tmax = math.ceil(call(sound, "Get end time"))
            
            # Get timestamps
            times = list(range(tmin, tmax))
            
            # Compute pitch
            pitch = sound.to_pitch(time_step, f0min, f0max)
            pitch_values = pitch.selected_array['frequency']
            if len(pitch_values) < tmax:
                pitch_values = np.append( # if missing time points, pad with NaN
                    pitch_values, np.zeros(tmax - len(pitch_values)) + np.nan
                )
            
            # Compute intensity
            intensity = sound.to_intensity(f0min, time_step)
            intensity_values = intensity.values[0]
            if len(intensity_values) < tmax:
                intensity_values = np.append( # if missing time points, pad with NaN
                    intensity_values, np.zeros(tmax - len(intensity_values)) + np.nan
                )
            
            # Compute jitter
            jitter = []
            for time in range(tmin, tmax):
                sound_part = call(sound, "Extract part", time - time_step, time, "rectangular", 1, "yes")
                pointProcess = call(sound_part, "To PointProcess (periodic, cc)", f0min, f0max)
                jitter_at_time = call(pointProcess, "Get jitter (local)", 0, 0, 0.0001, 0.02, 1.3)
                jitter.append(jitter_at_time)
                
            # Compute shimmer
            shimmer = []
            for time in range(tmin, tmax):
                sound_part = call(sound, "Extract part", time - time_step, time, "rectangular", 1, "yes")
                pointProcess = call(sound_part, "To PointProcess (periodic, cc)", f0min, f0max)
                shimmer_at_time = call([sound, pointProcess], "Get shimmer (local)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
                shimmer.append(shimmer_at_time)
                
            # Compute formants
            f1_list = []
            f2_list = []
            f3_list = []
            f4_list = []
            formants = call(sound, "To Formant (burg)", 0.0025, 5, 5000, 0.025, 50)
            for time in range(tmin, tmax):
                f1 = call(formants, "Get value at time", 1, time, 'Hertz', 'Linear')
                f2 = call(formants, "Get value at time", 2, time, 'Hertz', 'Linear')
                f3 = call(formants, "Get value at time", 3, time, 'Hertz', 'Linear')
                f4 = call(formants, "Get value at time", 4, time, 'Hertz', 'Linear')
                f1_list.append(f1)
                f2_list.append(f2)
                f3_list.append(f3)
                f4_list.append(f4)
             
            # Create data frame of accoustic features & append to master list
            accoustics_df = pd.DataFrame({
                'ID': ID,
                'disclosure': disclosure,
                'time': times, 
                'pitch': pitch_values, 
                'intensity': intensity_values,
                'jitter': jitter,
                'shimmer': shimmer,
                'f1': f1_list,
                'f2': f2_list,
                'f3': f3_list,
                'f4': f4_list
                })
            all_accoustics.append(accoustics_df)
# Save as csv
all_accoustics_df = pd.concat(all_accoustics, ignore_index=True)
all_accoustics_df.to_csv('Data/processed/accoustic_features.csv', index=False)<jupyter_output><empty_output><jupyter_text>Test code for one individual recording<jupyter_code># Import audio recording
sound = parselmouth.Sound('Data/test.wav')
# Set parameters
f0min = 75 #75 for males, 100 for females
f0max = 500 # 300 for males, 500 for females
time_step = 1.0
tmin = math.floor(call(sound, "Get start time"))
tmax = math.ceil(call(sound, "Get end time"))<jupyter_output><empty_output><jupyter_text>Compute pitch<jupyter_code>pitch = sound.to_pitch(time_step, f0min, f0max)
pitch_values = pitch.selected_array['frequency']
len(pitch_values)<jupyter_output><empty_output><jupyter_text>Compute intensity<jupyter_code>intensity = sound.to_intensity(f0min, time_step)
intensity_values = intensity.values[0]
len(intensity_values)<jupyter_output><empty_output><jupyter_text>Compute jitter (associated with rougher voices)<jupyter_code># local is most commonly used type of jitter
# looks like i'll have to use a for loop to get jitter over time...
# should check if this is kosher & what a good window size would be
# seems like betterup just used a one second window

jitter = []
for time in range(tmin, tmax):
    sound_part = call(sound, "Extract part", time - time_step, time, "rectangular", 1, "yes")
    pointProcess = call(sound_part, "To PointProcess (periodic, cc)", f0min, f0max)
    jitter_at_time = call(pointProcess, "Get jitter (local)", 0, 0, 0.0001, 0.02, 1.3)
    jitter.append(jitter_at_time)

#len(jitter)<jupyter_output><empty_output><jupyter_text>Compute shimmer<jupyter_code>shimmer = []
for time in range(tmin, tmax):
    sound_part = call(sound, "Extract part", time - time_step, time, "rectangular", 1, "yes")
    pointProcess = call(sound_part, "To PointProcess (periodic, cc)", f0min, f0max)
    shimmer_at_time = call([sound, pointProcess], "Get shimmer (local)", 0, 0, 0.0001, 0.02, 1.3, 1.6)
    shimmer.append(shimmer_at_time)

#len(shimmer)<jupyter_output><empty_output><jupyter_text>Compute formants<jupyter_code>formants = call(sound, "To Formant (burg)", 0.0025, 5, 5000, 0.025, 50)

f1_list = []
f2_list = []
f3_list = []
f4_list = []

for time in range(tmin, tmax):
    f1 = call(formants, "Get value at time", 1, time, 'Hertz', 'Linear')
    f2 = call(formants, "Get value at time", 2, time, 'Hertz', 'Linear')
    f3 = call(formants, "Get value at time", 3, time, 'Hertz', 'Linear')
    f4 = call(formants, "Get value at time", 4, time, 'Hertz', 'Linear')
    f1_list.append(f1)
    f2_list.append(f2)
    f3_list.append(f3)
    f4_list.append(f4)<jupyter_output><empty_output><jupyter_text>Save to data file<jupyter_code>times = list(range(tmin, tmax))
ID='01'
df = pd.DataFrame({
    'ID': ID,
    'time': times, 
    'pitch': pitch_values, 
    'intensity': intensity_values,
    'jitter': jitter,
    'shimmer': shimmer,
    'f1': f1_list,
    'f2': f2_list,
    'f3': f3_list,
    'f4': f4_list
    })
df.to_csv('Data/accoustics.csv', index=False)<jupyter_output><empty_output><jupyter_text>Play with parallel code<jupyter_code>import multiprocessing
from joblib import Parallel, delayed
from tqdm import tqdm
num_cores = multiprocessing.cpu_count()
recordings = [file for file in files for subdir, dirs, files in os.walk('Data/recordings') 
          if '/P' in os.path.join(subdir, file) and 'self' in os.path.join(subdir, file)]
inputs = tqdm(recordings)
def get_accoustic_features
if __name__ == "__main__":
    processed_list = Parallel(n_jobs=num_cores)(delayed(myfunction)(i,parameters) for i in inputs)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11601584606677985, 'fraction_numerical': 0.024448217317487267, 'mean_word_length': 2.6497315159025194, 'pattern_counts': {'":': 1, '<': 43, '<?xml version=': 0, '>': 41, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '395892', 'license': 'MIT', 'n_tokens_mistral': 3063, 'n_tokens_neox': 2849, 'n_words': 722, 'path': 'get_accoustic_features.ipynb', 'repo_name': 'eleanor-collier/syncpilot-data-pipeline', 'size': '8835'}