#!/usr/bin/env python3 from __future__ import annotations from psychopy import constants, core, event, gui, visual from collections import namedtuple import frensch_procedures import random from pprint import pprint import pandas as pd import pickle DisplayVariable = namedtuple("DisplayVariable", ["name", "values"]) DisplayProcedure = namedtuple("DisplayProcedure", ["procedure", "solution"]) intro_text = """Vielen Dank dass Sie bei unserem Experiment zum menschlichen Lernen teilnehmen! Im folgenden müssen sie verschiedene Rechenaufgaben lösen. Verwenden Sie die gewohnten Rechenregeln und geben sie Ihre Lösung bitte immer als zweistellige Zahl ein, und bestätigen mit Enter. Die Werte der Variablen werden oben am Bildschirm angezeigt. Manche Variablen haben mehrere mögliche Werte; "_2" besagt z.B. das der zweite Wert zu verwenden ist. Nach jeder Aufgabe können Sie kurz pausieren. Drücken Sie die Leertaste um zu beginnen""" def experiment_shutdown(): WIN.close() core.quit() WIN = visual.Window((2560, 1440), fullscr=True, units="pix") MONITOR_FPS = 60 TRAIN_TRIALS = 75 TEST_TRIALS = 50 TRAIN_TRIALS = 1 TEST_TRIALS = 1 ORDER_CONDITIONS = ["fixed", "random", "blocked"] PROCEDURE_KEYS = ["1", "2", "3", "4", "5", "6", "overall"] # Cancel experiment anytime with Esc event.globalKeys.add(key="escape", func=experiment_shutdown, name="shutdown") def generate_variable_display(varx: list[DisplayVariable], x_positions: list[int]): assert len(varx) == len(x_positions) stims = [] def gen_value_stims(values, x, y, offset): for value in values: y -= offset value_stim = visual.TextBox2( WIN, pos=(x, y), text=value, # size=200, letterHeight=100, alignment="center", ) stims.append(value_stim) y = 650 offset = 120 for var, x_pos in zip(varx, x_positions): stim_var = visual.TextBox2( WIN, pos=[x_pos, y], text=var.name, # size=[1000, 1000], letterHeight=50, alignment="center", ) stims.append(stim_var) gen_value_stims(var.values, x_pos, y, offset) return stims def generate_procedure_display(procedure: DisplayProcedure, position): stim_procedure = visual.TextBox2( WIN, pos=position, text=procedure.procedure, size=[1000000, 1000], letterHeight=50, alignment="center", ) return stim_procedure def generate_all_watersamples(n): samples = [] for _ in range(n): samples.append(frensch_procedures.constrained_WaterSample()) return samples def run_trial(water_sample, procedure_keys: list, condition): water_sample.print_all() if condition == "random": random.shuffle(procedure_keys) solid = DisplayVariable("Mineralien", [water_sample.solid]) algae = DisplayVariable("Algen", [water_sample.algae]) lime = DisplayVariable("Sandstein", water_sample.lime) toxin = DisplayVariable("Gifte", water_sample.toxin) x_positions = [-800, -400, 400, 800] stims = generate_variable_display([solid, algae, lime, toxin], x_positions) procedures = water_sample.procedure_dict() answers = [] proc_x = -600 proc_y = -100 answ_x = 200 answ_y = -100 y_offset = 80 for proc in procedure_keys: p = DisplayProcedure(procedures[proc][1], procedures[proc][0]) p = generate_procedure_display(p, (proc_x, proc_y)) stims.append(p) proc_y -= y_offset stim_answer_equals = visual.TextBox2( WIN, "=", letterHeight=50, pos=(answ_x - 100, answ_y), size=[150, 70], alignment="center", ) stims.append(stim_answer_equals) stim_answer_box = visual.TextBox2( WIN, "", letterHeight=50, pos=(answ_x, answ_y), size=[150, 70], editable=True, fillColor="white", color="black", alignment="center", ) stims.append(stim_answer_box) answ_y -= y_offset not_finished = True answer = "not answered" start_time = core.monotonicClock.getTime() while not_finished: stim_answer_box.hasFocus = True for stim in stims: stim.draw() WIN.flip() answer = stim_answer_box.text if "\n" in answer: if answer[0].isdigit() and answer[1].isdigit(): not_finished = False else: stim_answer_box.text = answer[:-1] if len(answer) > 2: stim_answer_box.text = stim_answer_box.text[:2] answer_time = core.monotonicClock.getTime() - start_time answers.append((answer.replace("\n", ""), answer_time)) # event.waitKeys(keyList=["space"]) return answers, procedure_keys pause = visual.TextBox2( WIN, """Drücken Sie die Leertaste um mit der nächsten Aufgabe fortzufahren""", letterHeight=50, alignment="center", ) intro = visual.TextBox2(WIN, intro_text, letterHeight=30, alignment="center") intro.draw() WIN.flip() event.waitKeys(keyList=["space"]) train_procedures = PROCEDURE_KEYS[:-1] random.shuffle(train_procedures) transfer_procedure = train_procedures[-1] train_procedures = train_procedures[:-1] train_procedures.append(PROCEDURE_KEYS[-1]) all_samples = generate_all_watersamples(TRAIN_TRIALS + TEST_TRIALS) pprint(all_samples) results = {} for i in range(TRAIN_TRIALS): print(train_procedures) answer, procedure_keys = run_trial( all_samples[i], train_procedures, ORDER_CONDITIONS[0] ) answer_dict = {} answer_dict["procedure_order"] = procedure_keys answer_dict["water_sample"] = all_samples[i].water_sample_dict() for proc, key in zip(answer, procedure_keys): answer_dict[key] = {"answer": proc[0], "time": proc[1]} results[f"train_{i}"] = answer_dict pause.draw() WIN.flip() event.waitKeys(keyList=["space"]) phase = visual.TextBox2( WIN, """Sie haben den ersten Teil geschafft! Der zweite Teil ist etwas kürzer als der erste Drücken Sie die Leertaste um anzufangen.""", letterHeight=50, alignment="center", ) phase.draw() WIN.flip() event.waitKeys(keyList=["space"]) train_procedures[2] = transfer_procedure for i in range(TEST_TRIALS): print(train_procedures) answer, procedure_keys = run_trial( all_samples[TRAIN_TRIALS + i], train_procedures, ORDER_CONDITIONS[0] ) answer_dict = {} answer_dict["procedure_order"] = procedure_keys answer_dict["water_sample"] = all_samples[TRAIN_TRIALS + i].water_sample_dict() for proc, key in zip(answer, procedure_keys): answer_dict[key] = {"answer": proc[0], "time": proc[1]} results[f"test_{i}"] = answer_dict pause.draw() WIN.flip() event.waitKeys(keyList=["space"]) pprint(results) df = pd.DataFrame.from_dict(results, orient="index") df.to_csv("vp_results.csv") with open("vp.pkl", "wb") as file: pickle.dump(results, file)