Picture-Word-N1/04 Analysis/raw_data/stim-pc/01_localiser_task.py

subj_id_str = input('Input participant id (integer): ')

if subj_id_str.isdigit():
    subj_id = int(subj_id_str)
else:
    subj_id = subj_id_str

# %% General Setup

from psychopy import parallel, visual, core, monitors, event
import os
import sys
import datetime
import numpy as np
from matplotlib import pyplot as plt
import pandas as pd
from string import ascii_lowercase
from random import choices

from stim import *
from randphase import draw, phase

# %% Participant Info

info_participants = pd.read_csv('participants.csv')
info_counterbalance = pd.read_csv('counterbalance.csv')

participant_info = {
    'subj_id': subj_id,
    'session_nr': int(info_participants.session_nr[info_participants.subj_id==subj_id_str]),
    'stim_grp': int(info_counterbalance.stim_grp[info_counterbalance.subj_id==subj_id_str]),  # 1/2
    'resp_grp': int(info_counterbalance.resp_grp[info_counterbalance.subj_id==subj_id_str]),  # 1/2
    'sex': str(info_participants.sex[info_participants.subj_id==subj_id_str].tolist()[0]),  # m/f/nb
    'age': int(info_participants.age[info_participants.subj_id==subj_id_str])
}

# %% Check participant info makes sense

def is_int(s):
    try:
        int(s)
        return True
    except ValueError:
        return False

pinfo_check = {
    'subj_id_is_int': isinstance(participant_info['subj_id'], int) | (participant_info['subj_id']=='0_test'),
    'stim_grp_1_or_2': participant_info['stim_grp'] in [1, 2],
    'resp_grp_1_or_2': participant_info['resp_grp'] in [1, 2],
    'sex_m_f_or_nb': participant_info['sex'] in ['m', 'f', 'nb'],
    'age_is_int': isinstance(participant_info['age'], int)
}

if not all(value for value in pinfo_check.values()):
    print('Participant info error!')
    print(pinfo_check)
    exit()

for f in participant_info:
    print('  {0}: {1}'.format(f, participant_info[f]))

# %% Output setup

subj_file_name = '{0}_{1}_{2}.csv'.format(participant_info['subj_id'], datetime.datetime.now().strftime('%d-%m-%y'), participant_info['session_nr'])

subj_data_path = os.path.join('data', 'localiser', subj_file_name)

if os.path.isfile(subj_data_path):
    sys.exit('File {} already exists!'.format(subj_data_path))

# %% Set up psychopy and eeg

# set up triggers
# (send trigger with `port.setData(x)`, where `x` is a number from 0 to 255)
port = parallel.ParallelPort()

# class fake_parallelport:
    # def setData(self, x):
        # print('Sent {} to fake parallel port'.format(int(x)))
# port = fake_parallelport()
# port.setData(99)

# set up ViewPix monitor
mon_Hz = 60  # refresh rate
mon_framelen = 1/mon_Hz  # length of one frame in seconds
mon = monitors.Monitor(name='ViewPix', notes='{0}Hz'.format(mon_Hz))
mon.setWidth(52)
mon.setDistance(48)
mon.setSizePix((1920, 1080))
# mon.setSizePix((750, 500))
mon.saveMon()

# set up the Psychopy window
win = visual.Window(fullscr=True,
                    size=mon.getSizePix(),
                    screen=1,
                    monitor=mon,
                    units='deg',
                    color=[0, 0, 0])

# hide the mouse
event.Mouse(visible=False)

win.refreshThreshold = mon_framelen + 0.001  # tolerance of 1 ms (any refresh that takes more than the tolerance longer than expected = dropped frame)
#win.recordFrameIntervals = True  # record actual frame lengths

# set up which button is affirmative, and which negative
if participant_info['resp_grp'] == 1:
    resp_setup = {'yes': 'rctrl', 'no': 'lctrl'}
elif participant_info['resp_grp'] == 2:
    resp_setup = {'yes': 'lctrl', 'no': 'rctrl'}

resp_setup_names = {'lctrl': 'Left Control', 'rctrl': 'Right Control'}

init_stim = visual.TextStim(
        win,
        units='norm',
        height=0.07,
        wrapWidth=1.75,
        text='Initialising...'
    )

init_stim.draw()
win.flip()

# %% Trial Setup

# get proportional difference between descenders' heights and other characters
# (psychopy seems to ignore descenders in calculating text height)
desc = ['g', 'j', 'p', 'q', 'y']
non_desc = [l for l in ascii_lowercase if l not in desc]
desc_im = draw.text(''.join(ascii_lowercase), font='cour.ttf', size=1000)
non_desc_im = draw.text(''.join(non_desc), font='cour.ttf', size=1000)
font_desc_prop_diff = desc_im.size[0] / non_desc_im.size[0]

# options for creating the text images
text_va_height = 1.5  # text height in visual angle
text_va_rad = (text_va_height) / (180/np.pi)
text_height_cm = 2 * mon.getDistance() * np.tan(text_va_rad / 2)
font_size = text_height_cm * 28.3464567 * font_desc_prop_diff  
font_size_ceil = np.uint8(np.ceil(font_size))
font_disp_prop = font_size / font_size_ceil  # the proportion of original height that the image should be presented at for the desired visual angle
text_dpi = 72

# import words for trials
trls_words = pd.read_csv(os.path.join('stim', 'localiser_stim.csv')).string.tolist()
practice_trls_words = pd.read_csv(os.path.join('stim', 'practice_localiser_stim.csv')).string.tolist()

# create a dataframe of all trials
trls = pd.DataFrame(data={'string': trls_words + trls_words + trls_words,
        'condition': ['word']*len(trls_words) + ['bacs']*len(trls_words) + ['noise']*len(trls_words),
        'item_nr': np.concatenate([np.arange(1, len(trls_words)+1),
                                   np.arange(1, len(trls_words)+1),
                                   np.arange(1, len(trls_words)+1)])})

# get the file names for each trial
trls['file'] = ['{0}{1:03d}.png'.format(trls.condition[i], trls.item_nr[i]) for i in range(len(trls))]
trls['file_green'] = ['green_{0}'.format(trls.file[i]) for i in range(len(trls))]

# store what the correct response for each trial will be
trls['corr_ans'] = [resp_setup['yes'] if trls.condition[trl_nr]=='word' else resp_setup['no'] for trl_nr in range(len(trls))]

# create and save each trial's image (noise images are unique to each session)
for i in set(trls.item_nr):
    print('Generating stimuli {0}/{1}'.format(i, len(set(trls.item_nr))))
    trls_i = trls[trls.item_nr==i]
    w_idx = (trls.item_nr==i) & (trls.condition=='word')
    b_idx = (trls.item_nr==i) & (trls.condition=='bacs')
    n_idx = (trls.item_nr==i) & (trls.condition=='noise')
    # save this item's word image
    w_i = draw.text(trls.string[w_idx].item(), font='cour.ttf', size=font_size_ceil, crop_x='font', crop_y='font')
    w_i.save(os.path.join('stim', 'img', trls.file[w_idx].item()))
    # save this item's word image in green
    w_gr_i = draw.text(trls.string[w_idx].item(), font='cour.ttf', size=font_size_ceil, crop_x='font', crop_y='font', colour=(0,255,0))
    w_gr_i.save(os.path.join('stim', 'img', trls.file_green[w_idx].item()))
    # save this item's BACS string image
    b_i = draw.text(trls.string[b_idx].item(), font='BACS2serif.otf', size=font_size_ceil, crop_x='font', crop_y='font')
    b_i.save(os.path.join('stim', 'img', trls.file[b_idx].item()))
    # save this item's BACS string image in green
    b_gr_i = draw.text(trls.string[b_idx].item(), font='BACS2serif.otf', size=font_size_ceil, crop_x='font', crop_y='font', colour=(0,255,0))
    b_gr_i.save(os.path.join('stim', 'img', trls.file_green[b_idx].item()))
    # get phase-randomised word image
    n_i = phase.randomise(w_i, noise='permute', contrast_adj=0.5)
    n_i.save(os.path.join('stim', 'img', trls.file[n_idx].item()))
    # get the phase-randomised image in green
    n_gr_i = greenify_noise(n_i)
    n_gr_i.save(os.path.join('stim', 'img', trls.file_green[n_idx].item()))
    
# generate the practice stimuli
practice_trls = pd.DataFrame(data={'string': practice_trls_words + practice_trls_words + practice_trls_words,
        'condition': ['word']*len(practice_trls_words) + ['bacs']*len(practice_trls_words) + ['noise']*len(practice_trls_words),
        'item_nr': np.concatenate([np.arange(1, len(practice_trls_words)+1),
                                   np.arange(1, len(practice_trls_words)+1),
                                   np.arange(1, len(practice_trls_words)+1)])})

# get the file names for each practice trial
practice_trls['file'] = ['prac_{0}{1:03d}.png'.format(practice_trls.condition[i], practice_trls.item_nr[i]) for i in range(len(practice_trls))]
practice_trls['file_green'] = ['green_{0}'.format(practice_trls.file[i]) for i in range(len(practice_trls))]

# store what the correct response for each trial will be
practice_trls['corr_ans'] = [resp_setup['yes'] if practice_trls.condition[trl_nr]=='word' else resp_setup['no'] for trl_nr in range(len(practice_trls))]

# create and save each practice trial's image
for i in set(practice_trls.item_nr):
    print('Generating practice stimuli {0}/{1}'.format(i, len(set(practice_trls.item_nr))))
    trls_i = practice_trls[practice_trls.item_nr==i]
    w_idx = (practice_trls.item_nr==i) & (practice_trls.condition=='word')
    b_idx = (practice_trls.item_nr==i) & (practice_trls.condition=='bacs')
    n_idx = (practice_trls.item_nr==i) & (practice_trls.condition=='noise')
    # save this item's word image
    w_i = draw.text(practice_trls.string[w_idx].item(), font='cour.ttf', size=font_size_ceil, crop_x='font', crop_y='font')
    w_i.save(os.path.join('stim', 'img', practice_trls.file[w_idx].item()), dpi=(text_dpi, text_dpi))
    # save this item's word image in green
    w_gr_i = draw.text(practice_trls.string[w_idx].item(), font='cour.ttf', size=font_size_ceil, crop_x='font', crop_y='font', colour=(0,255,0))
    w_gr_i.save(os.path.join('stim', 'img', practice_trls.file_green[w_idx].item()))
    # save this item's BACS string image
    b_i = draw.text(practice_trls.string[b_idx].item(), font='BACS2serif.otf', size=font_size_ceil, crop_x='font', crop_y='font')
    b_i.save(os.path.join('stim', 'img', practice_trls.file[b_idx].item()))
    # save this item's BACS string image in green
    b_gr_i = draw.text(practice_trls.string[b_idx].item(), font='BACS2serif.otf', size=font_size_ceil, crop_x='font', crop_y='font', colour=(0,255,0))
    b_gr_i.save(os.path.join('stim', 'img', practice_trls.file_green[b_idx].item()))
    # get phase-randomised word image
    n_i = phase.randomise(w_i, noise='permute', contrast_adj=0.5)
    n_i.save(os.path.join('stim', 'img', practice_trls.file[n_idx].item()))
    # get the phase-randomised image in green
    n_gr_i = greenify_noise(n_i)
    n_gr_i.save(os.path.join('stim', 'img', practice_trls.file_green[n_idx].item()))

# reorder trials randomly, and ensure no more than five of the same type follow each other
shuffle_iter = 0
max_consec_trltypes = 0
while shuffle_iter==0 or max_consec_trltypes>5:
    shuffle_iter+=1
    # shuffle randomly
    trls = trls.sample(frac=1).reset_index(drop=True)
    # count maximum number of consecutive trial types
    max_consec_trltypes = max(get_consec_lens(trls.condition))
    print('Reshuffled trials (iter {}), with a maximum of {} consecutive trial types'.format(shuffle_iter, max_consec_trltypes))
print('Successfully reshuffled trials\n')

# %% Instructions

def instructions(reminder = False, practice = False, wait_time = 3):

    print('DISPLAYING INSTRUCTIONS')

    if practice:
        practice_txt = u'For the practice trials, you will be given feedback on your accuracy for each trial.'
    else:
        practice_txt = u'Unlike the practice trials, you will not be given feedback on your accuracy for each trial.'

    instr_txt = u'In each trial, the following things will happen:\n\n1) You will be shown a picture of a real word, a fake font word, or a noise image.\n2) The image will turn green.\n3) When the image turns green:\n        Press the {0} key if the image is of a real word.\n           OR\n        Press the {1} key if it is not of a real word.\n\n\n Once the image changes colour, try to respond as quickly and accurately as possible.\n\n {2}\n\n\nWhen you have read these instructions, press the space key to begin...'.format(resp_setup_names[resp_setup['yes']], resp_setup_names[resp_setup['no']], practice_txt)

    if reminder:
        instr_txt = u'{0}{1}'.format('Here\'s a reminder of the task instructions.\n\n\n', instr_txt)

    instr_screen = visual.TextStim(
        win,
        units='norm',
        height=0.07,
        wrapWidth=1.75,
        text=instr_txt,
        anchorHoriz='center',
        alignText='left'
    )

    instr_screen.draw()
    win.flip()

    core.wait(wait_time)  # ensure that instructions have been read
    event.waitKeys(keyList='space')
    print('SPACE PRESSED; CONTINUING')

# %% Trials setup

# function that will be called for each trial

def trial(image_path=os.path.join('stim', 'img', 'word001.png'), green_image_path=os.path.join('stim', 'img', 'green_word001.png'), stim_trigger=101):

    # a fixation point in style of Thaler et al.
    thaler_fix_stim = thaler_list_fix(win=win)

    # a blank stimulus
    blank_stim = visual.TextStim(win=win, text='')

    # the image to display
    img_stim = visual.ImageStim(
        win = win,
        image = image_path,
        pos = (0.0, 0.0)
    )

    # the image in green
    img_stim_green = visual.ImageStim(
        win = win,
        image = green_image_path,
        pos = (0.0, 0.0)
    )

    # fixation 1
    display_list_ms(win, thaler_fix_stim, 300)  # note that the list version of the function is required, as thaler_list_fix returns a list object
    fix_jitt = display_jitter_ms(win, blank_stim, 300, 1300)

    # image
    display_x_ms(win, img_stim, 500, port = port, trigg = stim_trigger)

    # image in green
    img_stim_green.draw()
    win.flip()
    stim_rt_clock = core.MonotonicClock()

    # response
    stim_resp = event.waitKeys(keyList=['lctrl', 'rctrl'], timeStamped = stim_rt_clock)
    port.setData(0)

    trl_dat = {
        'fix_jitt': fix_jitt,
        'stim_resp': stim_resp
    }

    return(trl_dat)

def block_break():

    # read the subject's data
    subj_data_so_far = pd.read_csv(subj_data_path)

    accuracies = subj_data_so_far['acc'].to_numpy()
    rts = subj_data_so_far['rt'].to_numpy()
    blocks = subj_data_so_far['block_nr'].to_numpy()

    block_summ_txt = ''
    for block_nr in np.unique(blocks):
        block_summ_txt = '{0}\n  Block {1}: {2}%, {3} ms'.format(
            block_summ_txt,
            block_nr,
            np.round(np.mean(accuracies[blocks==block_nr] * 100), 1),
            np.round(np.median(rts[blocks==block_nr])), 1)

    # give subject summary of their data
    break_msg = visual.TextStim(
        win,
        units='norm',
        height=0.07,
        wrapWidth=1.75,
        text=u'Block {0} complete. Time for a break!\n\n\nYour overall accuracy so far is: {1}%\nYour median response time is {2} ms\n\nHere\'s a per-block summary of your experiment so far:{3}'.format(
            np.max(blocks),
            np.round(np.mean(accuracies * 100), 1),
            np.round(np.median(rts), 1),
            block_summ_txt)
    )

    break_msg.draw()
    win.flip()

    print('BREAK BEGUN, PRESS \'Y\' KEY TO CONTINUE THE EXPERIMENT')
    event.waitKeys(keyList='y')
    print('BREAK ENDED')
    
# %% Practice Trials

practice = True
this_is_a_repeat_practice_block = False

while practice:

    instructions(reminder = this_is_a_repeat_practice_block, practice = True)
    
    practice_accuracies = []
    practice_rts = []
    practice_trls = practice_trls.sample(frac=1).reset_index(drop=True)

    for trl_i in range(len(practice_trls.index)):

        condition = practice_trls.condition[trl_i]
        corr_ans = practice_trls.corr_ans[trl_i]

        trl_dat = trial(
            image_path = os.path.join('stim', 'img', practice_trls.file[trl_i]),
            green_image_path = os.path.join('stim', 'img', practice_trls.file_green[trl_i]),
            stim_trigger = 25
        )

        if corr_ans == trl_dat['stim_resp'][0][0]:
            trl_corr = 1
        else:
            trl_corr = 0
        
        practice_accuracies.append(trl_corr)
        practice_rts.append(round(trl_dat['stim_resp'][0][1] * 1000, 1))

        print('PRACTICE TRL', '{}/{}'.format(trl_i+1, len(practice_trls.index)), ', COND', condition, ', CORR_ANS', corr_ans, ', RESP', trl_dat['stim_resp'][0][0], ', ACC', trl_corr, ', RT', round(trl_dat['stim_resp'][0][1] * 1000, 1), ', IMAGE', practice_trls['file'][trl_i], ', IMAGE_GREEN', practice_trls['file_green'][trl_i], ', STRING', practice_trls['string'][trl_i], ', FIX_JITT', round(trl_dat['fix_jitt'][1], 1))

        if trl_corr == 1:
            feedback_txt = 'CORRECT!'
            feedback_col = [-1, 1, -1]
        else:
            feedback_txt = 'INCORRECT!'
            feedback_col = [1, -1, -1]

        feedback_msg = visual.TextStim(
            win = win,
            text = feedback_txt,
            height = 1.5,
            units = 'deg',
            color = feedback_col,
            colorSpace = 'rgb',
            font = 'Courier New'
        )

        display_x_ms(win, feedback_msg, 1000)

    wait_msg = visual.TextStim(
        win = win,
        text = 'Accuracy: {0}%\nMedian RT: {1} ms\n\nPlease wait for Jack...'.format(
            np.round(np.sum(practice_accuracies)/len(practice_trls.index)*100, 1),
            np.round(np.median(practice_rts), 1)),
        height = 0.07,
        units = 'norm',
        wrapWidth=1.75,
    )
    wait_msg.draw()
    win.flip()

    print('PRACTICE BLOCK ACCURACY: {0}/{1} ({2}%)'.format(np.sum(practice_accuracies), len(practice_trls.index), np.round(np.sum(practice_accuracies)/len(practice_trls.index)*100, 1)))
    print('PRACTICE BLOCK MEDIAN RT: {0} (MIN={1}, MAX={2})'.format(np.round(np.median(practice_rts), 1), np.min(practice_rts), np.max(practice_rts)))
    print('REPEAT PRACTICE TRIALS? Y/N')
    repeat_yn = event.waitKeys(keyList=['y', 'n'])
    
    if repeat_yn == ['y']:
        print('REPEATING PRACTICE TRIALS')
        this_is_a_repeat_practice_block = True
    else:
        print('ENDING PRACTICE TRIALS')
    practice = repeat_yn == ['y']

# %% Experimental Trials

instructions(reminder = True, wait_time = 2)

block_nr = 1

for trl_i in range(len(trls.index)):

    # at start of trial, check if time for a break
    # (as trl_i is an index, will take a break if previous trial was the last before a break is due)
    if trl_i % 60 == 0 and trl_i != 0:
        block_break()
        instructions(reminder = True, wait_time = 2)
        block_nr += 1

    condition = trls['condition'][trl_i]
    corr_ans = trls.corr_ans[trl_i]

    if condition == 'word':
        trigg = 101
    elif condition == 'bacs':
        trigg = 102
    elif condition == 'noise':
        trigg = 103
    else:
        print('UNKNOWN CONDITION IN TRIAL {0}! USING TRIGGER OF 50'.format(trl_i+1))
        trigg = 50
    
    trl_dat = trial(
        image_path = os.path.join('stim', 'img', trls.file[trl_i]),
        green_image_path = os.path.join('stim', 'img', trls.file_green[trl_i]),
        stim_trigger = trigg
    )

    if corr_ans == trl_dat['stim_resp'][0][0]:
        trl_corr = 1
    else:
        trl_corr = 0
    
    trl_dat_tidy = {
        'date': datetime.datetime.now().strftime('%d-%m-%y'),
        'trial_save_time': datetime.datetime.now().strftime('%H:%M:%S'),
        'subj_id': participant_info['subj_id'],
        'stim_grp': participant_info['stim_grp'],
        'resp_grp': participant_info['resp_grp'],
        'sex': participant_info['sex'],
        'age': participant_info['age'],
        'block_nr': block_nr,
        'trl_nr': trl_i + 1,
        'item_nr': trls['item_nr'][trl_i],
        'condition': condition,
        'eeg_trigg': trigg,
        'image': trls['file'][trl_i],
        'image_green': trls['file_green'][trl_i],
        'string': trls['string'][trl_i],
        'corr_ans': corr_ans,
        'resp': trl_dat['stim_resp'][0][0],
        'acc': trl_corr,
        'rt': trl_dat['stim_resp'][0][1] * 1000,
        'fix_jitt_flip': trl_dat['fix_jitt'][0],
        'fix_jitt_ms': trl_dat['fix_jitt'][1]
    }

    write_csv(subj_data_path, trl_dat_tidy)

    print('TRL', '{}/{}'.format(trl_i+1, len(trls.index)), ', COND', condition, ', CORR_ANS', corr_ans, ', RESP', trl_dat['stim_resp'][0][0], ', ACC', trl_corr, ', RT', round(trl_dat['stim_resp'][0][1] * 1000, 1), ', IMAGE', trls['file'][trl_i], ', WORD', trls['string'][trl_i], ', FIX_JITT', round(trl_dat['fix_jitt'][1], 1))

# %% Close

end_stim = visual.TextStim(
    win,
    units='norm',
    height=0.07,
    wrapWidth=1.75,
    text='Task Complete! Please wait...'
)

end_stim.draw()
win.flip()
print('EXPERIMENT END. Overall, {0} frames were dropped.'.format(win.nDroppedFrames))
core.wait(5)