cosacog/paired_tms4psychopy.py

## paired_tms4psychopy.py
#!/usr/bin/env python2
# -*- coding: utf-8 -*-

# import iroiro
from __future__ import division  # so that 1/3=0.333 instead of 1/3=0
from psychopy import locale_setup, visual, core, data, event, logging, sound, gui, parallel
from psychopy.constants import *  # things like STARTED, FINISHED
import numpy as np  # whole numpy lib is available, prepend 'np.'
import os  # handy system and path functions
import sys # to get file system encoding

#----------------- constants and settings-----------------------------
# ISIs condition
lst_dict_cond = []
lst_dict_cond.append({"name":"ISI20"  , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.020, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
lst_dict_cond.append({"name":"ISI30"  , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.030, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
lst_dict_cond.append({"name":"ISI40"  , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.040, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
lst_dict_cond.append({"name":"control", "prt_cond":int('00000000',2), "duration_cond":0.005, "ISI":0.020, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":15})

# Inter trial interval
lst_iti = [5, 7] # when identical numbers, constant interval

# constants: port address
PORT_ADDRESS_HEX = '0xCFF8'

# constants: this time must be longer than longest isi in lst_dict_cond
T_WAIT = 1.0

# constants: take some rest every "TRIAL_NUM_SESSION" trials
TRIAL_NUM_SESSION = 60

# setting: randomise or not
RANDOMISE_TRIALS = True

# constants: trial states-included in psychopy.constants
# STARTED = 1
# NOT_STARTED = 0
# FINISHED = -1

#--------- functions --------------------
def GetTrialSequence(lst_dict_cond):
    """
    create sequence of stimulus conditions
    params:
      lst_dict_cond: list of dict_cond - see above
    return:
      lst_dict_stim: lst stim conditions for each trial
    """

    lst_dict_stim = []
    for item in lst_dict_cond:
        trialN = item['trialN']
        item_subset = {k:item[k] for k in ['prt_cond','duration_cond','ISI','prt_test','duration_test']}
        for idx in range(trialN):
            lst_dict_stim.append(item_subset)
    return lst_dict_stim

def take_rest(msg_txts):
    """
    take some rest before the 1st trial and between sessions
    params:
        msg_txts
    """
    msgInit.setText(msg_txts)
    print(msg_txts)
    #msgInit.setAutoDraw(True)
    continueRoutine = True
    while continueRoutine:
        win.flip()
        theseKeys = event.getKeys(keyList=['return'])
        if "return" in theseKeys:
            continueRoutine = False

    msgInit.setText("")
    win.flip()
    core.wait(2.0)

def _get_iti(lst_iti):
    """
    set random number for inter trial interval
    params:
        iti: list of 2 numbers, shortest and longest time (sec)
        iti[0] <= iti[1]
    return:
        iti_out: float: iti for single trial
    """
    import numpy as np
    if lst_iti[0] == lst_iti[1]:
        iti_out = float(lst_iti[0])
        return iti_out
    iti_out = np.random.rand(1) * (lst_iti[1] - lst_iti[0]) + lst_iti[0]
    return iti_out

#---------- create trial sequences ----------------
lst_dict_stim = GetTrialSequence(lst_dict_cond) # condition info (e.g. DARK_COND, BRIGHT_SINGLE) sequence
idxsTrial = np.arange(len(lst_dict_stim)) # index to define each trial

if RANDOMISE_TRIALS:
    idxsTrialRandom = np.random.permutation(idxsTrial)
else:
    idxsTrialRandom = idxsTrial

#-------- Start Code - component code to be run before the window creation -----
# Setup the Window
win = visual.Window(size=[1024, 768], fullscr= False, screen=1, allowGUI=True, allowStencil=False,
    monitor=u'testMonitor', color=[0,0,0], colorSpace='rgb',
    blendMode='avg', useFBO=True,
    )

# Initialize components for Routine "trial"
trialClock = core.Clock()

msgInit = visual.TextStim(win=win, ori=0, name='text',
    text=u'',    font=u'Arial', alignHoriz ='center',
    units='pix', pos=[0, 0], height=50, wrapWidth=1024,
    color=u'white', colorSpace='rgb', opacity=1,
    depth=-3.0)
msgInit.setAutoDraw(True)

p_port = parallel.ParallelPort(address=PORT_ADDRESS_HEX)

#--------- start main processing -----------------------
len_trls = len(lst_dict_stim)
sessionNtotal = int(np.ceil(len_trls / TRIAL_NUM_SESSION))

# show message to be ready for trials
msgTxt = "{0} sessions of {1} trials in total.\nPress 'ENTER' to start.".format(sessionNtotal, len_trls)
take_rest(msgTxt)

#----------- for-loop of each trial ---------------
idxTrl = 0
sessionN = 0
for idx in idxsTrialRandom:
    idxTrl += 1
    #--- prepare for a single trial ------------
    # collect necessary info
    dict_stim = lst_dict_stim[idx]
    prt_cond = dict_stim['prt_cond']
    dur_cond = dict_stim['duration_cond']
    prt_test = dict_stim['prt_test']
    dur_test = dict_stim['duration_test']
    isi_cond_test = dict_stim['ISI']
    iti = _get_iti(lst_iti)
    iti_pretms = iti - T_WAIT
    # show message of trial number
    msg_txt = "Next Trial\n{0}/{1}".format(idxTrl,len_trls)
    msgInit.setText(msg_txt)
    win.flip()

    #------Prepare to start Routine "trial": reset timer and frame number -------
    p_port.setData(0)
    t = 0
    trialClock.reset()  # clock
    # status
    cond_status = NOT_STARTED
    test_status = NOT_STARTED
    continueRoutine = True
    #-------Start Routine "trial"-------
    while continueRoutine:
        # get current time and frame number
        t = trialClock.getTime()

        # conditioning stimulus
        if cond_status == NOT_STARTED and t >= iti_pretms:
            # keep track of start time/frame for later
            # circlePre.tStart = t  # underestimates by a little under one frame
            p_port.setData(prt_cond)
            cond_status = STARTED

        if cond_status == STARTED and t >= iti_pretms + dur_cond:
            p_port.setData(0)
            cond_status = FINISHED

        # test stimulus
        if test_status == NOT_STARTED and t >= iti_pretms + isi_cond_test:
            p_port.setData(prt_test)
            test_status = STARTED

        if test_status == STARTED and t >= iti_pretms + isi_cond_test + dur_test:
            p_port.setData(0)
            test_status = FINISHED

        # quit script when pressed "esc"
        theseKeys = event.getKeys(keyList=['escape'])
        if "escape" in theseKeys:
            win.close()
            core.quit()

        if t > iti:
            continueRoutine = False

    if idxTrl % TRIAL_NUM_SESSION == 0 and idxTrl < len_trls:
        sessionN += 1
        msgTxtBreak = "Take some Rest\n Finished Session {0}/{1}.\n Press 'ENTER' to continue.".format(sessionN, sessionNtotal)
        take_rest(msgTxtBreak)

core.wait(T_WAIT)
#-------Ending Routine "trial"-------
msgTxtEnd = "FINISHED.\nWindow will be closed.\nPress 'ENTER' to quit."
take_rest(msgTxtEnd)

# end the script
win.close()
core.quit()
	#!/usr/bin/env python2
	# -- coding: utf-8 --

	# import iroiro
	from __future__ import division # so that 1/3=0.333 instead of 1/3=0
	from psychopy import locale_setup, visual, core, data, event, logging, sound, gui, parallel
	from psychopy.constants import * # things like STARTED, FINISHED
	import numpy as np # whole numpy lib is available, prepend 'np.'
	import os # handy system and path functions
	import sys # to get file system encoding

	#----------------- constants and settings-----------------------------
	# ISIs condition
	lst_dict_cond = []
	lst_dict_cond.append({"name":"ISI20" , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.020, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
	lst_dict_cond.append({"name":"ISI30" , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.030, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
	lst_dict_cond.append({"name":"ISI40" , "prt_cond":int('00000110',2), "duration_cond":0.005, "ISI":0.040, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":12})
	lst_dict_cond.append({"name":"control", "prt_cond":int('00000000',2), "duration_cond":0.005, "ISI":0.020, "prt_test":int('00110000',2), "duration_test":0.005, "trialN":15})

	# Inter trial interval
	lst_iti = [5, 7] # when identical numbers, constant interval

	# constants: port address
	PORT_ADDRESS_HEX = '0xCFF8'

	# constants: this time must be longer than longest isi in lst_dict_cond
	T_WAIT = 1.0

	# constants: take some rest every "TRIAL_NUM_SESSION" trials
	TRIAL_NUM_SESSION = 60

	# setting: randomise or not
	RANDOMISE_TRIALS = True

	# constants: trial states-included in psychopy.constants
	# STARTED = 1
	# NOT_STARTED = 0
	# FINISHED = -1

	#--------- functions --------------------
	def GetTrialSequence(lst_dict_cond):
	"""
	create sequence of stimulus conditions
	params:
	lst_dict_cond: list of dict_cond - see above
	return:
	lst_dict_stim: lst stim conditions for each trial
	"""

	lst_dict_stim = []
	for item in lst_dict_cond:
	trialN = item['trialN']
	item_subset = {k:item[k] for k in ['prt_cond','duration_cond','ISI','prt_test','duration_test']}
	for idx in range(trialN):
	lst_dict_stim.append(item_subset)
	return lst_dict_stim

	def take_rest(msg_txts):
	"""
	take some rest before the 1st trial and between sessions
	params:
	msg_txts
	"""
	msgInit.setText(msg_txts)
	print(msg_txts)
	#msgInit.setAutoDraw(True)
	continueRoutine = True
	while continueRoutine:
	win.flip()
	theseKeys = event.getKeys(keyList=['return'])
	if "return" in theseKeys:
	continueRoutine = False

	msgInit.setText("")
	win.flip()
	core.wait(2.0)

	def _get_iti(lst_iti):
	"""
	set random number for inter trial interval
	params:
	iti: list of 2 numbers, shortest and longest time (sec)
	iti[0] <= iti[1]
	return:
	iti_out: float: iti for single trial
	"""
	import numpy as np
	if lst_iti[0] == lst_iti[1]:
	iti_out = float(lst_iti[0])
	return iti_out
	iti_out = np.random.rand(1) * (lst_iti[1] - lst_iti[0]) + lst_iti[0]
	return iti_out

	#---------- create trial sequences ----------------
	lst_dict_stim = GetTrialSequence(lst_dict_cond) # condition info (e.g. DARK_COND, BRIGHT_SINGLE) sequence
	idxsTrial = np.arange(len(lst_dict_stim)) # index to define each trial

	if RANDOMISE_TRIALS:
	idxsTrialRandom = np.random.permutation(idxsTrial)
	else:
	idxsTrialRandom = idxsTrial

	#-------- Start Code - component code to be run before the window creation -----
	# Setup the Window
	win = visual.Window(size=[1024, 768], fullscr= False, screen=1, allowGUI=True, allowStencil=False,
	monitor=u'testMonitor', color=[0,0,0], colorSpace='rgb',
	blendMode='avg', useFBO=True,
	)

	# Initialize components for Routine "trial"
	trialClock = core.Clock()

	msgInit = visual.TextStim(win=win, ori=0, name='text',
	text=u'', font=u'Arial', alignHoriz ='center',
	units='pix', pos=[0, 0], height=50, wrapWidth=1024,
	color=u'white', colorSpace='rgb', opacity=1,
	depth=-3.0)
	msgInit.setAutoDraw(True)

	p_port = parallel.ParallelPort(address=PORT_ADDRESS_HEX)

	#--------- start main processing -----------------------
	len_trls = len(lst_dict_stim)
	sessionNtotal = int(np.ceil(len_trls / TRIAL_NUM_SESSION))

	# show message to be ready for trials
	msgTxt = "{0} sessions of {1} trials in total.\nPress 'ENTER' to start.".format(sessionNtotal, len_trls)
	take_rest(msgTxt)

	#----------- for-loop of each trial ---------------
	idxTrl = 0
	sessionN = 0
	for idx in idxsTrialRandom:
	idxTrl += 1
	#--- prepare for a single trial ------------
	# collect necessary info
	dict_stim = lst_dict_stim[idx]
	prt_cond = dict_stim['prt_cond']
	dur_cond = dict_stim['duration_cond']
	prt_test = dict_stim['prt_test']
	dur_test = dict_stim['duration_test']
	isi_cond_test = dict_stim['ISI']
	iti = _get_iti(lst_iti)
	iti_pretms = iti - T_WAIT
	# show message of trial number
	msg_txt = "Next Trial\n{0}/{1}".format(idxTrl,len_trls)
	msgInit.setText(msg_txt)
	win.flip()

	#------Prepare to start Routine "trial": reset timer and frame number -------
	p_port.setData(0)
	t = 0
	trialClock.reset() # clock
	# status
	cond_status = NOT_STARTED
	test_status = NOT_STARTED
	continueRoutine = True
	#-------Start Routine "trial"-------
	while continueRoutine:
	# get current time and frame number
	t = trialClock.getTime()

	# conditioning stimulus
	if cond_status == NOT_STARTED and t >= iti_pretms:
	# keep track of start time/frame for later
	# circlePre.tStart = t # underestimates by a little under one frame
	p_port.setData(prt_cond)
	cond_status = STARTED

	if cond_status == STARTED and t >= iti_pretms + dur_cond:
	p_port.setData(0)
	cond_status = FINISHED

	# test stimulus
	if test_status == NOT_STARTED and t >= iti_pretms + isi_cond_test:
	p_port.setData(prt_test)
	test_status = STARTED

	if test_status == STARTED and t >= iti_pretms + isi_cond_test + dur_test:
	p_port.setData(0)
	test_status = FINISHED

	# quit script when pressed "esc"
	theseKeys = event.getKeys(keyList=['escape'])
	if "escape" in theseKeys:
	win.close()
	core.quit()

	if t > iti:
	continueRoutine = False

	if idxTrl % TRIAL_NUM_SESSION == 0 and idxTrl < len_trls:
	sessionN += 1
	msgTxtBreak = "Take some Rest\n Finished Session {0}/{1}.\n Press 'ENTER' to continue.".format(sessionN, sessionNtotal)
	take_rest(msgTxtBreak)

	core.wait(T_WAIT)
	#-------Ending Routine "trial"-------
	msgTxtEnd = "FINISHED.\nWindow will be closed.\nPress 'ENTER' to quit."
	take_rest(msgTxtEnd)

	# end the script
	win.close()
	core.quit()