main.py

from __future__ import annotations
from contextlib import nullcontext
from functools import partial
from pathlib import Path
import pandas as pd
from psychopy import core
from psyflow import BlockUnit,StimBank,StimUnit,SubInfo,TaskSettings,context_from_config,initialize_exp,initialize_triggers,load_config,parse_task_run_options,runtime_context
from src import generate_study_plans,generate_test_plans,run_trial,summarize_test
MODES=("human","qa","sim");DEFAULT_CONFIG_BY_MODE={"human":"config/config.yaml","qa":"config/config_qa.yaml","sim":"config/config_scripted_sim.yaml"}
def block(name,idx,plans,settings,win,kb,bank,tr,rows):(BlockUnit(block_id=name,block_idx=idx,settings=settings,window=win,keyboard=kb).add_condition(plans).on_start(lambda _:tr.send(settings.triggers.get("block_start"))).on_end(lambda _:tr.send(settings.triggers.get("block_end"))).run_trial(partial(run_trial,stim_bank=bank,trigger_runtime=tr,block_id=name,block_idx=idx)).to_dict(rows))
def run(options):
 root=Path(__file__).resolve().parent;cfg=load_config(str(options.config_path));out=None;scope=nullcontext();ctx=None
 if options.mode in ("qa","sim"):ctx=context_from_config(task_dir=root,config=cfg,mode=options.mode);out=ctx.output_dir;scope=runtime_context(ctx)
 with scope:
  sub={"subject_id":"qa"} if options.mode=="qa" else ({"subject_id":str(ctx.session.participant_id or "sim")} if options.mode=="sim" else SubInfo(cfg["subform_config"]).collect());s=TaskSettings.from_dict(cfg["task_config"]);s.add_subinfo(sub)
  if out is not None:s.save_path=str(out)
  if options.mode=="qa" and out is not None:out.mkdir(parents=True,exist_ok=True);s.res_file=str(out/"qa_trace.csv");s.log_file=str(out/"qa_psychopy.log");s.json_file=str(out/"qa_settings.json")
  s.triggers=cfg["trigger_config"];tr=initialize_triggers(mock=True) if options.mode in ("qa","sim") else initialize_triggers(cfg);win,kb=initialize_exp(s);bank=StimBank(win,cfg["stim_config"]).preload_all();s.save_to_json();tr.send(s.triggers.get("experiment_start"));StimUnit("instruction",win,kb,runtime=tr).add_stim(bank.get("instruction")).wait_and_continue();study=generate_study_plans(root/"assets"/"stimuli.csv",per_cue=int(s.study_items_per_cue),base_seed=int(s.plan_seed));rows=[];block("study",0,study,s,win,kb,bank,tr,rows);StimUnit("test_instruction",win,kb,runtime=tr).add_stim(bank.get("test_instruction")).wait_and_continue();test=generate_test_plans(root/"assets"/"stimuli.csv",studied=study,foils=int(s.recognition_foils),base_seed=int(s.plan_seed));block("recognition",1,test,s,win,kb,bank,tr,rows);x=summarize_test(rows);StimUnit("good_bye",win,kb,runtime=tr).add_stim(bank.get_and_format("good_bye",remember=f"{x['remember_hit_rate']:.1%}",forget=f"{x['forget_hit_rate']:.1%}",effect=f"{x['directed_forgetting_effect']:.1%}",false_alarm=f"{x['false_alarm_rate']:.1%}")).wait_and_continue(terminate=True);tr.send(s.triggers.get("experiment_end"));pd.DataFrame(rows).to_csv(s.res_file,index=False);tr.close();core.quit()
def main():run(parse_task_run_options(task_root=Path(__file__).resolve().parent,description="Run directed forgetting task",default_config_by_mode=DEFAULT_CONFIG_BY_MODE,modes=MODES))
if __name__=="__main__":main()