preprocess_ipu_fto.py

import pandas as pd

# Pre-process Bumblebee CSV files for use in Praat, adding a flag for IPUs that precede a turn-yield where...
# 1. there is no overlapping speech during floor transfer, i.e. FTO duration is greater than 0
# 2. the current IPU preceding the FT is not a lone backchannel, or is a short IPU but as part of a longer turn
# 3. the IPU after the FT is not a lone backchannel ("Smooth Switch")
# 4. the IPU after the FT is a short IPU but as part of a longer turn ("Hesitant Switch")

# Input CSV file formats (slight alterations to original Bumblebee CSV files)...
#
# Columns in input IPU CSV:
#  1. Original_IPU_Row: the row number from the input IPU CSV that each IPU belongs to
#  2. Condition: one of "Quiet" or "70"
#  3. Type: only "Task"
#  4. Group
#  5. Replicate
#  6. Talker
#  7. IPU_Start
#  8. IPU_Stop
#  9. IPU_Duration
# 10. NumSyll
# 11. LeveldB
# 13. isOverlapWithin
#
# Columns in input FTO CSV:
#  1. Original_FTO_Row: the row number from the input FTO CSV that each FTO belongs to
#  2. Condition: one of "Quiet" or "70"
#  3. Type: only "Task"
#  4. Group
#  5. Replicate
#  6. TalkerThatTookFloor
#  7. Pre_FT_IPU_Dur
#  8. Pre_FT_Num_Syll
#  9. Pre_FT_leveldB
# 10. Post_FT_IPU_Dur
# 11. Post_FT_Num_Syll
# 12. Post_FT_leveldB
# 13. FTO

# Output CSV file format
#
# Columns in output CSV:
#  1. Sub_Dir: subdirectory folder name of that contains the audio recording
#  2. WAV_Filename: the corresponding audio recording that each IPU belongs to
#  3. Original_IPU_Row
#  4. Condition
#  5. Type
#  6. Group
#  7. Replicate
#  8. Talker
#  9. IPU_Start
# 10. IPU_Stop
# 11. IPU_Duration
# 12. NumSyll
# 13. LeveldB
# 14. isOverlapWithin
# 15. Post_IPU_FTO: the FTO that follows each IPU, as obtained from the FTO CSV
# 16. Original_FTO_Row
# 17. isTargetIPU: whether the IPU in this row is one we should analyze
# 18. postIPU_Switch_Type: the nature of the switch, either "Smooth" or "Hesitant"

# How to use...
# 1. Filter the CSV data (e.g. in Excel) to produce 4 new CSV files that only has Task (no Free conversation), and one of Quiet or 70
# 2. At the end of this script, type in the duration below which an IPU will be considered as a backchannel
# 3. - - , type in the path to the IPU CSV file, FTO CSV file, both being for either Quiet or 70
# 4. - - , type in the path where the output CSV should go

# Dictionary to match each replicate to an audio recording, key being TaskConditionGroupReplicate
wav_file_listing = dict(
                        TN21= "G2_T_N_4",
                        TN22= "G2_T_N_6",
                        TN23= "G2_T_N_12",
                        TQ21= "G2_T_Q_1",
                        TQ22= "G2_T_Q_8",
                        TQ23= "G2_T_Q_10",
                        
                        TN51= "G5_T_N_5",
                        TN52= "G5_T_N_8",
                        TN53= "G5_T_N_12",
                        TQ51= "G5_T_Q_1",
                        TQ52= "G5_T_Q_3",
                        TQ53= "G5_T_Q_9",

                        TN61= "G6_T_N_2",
                        TN62= "G6_T_N_10",
                        TN63= "G6_T_N_12",
                        TQ61= "G6_T_Q_4",
                        TQ62= "G6_T_Q_6",
                        TQ63= "G6_T_Q_7",

                        TN71= "G7_T_N_1",
                        TN72= "G7_T_N_5",
                        TN73= "G7_T_N_10",
                        TQ71= "G7_T_Q_4",
                        TQ72= "G7_T_Q_8",
                        TQ73= "G7_T_Q_12",

                        TN81= "G8_T_N_2",
                        TN82= "G8_T_N_3",
                        TN83= "G8_T_N_6",
                        TQ81= "G8_T_Q_7",
                        TQ82= "G8_T_Q_9",
                        TQ83= "G8_T_Q_11",

                        TN91= "G9_T_N_4",
                        TN92= "G9_T_N_7",
                        TN93= "G9_T_N_10",
                        TQ91= "G9_T_Q_1",
                        TQ92= "G9_T_Q_5",
                        TQ93= "G9_T_Q_11",

                        TN101= "G10_T_N_2",
                        TN102= "G10_T_N_6",
                        TN103= "G10_T_N_10",
                        TQ101= "G10_T_Q_3",
                        TQ102= "G10_T_Q_8",
                        TQ103= "G10_T_Q_12",

                        TN111= "G11_T_N_2",
                        TN112= "G11_T_N_5",
                        TN113= "G11_T_N_7",
                        TQ111= "G11_T_Q_4",
                        TQ112= "G11_T_Q_10",
                        TQ113= "G11_T_Q_12",

                        TN121= "G12_T_N_1",
                        TN122= "G12_T_N_5",
                        TN123= "G12_T_N_8",
                        TQ121= "G12_T_Q_4",
                        TQ122= "G12_T_Q_10",
                        TQ123= "G12_T_Q_12",

                        TN131= "G13_T_N_1",
                        TN132= "G13_T_N_5",
                        TN133= "G13_T_N_8",
                        TQ131= "G13_T_Q_4",
                        TQ132= "G13_T_Q_10",
                        TQ133= "G13_T_Q_12",

                        TN141= "G14_T_N_1",
                        TN142= "G14_T_N_3",
                        TN143= "G14_T_N_11",
                        TQ141= "G14_T_Q_5",
                        TQ142= "G14_T_Q_7",
                        TQ143= "G14_T_Q_9",

                        TN151= "G15_T_N_2",
                        TN152= "G15_T_N_3",
                        TN153= "G15_T_N_6",
                        TQ151= "G15_T_Q_8",
                        TQ152= "G15_T_Q_10",
                        TQ153= "G15_T_Q_12",
)



def preprocess_ipu_fto(thres_BC, ipu_file_path, fto_file_path, output_path):
    # read file1 and file2 into DataFrames
    df1 = pd.read_csv(ipu_file_path)   # the IPUs data as a DataFrame
    df2 = pd.read_csv(fto_file_path)   # the FTOs data as a DataFrame

    # get number of rows in both DataFrames
    num_rows_IPU = df1.shape[0]
    num_rows_FTO = df2.shape[0]

    # number of columns in the IPU DataFrame
    n_cols_IPU = df1.shape[1]


    # a list for the Condition taken from the IPUs data CSV
    ipu_conditions = [0] * num_rows_IPU
    for index, value in df1["Condition"].items():
        if value == 70:
            value = "N"
        elif value == "Quiet":
            value = "Q"
        ipu_conditions[index] = value

    # a list for the Group taken from the IPUs data CSV
    ipu_group = df1["Group"].tolist()

    # a list for the Replicate taken from the IPUs data CSV
    ipu_replicate = df1["Replicate"].tolist()

    # a list for the IPU durations taken from the IPUs data CSV
    ipu_durations = df1["IPU_Duration"].tolist()
    
    # a list for the pre-FT IPU durations taken from the FTOs data CSV
    preFT_durations = df2["Pre_FT_IPU_Dur"].tolist()

    # a list for the FTOs from the FTOs data CSV
    ftos = df2["FTO"].tolist()

    # a list for the original row numbers from the FTOs data CSV
    fto_rows = df2["Original_FTO_Row"].tolist()


    # create new lists for the new columns
    sub_dir = [0] * num_rows_IPU              # the subdirectory containing the audio recording
    wav_filename = [0] * num_rows_IPU         # the corresponding audio recording for each IPU
    postIPU_FTOs = [0] * num_rows_IPU         # to store the FTOs that will be matched
    matched_fto_rows = [0] * num_rows_IPU     # to store the original row numbers of the matched FTOs
    target_flag = [0] * num_rows_IPU          # to denote if this IPU is one we should analyze that will be matched
    switch_type = [0] * num_rows_IPU          # to denote what kind of switch in talker occurs after this IPU
    

    # iterate through the CSV files (note: there are less FTOs than IPUs since there are less FTs)
    index2 = 0
    index1 = 0
    while index2 < len(preFT_durations) and index1 < len(ipu_durations):

        # search through IPU durations
        while index1 < len(ipu_durations):
            
            # match a pre-FT duration in the FTOs data with an IPU in the IPUs data
            if (ipu_durations[index1] == preFT_durations[index2]):

                # eventually put the FTO in the row where the IPU duration was matched in the IPUs data
                postIPU_FTOs[index1] = ftos[index2]
                matched_fto_rows[index1] = fto_rows[index2]

                index1 += 1
                break
            else:
                index1 += 1
            
        # now search for a match for the next pre-FT IPU duration
        index2 += 1
    

    # match the WAV file from the big dictionary, key being TaskConditionGroupReplicate
    for i in range(len(ipu_durations)):
        key = "T" + ipu_conditions[i] + str(ipu_group[i]) + str(ipu_replicate[i])
        listed_file = wav_file_listing.get(key)
        wav_filename[i] = listed_file

        sub_dir[i] = "Group " + str(ipu_group[i])


    # setting a flag to label which IPUs whose final section we want to examine for certain features
    for i in range(len(ipu_durations) - 1):
        curr_IPU_not_BC         = (ipu_durations[i] > thres_BC)       # whether the current IPU is a backchannel
        no_overlap_FTO          = (postIPU_FTOs[i] > 0)               # whether there is a talker change and overlapping speech
        next_IPU_not_BC         = (ipu_durations[i+1] > thres_BC)     # whether the next IPU is a backchannel
        no_FT_after_next_IPU    = (postIPU_FTOs[i+1] == 0)            # whether the new talker keeps turn for more than 1 IPU
        if i > 0:
            no_FT_before_curr_IPU = (postIPU_FTOs[i-1] == 0)          # whether the current talker kept the turn for more than 1 IPU
        else:
            no_FT_before_curr_IPU = 0                                 # the 1st IPU by definition does not come after any switch in talkers

        # before the floor transfer, the IPU is not too short, or is a short IPU at the end of a longer turn
        preFT_turn_ok = ((curr_IPU_not_BC and no_overlap_FTO) or ((not curr_IPU_not_BC) and no_FT_before_curr_IPU and no_overlap_FTO))

        # no overlapping speech, response is not a backchannel
        if preFT_turn_ok and next_IPU_not_BC:
            target_flag[i] = 1
            switch_type[i] = "Smooth"

        # no overlapping speech, response is a backchannel at the start of a longer turn
        elif preFT_turn_ok and ((not next_IPU_not_BC) and no_FT_after_next_IPU):
            target_flag[i] = 1
            switch_type[i] = "Hesitant"

        else:
            # target_flag[] list was initialized with all 0s
            switch_type[i] = "Other"
    
    switch_type[num_rows_IPU - 1] = "N/A"  # the last IPU by definition does not precede any switch in talkers


    # add columns to the original IPUs DataFrame
    df1.insert(loc=0, column="WAV_Filename", value=wav_filename, allow_duplicates=False)
    df1.insert(loc=0, column="Sub_Dir", value=sub_dir, allow_duplicates=False)
    df1.insert(loc=n_cols_IPU+2, column="Post_IPU_FTO", value=postIPU_FTOs, allow_duplicates=False)
    df1.insert(loc=n_cols_IPU+3, column="Original_FTO_Row", value=matched_fto_rows, allow_duplicates=False)
    df1.insert(loc=n_cols_IPU+4, column="isTargetIPU", value=target_flag, allow_duplicates=False)
    df1.insert(loc=n_cols_IPU+5, column="postIPU_Switch_Type", value=switch_type, allow_duplicates=False)


    # export the updated DataFrame to a new CSV file
    df1.to_csv(output_path, index=True)
    print(f"Preprocessing complete, file saved as '{output_path}'")



if __name__ == "__main__":
    thres_BC = 0.35     # duration below which a unit of speech is considered a backchannel, in seconds

    file1 = "BB_IPUs_T_Q.csv"                                       # the IPUs data, here being the full TaskQuiet
    file2 = "BB_FTOs_T_Q.csv"                                       # the FTOs data, here being the full TaskQuiet
    output_file = "preproc_IPUs_T_Q.csv"                            # path where output CSV should go
    preprocess_ipu_fto(thres_BC, file1, file2, output_file)

    file1 = "BB_IPUs_T_N.csv"                                       # the IPUs data, here being the full TaskQuiet
    file2 = "BB_FTOs_T_N.csv"                                       # the FTOs data, here being the full TaskQuiet
    output_file = "preproc_IPUs_T_N.csv"                            # path where output CSV should go
    preprocess_ipu_fto(thres_BC, file1, file2, output_file)