imposedDirectionTask.py

import json
import os
import platform
import random
import sys
import time
from string import ascii_letters, digits
import numpy as np
import pandas as pd
import pylink
from psychopy import core, event, gui, logging, monitors, visual, clock
from psychopy.tools.monitorunittools import deg2pix, pix2cm, cm2pix
from scipy import stats
from screeninfo import get_monitors
from EyeLinkCoreGraphicsPsychoPy import EyeLinkCoreGraphicsPsychoPy

# Create a wx.App instance
# app = wx.App(False)
# Switch to the script folder
script_path = os.path.dirname(sys.argv[0])
if len(script_path) != 0:
    os.chdir(script_path)

# Show only critical log message in the PsychoPy console

logging.console.setLevel(logging.CRITICAL)

# Set this variable to True if you use the built-in retina screen as your
# primary display device on macOS. If have an external monitor, set this
# variable True if you   choose to "Optimize for Built-in Retina Display"
# in the Displays preference settings.
use_retina = False

# Set this variable to True to run the script in "Dummy Mode"
dummy_mode = False
# Set this variable to True to run the task in full screen mode
# It is easier to debug the script in non-fullscreen mode
full_screen = True
# Set up EDF data file name and local data folder
nameOfTheTask = "imposeDirection"
# The EDF data filename should not exceed 8 alphanumeric characters
# use ONLY number 0-9, letters, & _ (underscore) in the filename
edf_fname = "sub"
# The proba_up
proba_up = 100
# Prompt user to specify an EDF data filename
# before we open a fullscreen window
dlg_title = "Subject Metadata Entry"
dlg_prompt = (
    "Please enter a file name with 8 or fewer characters\n"
    + "[letters, numbers, and underscore]."
)
# loop until we get a valid filename
while True:
    dlg = gui.Dlg(dlg_title)
    dlg.addText(dlg_prompt)
    dlg.addField(f"{edf_fname}", "Subject Number:")
    dlg.addField("session", "Session:")
    dlg.addField("age", "Age:")
    dlg.addField("sex", "Sex", choices=["F", "M", "NB"])
    dlg.addField("training", "Training", choices=["no", "yes"])
    dlg.addField("proba", "Probability", choices=[0.0, 0.25, 0.5, 0.75, 1.0])
    # show dialog and wait for OK or Cancel
    ok_data = dlg.show()
    # print(dlg)
    print(ok_data)
    if ok_data is not None and dlg.OK:  # if ok_data is not None
        proba_up = ok_data["proba"]
        print(ok_data)
    else:
        print("user cancelled")
        core.quit()
        sys.exit()

    # strip trailing characters, ignore the ".edf" extension
    edf_fname = f"{ok_data['sub']}{ok_data['session']}{int(ok_data['proba']*100)}"

    # check if the filename is valid (length <= 8 & no special char)
    allowed_char = ascii_letters + digits + "-" + "_"
    if not all([c in allowed_char for c in edf_fname]):
        print("ERROR: Invalid EDF filename")
    elif len(edf_fname) > 8:
        print("ERROR: EDF filename should not exceed 8 characters")
    else:
        break

# Creating a folder for the results
# e.g., files defining the interest areas used in each trial
results_folder = f"results_{nameOfTheTask}"
if not os.path.exists(results_folder):
    os.makedirs(results_folder)

# We download EDF data file from the EyeLink Host PC to the local hard
# session_identifier = edf_fname + time_str
subject_identifier = f"sub-{ok_data['sub']}"

# create a folder for the current testing session in the "results" folder
# This folder refers to the subject folders that contains all the conditions done by them.
subject_folder = os.path.join(results_folder, subject_identifier)
if not os.path.exists(subject_folder):
    os.makedirs(subject_folder)

# condition folder
condition_folder = os.path.join(subject_folder, f"session-{ok_data['session']}")
if not os.path.exists(condition_folder):
    os.makedirs(condition_folder)


# Step 1: Connect to the EyeLink Host PC
#
# The Host IP address, by default, is "100.1.1.1".
# the "el_tracker" objected created here can be accessed through the Pylink
# Set the Host PC address to "None" (without quotes) to run the script
# in "Dummy Mode"
if dummy_mode:
    el_tracker = pylink.EyeLink(None)
else:
    try:
        el_tracker = pylink.EyeLink("100.1.1.1")
    except RuntimeError as error:
        print("ERROR:", error)
        core.quit()
        sys.exit()

# Step 2: Open an EDF data file on the Host PC
edf_file = f"sub-{ok_data['sub']}_ses-{ok_data['session']}_proba-{ok_data['proba']}.EDF"
edf_file = f"{edf_fname}.EDF"
try:
    el_tracker.openDataFile(edf_file)
except RuntimeError as err:
    print("ERROR:", err)
    # close the link if we have one open
    if el_tracker.isConnected():
        el_tracker.close()
    core.quit()
    sys.exit()

# Add a header text to the EDF file to identify the current experiment name
# This is OPTIONAL. If your text starts with "RECORDED BY " it will be
# available in DataViewer's Inspector window by clicking
# the EDF session node in the top panel and looking for the "Recorded By:"
# field in the bottom panel of the Inspector.
preamble_text = "RECORDED BY %s" % os.path.basename(__file__)
el_tracker.sendCommand("add_file_preamble_text '%s'" % preamble_text)

# Step 3: Configure the tracker
#
# Put the tracker in offline mode before we change tracking parameters
el_tracker.setOfflineMode()

# Get the software version:  1-EyeLink I, 2-EyeLink II, 3/4-EyeLink 1000,
# 5-EyeLink 1000 Plus, 6-Portable DUO
eyelink_ver = 0  # set version to 0, in case running in Dummy mode
if not dummy_mode:
    vstr = el_tracker.getTrackerVersionString()
    eyelink_ver = int(vstr.split()[-1].split(".")[0])
    # print out some version info in the shell
    print("Running experiment on %s, version %d" % (vstr, eyelink_ver))

# File and Link data control
# what eye events to save in the EDF file, include everything by default
file_event_flags = "LEFT,RIGHT,FIXATION,SACCADE,BLINK,MESSAGE,BUTTON,INPUT"
# what eye events to make available over the link, include everything by default
link_event_flags = "LEFT,RIGHT,FIXATION,SACCADE,BLINK,BUTTON,FIXUPDATE,INPUT"
# what sample data to save in the EDF data file and to make available
# over the link, include the 'HTARGET' flag to save head target sticker
# data for supported eye trackers
if eyelink_ver > 3:
    file_sample_flags = (
        "LEFT,RIGHT,GAZE,HREF,RAW,AREA,HTARGET,GAZERES,BUTTON,STATUS,INPUT"
    )
    link_sample_flags = "LEFT,RIGHT,GAZE,GAZERES,AREA,HTARGET,STATUS,INPUT"
else:
    file_sample_flags = "LEFT,RIGHT,GAZE,HREF,RAW,AREA,GAZERES,BUTTON,STATUS,INPUT"
    link_sample_flags = "LEFT,RIGHT,GAZE,GAZERES,AREA,STATUS,INPUT"
el_tracker.sendCommand("file_event_filter = %s" % file_event_flags)
el_tracker.sendCommand("file_sample_data = %s" % file_sample_flags)
el_tracker.sendCommand("link_event_filter = %s" % link_event_flags)
el_tracker.sendCommand("link_sample_data = %s" % link_sample_flags)

# Optional tracking parameters
# Sample rate, 250, 500, 1000, or 2000, check your tracker specification
# if eyelink_ver > 2n
#     el_tracker.sendCommand("sample_rate 1000")

# Choose a calibration type, H3, HV3, HV5, HV13 (HV = horizontal/vertical),
el_tracker.sendCommand("calibration_type = HV5")
# Set a gamepad button to accept calibration/drift check target
# You need a supported gamepad/button box that is connected to the Host PC
el_tracker.sendCommand("button_function 5 'accept_target_fixation'")

# Step 4: set up a graphics environment for calibration
#
# Get a list of all connected displays from screeninfo
all_screens = get_monitors()
print(all_screens)

# Specify the index for your secondary display
display_index = 1

# Check if the specified index is valid
if display_index < len(all_screens):
    # Get the screen properties for the specified display
    secondary_screen = all_screens[display_index]
    scn_x = secondary_screen.x
    print(scn_x)

    scn_y = secondary_screen.y

    print(scn_y)

else:
    display_index = 0


print(display_index)

# Get the monitor size in pixels
monitor_size_pix = all_screens[display_index].width, all_screens[display_index].height
print(monitor_size_pix)
# Open a window, be sure to specify monitor parameters
mon = monitors.Monitor("myMonitor", width=70, distance=57.0)
# Set the monitor size in pixels
mon.setSizePix(monitor_size_pix)
print("width of the monitor in CM is:", pix2cm(all_screens[display_index].width, mon))
print("width of the monitor in pix is:", cm2pix(70, mon))
# Defining the velocity in deg/s
velocity = 11
vel_pix = deg2pix(velocity, monitor=mon)
print("Velocity in pix/s:", vel_pix)
print("1 deg in pix:", deg2pix(1, monitor=mon))
# Create a PsychoPy window
win = visual.Window(
    fullscr=full_screen,
    monitor=mon,
    winType="pyglet",
    units="pix",
    screen=display_index,
)
# Allow the window to draw several frames before getting the frame rate
max_attempts = 60
frame_rate = None
for attempt in range(max_attempts):
    win.flip()
    time.sleep(0.05)  # Short sleep to let the system process
    frame_rate = win.getActualFrameRate()
    if frame_rate is not None:
        break

print("Frame rate:", frame_rate)
sec_per_frame = win.getMsPerFrame()[0] / 1000
print("Second per frame:", sec_per_frame)
print("FR:", 1 / sec_per_frame)
# Forcing the frame rate to be a 120Hz
# frame_rate=60
if display_index != 0:
    # Get the Pyglet window object associated with the Psychopy window
    pyglet_win = win.winHandle
    pyglet_win.set_location(scn_x, -scn_y)

# get the native screen resolution used by PsychoPy
scn_width, scn_height = win.size
print("window size:", scn_width, scn_height)
# resolution fix for Mac retina displays
if "Darwin" in platform.system():
    if use_retina:
        scn_width = int(scn_width / 2.0)
        scn_height = int(scn_height / 2.0)

# Pass the display pixel coordinates (left, top, right, bottom) to the tracker
# see the EyeLink Installation Guide, "Customizing Screen Settings"
el_coords = "screen_pixel_coords = 0 0 %d %d" % (scn_width - 1, scn_height - 1)
el_tracker.sendCommand(el_coords)

# Write a DISPLAY_COORDS message to the EDF file
# Data Viewer needs this piece of info for proper visualization, see Data
# Viewer User Manual, "Protocol for EyeLink Data to Viewer Integration"
dv_coords = "DISPLAY_COORDS  0 0 %d %d" % (scn_width - 1, scn_height - 1)
el_tracker.sendMessage(dv_coords)

# Configure a graphics environment (genv) for tracker calibration
genv = EyeLinkCoreGraphicsPsychoPy(el_tracker, win)
print(genv)  # print out the version number of the CoreGraphics library

# Set background and foreground colors for the calibration target
# in PsychoPy, (-1, -1, -1)=black, (1, 1, 1)=white, (0, 0, 0)=mid-gray
foreground_color = (-1, -1, -1)
background_color = win.color
genv.setCalibrationColors(foreground_color, background_color)

# Use the default calibration target ('circle')
genv.setTargetType("picture")
genv.setPictureTarget(os.path.join("picture", "images", "fixTarget.bmp"))
# Configure the size of the calibration target (in pixels)
# this option applies only to "circle" and "spiral" targets
# genv.setTargetSize(24) def su
# Beeps to play during calibration, validation and drift correction
# parameters: target, good, error
#     target -- sound to play when target moves
#     good -- sound to play on successful operation
#     error -- sound to play on failure or interruption
# Each parameter could be ''--default sound, 'off'--no sound, or a wav file
# genv.setCalibrationSounds("", "", "")

# resolution fix for macOS retina display issues
if use_retina:
    genv.fixMacRetinaDisplay()

# Request Pylink to use the PsychoPy window we opened above for calibration
pylink.openGraphicsEx(genv)


# define a few helper functions for trial handling


def clear_screen(win):
    """clear up the PsychoPy window"""

    win.fillColor = genv.getBackgroundColor()
    win.flip()


def show_msg(win, text, wait_for_keypress=True):
    """Show task instructions on screen"""

    msg = visual.TextStim(
        win, text, color=genv.getForegroundColor(), wrapWidth=scn_width / 2
    )
    clear_screen(win)
    msg.draw()
    win.flip()

    # wait indefinitely, terminates upon any key press
    if wait_for_keypress:
        event.waitKeys()
        clear_screen(win)


def terminate_task():
    """Terminate the task gracefully and retrieve the EDF data file

    file_to_retrieve: The EDF on the Host that we would like to download
    win: the current window used by the experimental script
    """

    el_tracker = pylink.getEYELINK()

    if el_tracker.isConnected():
        # Terminate the current trial first if the task terminated prematurely
        error = el_tracker.isRecording()
        if error == pylink.TRIAL_OK:
            abort_trial()

        # Put tracker in Offline mode
        el_tracker.setOfflineMode()

        # Clear the Host PC screen and wait for 500 ms
        el_tracker.sendCommand("clear_screen 0")
        pylink.msecDelay(500)

        # Close the edf data file on the Host
        el_tracker.closeDataFile()

        # Show a file transfer message on the screen
        msg = "EDF data is transferring from EyeLink Host PC..."
        show_msg(win, msg, wait_for_keypress=False)

        # Download the EDF data file from the Host PC to a local data folder
        # parameters: source_file_on_the_host, destination_file_on_local_drive
        local_edf = os.path.join(
            condition_folder,
            f"sub-{ok_data['sub']}_ses-{ok_data['session']}_proba-{int(ok_data['proba']*100)}.EDF",
        )
        try:
            el_tracker.receiveDataFile(edf_file, local_edf)
        except RuntimeError as error:
            print("ERROR:", error)

        # Close the link to the tracker.
        el_tracker.close()

    # close the PsychoPy window
    win.close()

    # quit PsychoPy
    core.quit()
    sys.exit()


def abort_trial():
    """Ends recording"""

    el_tracker = pylink.getEYELINK()

    # Stop recording
    if el_tracker.isRecording():
        # add 100 ms to catch final trial events
        pylink.pumpDelay(100)
        el_tracker.stopRecording()

    # clear the screen
    clear_screen(win)

    # send a message to mark trial end
    el_tracker.sendMessage("TRIAL_RESULT %d" % pylink.TRIAL_ERROR)

    return pylink.TRIAL_ERROR


def run_trial(trial_index, p_up=1.0, p_down=0.0, arrow=None):
    """Helper function specifying the events that will occur in a single trial"""

    # get a reference to the currently active EyeLink connection
    el_tracker = pylink.getEYELINK()

    # put the tracker in the offline mode first
    el_tracker.setOfflineMode()
    # clear the Host PC screen before we draw the backdrop
    el_tracker.sendCommand("clear_screen 0")
    # send a "TRIALID" message to mark the start of a trial, see Data
    # Viewer User Manual, "Protocol for EyeLink Data to Viewer Integration"
    el_tracker.sendMessage("TRIALID %d" % trial_index)

    # record_status_message : show some info on the Host PC
    # here we show how many trial has been tested
    status_msg = "TRIAL number %d" % trial_index
    el_tracker.sendCommand("record_status_message '%s'" % status_msg)

    # Step 6: Run a single trial

    # Create the fixation cross
    fixation_cross = visual.TextStim(win, text="+", color="white", height=25)

    # Create two arrow buttons for color selection

    # Arrow parameters
    arrow_width = 10
    arrow_height = 30
    stem_height = 30

    # Up arrow
    up_arrow = visual.ShapeStim(
        win=win,
        vertices=[
            (0, arrow_height),  # Top of the arrowhead
            (-arrow_width, 0),  # Left base of the arrowhead
            (-arrow_width / 2, 0),  # Left top of the stem
            (-arrow_width / 2, -stem_height),  # Bottom of the stem
            (arrow_width / 2, -stem_height),  # Right bottom of the stem
            (arrow_width / 2, 0),  # Right top of the stem
            (arrow_width, 0),  # Right base of the arrowhead
        ],
        lineWidth=3,
        lineColor="white",
        fillColor="white",
        pos=(0, 0),
    )

    # Down arrow
    down_arrow = visual.ShapeStim(
        win=win,
        vertices=[
            (0, -arrow_height),  # Bottom of the arrowhead
            (-arrow_width, 0),  # Left base of the arrowhead
            (-arrow_width / 2, 0),  # Left top of the stem
            (-arrow_width / 2, stem_height),  # Top of the stem
            (arrow_width / 2, stem_height),  # Right top of the stem
            (arrow_width / 2, 0),  # Right base of the stem
            (arrow_width, 0),  # Right base of the arrowhead
        ],
        lineWidth=3,
        lineColor="white",
        fillColor="white",
        pos=(0, 0),
    )

    # # checking where the arrow pointing up is positioned
    # if button_pos > 0:
    #     button_pos = "up"
    # else:
    #     button_pos = "down"
    #
    # if button_pos == "up":
    #     print("the arrow pointing up is: UP")
    # else:
    #     print("the arrow pointing up is: DOWN")

    while (not dummy_mode) and (
        trial_index == 1
        or trial_index == 50
        or trial_index == 100
        or trial_index == 150
        or trial_index == 200
    ):
        # terminate the task if no longer connected to the tracker or
        # user pressed Ctrl-C to terminate the task
        if (not el_tracker.isConnected()) or el_tracker.breakPressed():
            terminate_task()
            return pylink.ABORT_EXPT

        # drift-check and re-do the camera setup if ESC is pressed

        try:
            error = el_tracker.doDriftCorrect(
                int(scn_width / 2.0), int(scn_height / 2.0), 1, 1
            )

            # break following a successful drift-check
            if error is not pylink.ESC_KEY:
                break
        except:
            pass

    # put the tracker in the offline before starting recording
    el_tracker.setOfflineMode()

    try:
        el_tracker.startRecording(1, 1, 1, 1)
    except RuntimeError as error:
        print("ERROR:", error)
        abort_trial()
        return pylink.TRIAL_ERROR

    # Allocate some time for the tracker to cache some samples
    pylink.pumpDelay(100)
    # clear the screen
    clear_screen(win)
    # Show the color selection buttons
    # up_arrow.draw()
    # down_arrow.draw()
    # win.flip()
    # el_tracker.sendMessage("cue_selection")
    stimInterval = random.uniform(0.5, 0.75)
    chosen_arrow = arrow
    if chosen_arrow == "up":
        up_arrow.draw()
    else:
        down_arrow.draw()
    win.flip()
    el_tracker.sendMessage("StimOn")
    core.wait(stimInterval)
    el_tracker.sendMessage("StimOff")
    win.flip()
    
    # Probability of the dot going right given that the arrow chosen is the one pointing UP
    v_up = vel_pix * (2 * stats.bernoulli.rvs(p_up) - 1)
    # Probability of the dot going right given that the arrow chosen is the one pointing DOWN
    v_down = vel_pix * (2 * stats.bernoulli.rvs(p_down) - 1)
    print("the arrow displayed is:", chosen_arrow)

    if chosen_arrow == "up":
        target_direction = np.sign(v_up)
    else:
        target_direction = np.sign(v_down)

    # position of the dot depending on the arrow chosen.
    upper_pos = [0, 0]
    lower_pos = [0, 0]
    # Initialize upper and lower circle dot
    upper_circle = visual.Circle(
        win, radius=5, fillColor="white", lineColor="white", pos=upper_pos
    )
    lower_circle = visual.Circle(
        win, radius=5, fillColor="white", lineColor="white", pos=lower_pos
    )

    #fixation_cross.draw()
    #win.flip()
    #el_tracker.sendMessage("FixOn")
    # Random time between 500ms and 570ms
    #random_time_interval = random.uniform(0.5, 0.75)
    #core.wait(random_time_interval)
    #el_tracker.sendMessage("FixOff")
    #win.flip()
    # GAP
    gap = 0.2
    core.wait(gap)
    win.flip()
    # Create a high-resolution clock
    frame_clock = clock.Clock()

    last_frame_time = frame_clock.getTime()
    if chosen_arrow == "up":
        upper_circle.draw()
    else:
        lower_circle.draw()

    win.flip()

    el_tracker.sendMessage("TargetOnSet")

    # Animation loop for each trial

    # Duration of the trial:
    deltaT = 0.6
    # timer = core.CountdownTimer(deltaT)  # Set timer
    elapsed_time = 0
    num_frames = 0
    while elapsed_time < deltaT:
        frame_time = frame_clock.getTime()
        dt = frame_time - last_frame_time
        if chosen_arrow == "up":
            upper_pos[0] += target_direction * vel_pix * dt
            upper_circle.pos = upper_pos
            upper_circle.draw()

        else:
            lower_pos[0] += target_direction * vel_pix * dt
            lower_circle.pos = lower_pos
            lower_circle.draw()
        # Update the window
        win.flip()
        elapsed_time += dt
        last_frame_time = frame_time
        # print("dt:",dt)
        # print('frame rate:',1/dt)
        num_frames += 1

    el_tracker.sendMessage("TargetOffSet")
    # print the number of frames
    print("Number of frames:", num_frames)
    # print the elapsed time
    print("Elapsed time:", elapsed_time)

    # Send a message to clear the Data Viewer screen, get it ready for
    # drawing the pictures during visualization
    # bgcolor_RGB = (116, 116, 116)
    # el_tracker.sendMessage('!V CLEAR %d %d %d' % bgcolor_RGB)

    # abort the current trial if the tracker is no longer recording
    error = el_tracker.isRecording()
    if error is not pylink.TRIAL_OK:
        el_tracker.sendMessage("tracker_disconnected")
        abort_trial()
        return error

    # check keyboard events
    for keycode, modifier in event.getKeys(modifiers=True):
        # Abort a trial if "ESCAPE" is pressed
        if keycode == "escape":
            el_tracker.sendMessage("trial_skipped_by_user")
            # clear the screen
            clear_screen(win)
            # abort trial
            abort_trial()
            return pylink.SKIP_TRIAL

        # Terminate the task if Ctrl-c
        if keycode == "c" and (modifier["ctrl"] is True):
            el_tracker.sendMessage("terminated_by_user")
            terminate_task()
            return pylink.ABORT_EXPT

    # clear the screen
    clear_screen(win)
    el_tracker.sendMessage("blank_screen")
    # stop recording; add 100 msec to catch final events before stopping
    pylink.pumpDelay(100)
    el_tracker.stopRecording()

    return chosen_arrow, target_direction


def run_exp(
    NumbOfTrials=4,
    proba_up=proba_up,
    proba_down=1 - proba_up,
    filename=f"sub-{ok_data['sub']}_ses-{ok_data['session']}_proba-{int(ok_data['proba']*100)}",
):

    # Initialize the DataFrame with the number of trials
    events = pd.DataFrame(
        index=range(NumbOfTrials),
        columns=[
            "trial",
            "proba",
            "arrow",
            "target_direction",
        ],
    )

    # Show the task instructions
    task_msg = (
        "In the task, one arrow will appear, pointing UP or DOWN,\n"
        + "after which a dot will appear.\n"
        + "Follow the movement of the dot with your eyes until the end.\n"
    )

    show_msg(win, task_msg)
    seq = ["down"] * (NumbOfTrials // 2) + ["up"] * (NumbOfTrials // 2)
    np.random.shuffle(seq)
    print(seq)
    for i in range(NumbOfTrials):
        arrow, target_direction = run_trial(i + 1, proba_up, proba_down, arrow=seq[i])
        events.loc[i] = [
            i + 1,
            proba_up,
            arrow,
            target_direction,
        ]
        # Printing the ration of the chosen_arrow
        print(
            "the ratio of the up arrow is: ",
            events["arrow"].value_counts(normalize=True),
        )

    # Create the full file path for the CSV and json files

    csv_filepath = os.path.join(condition_folder, f"{filename}.csv")
    json_path = os.path.join(condition_folder, f"{filename}.json")
    with open(json_path, "w") as f:
        json.dump(ok_data, f)

    # Save the DataFrame to the CSV file
    events.to_csv(csv_filepath, index=False)


# Show the task instructions for calibration
task_msg = "Follow the Bull's Eye target with your eyes.\n"
show_msg(win, task_msg)


# skip this step if running the script in Dummy Mode
if not dummy_mode:
    try:
        el_tracker.doTrackerSetup()
    except RuntimeError as err:
        print("ERROR:", err)
        el_tracker.exitCalibration()


# Step 6: Training

# Step 7: Run the experiment trials
NumbOfTrials = 10
# Run the experiment
run_exp(NumbOfTrials, proba_up)

# Step 7: Terminate the task
terminate_task()