CNN regressor/2.shape_data.py

# Copyright 2026 Yokohama National University. All Rights Reserved.

# This is a program that shapes the behavioral data into inputs for the regression model as short time-series segments of approximately 1 second (32 frames), centered on the target frame to be predicted.
# Generates the cnn_dataset folder containing subfolders for feature, label_1 to label_6, and frame.


#import public module
import numpy as np
import os
import itertools

def create_cnn_Data_label(debate,window_size,label,number_of_interlocutors,N):
    # Generates the feature datasets (for CNN input) of the speaker, addressee, and other listeners for a specified debate, along with the corresponding unified functional spectrum label datasets.
    """
    window_size : int   
        Number of frames used as a single input data (window size for feature extraction, i.e., 32 frames)
    number_of_interlocutors : int   
        Number of interlocutors
    N : int 
        Time window for averaging
    frame_forward : int
        Frame width in the future direction from the center when including multiple frames in one data sample
    frame_back : int
        Frame width in the past direction from the center when including multiple frames in one data sample
    frame_win : list(int)
        Index array of frame numbers to include when including multiple frames in one data sample
    label : int
        Number of bases (1–6)
    floorholder : int
        The floor holder (i.e., the speaker) is labeled as 800, and all others are labeled as −1.
    gaze_direction : int
        Label indicating whom the participant is looking at.
        0 corresponds to P1, 1 to P2, 2 to P3, and 3 to P4.

    Final outputs      
    frame 
        Create 24 .npy files for all sessions (6 groups × 2 sessions per group = 12 sessions) containing the test frames and total frames (frame folder).
    Feature data
        Create x_data_debate_part1 and x_data_debate_part2 (shape: number of samples × 32 × 96 × 1) for all sessions (6 groups × 2 sessions per group = 12 sessions), resulting in a total of 24 x-label files (feature folder).
        Note that part1 and part2 correspond to swapping the order of the other two listeners.
    Label data
        Create y_data_debate_part1 and y_data_debate_part2 (shape: number of samples × 1) for each label 1–6, for all sessions (6 groups × 2 sessions per group = 12 sessions), resulting in a total of 24 y-label files (label 1–6 folders).
        Note that part1 and part2 correspond to swapping the order of the other two listeners.
    """ 
    # Frame range for analysis
    frame_span = {'2024_0807session1F':[4452,24836],'2024_0807session3F':[3105,20313],
                  '2024_0809session1F':[4174,23565],'2024_0809session3F':[3395,25038],
                  '2024_0826session1F':[4320,23311],'2024_0826session3F':[3648,22231],
                  '2024_0829session1F':[4440,24180],'2024_0829session3F':[3671,24835],
                  '2024_0905session1F':[4825,26339],'2024_0905session3F':[3708,17524],
                  '2024_0910session1F':[4650,21481],'2024_0910session3F':[3658,20479]} 
    
    frame_start = frame_span[debate][0]
    frame_end = frame_span[debate][1]
    #Person ID
    person_num = {f'P{person}':person for person in range(1,number_of_interlocutors+1)}
    #Specified window width
    frame_forward = window_size // 2 + 1
    if window_size % 2 ==0:
        frame_back = - (window_size // 2) + 1
    else:
        frame_back = - (window_size // 2)
    frame_win = np.arange(frame_back,frame_forward)

    #Data loading
    #FloorHolder
    floorholder={person : np. load('processed_data/FloorHolder/' + debate + person + '.npy') for person in person_num}
    #gaze_direction
    gaze_direction = {person : np. load('processed_data/gaze_direction/' + debate + person + '.npy') for person in person_num}
    #head_pose_angle
    head_pose_angle = {person : np. load('processed_data/head_pose_angle/' + debate + person + '.npy') for person in person_num}
    #utter
    utter = {person : np. load('processed_data/utter/' + debate + person + '.npy') for person in person_num}
    #gaze_status_GT
    gaze_status_GT = {person : np. load('processed_data/gaze_status_GT/' + debate + person + '.npy') for person in person_num}
    #eyeball_direction
    eyeball = {person : np. load('processed_data/eyeball/' + debate + person + '.npy') for person in person_num}
    #AU
    au = {person : np. load('processed_data/au/' + debate + person + '.npy') for person in person_num}


    #Label data
    Unified_functional_spectrum_pair_label={}
    for pair in itertools.combinations(person_num.keys(),2):
        pair=list(pair)
        others=[p for p in person_num.keys() if p not in pair]
        Unified_functional_spectrum_pair_label['_'.join(pair)+'_'+''.join(others)]=np.load('processed_data/Unified_functional_spectrum_pair_label_' + str(label) + '/' + debate + pair[0] + '_' + pair[1] + '_' + ''.join(others) + '.npy')

    # Feature and label data
    x_data_debate_part1 = []
    x_data_debate_part2 = []
    y_data_debate_part1 = []
    y_data_debate_part2 = []

    # List storing the frame numbers of the target debate
    total_frame_debate = list(range(len(Unified_functional_spectrum_pair_label['P1_P2_P3P4'])))

    # List storing the frame numbers of the test set
    test_frame_debate = []

    min_len=min(min([len(_) for _ in head_pose_angle.values()]),
                min([len(_) for _ in utter.values()]),
                min([len(_) for _ in gaze_status_GT.values()]),
                min([len(_) for _ in eyeball.values()]),
                min([len(_) for _ in au.values()]))
    
    # List storing speaker–addressee pairs
    speaker_gaze_match = []

    for frame in range(frame_start, frame_end + 1):
        # Whether the window centered on a frame falls within a valid range
        if (frame + frame_win[0] >= 0) and (frame + frame_win[-1] < min_len):
            if floorholder['P1'][frame] == 800: #When P1 is the Floor Holder (speaker)
                if gaze_direction['P1'][frame] == person_num['P2']-1: #When P1 is looking at P2
                    speaker_gaze_match.append(12) 
                elif gaze_direction['P1'][frame] == person_num['P3']-1: #When P1 is looking at P3
                    speaker_gaze_match.append(13) 
                elif gaze_direction['P1'][frame] == person_num['P4']-1: #When P1 is looking at P4
                    speaker_gaze_match.append(14)  
                else: #When not looking at anyone
                    speaker_gaze_match.append(10) 
            elif floorholder['P2'][frame] == 800: #When P2 is the Floor Holder (speaker)
                if gaze_direction['P2'][frame] == person_num['P1']-1: #When P2 is looking at P1
                    speaker_gaze_match.append(21) 
                elif gaze_direction['P2'][frame] == person_num['P3']-1: #When P2 is looking at P3
                    speaker_gaze_match.append(23) 
                elif gaze_direction['P2'][frame] == person_num['P4']-1: #When P2 is looking at P4
                    speaker_gaze_match.append(24)  
                else: #When not looking at anyone
                    speaker_gaze_match.append(20)  
            elif floorholder['P3'][frame] == 800: #When P3 is the Floor Holder (speaker)
                if gaze_direction['P3'][frame] == person_num['P1']-1: #When P3 is looking at P1
                        speaker_gaze_match.append(31)  
                elif gaze_direction['P3'][frame] == person_num['P2']-1: #When P3 is looking at P2
                        speaker_gaze_match.append(32)  
                elif gaze_direction['P3'][frame] == person_num['P4']-1: #When P3 is looking at P4
                        speaker_gaze_match.append(34)  
                else: #When not looking at anyone
                    speaker_gaze_match.append(30) 
            elif floorholder['P4'][frame] == 800: #When P4 is the Floor Holder (speaker)
                if gaze_direction['P4'][frame] == person_num['P1']-1: #When P4 is looking at P1
                        speaker_gaze_match.append(41)  
                elif gaze_direction['P4'][frame] == person_num['P2']-1: #When P4 is looking at P2
                        speaker_gaze_match.append(42)  
                elif gaze_direction['P4'][frame] == person_num['P3']-1: #When P4 is looking at P3
                        speaker_gaze_match.append(43)  
                else: #When not looking at anyone
                    speaker_gaze_match.append(40)       
            else: #When there is no speaker
                speaker_gaze_match.append(00)

    speaker_gaze_match = np.array(speaker_gaze_match)

    label_dict={}
    for pair in itertools.combinations(person_num.keys(),2):
        pair=list(pair)
        others=[p for p in person_num.keys() if p not in pair]
        label_dict[person_num[pair[0]]*10+person_num[pair[1]]]='_'.join(pair)+'_'+''.join(others)
        label_dict[person_num[pair[1]]*10+person_num[pair[0]]]='_'.join(pair)+'_'+''.join(others)
    
    # Track the current consecutive values
    current_value = None
    current_sequence = []

    # Processing loop
    for i, value in enumerate(speaker_gaze_match):
        if value == current_value:  # In case of consecutive values
            current_sequence.append(i)
        else:  # When a new value appears
            # Check if there are at least N frames and whether the consecutive value is not 00, 10, 20, 30, or 40 (values indicating Avert)
            if len(current_sequence) >= N:
                if current_value in label_dict:  # Case 1: For the corresponding pair
                    # ---- (A) Retrieve the label array corresponding to current_value ----
                    label_array = Unified_functional_spectrum_pair_label[label_dict[current_value]]
                    # ---- (B) Process num_to_extract frames starting from the beginning of the consecutive frames ----
                    num_to_extract = len(current_sequence) - (N-1)
                    extract_indices = current_sequence[:num_to_extract]

                    for seq_idx in extract_indices:
                        target_frame = frame_start + seq_idx  # Actual frame numbers
                        # Save the frame numbers
                        test_frame_debate.append(target_frame)

                        # ---- (C) Obtain the correct ground-truth labels ----
                        y_label = label_array[target_frame]

                        # ---- (D) Feature extraction ----
                        # Inside the speaker_gaze_match loop
                        speaker = current_value // 10
                        listener = current_value % 10
                        others = [p for p in range(1,number_of_interlocutors+1) if p not in [speaker, listener]]
                        part1_order = [speaker, listener] + others
                        part2_order = [speaker, listener] + others[::-1]
                        # --- part1 ---
                        for win in frame_win:
                            idx = target_frame + win
                            for p in part1_order:
                                x_data_debate_part1.extend(head_pose_angle[f'P{p}'][idx][:])
                                x_data_debate_part1.append(utter[f'P{p}'][idx])
                                x_data_debate_part1.append(gaze_status_GT[f'P{p}'][idx])
                                x_data_debate_part1.extend(eyeball[f'P{p}'][idx][:])
                                x_data_debate_part1.extend(au[f'P{p}'][idx][:])

                        y_data_debate_part1.append(y_label)

                        # --- part2 ---
                        for win in frame_win:
                            idx = target_frame + win
                            for p in part2_order:
                                x_data_debate_part2.extend(head_pose_angle[f'P{p}'][idx][:])
                                x_data_debate_part2.append(utter[f'P{p}'][idx])
                                x_data_debate_part2.append(gaze_status_GT[f'P{p}'][idx])
                                x_data_debate_part2.extend(eyeball[f'P{p}'][idx][:])
                                x_data_debate_part2.extend(au[f'P{p}'][idx][:])

                        y_data_debate_part2.append(y_label)

            # Update to the new value
            current_value = value
            current_sequence = [i]

    # Process the last sequence
    if len(current_sequence) >= N:
                if current_value in label_dict:  # Case 1: For the corresponding pair
                    # ---- (A) Retrieve the label array corresponding to current_value ----
                    label_array = Unified_functional_spectrum_pair_label[label_dict[current_value]]
                    # ---- (B) Process num_to_extract frames starting from the beginning of the consecutive frames ----
                    num_to_extract = len(current_sequence) - (N-1)
                    extract_indices = current_sequence[:num_to_extract]

                    for seq_idx in extract_indices:
                        target_frame = frame_start + seq_idx  # Actual frame numbers
                        test_frame_debate.append(target_frame)

                        # ---- (C) Obtain the correct ground-truth labels ----
                        y_label = label_array[target_frame]

                        # ---- (D) Feature extraction ----
                        # Inside the speaker_gaze_match loop
                        speaker = current_value // 10
                        listener = current_value % 10
                        others = [p for p in range(1,number_of_interlocutors+1) if p not in [speaker, listener]]
                        part1_order = [speaker, listener] + others
                        part2_order = [speaker, listener] + others[::-1]
                        # part1
                        for win in frame_win:
                            idx = target_frame + win
                            for p in part1_order:
                                x_data_debate_part1.extend(head_pose_angle[f'P{p}'][idx][:])
                                x_data_debate_part1.append(utter[f'P{p}'][idx])
                                x_data_debate_part1.append(gaze_status_GT[f'P{p}'][idx])
                                x_data_debate_part1.extend(eyeball[f'P{p}'][idx][:])
                                x_data_debate_part1.extend(au[f'P{p}'][idx][:])

                        y_data_debate_part1.append(y_label)

                        # --- part2 ---
                        for win in frame_win:
                            idx = target_frame + win
                            for p in part2_order:
                                x_data_debate_part2.extend(head_pose_angle[f'P{p}'][idx][:])
                                x_data_debate_part2.append(utter[f'P{p}'][idx])
                                x_data_debate_part2.append(gaze_status_GT[f'P{p}'][idx])
                                x_data_debate_part2.extend(eyeball[f'P{p}'][idx][:])
                                x_data_debate_part2.extend(au[f'P{p}'][idx][:])

                        y_data_debate_part2.append(y_label)

    x_data_debate_part1 = np.array(x_data_debate_part1).reshape(-1,window_size,24*number_of_interlocutors,1)
    x_data_debate_part2 = np.array(x_data_debate_part2).reshape(-1,window_size,24*number_of_interlocutors,1)
    y_data_debate_part1 = np.array(y_data_debate_part1).reshape(-1,1)
    y_data_debate_part2 = np.array(y_data_debate_part2).reshape(-1,1)

    os.makedirs('cnn_dataset/Unified_functional_spectrum_pair/feature/',exist_ok=True)
    os.makedirs('cnn_dataset/Unified_functional_spectrum_pair/label_' + str(label) + '/',exist_ok=True)
    os.makedirs('cnn_dataset/Unified_functional_spectrum_pair/frame/',exist_ok=True)

    #Save the created feature and label data → Create the training dataset for the deep learning model
    np.save('cnn_dataset/Unified_functional_spectrum_pair/feature/x_' + debate + '_part1.npy', x_data_debate_part1)
    np.save('cnn_dataset/Unified_functional_spectrum_pair/feature/x_' + debate + '_part2.npy', x_data_debate_part2)
    np.save('cnn_dataset/Unified_functional_spectrum_pair/label_' + str(label) + '/y_' + debate + '_part1.npy', y_data_debate_part1)
    np.save('cnn_dataset/Unified_functional_spectrum_pair/label_' + str(label) + '/y_' + debate + '_part2.npy', y_data_debate_part2)
    np.save('cnn_dataset/Unified_functional_spectrum_pair/frame/total_frame_' + debate + '.npy', total_frame_debate)
    np.save('cnn_dataset/Unified_functional_spectrum_pair/frame/test_frame_' + debate + '.npy', test_frame_debate)

#Main processing loop
label_set = [1, 2, 3, 4, 5, 6]

for label in label_set:
    for date in ["2024_0807","2024_0809","2024_0826","2024_0829","2024_0905","2024_0910"]:
        for session in ["session1F","session3F"]:
            debate = date + session

            #Function call
            create_cnn_Data_label(
                debate=debate,
                window_size=32,
                label=label,
                number_of_interlocutors=4,
                N=7
            )