File size: 5,270 Bytes
95456ed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """
Create the data for training ScanDL on all data.
"""
import argparse
import os
import json
import numpy as np
import pandas as pd
import sys
from ScanDL2.scandl_module.scripts.sp_load_celer_zuco import (
load_celer,
load_celer_speakers,
process_celer,
)
from ScanDL2.scandl_module.scripts.sp_load_celer_zuco import (
load_zuco,
process_zuco,
get_kfold,
get_kfold_indices_combined,
)
from ScanDL2.scandl_module.scripts.sp_load_celer_zuco import load_emtec, process_emtec
from ScanDL2.scandl_module.scripts.sp_load_celer_zuco import load_bsc, process_bsc
from ScanDL2.scandl_module.scripts.sp_load_celer_zuco import flatten_data, unflatten_data
from transformers import set_seed, BertTokenizerFast
sys.path.append("./")
sys.path.append("../")
def create_argparser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
parser.add_argument(
"--folder-name",
type=str,
default="processed_data_all",
help="Name of the folder to save the processed data in.",
)
parser.add_argument(
"--max-fix-dur",
type=int,
help="max fixatino duration value. greater fixation durations are replaced with this value.",
default=999,
)
parser.add_argument(
"--data",
type=str,
choices=["celer", "emtec", "bsc"],
required=True,
)
defaults = dict()
defaults.update(load_defaults_config(parser.parse_args()))
add_dict_to_argparser(parser, defaults)
return parser
def load_defaults_config(args):
"""
Load defaults for training args.
"""
if args.data == "emtec":
config_name = "config_emtec.json"
elif args.data == "bsc":
config_name = "config_bsc.json"
else:
config_name = "config.json"
with open(f"diffusion_only/scandl_diff_dur/{config_name}", "r") as f:
return json.load(f)
def add_dict_to_argparser(parser, default_dict):
for k, v in default_dict.items():
v_type = type(v)
if v is None:
v_type = str
elif isinstance(v, bool):
v_type = str2bool
parser.add_argument(f"--{k}", default=v, type=v_type)
def str2bool(v):
"""
https://stackoverflow.com/questions/15008758/parsing-boolean-values-with-argparse
"""
if isinstance(v, bool):
return v
if v.lower() in ("yes", "true", "t", "y", "1"):
return True
elif v.lower() in ("no", "false", "f", "n", "0"):
return False
else:
raise argparse.ArgumentTypeError("boolean value expected")
def main():
base_folder_name = "scandl2_pkg"
print("Loading argument parser...")
args = create_argparser().parse_args()
set_seed(args.seed)
if args.data == "celer":
tokenizer = BertTokenizerFast.from_pretrained(args.config_name)
data_path = args.folder_name + "_celer"
if not os.path.exists(os.path.join(base_folder_name, data_path)):
os.makedirs(os.path.join(base_folder_name, data_path))
# load Celer data
word_info_df, eyemovement_df = load_celer()
reader_list = load_celer_speakers(only_native_speakers=args.celer_only_L1)
sn_list = np.unique(
word_info_df[word_info_df["list"].isin(reader_list)].sentenceid.values
).tolist()
data, splitting_IDs_dict = process_celer(
sn_list=sn_list,
reader_list=reader_list,
word_info_df=word_info_df,
eyemovement_df=eyemovement_df,
tokenizer=tokenizer,
args=args,
inference="cv",
max_fix_dur=args.max_fix_dur,
)
flattened_data = flatten_data(data)
flattened_data = np.array(flattened_data, dtype=object).tolist()
train_data = unflatten_data(flattened_data=flattened_data, split="train")
train_data.save_to_disk(os.path.join(base_folder_name, data_path))
elif args.data == "bsc":
raise NotImplementedError("BSC data not implemented yet.")
elif args.data == "emtec":
tokenizer = BertTokenizerFast.from_pretrained(args.config_name)
data_path = args.folder_name + "_emtec"
if not os.path.exists(os.path.join(base_folder_name, data_path)):
os.makedirs(os.path.join(base_folder_name, data_path))
# load EMTeC data
print("Loading EMTeC data...")
fixations_df, stimuli_df = load_emtec()
data, splitting_IDs_dict = process_emtec(
fixations_df=fixations_df,
stimuli_df=stimuli_df,
tokenizer=tokenizer,
args=args,
inference="cv",
max_fix_dur=args.max_fix_dur,
)
flattened_data = flatten_data(data)
flattened_data = np.array(flattened_data, dtype=object).tolist()
train_data = unflatten_data(flattened_data=flattened_data, split="train")
train_data.save_to_disk(os.path.join(base_folder_name, data_path))
else:
raise NotImplementedError("Data not implemented yet.")
if __name__ == "__main__":
raise SystemExit(main())
|