Pre-Processing I: from XDF to raw files

Overview

Data for this experiment has been collected been collected using LabStreamLayer (Kothe et al., 2025), allowing for recording of precisely synchronized streams (see documentation).

These streams include: - Audio stream (16kHz) - Webcam Frame stream (ca. 60 Hz) - Marker stream (sent from custom buttonbox) - Balance board stream (500 Hz)

LSL outputs single file in Extensible Data Format (XDF). This file contains all the data streams in form of a timeseries. In this script, we extract all the streams from the file and create raw (audio, video, and other) files for further processing.

More information about the complete lab setup can be found in our preregistration where we archived list of all equipment and software used in the experiment. It is available at OSF Registries.

# Import packages
import os
import pyxdf
import glob
import pandas as pd
import numpy as np
import wave, struct, random
from scipy.io import wavfile
import noisereduce as nr
import cv2

# Set folders
curfolder = os.getcwd()

# If folder data doesn't exist, create it
if not os.path.exists(os.path.join(curfolder,'data')):
    os.makedirs(os.path.join(curfolder,'data'))
datafolder = os.path.join(curfolder,'data')

# Also error_logs
if not os.path.exists(os.path.join(datafolder,'error_logs')):
    os.makedirs(os.path.join(datafolder,'error_logs'))
errorlogs = os.path.join(datafolder,'error_logs')

experiment_to_process = os.path.join(curfolder,'..','00_raw')

# Also CsvDataTS_raw
if not os.path.exists(os.path.join(datafolder,'Data_processed','CsvDataTS_raw')):
    os.makedirs(os.path.join(datafolder,'Data_processed','CsvDataTS_raw'))
outputfolder = os.path.join(datafolder,'Data_processed','CsvDataTS_raw') # outputfolder raw

# Also Data_trials
if not os.path.exists(os.path.join(datafolder,'Data_processed','Data_trials')):
    os.makedirs(os.path.join(datafolder,'Data_processed','Data_trials'))
trialfolder = os.path.join(datafolder,'Data_processed','Data_trials') # outputfolder trialbased

# get all the folders in the target folder
datafolders = glob.glob(os.path.join(experiment_to_process, '*'))

# Extract the folder IDs
datafolders_id = [x.split(os.sep)[-1] for x in datafolders]

# print(curfolder)
# print(targetfolder)
print(datafolders_id)

# Identify all xdf files and all the associated triallist info
xdffiles = []
trialdatas = []

for i in datafolders_id:
    file = glob.glob(os.path.join(experiment_to_process, i, '*.xdf'))
    trialfile = os.path.join(experiment_to_process, i, i+'_results'+'.csv')
    trialdatas.append(trialfile)
    xdffiles.extend(file)

# These are the xdf files we need to process
print(xdffiles[0:10])
print(trialdatas[0:10])
print(f'Number of xdf files: {len(datafolders_id)}')

Extracting streams from XDF file

Using pyxdf package, we can extract the streams from the XDF file.

First, we extract streams for the whole session. Then, we cut these streams into trial-sized chunks, based on the marker stream that indicates the start of each trial. Moreover, we connect the trial with information from the PsychoPy log file, which includes metadata about the trial (i.e., concept, correction number, etc.)

Note that we have to use relative timing for each stream, because there was an error in the camera recording script. The original pre-registered pipeline counted with absolute time but due to the error, this would result in de-synchronized data as inspected in the first run of this script.

(This code takes some time to execute.)

# Audio write function
def to_audio(fileloc, timeseries, samplerate = 16000, channels = 1):
    """
    Writes a time series to an audio file in WAV format.

    Parameters:
    - fileloc (str): Path to save the output WAV file.
    - timeseries (array-like): Time series data to write to the audio file.
    - samplerate (int, optional): Sampling rate of the audio. Default is 16000.
    - channels (int, optional): Number of audio channels. Default is 1 (mono).
    """

    obj = wave.open(fileloc,'w')
    obj.setnchannels(channels) # mono
    obj.setsampwidth(2)
    obj.setframerate(float(samplerate))
    for i in timeseries:
        data = struct.pack('<h', int(i))
        obj.writeframesraw( data )
    obj.close()

# Function to retrieve closest value
def find_closest_value_and_retrieve(df, target_number, target_column, retrieve_column):
    """
    Finds the closest value in a DataFrame column to a target number and retrieves the corresponding value from another column.

    Parameters:
    - df (pd.DataFrame): The DataFrame to search.
    - target_number (float): The target number to find the closest value to.
    - target_column (str): The column name to search for the closest value.
    - retrieve_column (str): The column name to retrieve the corresponding value from.

    """
    # Get the absolute differences between a value in column and the target number
    differences = abs(df[target_column] - target_number)
    
    # Find the index of the minimum difference
    min_difference_index = differences.idxmin()
    
    # Retrieve the corresponding value from the column
    result_value = df.loc[min_difference_index, retrieve_column]

    return result_value

def write_video(vidloc, fourcc, originalfps, frameWidth, frameHeight, capture, frames):
    """
    Writes a video file from a list of frames captured from a video source.

    Parameters:
    - vidloc (str): Path to save the output video file.
    - fourcc (int): FourCC code for video codec.
    - originalfps (float): Frames per second of the output video.
    - frameWidth (int): Width of the video frames.
    - frameHeight (int): Height of the video frames.
    - capture (cv2.VideoCapture): Video capture object.
    - frames (list): List of frame indices to write to the video.
    """

    out = cv2.VideoWriter(vidloc, fourcc, fps = originalfps, frameSize = (int(frameWidth), int(frameHeight)))

    print('Looping over frames')

    # index the frames of the current trial ##FLAG! I have the feeling you are repeating this type of video writing a lot, why not a handy function
    for fra in frames:
        capture.set(cv2.CAP_PROP_POS_FRAMES, fra)
        ret, frame = capture.read()
        if ret:
            out.write(frame)
            
        if not ret:
            print('a frame was dropped: ' + str(fra))
        
    capture.release()
    out.release()

def processXDFtoCSV(stream, dat, outputfolder, relative_time=False):
    """
    Processes an XDF stream and saves it as a CSV file. Optionally includes relative time in the output.
    If the stream is audio data ('Mic'), also creates a WAV file and performs noise reduction.

    Parameters:
    - stream: The XDF stream to process.
    - dat: The data ID associated with the stream.
    - outputfolder: The folder where the CSV and WAV files will be saved.
    """

    timeseriestype = stream['info']['name'][0]
    samplerate = round(float(stream['info']['nominal_srate'][0]))
    # In the xdf loop over the streams and save it as csv if not yet exists
    channelcount = stream['info']['channel_count'][0]
    print('working on stream: ' + timeseriestype + '  with a channel count of ' + str(channelcount) +'\n and a sampling rate of ' + str(samplerate))
    timevec = stream['time_stamps']
    # Add relative time
    if relative_time:
        timevec_rel = stream['time_stamps'] - stream['time_stamps'][0]
        timeseries = stream['time_series']
        matrix_aux = np.vstack([np.transpose(timevec),np.transpose(timeseries), np.transpose(timevec_rel)])
    else:
        timeseries = stream['time_series']
        matrix_aux = np.vstack([np.transpose(timevec),np.transpose(timeseries)])
        
    matrix = np.transpose(matrix_aux)
    df_lab = pd.DataFrame(matrix)
    df_lab.to_csv(outputfolder+dat+'_'+timeseriestype+'_nominal_srate'+str(samplerate)+'.csv',index=False)


    # For audio data also create a wav file
    if timeseriestype == 'Mic':
        if not os.path.exists(outputfolder+'Audio\\'):
            os.makedirs(outputfolder+'Audio\\')
        wavloc = outputfolder+'Audio/'+dat+'_'+timeseriestype+'_nominal_srate'+str(samplerate)+'.wav'
        to_audio(wavloc, timeseries)
        # Load data
        rate, data = wavfile.read(wavloc)
        # Perform noise reduction
        reduced_noise = nr.reduce_noise(y=data, sr=rate, n_std_thresh_stationary=1.5,stationary=True)
        # If folder Audio doesn't exist, create it
        wavloc2 = outputfolder+'Audio/'+dat+'_'+timeseriestype+'_nominal_srate'+str(samplerate)+'_denoised.wav'
        wavfile.write(wavloc2, rate, reduced_noise)

        return df_lab, reduced_noise
    
    else:
        return df_lab
    
def processREXDFtoCSV(stream, dat, outputfolder, relative_time=False):
    """
    This function processes a stream from an XDF file and saves it as a CSV file.
    It handles both regular time series data and audio data (Mic) with special processing.
    For audio data, it also creates a WAV file and applies noise reduction.

    Parameters:
    - stream: The stream data from the XDF file
    - dat: A string identifier for the data
    - outputfolder: The folder where output files will be saved
    - relative_time: Boolean flag to include relative time in the output
    Returns:
    - For regular data: A pandas DataFrame containing the processed data
    - For audio data: A tuple containing the DataFrame and the denoised audio data
    """
    timeseriestype = stream['info']['name'][0]
    samplerate = round(float(stream['info']['nominal_srate'][0]))
    # In the xdf loop over the streams and save it as csv if not yet exists
    channelcount = stream['info']['channel_count'][0]
    print('working on stream: ' + timeseriestype + '  with a channel count of ' + str(channelcount) +'\n and a sampling rate of ' + str(samplerate))
    timevec = stream['time_stamps']
    # Add relative time
    if relative_time:
        timevec_rel = stream['time_stamps'] - stream['time_stamps'][0]
        timeseries = stream['time_series']
        matrix_aux = np.vstack([np.transpose(timevec),np.transpose(timeseries), np.transpose(timevec_rel)])
    else:
        timeseries = stream['time_series']
        matrix_aux = np.vstack([np.transpose(timevec),np.transpose(timeseries)])
        
    matrix = np.transpose(matrix_aux)
    df_lab = pd.DataFrame(matrix)
    df_lab.to_csv(os.path.join(outputfolder, dat+'_rein_'+timeseriestype+'_nominal_srate'+str(samplerate)+'.csv'),index=False)

    # For audio data also create a wav file
    if timeseriestype == 'Mic':
        if not os.path.exists(os.path.join(outputfolder, 'Audio')):
            os.makedirs(os.path.join(outputfolder, 'Audio'))
        wavloc = os.path.join(outputfolder, 'Audio', dat+'_rein_'+timeseriestype+'_nominal_srate'+str(samplerate)+'.wav')
        to_audio(wavloc, timeseries)
        # Load data
        rate, data = wavfile.read(wavloc)
        # Perform noise reduction
        reduced_noise = nr.reduce_noise(y=data, sr=rate, n_std_thresh_stationary=1.5,stationary=True)
        # If folder Audio doesn't exist, create it
        wavloc2 = os.path.join(outputfolder, 'Audio', dat+'_rein_'+timeseriestype+'_nominal_srate'+str(samplerate)+'_denoised.wav')
        wavfile.write(wavloc2, rate, reduced_noise)

        return df_lab, reduced_noise
    
    else:
        return df_lab

Handle exceptions

bug = ['71_1', '42_2', '42_1', '43_2'] # some problem with balance board, 42_1+43_2 is also reinitiated
bug2 = ['21_2', '25_1', '25_2', '35_1', '47_2'] # problems with video
bug3 = ['4_2']  # xdf recorded only for first block, problem to associate it
bug4 = ['44_2'] # reinitiated csv file for results, missing markers in the end
bug6 = ['23_2'] # wrongl xdf
bug7 = ['8_1'] # faulty video (but rein ok)

bugs = bug + bug2 + bug3 + bug4 + bug6 + bug7

rein = ['54_2', '49_2', '43_2', '42_1', '36_1', '36_2', '31_2', '18_2', '8_1', '1_2', '50_1', '43_1', '49_1']
rein_bug = ['42_1', '43_2'] # problem with bb

Pipeline for xdf processing

errorlist = []

for dat in datafolders_id:
    print('Processing session: ' + dat)

    # Skip bugs
    if dat in bugs:
        print('This participant has a bug, we will skip it')
        continue
    
    print('Loading in data from participant: ' + dat)
    trialdata = pd.read_csv(os.path.join(experiment_to_process, dat, dat+'_results.csv'), sep=",") # this also ignores the reinitiated files
    
    # Get the xdf file
    files = glob.glob(os.path.join(experiment_to_process, dat, '*.xdf'))

    #### Exception handling ###########################
    
    # If this session has a reinitiated file, mark it as exception
    if any('reinitiated' in file for file in files):
        print('There is a reinitiated file, we will ignore this one')
        # Remove the reinitiated file from the list
        files = [x for x in files if 'reinitiated' not in x]
        exception = True

    elif any ('onlytpose' in file for file in files):
        # Ignore this file
        files = [x for x in files if 'onlytpose' not in x]
        exception = False # we can treat it as regular file

    else: 
        # If there is no reinitiated file, we can proceed
        print('There is no reinitiated file, we can proceed')
        exception = False

    if len(files) != 1:
        print('There is more than one file')
        break

    #########################################

    file = files[0]

    # Load the file
    streams, header = pyxdf.load_xdf(
        file,
        synchronize_clocks=True,
        dejitter_timestamps=False)
    
    mic_stream = next(s for s in streams if s['info']['name'][0] == 'Mic')

    ##### Mic exception handling #####
    if dat == '18_1':
        print('This is 18_1 and we need to correct some rows (bug)')
        sr = mic_stream['info']['nominal_srate'][0]
        sr = float(sr)
        dt = 1/sr

        timestamps = mic_stream['time_stamps']

        t66 = timestamps[66]
        print('t66: ' + str(t66)) # this is the first correct timestamp

        for i in range(0, 65):
            timestamps[i] = t66 - dt * (66 - (i+1))

        mic_stream['time_stamps'] = timestamps
        print('timestamps after correction: ')
        print(mic_stream['time_stamps'])
    
    #######################################

    # Load in the individual streams
    webcam_stream = next(s for s in streams if s['info']['name'][0] == 'MyWebcamFrameStream')
    marker_stream = next(s for s in streams if s['info']['name'][0] == 'MyMarkerStream')
    bb_stream = next(s for s in streams if s['info']['name'][0] == 'BalanceBoard_stream')
    
    # Marker stream
    print("Processing MyMarkerStream...")
    processXDFtoCSV(marker_stream, dat, outputfolder, relative_time=False)
    markers = pd.read_csv(outputfolder+dat+'_MyMarkerStream_nominal_srate0.csv')

    # Check for buggy last marker Trial start
    if markers.iloc[-1][1] != 'Next_word' and markers.iloc[-1][1] != 'Experiment_end':
        if pd.isna(markers.iloc[-1][1]):
            print('Last row is nan, remove...')
            markers = markers.iloc[:-1]
        elif 'Trial_start' in markers.iloc[-1][1]:
            print('Last row is buggy marker, remove...')
            markers = markers.iloc[:-1]
    
    # Process webcam stream
    print("Processing MyWebcamFrameStream...")
    webcam_ts = processXDFtoCSV(webcam_stream, dat, outputfolder, relative_time=True)
    
    # Process audio stream
    print("Processing Mic...")
    mic_ts, reduced_noise = processXDFtoCSV(mic_stream, dat, outputfolder, relative_time=True)

    # Process BB stream
    print("Processing BalanceBoard_stream...")
    bb_ts = processXDFtoCSV(bb_stream, dat, outputfolder, relative_time=True) 

    print('Done with processing a complete time series and audio data; we will now start making trial snipped data')
  
   # Initialize lists to store trial start and end times
    beginlist = []
    endlist = []
    timestamps = []
    timestamps_2 = []
    tpose_starts = []
    tpose_ends = []

    # Iterate over markers and save times of trial starts and ends
    for row in markers.iterrows():
        if 'Trial_start' in row[1].iloc[1] or 'Practice trial starts' in row[1].iloc[1]:
            beginlist.append(row[1].iloc[0])
        if 'Trial_end' in row[1].iloc[1] or 'Practice trial ends' in row[1].iloc[1]:
            endlist.append(row[1].iloc[0])
        if 'Experiment_start' in row[1].iloc[1]:
            timestamps.append(row[1].iloc[0])
        if 'New block starts' in row[1].iloc[1]:
            timestamps_2.append(row[1].iloc[0])
        if 'Tpose starts' in row[1].iloc[1]:
            tpose_starts.append(row[1].iloc[0])
        if 'Tpose ends' in row[1].iloc[1]:
            tpose_ends.append(row[1].iloc[0])
                    
    # Converting coefficient for lsl to psychopy time
    exp_start_pp = float(trialdata['exp_start'][0])

    ####### Exception handling #######
    if exception == False:
 
        if timestamps != []:
            print('There are timestamps in timestamps, we can use the first one')
            lsl_to_pp = timestamps[0] - exp_start_pp
        else:
            # If there is three timestamps in timestamps2, then all is fine and we can use the first one
            if len(timestamps_2) == 3:
                print('There are three timestamps in timestamps_2, we can use the first one')
                block_start_pp = float(trialdata['block_start'][0])
            # If there is only two timestamps, then we need to assess the second unique value from trialdata
            elif len(timestamps_2) == 2:
                print('There are two timestamps in timestamps_2, we can use the second one')
                unique_values = trialdata['block_start'].unique()
                block_start_pp = unique_values[1]
            else:
                print('There are no timestamps in timestamps_2, we can use the third one')
                unique_values = trialdata['block_start'].unique()
                block_start_pp = unique_values[2]
                
            lsl_to_pp = timestamps_2[0] - block_start_pp

    elif exception == True and dat == '36_2':
        if timestamps != []:
            lsl_to_pp = timestamps[0] - exp_start_pp
        else:
            if len(timestamps_2) == 3:
                block_start_pp = float(trialdata['block_start'][0])
            else:
                unique_values = trialdata['block_start'].unique()
                block_start_pp = unique_values[1]

            lsl_to_pp = timestamps_2[0] - block_start_pp
    
    elif exception == True and dat == '36_1':
        print('This is 36_1 and this should be ran')
        unique_values = trialdata['block_start'].unique()
        block_start_pp = unique_values[1] # first xdf has marker only for the second block
        lsl_to_pp = timestamps_2[0] - block_start_pp
 
    elif exception == True and dat == '8_1':
        print('This is 8_1 and the first file has faulty video')
        continue
        
    else: # this is for exceptions that are 'regular'
        if timestamps != []:
            print('There is experiment start in timestamps')
            lsl_to_pp = timestamps[0] - exp_start_pp
        else:
            if len(timestamps_2) == 3:
                block_start_pp = float(trialdata['block_start'][0])
            else:
                unique_values = trialdata['block_start'].unique()
                block_start_pp = unique_values[0]

            lsl_to_pp = timestamps_2[0] - block_start_pp

    ##################################################
    
    # Now we can proceed to cutting   
    for i in range(len(beginlist)):
        # Prepare the range of the trial
        begin = beginlist[i]
        end = endlist[i]

        ##########################################
        # Convert the beginst to psychopy time
        beginst_pp = begin - lsl_to_pp
        # Now find in trialdata the closest time to the beginst_pp to gather info
        # Whether it is practice or trial
        practice = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'practice')
        if practice == 'practice':
            trialtype = 'pr'
        else:
            trialtype = 'trial'
        
        # Which participant it is
        cycle = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'cycle')
        if cycle == 0:
            participant = 'p0'
        else:
            participant = 'p1'

        # What concept it is
        word = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'word')
        # Modality
        modality = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'modality')
        # Correction, if applicable
        correction_info = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'correction')
        if correction_info == 0:
            correction = '_c0'
        elif correction_info == 1:
            correction = '_c1'
        elif correction_info == 2:
            correction = '_c2'
        else:
            correction = ''

        ##########################################

        # get frame number of this timestamp from webcam stream
        start_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], begin)
        end_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], end)
        start_frame_number = webcam_ts.iloc[:,1][start_frame_idx]
        end_frame_number = webcam_ts.iloc[:,1][end_frame_idx]

        # Get corresponding timestamps (relative)
        start_time = webcam_ts.iloc[:,2][start_frame_idx]
        end_time = webcam_ts.iloc[:,2][end_frame_idx]

        # Cut the video stream csv in this subset
        webcam_subset = webcam_ts.iloc[start_frame_idx:end_frame_idx+1, :]

        if len(webcam_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_'+trialtype+'_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
        else:
            # Save subset to csv
            webcam_subset.to_csv(trialfolder+dat+'_'+trialtype+'_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'_'+participant+'_'+word+'_'+modality+correction+'.csv', index=False)

        # Cut the audio stream csv from start_time to end_time according to relative time
        mic_start_idx = np.searchsorted(mic_ts.iloc[:,2], start_time)
        mic_end_idx = np.searchsorted(mic_ts.iloc[:,2], end_time)
        mic_subset = mic_ts.iloc[mic_start_idx:mic_end_idx+1, :]

        if len(mic_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
            
        else:  
            # Save subset to csv
            mic_subset.to_csv(trialfolder+dat+'_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.csv', index=False)
            # Also create audio
            wavloc = trialfolder+'Audio/'+dat+'_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.wav'
            to_audio(wavloc, mic_subset.iloc[:, 1])
            # Also apply denoising
            reduced_noiseclip = reduced_noise[mic_start_idx:mic_end_idx+1]
            wavloc2 = trialfolder+'Audio/'+dat+'_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'_denoised.wav'
            wavfile.write(wavloc2, 16000, reduced_noiseclip)

        # Cut the balance board stream csv from start_time to end_time according to relative time
        bb_start_idx = np.searchsorted(bb_ts.iloc[:,5], start_time)
        bb_end_idx = np.searchsorted(bb_ts.iloc[:,5], end_time)
        bb_subset = bb_ts.iloc[bb_start_idx:bb_end_idx+1, :]

        if len(bb_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_'+trialtype+'_'+ str(i) +'_'+'BalanceBoard_stream'+'_nominal_srate'+str(100)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
        else:
            # Save subset to csv
            bb_subset.to_csv(trialfolder+dat+'_'+trialtype+'_'+ str(i) +'_'+'BalanceBoard_stream'+'_nominal_srate'+str(100)+'_'+participant+'_'+word+'_'+modality+correction+'.csv', index=False)
  
    # Get information about the tpose for camera
    for i in range(len(tpose_starts)):
        begin = tpose_starts[i]
        end = tpose_ends[i]
        
        # Get frame number of this timestamp from webcam stream
        start_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], begin)
        end_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], end)
        start_frame_number = webcam_ts.iloc[:,1][start_frame_idx]
        end_frame_number = webcam_ts.iloc[:,1][end_frame_idx]

        # Find indices of the specified frames
        try:
            start_frame_idx = webcam_ts.index[webcam_ts.iloc[:,1] == start_frame_number][0]
            end_frame_idx = webcam_ts.index[webcam_ts.iloc[:,1] == end_frame_number][0]
        except IndexError:
            print("Frame number not found in DataFrame.")
            errorlist.append(dat+'_'+'tpose_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv')
            continue

        # Cut the vide stream csv in this subset
        webcam_ts_subset = webcam_ts.iloc[start_frame_idx:end_frame_idx+1, :]

        if len(webcam_ts_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_'+'tpose_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv')
            continue
        else:
            # Save subset to csv
            if dat == '43_1' or dat == '36_1':
                webcam_ts_subset.to_csv(trialfolder+dat+'_'+'tpose_1_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv', index=False) # in the xdf file is only tpose of p1
            else:
                webcam_ts_subset.to_csv(trialfolder+dat+'_'+'tpose_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv', index=False)

# Save the error log
errors = pd.DataFrame(errorlist, columns=['file_error'])
today = pd.Timestamp("today").strftime("%Y_%m")
file_path = errorlogs+'error_log_cuttingtrails' + today + '.csv'
errors.to_csv(file_path, index=False) 

print('We are done: proceed to snipping videos to triallevel')

Pipeline for reinitiated files

In some cases, the recording crashed in the middle of recording (due the the above mentioned error in camera recording script). For these cases, we have more XDF files as the recording was restarted and the experiment continued. In the previous chunk, we ignore those files. Now we process them with similar procedure.

errorlist = []

for dat in datafolders_id:
    print('Loading in data from participant: ' + dat)

    if dat not in rein:
        print('Not rein')
        continue

    if dat in rein_bug:
        print('There is a bug in this file')
        continue

    if '18_2' in dat:
        trialdata = pd.read_csv(os.path.join(experiment_to_process, dat, dat+'_reinitiated_results.csv'), sep=",")
    else:
        trialdata = pd.read_csv(os.path.join(experiment_to_process, dat, dat+'_results.csv'), sep=",")
    
    # Get the xdf file
    files = glob.glob(os.path.join(experiment_to_process, dat, '*.xdf'))
    files = [x for x in files if 'reinitiated' in x or 'onlytpose' in x]
    file = files[0]

    # Load the xdf file
    streams, header = pyxdf.load_xdf(
        file,
        synchronize_clocks=True,
        dejitter_timestamps=False)
    
    # Get the streams
    mic_stream = next(s for s in streams if s['info']['name'][0] == 'Mic')
    webcam_stream = next(s for s in streams if s['info']['name'][0] == 'MyWebcamFrameStream')
    marker_stream = next(s for s in streams if s['info']['name'][0] == 'MyMarkerStream')
    bb_stream = next(s for s in streams if s['info']['name'][0] == 'BalanceBoard_stream')

    # Marker stream
    print("Processing MyMarkerStream...")
    processREXDFtoCSV(marker_stream, dat, outputfolder, relative_time=False)
    markers = pd.read_csv(os.path.join(outputfolder, dat+'_rein_MyMarkerStream_nominal_srate0.csv'))

    # Handle buggy last marker
    if markers.iloc[-1][1] != 'Next_word' and markers.iloc[-1][1] != 'Experiment_end':
        if pd.isna(markers.iloc[-1][1]):
            print('Last row is nan, remove...')
            markers = markers.iloc[:-1]
        elif 'Trial_start' in markers.iloc[-1][1] or 'Practice trial starts' in markers.iloc[-1][1]:
            print('Last row is buggy marker, remove...')
            markers = markers.iloc[:-1]

    # Process webcam stream
    print("Processing MyWebcamFrameStream...")
    webcam_ts = processREXDFtoCSV(webcam_stream, dat, outputfolder, relative_time=True)
    
    # Process audio stream
    print("Processing Mic...")
    mic_ts, reduced_noise = processREXDFtoCSV(mic_stream, dat, outputfolder, relative_time=True)

    # Process BB stream
    print("Processing BalanceBoard_stream...")
    bb_ts = processREXDFtoCSV(bb_stream, dat, outputfolder, relative_time=True) 

    print('done with processing a complete time series and audio data; we will now start making trial snipped data')
 
    beginlist = []
    endlist = []
    timestamps = []
    timestamps_2 = []
    tpose_starts = []
    tpose_ends = []

    # Iterate over markers and save times of trial starts and ends
    for row in markers.iterrows():
        if 'Trial_start' in row[1].iloc[1] or 'Practice trial starts' in row[1].iloc[1]:
            beginlist.append(row[1].iloc[0])
        if 'Trial_end' in row[1].iloc[1] or 'Practice trial ends' in row[1].iloc[1]:
            endlist.append(row[1].iloc[0])
        if 'Experiment_start' in row[1].iloc[1]:
            timestamps.append(row[1].iloc[0])
        if 'New block starts' in row[1].iloc[1]:
            timestamps_2.append(row[1].iloc[0])
        if 'Tpose starts' in row[1].iloc[1]:
            tpose_starts.append(row[1].iloc[0])
        if 'Tpose ends' in row[1].iloc[1]:
            tpose_ends.append(row[1].iloc[0])

    # Get to lsl_to_pp coefficient
    if timestamps != []:
        print('There is experiment start in timestamps')
        lsl_to_pp = timestamps[0] - float(trialdata['exp_start'][0])
        
    if timestamps_2 != []:
        unique_values = trialdata['block_start'].unique()
        # Get rid of nan values
        unique_values = unique_values[~np.isnan(unique_values)]
        block_start_pp = unique_values[-1]
        lsl_to_pp = timestamps_2[-1] - block_start_pp

    else:
        print('Check: this should be 8_1')
        # Get first concept in modality geluiden and practice none
        first_concept = trialdata.loc[(trialdata['modality'] == 'geluiden') & (trialdata['practice'] == 'none')].iloc[0]['trial_start']
        first_concept_lsl = beginlist[0] 
            
        lsl_to_pp = first_concept_lsl - first_concept

    # Now we can proceed to cutting   
    for i in range(len(beginlist)):
        begin = beginlist[i]
        end = endlist[i]

        ##########################################
        # Convert the beginst to psychopy time
        beginst_pp = begin - lsl_to_pp

        # Now find in trialdata the closest time to the beginst_pp to gather info
        # Whether it is practice or trial
        practice = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'practice')
        if practice == 'practice':
            trialtype = 'pr'
        else:
            trialtype = 'trial'
        
        # Which participant it is
        cycle = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'cycle')
        if cycle == 0:
            participant = 'p0'
        else:
            participant = 'p1'

        # What concept it is
        word = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'word')
        # Modality
        modality = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'modality')
        # Correction, if applicable
        correction_info = find_closest_value_and_retrieve(trialdata, beginst_pp, 'trial_start', 'correction')
        if correction_info == 0:
            correction = '_c0'
        elif correction_info == 1:
            correction = '_c1'
        elif correction_info == 2:
            correction = '_c2'
        else:
            correction = ''

        ##########################################

        # Get frame number of this timestamp from webcam stream
        start_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], begin)
        end_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], end)
        start_frame_number = webcam_ts.iloc[:,1][start_frame_idx]
        end_frame_number = webcam_ts.iloc[:,1][end_frame_idx]

        # print('start frame number: ' + str(start_frame_number))
        # print('end frame number: ' + str(end_frame_number))

        # Get corresponding timestamps (relative)
        start_time = webcam_ts.iloc[:,2][start_frame_idx]
        end_time = webcam_ts.iloc[:,2][end_frame_idx]

        # Cut the video stream csv in this subset
        webcam_subset = webcam_ts.iloc[start_frame_idx:end_frame_idx+1, :]

        if len(webcam_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
        else:
            # Save subset to csv
            webcam_subset.to_csv(os.path.join(trialfolder, dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'_'+participant+'_'+word+'_'+modality+correction+'.csv'), index=False)

        # Cut the audio stream csv from start_time to end_time according to relative time
        mic_start_idx = np.searchsorted(mic_ts.iloc[:,2], start_time)
        mic_end_idx = np.searchsorted(mic_ts.iloc[:,2], end_time)
        mic_subset = mic_ts.iloc[mic_start_idx:mic_end_idx+1, :]

        if len(mic_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
            
        else:  
            # Save subset to csv
            mic_subset.to_csv(os.path.join(trialfolder, dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.csv'), index=False)
            # Also create audio
            wavloc = os.path.join(trialfolder, 'Audio', dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'.wav')
            to_audio(wavloc, mic_subset.iloc[:, 1])
            # Also apply denoising
            reduced_noiseclip = reduced_noise[mic_start_idx:mic_end_idx+1]
            wavloc2 = os.path.join(trialfolder, 'Audio', dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'Mic'+'_nominal_srate'+str(16000)+'_'+participant+'_'+word+'_'+modality+correction+'_denoised.wav')
            wavfile.write(wavloc2, 16000, reduced_noiseclip)

        # Cut the balance board stream csv from start_time to end_time according to relative time
        bb_start_idx = np.searchsorted(bb_ts.iloc[:,5], start_time)
        bb_end_idx = np.searchsorted(bb_ts.iloc[:,5], end_time)
        bb_subset = bb_ts.iloc[bb_start_idx:bb_end_idx+1, :]

        if len(bb_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'BalanceBoard_stream'+'_nominal_srate'+str(100)+'_'+participant+'_'+word+'_'+modality+correction+'.csv')
            continue
        else:
            # Save subset to csv
            bb_subset.to_csv(os.path.join(trialfolder, dat+'_rein_'+trialtype+'_'+ str(i) +'_'+'BalanceBoard_stream'+'_nominal_srate'+str(100)+'_'+participant+'_'+word+'_'+modality+correction+'.csv'), index=False)
  
    if len(tpose_starts) < 2:
        index = 1
    elif len(tpose_starts) == 2:
        index = 0
    else: 
        break

    # Get information about the tpose for camera
    for i in range(len(tpose_starts)):
        begin = tpose_starts[i]
        end = tpose_ends[i]
        
        # get frame number of this timestamp from webcam stream
        start_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], begin)
        end_frame_idx = np.searchsorted(webcam_ts.iloc[:,0], end)
        start_frame_number = webcam_ts.iloc[:,1][start_frame_idx]
        end_frame_number = webcam_ts.iloc[:,1][end_frame_idx]

        # print('start frame number: ' + str(start_frame_number))
        # print('end frame number: ' + str(end_frame_number))

        # Find indices of the specified frames
        try:
            start_frame_idx = webcam_ts.index[webcam_ts.iloc[:,1] == start_frame_number][0]
            end_frame_idx = webcam_ts.index[webcam_ts.iloc[:,1] == end_frame_number][0]
            
        except IndexError:
            print("Frame number not found in DataFrame.")
            errorlist.append(dat+'_rein_'+'tpose_'+ str(index) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv')
            continue

        # Cut the vide stream csv in this subset
        webcam_ts_subset = webcam_ts.iloc[start_frame_idx:end_frame_idx+1, :]

        if len(webcam_ts_subset.axes[0]) < 2:
            print('No data within range...')
            errorlist.append(dat+'_rein_'+'tpose_'+ str(index) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv')
            continue
        else:
            # Save subset to csv
           webcam_ts_subset.to_csv(os.path.join(trialfolder, dat+'_rein_'+'tpose_'+ str(index) +'_'+'MyWebcamFrameStream'+'_nominal_srate'+str(500)+'.csv'), index=False)
             
# Save the error log
errors = pd.DataFrame(errorlist, columns=['file_error'])
# Get todays date
today = pd.Timestamp("today").strftime("%Y_%m")
file_path = os.path.join(errorlogs,'error_log_cuttingtrails' + today + '.csv')
errors.to_csv(file_path, index=False) 
        
print('We are done: proceed to snipping videos to triallevel')

Examples

Here is an audio example

from IPython.display import Audio

wavfiles = glob.glob(os.path.join(trialfolder, 'Audio', '*.wav'))
# Keep only the denoised files
wavfiles = [x for x in wavfiles if 'denoised' in x]
# Pick random one
randomfile = random.choice(wavfiles)

Audio(randomfile)

Video writing

Now we have raw data for each trial in csv and wav, but we still need to write videos for each trial. We will use the trial-cut files to access the range of frames in the original raw video file and cut it out from it

# This is a folder with timeseries
tsfolder = os.path.join(datafolder, 'Data_processed', 'Data_trials')
# This is where the raw long video is
videofolder = os.path.join(curfolder, '..', '00_raw')

# Loop through the csv's in tsfolder that has string 'MyWebcamStream' in name
for file in os.listdir(tsfolder):
    print('Now processing file: ' + file)

    if 'MyWebcamFrameStream' in file:

        if '_rein_' in file:
            print('Now processing file '+ file)

            if 'tpose' in file:
                dyadIndex = file.split('_')[0]   # this is dyad number
                partIndex = file.split('_')[1]
                sessionIndex = dyadIndex + '_' + partIndex
                participant = 'p' + file.split('_')[4] # this is participant 0/1

            else: 
                dyadIndex = file.split('_')[0]   # this is dyad number
                partIndex = file.split('_')[1]   # this is part of the session
                sessionIndex = dyadIndex + '_' + partIndex # this is the session index
                trialIndex = file.split('_')[4] # this is trial number
                participant = file.split('_')[8] # this is participant 0/1
                word = file.split('_')[9] # this is the concept
                modality = file.split('_')[10].split('.')[0] # this is the modality

            # Assess the correction
            if 'c0' in file:
                correction = '_c0'
            elif 'c1' in file:
                correction = '_c1'
            elif 'c2' in file:
                correction = '_c2'
            else:
                correction = ''

            # Assess the trial type
            if '_pr_' in file:
                trialtype = 'pr'
            elif '_tpose_' in file:
                trialtype = 'tpose'
            else:
                trialtype = 'trial'

            trialdata = pd.read_csv(os.path.join(trialfolder, file))

            # This is the location where the video will be saved
            if '_tpose_' in file:
                vidloc = os.path.join(trialfolder, sessionIndex + '_rein_tpose_' + participant+ '_video_raw'+'.avi')
            else:
                vidloc = os.path.join(trialfolder, sessionIndex + '_rein_' + trialtype + '_' + str(trialIndex) + '_' + participant + '_' + word + '_' + modality + correction + '_video_raw' + '.avi')

            # Check if it exists, if yes, skip
            if os.path.exists(vidloc):
                print('Video already exists, skipping...')
                continue
            
            videolong = os.path.join(videofolder, sessionIndex, sessionIndex + '_reinitiated-video.avi') # this is the long video 

        else:
            print('Now processing file '+ file)
            if 'tpose' in file:
                dyadIndex = file.split('_')[0]   # this is dyad number
                partIndex = file.split('_')[1]   # this is part of the session
                sessionIndex = dyadIndex + '_' + partIndex # this is the session ID
                participant = 'p' + file.split('_')[3] # this is participant 0/1

            else:
                dyadIndex = file.split('_')[0]   # this is dyad number
                partIndex = file.split('_')[1]   # this is part of the session
                sessionIndex = dyadIndex + '_' + partIndex # this is the session index
                trialIndex = file.split('_')[3] # this is trial number
                participant = file.split('_')[7] # this is participant 0/1
                word = file.split('_')[8] # this is the concept
                modality = file.split('_')[9].split('.')[0] # this is the modality

            # Assess the correction
            if 'c0' in file:
                correction = '_c0'
            elif 'c1' in file:
                correction = '_c1'
            elif 'c2' in file:
                correction = '_c2'
            else:
                correction = ''

            # Assess the trial type
            if 'pr' in file:
                trialtype = 'pr'
            elif 'tpose' in file:
                trialtype = 'tpose'
            else:
                trialtype = 'trial'

            trialdata = pd.read_csv(os.path.join(trialfolder, file))

            # This is the location where the video will be saved
            if 'tpose' in file:
                vidloc = os.path.join(trialfolder, sessionIndex + '_tpose_'+ participant + '_video_raw'+'.avi')
            else:
                vidloc = os.path.join(trialfolder, sessionIndex+'_'+trialtype+'_'+ str(trialIndex) +'_'+participant+'_'+word+'_'+modality+correction+'_video_raw'+'.avi')

            # Check if it exists, if yes, skip
            if os.path.exists(vidloc):
                print('Video already exists, skipping...')
                continue
            
            videolong = os.path.join(videofolder, sessionIndex, sessionIndex + '-video.avi') # this is the long video 
        
        begin_time = trialdata['0'].min() # begin time of the trial
        end_time = trialdata['0'].max() # end time of the trial
        # Get the begin and end frame
        begin_frame = trialdata['1'].min().astype(int)
        end_frame = trialdata['1'].max().astype(int)
        totframes = end_frame-begin_frame # total number of frames in the trial
        frames = range(begin_frame, end_frame) # get all the frames in trial
        print(frames)
        # Load in the long video
        print('Loading the original video')
        capture = cv2.VideoCapture(videolong) 

        # What is the original fps of the video
        originalfps = round((totframes/(end_time-begin_time)),3)
        #print('original fps: '+str(originalfps))
        
        # Metadata
        fourcc = cv2.VideoWriter_fourcc(*'XVID')
        frameWidth = capture.get(cv2.CAP_PROP_FRAME_WIDTH)
        frameHeight = capture.get(cv2.CAP_PROP_FRAME_HEIGHT)

        # Start writing video
        print('Starting to write the video')
        write_video(vidloc, fourcc, originalfps, frameWidth, frameHeight, capture, frames)
        print('Video is done')

print('All done!')

Here is a video example

from IPython.display import Video

# Pick one video in tsfolder
avifiles = glob.glob(os.path.join(trialfolder, '*.avi'))
sample = random.choice(avifiles)

# adjust the size
Video(sample, width=450, height=250)

Note:

In the pre-registered script, we had here a step to correcting trials with faulty start or end. Because the pipeline needed a more robust workflow, we moved this step into separate script 03_Correct_cuts.ipynb(here)

Similarly, we moved the step to align high-sampling audio and audio-video alignment into separate script 02_Audiosync.ipynb (here)

References

Kothe, C., Shirazi, S. Y., Stenner, T., Medine, D., Boulay, C., Grivich, M. I., Artoni, F., Mullen, T., Delorme, A., & Makeig, S. (2025). The lab streaming layer for synchronized multimodal recording. Imaging Neuroscience, 3, IMAG.a.136. https://doi.org/10.1162/IMAG.a.136