#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
This experiment was created using PsychoPy3 Experiment Builder (v2023.2.3),
on Mon Jan 15 19:29:12 2024
If you publish work using this script the most relevant publication is:
Peirce J, Gray JR, Simpson S, MacAskill M, Höchenberger R, Sogo H, Kastman E, Lindeløv JK. (2019)
PsychoPy2: Experiments in behavior made easy Behav Res 51: 195.
https://doi.org/10.3758/s13428-018-01193-y
"""
# --- Import packages ---
from psychopy import locale_setup
from psychopy import prefs
from psychopy import plugins
plugins.activatePlugins()
prefs.hardware['audioLib'] = 'ptb'
prefs.hardware['audioLatencyMode'] = '3'
from psychopy import sound, gui, visual, core, data, event, logging, clock, colors, layout
from psychopy.tools import environmenttools
from psychopy.constants import (NOT_STARTED, STARTED, PLAYING, PAUSED,
STOPPED, FINISHED, PRESSED, RELEASED, FOREVER, priority)
import numpy as np # whole numpy lib is available, prepend 'np.'
from numpy import (sin, cos, tan, log, log10, pi, average,
sqrt, std, deg2rad, rad2deg, linspace, asarray)
from numpy.random import random, randint, normal, shuffle, choice as randchoice
import os # handy system and path functions
import sys # to get file system encoding
from psychopy.hardware import keyboard
# Run 'Before Experiment' code from config
import os
import datetime
from time import sleep
import argparse
from device.data_client import NeuracleDataClient
from device.trigger_box import TriggerNeuracle
from device.fubo_pneumatic_finger import FuboPneumaticFingerClient
from settings.config import settings
from bci_core.online import Controller, model_loader
from settings.config import settings
config_info = settings.CONFIG_INFO
# get train params
def parse_args():
parser = argparse.ArgumentParser(
description='Hand gesture train'
)
parser.add_argument(
'--subj',
dest='subj',
help='Subject name',
default=None,
type=str
)
parser.add_argument(
'--side',
dest='side',
help='train side',
default=None,
type=str
)
parser.add_argument(
'--n-trials',
dest='n_trials',
help='Trial number',
type=int,
)
parser.add_argument(
'--hand-feedback',
dest='hand_feedback',
action='store_true',
)
parser.add_argument(
'--hand-port',
dest='hand_port',
help='Peripheral serial port',
type=str
)
parser.add_argument(
'--trigger-port',
dest='trigger_port',
help='Triggerbox serial port',
type=str
)
parser.add_argument(
'--finger-model',
'-fm',
dest='finger_model',
help='Gesture to train',
type=str
)
parser.add_argument(
'--virtual-feedback-rate',
'-vfr',
dest='virtual_feedback_rate',
help='Virtual feedback rate',
type=float
)
parser.add_argument(
'--difficulty',
help='Task difficultys',
type=str
)
parser.add_argument(
'--model-filename',
dest='model_filename',
help='Model file',
default=None,
type=str
)
return parser.parse_args()
args = parse_args()
# connect neo
receiver = NeuracleDataClient(n_channel=len(config_info['channel_labels']),
samplerate=config_info['sample_rate'],
host=config_info['host'],
port=config_info['port'],
buffer_len=config_info['buffer_length'])
# connect to trigger box
trigger = TriggerNeuracle(port=args.trigger_port)
if args.hand_feedback:
# connect to mechanical hand
hand_device = FuboPneumaticFingerClient({'port': args.hand_port})
# build bci controller
if args.model_filename is not None:
model_path = os.path.join(settings.MODEL_PATH, args.subj, args.model_filename)
control_model = model_loader(model_path)
else:
control_model = None
controller = Controller(args.virtual_feedback_rate,
control_model,
reref_method=config_info['reref'])
# Run 'Before Experiment' code from decision
cnt_threshold_table = {
'easy': 3,
'mid': 4,
'hard': 5
}
cnt_threshold = cnt_threshold_table[args.difficulty]
correct_cnt = 0
# --- Setup global variables (available in all functions) ---
# Ensure that relative paths start from the same directory as this script
_thisDir = os.path.dirname(os.path.abspath(__file__))
# Store info about the experiment session
psychopyVersion = '2023.2.3'
expName = 'train' # from the Builder filename that created this script
expInfo = {
'participant': f"{randint(0, 999999):06.0f}",
'session': '001',
'date': data.getDateStr(), # add a simple timestamp
'expName': expName,
'psychopyVersion': psychopyVersion,
}
def showExpInfoDlg(expInfo):
"""
Show participant info dialog.
Parameters
==========
expInfo : dict
Information about this experiment, created by the `setupExpInfo` function.
Returns
==========
dict
Information about this experiment.
"""
# temporarily remove keys which the dialog doesn't need to show
poppedKeys = {
'date': expInfo.pop('date', data.getDateStr()),
'expName': expInfo.pop('expName', expName),
'psychopyVersion': expInfo.pop('psychopyVersion', psychopyVersion),
}
# show participant info dialog
dlg = gui.DlgFromDict(dictionary=expInfo, sortKeys=False, title=expName)
if dlg.OK == False:
core.quit() # user pressed cancel
# restore hidden keys
expInfo.update(poppedKeys)
# return expInfo
return expInfo
def setupData(expInfo, dataDir=None):
"""
Make an ExperimentHandler to handle trials and saving.
Parameters
==========
expInfo : dict
Information about this experiment, created by the `setupExpInfo` function.
dataDir : Path, str or None
Folder to save the data to, leave as None to create a folder in the current directory.
Returns
==========
psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
"""
# data file name stem = absolute path + name; later add .psyexp, .csv, .log, etc
if dataDir is None:
dataDir = _thisDir
filename = u'data/%s_%s_%s' % (expInfo['participant'], expName, expInfo['date'])
# make sure filename is relative to dataDir
if os.path.isabs(filename):
dataDir = os.path.commonprefix([dataDir, filename])
filename = os.path.relpath(filename, dataDir)
# an ExperimentHandler isn't essential but helps with data saving
thisExp = data.ExperimentHandler(
name=expName, version='',
extraInfo=expInfo, runtimeInfo=None,
originPath='/Users/dingkunliu/Projects/MI-BCI-Proj/kraken/backend/general_grasp_training.py',
savePickle=True, saveWideText=True,
dataFileName=dataDir + os.sep + filename, sortColumns='time'
)
thisExp.setPriority('thisRow.t', priority.CRITICAL)
thisExp.setPriority('expName', priority.LOW)
# return experiment handler
return thisExp
def setupLogging(filename):
"""
Setup a log file and tell it what level to log at.
Parameters
==========
filename : str or pathlib.Path
Filename to save log file and data files as, doesn't need an extension.
Returns
==========
psychopy.logging.LogFile
Text stream to receive inputs from the logging system.
"""
# this outputs to the screen, not a file
logging.console.setLevel(logging.DEBUG)
# save a log file for detail verbose info
logFile = logging.LogFile(filename+'.log', level=logging.DEBUG)
return logFile
def setupWindow(expInfo=None, win=None):
"""
Setup the Window
Parameters
==========
expInfo : dict
Information about this experiment, created by the `setupExpInfo` function.
win : psychopy.visual.Window
Window to setup - leave as None to create a new window.
Returns
==========
psychopy.visual.Window
Window in which to run this experiment.
"""
if win is None:
# if not given a window to setup, make one
win = visual.Window(
size=[1440, 900], fullscr=True, screen=0,
winType='pyglet', allowStencil=False,
monitor='testMonitor', color=[1,1,1], colorSpace='rgb',
backgroundImage='', backgroundFit='none',
blendMode='avg', useFBO=True,
units='height'
)
if expInfo is not None:
# store frame rate of monitor if we can measure it
expInfo['frameRate'] = win.getActualFrameRate()
else:
# if we have a window, just set the attributes which are safe to set
win.color = [1,1,1]
win.colorSpace = 'rgb'
win.backgroundImage = ''
win.backgroundFit = 'none'
win.units = 'height'
win.mouseVisible = False
win.hideMessage()
return win
def setupInputs(expInfo, thisExp, win):
"""
Setup whatever inputs are available (mouse, keyboard, eyetracker, etc.)
Parameters
==========
expInfo : dict
Information about this experiment, created by the `setupExpInfo` function.
thisExp : psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
win : psychopy.visual.Window
Window in which to run this experiment.
Returns
==========
dict
Dictionary of input devices by name.
"""
# --- Setup input devices ---
inputs = {}
ioConfig = {}
ioSession = ioServer = eyetracker = None
# create a default keyboard (e.g. to check for escape)
defaultKeyboard = keyboard.Keyboard(backend='ptb')
# return inputs dict
return {
'ioServer': ioServer,
'defaultKeyboard': defaultKeyboard,
'eyetracker': eyetracker,
}
def pauseExperiment(thisExp, inputs=None, win=None, timers=[], playbackComponents=[]):
"""
Pause this experiment, preventing the flow from advancing to the next routine until resumed.
Parameters
==========
thisExp : psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
inputs : dict
Dictionary of input devices by name.
win : psychopy.visual.Window
Window for this experiment.
timers : list, tuple
List of timers to reset once pausing is finished.
playbackComponents : list, tuple
List of any components with a `pause` method which need to be paused.
"""
# if we are not paused, do nothing
if thisExp.status != PAUSED:
return
# pause any playback components
for comp in playbackComponents:
comp.pause()
# prevent components from auto-drawing
win.stashAutoDraw()
# run a while loop while we wait to unpause
while thisExp.status == PAUSED:
# make sure we have a keyboard
if inputs is None:
inputs = {
'defaultKeyboard': keyboard.Keyboard(backend='PsychToolbox')
}
# check for quit (typically the Esc key)
if inputs['defaultKeyboard'].getKeys(keyList=['escape']):
endExperiment(thisExp, win=win, inputs=inputs)
# flip the screen
win.flip()
# if stop was requested while paused, quit
if thisExp.status == FINISHED:
endExperiment(thisExp, inputs=inputs, win=win)
# resume any playback components
for comp in playbackComponents:
comp.play()
# restore auto-drawn components
win.retrieveAutoDraw()
# reset any timers
for timer in timers:
timer.reset()
def run(expInfo, thisExp, win, inputs, globalClock=None, thisSession=None):
"""
Run the experiment flow.
Parameters
==========
expInfo : dict
Information about this experiment, created by the `setupExpInfo` function.
thisExp : psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
psychopy.visual.Window
Window in which to run this experiment.
inputs : dict
Dictionary of input devices by name.
globalClock : psychopy.core.clock.Clock or None
Clock to get global time from - supply None to make a new one.
thisSession : psychopy.session.Session or None
Handle of the Session object this experiment is being run from, if any.
"""
# mark experiment as started
thisExp.status = STARTED
# make sure variables created by exec are available globally
exec = environmenttools.setExecEnvironment(globals())
# get device handles from dict of input devices
ioServer = inputs['ioServer']
defaultKeyboard = inputs['defaultKeyboard']
eyetracker = inputs['eyetracker']
# make sure we're running in the directory for this experiment
os.chdir(_thisDir)
# get filename from ExperimentHandler for convenience
filename = thisExp.dataFileName
frameTolerance = 0.001 # how close to onset before 'same' frame
endExpNow = False # flag for 'escape' or other condition => quit the exp
# get frame duration from frame rate in expInfo
if 'frameRate' in expInfo and expInfo['frameRate'] is not None:
frameDur = 1.0 / round(expInfo['frameRate'])
else:
frameDur = 1.0 / 60.0 # could not measure, so guess
# Start Code - component code to be run after the window creation
# --- Initialize components for Routine "before_mi" ---
train_position = visual.TextStim(win=win, name='train_position',
text=None,
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='black', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=0.0);
instruction = visual.TextStim(win=win, name='instruction',
text='准备进行一般抓握训练，\n按空格键继续',
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='black', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=-1.0);
key_resp = keyboard.Keyboard()
# --- Initialize components for Routine "mi_prepare" ---
text = visual.TextStim(win=win, name='text',
text='请准备',
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='black', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=-1.0);
# --- Initialize components for Routine "mi_begin" ---
img_move = visual.ImageStim(
win=win,
name='img_move',
image=None, mask=None, anchor='center',
ori=0.0, pos=(0, 0), size=None,
color=[1,1,1], colorSpace='rgb', opacity=None,
flipHoriz=False, flipVert=False,
texRes=128.0, interpolate=True, depth=0.0)
# --- Initialize components for Routine "decision" ---
# --- Initialize components for Routine "mi_feedback" ---
feedback = visual.TextStim(win=win, name='feedback',
text=None,
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='black', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=0.0);
# --- Initialize components for Routine "mi_feedback_2" ---
feedback_2 = visual.TextStim(win=win, name='feedback_2',
text=None,
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='white', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=0.0);
# --- Initialize components for Routine "mi_rest" ---
img_rest = visual.ImageStim(
win=win,
name='img_rest',
image='static/images/rest.png', mask=None, anchor='center',
ori=0.0, pos=(0, 0), size=None,
color=[1,1,1], colorSpace='rgb', opacity=None,
flipHoriz=False, flipVert=False,
texRes=128.0, interpolate=True, depth=0.0)
# --- Initialize components for Routine "end" ---
mi_end = visual.TextStim(win=win, name='mi_end',
text='结束实验',
font='Open Sans',
pos=(0, 0), height=0.05, wrapWidth=None, ori=0.0,
color='black', colorSpace='rgb', opacity=None,
languageStyle='LTR',
depth=0.0);
# create some handy timers
if globalClock is None:
globalClock = core.Clock() # to track the time since experiment started
if ioServer is not None:
ioServer.syncClock(globalClock)
logging.setDefaultClock(globalClock)
routineTimer = core.Clock() # to track time remaining of each (possibly non-slip) routine
win.flip() # flip window to reset last flip timer
# store the exact time the global clock started
expInfo['expStart'] = data.getDateStr(format='%Y-%m-%d %Hh%M.%S.%f %z', fractionalSecondDigits=6)
# --- Prepare to start Routine "before_mi" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('before_mi.started', globalClock.getTime())
# Run 'Begin Routine' code from config
if args.side == "right":
train_position.text = '训练部位：右手'
else:
train_position.text = '训练部位：左手'
key_resp.keys = []
key_resp.rt = []
_key_resp_allKeys = []
# keep track of which components have finished
before_miComponents = [train_position, instruction, key_resp]
for thisComponent in before_miComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "before_mi" ---
routineForceEnded = not continueRoutine
while continueRoutine:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *train_position* updates
# if train_position is starting this frame...
if train_position.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
train_position.frameNStart = frameN # exact frame index
train_position.tStart = t # local t and not account for scr refresh
train_position.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(train_position, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'train_position.started')
# update status
train_position.status = STARTED
train_position.setAutoDraw(True)
# if train_position is active this frame...
if train_position.status == STARTED:
# update params
pass
# if train_position is stopping this frame...
if train_position.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > train_position.tStartRefresh + 2-frameTolerance:
# keep track of stop time/frame for later
train_position.tStop = t # not accounting for scr refresh
train_position.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'train_position.stopped')
# update status
train_position.status = FINISHED
train_position.setAutoDraw(False)
# *instruction* updates
# if instruction is starting this frame...
if instruction.status == NOT_STARTED and tThisFlip >= 2-frameTolerance:
# keep track of start time/frame for later
instruction.frameNStart = frameN # exact frame index
instruction.tStart = t # local t and not account for scr refresh
instruction.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(instruction, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'instruction.started')
# update status
instruction.status = STARTED
instruction.setAutoDraw(True)
# if instruction is active this frame...
if instruction.status == STARTED:
# update params
pass
# *key_resp* updates
waitOnFlip = False
# if key_resp is starting this frame...
if key_resp.status == NOT_STARTED and tThisFlip >= 2-frameTolerance:
# keep track of start time/frame for later
key_resp.frameNStart = frameN # exact frame index
key_resp.tStart = t # local t and not account for scr refresh
key_resp.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(key_resp, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'key_resp.started')
# update status
key_resp.status = STARTED
# keyboard checking is just starting
waitOnFlip = True
win.callOnFlip(key_resp.clock.reset) # t=0 on next screen flip
win.callOnFlip(key_resp.clearEvents, eventType='keyboard') # clear events on next screen flip
if key_resp.status == STARTED and not waitOnFlip:
theseKeys = key_resp.getKeys(keyList=['space'], ignoreKeys=["escape"], waitRelease=False)
_key_resp_allKeys.extend(theseKeys)
if len(_key_resp_allKeys):
key_resp.keys = _key_resp_allKeys[-1].name # just the last key pressed
key_resp.rt = _key_resp_allKeys[-1].rt
key_resp.duration = _key_resp_allKeys[-1].duration
# a response ends the routine
continueRoutine = False
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in before_miComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "before_mi" ---
for thisComponent in before_miComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('before_mi.stopped', globalClock.getTime())
# check responses
if key_resp.keys in ['', [], None]: # No response was made
key_resp.keys = None
thisExp.addData('key_resp.keys',key_resp.keys)
if key_resp.keys != None: # we had a response
thisExp.addData('key_resp.rt', key_resp.rt)
thisExp.addData('key_resp.duration', key_resp.duration)
thisExp.nextEntry()
# the Routine "before_mi" was not non-slip safe, so reset the non-slip timer
routineTimer.reset()
# set up handler to look after randomisation of conditions etc
trials = data.TrialHandler(nReps=args.n_trials, method='sequential',
extraInfo=expInfo, originPath=-1,
trialList=[None],
seed=None, name='trials')
thisExp.addLoop(trials) # add the loop to the experiment
thisTrial = trials.trialList[0] # so we can initialise stimuli with some values
# abbreviate parameter names if possible (e.g. rgb = thisTrial.rgb)
if thisTrial != None:
for paramName in thisTrial:
globals()[paramName] = thisTrial[paramName]
for thisTrial in trials:
currentLoop = trials
thisExp.timestampOnFlip(win, 'thisRow.t')
# pause experiment here if requested
if thisExp.status == PAUSED:
pauseExperiment(
thisExp=thisExp,
inputs=inputs,
win=win,
timers=[routineTimer],
playbackComponents=[]
)
# abbreviate parameter names if possible (e.g. rgb = thisTrial.rgb)
if thisTrial != None:
for paramName in thisTrial:
globals()[paramName] = thisTrial[paramName]
# --- Prepare to start Routine "mi_prepare" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('mi_prepare.started', globalClock.getTime())
# Run 'Begin Routine' code from initialize_buffer
decision_buffer = []
# keep track of which components have finished
mi_prepareComponents = [text]
for thisComponent in mi_prepareComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "mi_prepare" ---
routineForceEnded = not continueRoutine
while continueRoutine and routineTimer.getTime() < 1.5:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *text* updates
# if text is starting this frame...
if text.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
text.frameNStart = frameN # exact frame index
text.tStart = t # local t and not account for scr refresh
text.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(text, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'text.started')
# update status
text.status = STARTED
text.setAutoDraw(True)
# if text is active this frame...
if text.status == STARTED:
# update params
pass
# if text is stopping this frame...
if text.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > text.tStartRefresh + 1.5-frameTolerance:
# keep track of stop time/frame for later
text.tStop = t # not accounting for scr refresh
text.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'text.stopped')
# update status
text.status = FINISHED
text.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in mi_prepareComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "mi_prepare" ---
for thisComponent in mi_prepareComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('mi_prepare.stopped', globalClock.getTime())
# using non-slip timing so subtract the expected duration of this Routine (unless ended on request)
if routineForceEnded:
routineTimer.reset()
else:
routineTimer.addTime(-1.500000)
# set up handler to look after randomisation of conditions etc
classification = data.TrialHandler(nReps=5.0, method='sequential',
extraInfo=expInfo, originPath=-1,
trialList=[None],
seed=None, name='classification')
thisExp.addLoop(classification) # add the loop to the experiment
thisClassification = classification.trialList[0] # so we can initialise stimuli with some values
# abbreviate parameter names if possible (e.g. rgb = thisClassification.rgb)
if thisClassification != None:
for paramName in thisClassification:
globals()[paramName] = thisClassification[paramName]
for thisClassification in classification:
currentLoop = classification
thisExp.timestampOnFlip(win, 'thisRow.t')
# pause experiment here if requested
if thisExp.status == PAUSED:
pauseExperiment(
thisExp=thisExp,
inputs=inputs,
win=win,
timers=[routineTimer],
playbackComponents=[]
)
# abbreviate parameter names if possible (e.g. rgb = thisClassification.rgb)
if thisClassification != None:
for paramName in thisClassification:
globals()[paramName] = thisClassification[paramName]
# --- Prepare to start Routine "mi_begin" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('mi_begin.started', globalClock.getTime())
# Run 'Begin Routine' code from algo
if args.side == "right":
img_move.image = 'static/images/hand_move_right.png'
else:
img_move.image = 'static/images/hand_move_left.png'
# send trigger
current_true_label = settings.FINGERMODEL_IDS[args.finger_model]
win.callOnFlip(trigger.send_trigger, current_true_label)
# keep track of which components have finished
mi_beginComponents = [img_move]
for thisComponent in mi_beginComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "mi_begin" ---
routineForceEnded = not continueRoutine
while continueRoutine and routineTimer.getTime() < 1.0:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *img_move* updates
# if img_move is starting this frame...
if img_move.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
img_move.frameNStart = frameN # exact frame index
img_move.tStart = t # local t and not account for scr refresh
img_move.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(img_move, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'img_move.started')
# update status
img_move.status = STARTED
img_move.setAutoDraw(True)
# if img_move is active this frame...
if img_move.status == STARTED:
# update params
pass
# if img_move is stopping this frame...
if img_move.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > img_move.tStartRefresh + 1-frameTolerance:
# keep track of stop time/frame for later
img_move.tStop = t # not accounting for scr refresh
img_move.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'img_move.stopped')
# update status
img_move.status = FINISHED
img_move.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in mi_beginComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "mi_begin" ---
for thisComponent in mi_beginComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('mi_begin.stopped', globalClock.getTime())
# Run 'End Routine' code from algo
data_from_buffer = receiver.get_trial_data(clear=False)
decision = controller.step_decision(data_from_buffer, current_true_label)
decision_buffer.append(decision)
# write decision to data
logging.exp('decision: {}'.format(decision))
# using non-slip timing so subtract the expected duration of this Routine (unless ended on request)
if routineForceEnded:
routineTimer.reset()
else:
routineTimer.addTime(-1.000000)
# completed 5.0 repeats of 'classification'
# --- Prepare to start Routine "decision" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('decision.started', globalClock.getTime())
# Run 'Begin Routine' code from decision
cnt = 0
for d in decision_buffer:
if d == current_true_label:
cnt += 1
success = False
feedback_time = 2
global correct_cnt
if cnt >= cnt_threshold:
success = True
correct_cnt += 1 # count all correct trials
if args.hand_feedback:
feedback_time = 10
# keep track of which components have finished
decisionComponents = []
for thisComponent in decisionComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "decision" ---
routineForceEnded = not continueRoutine
while continueRoutine:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in decisionComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "decision" ---
for thisComponent in decisionComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('decision.stopped', globalClock.getTime())
# the Routine "decision" was not non-slip safe, so reset the non-slip timer
routineTimer.reset()
# --- Prepare to start Routine "mi_feedback" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('mi_feedback.started', globalClock.getTime())
# Run 'Begin Routine' code from code
if success:
feedback.text = '恭喜！'
if args.hand_feedback:
hand_device.start(args.finger_model)
else:
feedback.text = '继续努力！'
# keep track of which components have finished
mi_feedbackComponents = [feedback]
for thisComponent in mi_feedbackComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "mi_feedback" ---
routineForceEnded = not continueRoutine
while continueRoutine:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *feedback* updates
# if feedback is starting this frame...
if feedback.status == NOT_STARTED and tThisFlip >= 0-frameTolerance:
# keep track of start time/frame for later
feedback.frameNStart = frameN # exact frame index
feedback.tStart = t # local t and not account for scr refresh
feedback.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(feedback, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'feedback.started')
# update status
feedback.status = STARTED
feedback.setAutoDraw(True)
# if feedback is active this frame...
if feedback.status == STARTED:
# update params
pass
# if feedback is stopping this frame...
if feedback.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > feedback.tStartRefresh + feedback_time / 2-frameTolerance:
# keep track of stop time/frame for later
feedback.tStop = t # not accounting for scr refresh
feedback.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'feedback.stopped')
# update status
feedback.status = FINISHED
feedback.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in mi_feedbackComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "mi_feedback" ---
for thisComponent in mi_feedbackComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('mi_feedback.stopped', globalClock.getTime())
# the Routine "mi_feedback" was not non-slip safe, so reset the non-slip timer
routineTimer.reset()
# --- Prepare to start Routine "mi_feedback_2" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('mi_feedback_2.started', globalClock.getTime())
# Run 'Begin Routine' code from code_2
if success:
feedback_2.text = '恭喜！'
if args.hand_feedback:
hand_device.start('extend')
else:
feedback_2.text = '继续努力！'
# keep track of which components have finished
mi_feedback_2Components = [feedback_2]
for thisComponent in mi_feedback_2Components:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "mi_feedback_2" ---
routineForceEnded = not continueRoutine
while continueRoutine:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *feedback_2* updates
# if feedback_2 is starting this frame...
if feedback_2.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
feedback_2.frameNStart = frameN # exact frame index
feedback_2.tStart = t # local t and not account for scr refresh
feedback_2.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(feedback_2, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'feedback_2.started')
# update status
feedback_2.status = STARTED
feedback_2.setAutoDraw(True)
# if feedback_2 is active this frame...
if feedback_2.status == STARTED:
# update params
pass
# if feedback_2 is stopping this frame...
if feedback_2.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > feedback_2.tStartRefresh + feedback_time / 2-frameTolerance:
# keep track of stop time/frame for later
feedback_2.tStop = t # not accounting for scr refresh
feedback_2.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'feedback_2.stopped')
# update status
feedback_2.status = FINISHED
feedback_2.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in mi_feedback_2Components:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "mi_feedback_2" ---
for thisComponent in mi_feedback_2Components:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('mi_feedback_2.stopped', globalClock.getTime())
# the Routine "mi_feedback_2" was not non-slip safe, so reset the non-slip timer
routineTimer.reset()
# --- Prepare to start Routine "mi_rest" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('mi_rest.started', globalClock.getTime())
# Run 'Begin Routine' code from trigger_rest
# send trigger
win.callOnFlip(trigger.send_trigger, 0)
# keep track of which components have finished
mi_restComponents = [img_rest]
for thisComponent in mi_restComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "mi_rest" ---
routineForceEnded = not continueRoutine
while continueRoutine and routineTimer.getTime() < 5.0:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *img_rest* updates
# if img_rest is starting this frame...
if img_rest.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
img_rest.frameNStart = frameN # exact frame index
img_rest.tStart = t # local t and not account for scr refresh
img_rest.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(img_rest, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'img_rest.started')
# update status
img_rest.status = STARTED
img_rest.setAutoDraw(True)
# if img_rest is active this frame...
if img_rest.status == STARTED:
# update params
pass
# if img_rest is stopping this frame...
if img_rest.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > img_rest.tStartRefresh + 5-frameTolerance:
# keep track of stop time/frame for later
img_rest.tStop = t # not accounting for scr refresh
img_rest.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'img_rest.stopped')
# update status
img_rest.status = FINISHED
img_rest.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in mi_restComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "mi_rest" ---
for thisComponent in mi_restComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('mi_rest.stopped', globalClock.getTime())
# using non-slip timing so subtract the expected duration of this Routine (unless ended on request)
if routineForceEnded:
routineTimer.reset()
else:
routineTimer.addTime(-5.000000)
thisExp.nextEntry()
if thisSession is not None:
# if running in a Session with a Liaison client, send data up to now
thisSession.sendExperimentData()
# completed args.n_trials repeats of 'trials'
# --- Prepare to start Routine "end" ---
continueRoutine = True
# update component parameters for each repeat
thisExp.addData('end.started', globalClock.getTime())
# Run 'Begin Routine' code from score
mi_end.text = f"实验结束，\n得分：{int(correct_cnt / args.n_trials * 100)}"
# keep track of which components have finished
endComponents = [mi_end]
for thisComponent in endComponents:
thisComponent.tStart = None
thisComponent.tStop = None
thisComponent.tStartRefresh = None
thisComponent.tStopRefresh = None
if hasattr(thisComponent, 'status'):
thisComponent.status = NOT_STARTED
# reset timers
t = 0
_timeToFirstFrame = win.getFutureFlipTime(clock="now")
frameN = -1
# --- Run Routine "end" ---
routineForceEnded = not continueRoutine
while continueRoutine and routineTimer.getTime() < 5.0:
# get current time
t = routineTimer.getTime()
tThisFlip = win.getFutureFlipTime(clock=routineTimer)
tThisFlipGlobal = win.getFutureFlipTime(clock=None)
frameN = frameN + 1 # number of completed frames (so 0 is the first frame)
# update/draw components on each frame
# *mi_end* updates
# if mi_end is starting this frame...
if mi_end.status == NOT_STARTED and tThisFlip >= 0.0-frameTolerance:
# keep track of start time/frame for later
mi_end.frameNStart = frameN # exact frame index
mi_end.tStart = t # local t and not account for scr refresh
mi_end.tStartRefresh = tThisFlipGlobal # on global time
win.timeOnFlip(mi_end, 'tStartRefresh') # time at next scr refresh
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'mi_end.started')
# update status
mi_end.status = STARTED
mi_end.setAutoDraw(True)
# if mi_end is active this frame...
if mi_end.status == STARTED:
# update params
pass
# if mi_end is stopping this frame...
if mi_end.status == STARTED:
# is it time to stop? (based on global clock, using actual start)
if tThisFlipGlobal > mi_end.tStartRefresh + 5-frameTolerance:
# keep track of stop time/frame for later
mi_end.tStop = t # not accounting for scr refresh
mi_end.frameNStop = frameN # exact frame index
# add timestamp to datafile
thisExp.timestampOnFlip(win, 'mi_end.stopped')
# update status
mi_end.status = FINISHED
mi_end.setAutoDraw(False)
# check for quit (typically the Esc key)
if defaultKeyboard.getKeys(keyList=["escape"]):
thisExp.status = FINISHED
if thisExp.status == FINISHED or endExpNow:
endExperiment(thisExp, inputs=inputs, win=win)
return
# check if all components have finished
if not continueRoutine: # a component has requested a forced-end of Routine
routineForceEnded = True
break
continueRoutine = False # will revert to True if at least one component still running
for thisComponent in endComponents:
if hasattr(thisComponent, "status") and thisComponent.status != FINISHED:
continueRoutine = True
break # at least one component has not yet finished
# refresh the screen
if continueRoutine: # don't flip if this routine is over or we'll get a blank screen
win.flip()
# --- Ending Routine "end" ---
for thisComponent in endComponents:
if hasattr(thisComponent, "setAutoDraw"):
thisComponent.setAutoDraw(False)
thisExp.addData('end.stopped', globalClock.getTime())
# using non-slip timing so subtract the expected duration of this Routine (unless ended on request)
if routineForceEnded:
routineTimer.reset()
else:
routineTimer.addTime(-5.000000)
# Run 'End Experiment' code from config
receiver.close()
# mark experiment as finished
endExperiment(thisExp, win=win, inputs=inputs)
def saveData(thisExp):
"""
Save data from this experiment
Parameters
==========
thisExp : psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
"""
filename = thisExp.dataFileName
# these shouldn't be strictly necessary (should auto-save)
thisExp.saveAsWideText(filename + '.csv', delim='auto')
thisExp.saveAsPickle(filename)
def endExperiment(thisExp, inputs=None, win=None):
"""
End this experiment, performing final shut down operations.
This function does NOT close the window or end the Python process - use `quit` for this.
Parameters
==========
thisExp : psychopy.data.ExperimentHandler
Handler object for this experiment, contains the data to save and information about
where to save it to.
inputs : dict
Dictionary of input devices by name.
win : psychopy.visual.Window
Window for this experiment.
"""
if win is not None:
# remove autodraw from all current components
win.clearAutoDraw()
# Flip one final time so any remaining win.callOnFlip()
# and win.timeOnFlip() tasks get executed
win.flip()
# mark experiment handler as finished
thisExp.status = FINISHED
# shut down eyetracker, if there is one
if inputs is not None:
if 'eyetracker' in inputs and inputs['eyetracker'] is not None:
inputs['eyetracker'].setConnectionState(False)
logging.flush()
def quit(thisExp, win=None, inputs=None, thisSession=None):
"""
Fully quit, closing the window and ending the Python process.
Parameters
==========
win : psychopy.visual.Window
Window to close.
inputs : dict
Dictionary of input devices by name.
thisSession : psychopy.session.Session or None
Handle of the Session object this experiment is being run from, if any.
"""
thisExp.abort() # or data files will save again on exit
# make sure everything is closed down
if win is not None:
# Flip one final time so any remaining win.callOnFlip()
# and win.timeOnFlip() tasks get executed before quitting
win.flip()
win.close()
if inputs is not None:
if 'eyetracker' in inputs and inputs['eyetracker'] is not None:
inputs['eyetracker'].setConnectionState(False)
logging.flush()
if thisSession is not None:
thisSession.stop()
# terminate Python process
core.quit()
# if running this experiment as a script...
if __name__ == '__main__':
# call all functions in order
expInfo = showExpInfoDlg(expInfo=expInfo)
thisExp = setupData(expInfo=expInfo)
logFile = setupLogging(filename=thisExp.dataFileName)
win = setupWindow(expInfo=expInfo)
inputs = setupInputs(expInfo=expInfo, thisExp=thisExp, win=win)
run(
expInfo=expInfo,
thisExp=thisExp,
win=win,
inputs=inputs
)
saveData(thisExp=thisExp)
quit(thisExp=thisExp, win=win, inputs=inputs)