Merge pull request #2 from colinquirk/add-eeg-support

Add eeg support, mock tracker object

README.md

@@ -4,4 +4,8 @@ template/template.py -- This module provides a class that all of my experiments
 
 eyelinker/eyelinker.py -- This is a wrapper for pylink (from SR research) that makes it easy to control basic eyetracking experiments using eyelink trackers.
 
-eyelinker/PsychoPyCustomDisplay.py -- This is a module that connects psychopy and pylink so that eyelink can show graphics, play sounds, etc. If you use eyelinker, have it avaliable on the path and you will never need to use it directly.
+eyelinker/PsychoPyCustomDisplay.py -- This is a module that connects psychopy and pylink so that eyelink can show graphics, play sounds, etc. If you use eyelinker, have it avaliable on the path and you will never need to use it directly.
+
+pyplugger/pyplugger.py -- Various functions for controlling a PyCorder session from pyschopy
+
+pyplugger/inpout32.dll -- Required for working with parallel ports

eyelinker/eyelinker.py

@@ -9,8 +9,65 @@
 import psychopy.visual
 
 
-class EyeLinker:
-    def __init__(self, window, filename, eye):
+def try_connection(window):
+    print('Attempting to connect to eye tracker...')
+    try:
+        pl.EyeLink()
+        return True, None
+    except RuntimeError as e:
+        return False, e
+
+
+def display_not_connected_text(window):
+    warning_text = ('WARNING: Eyetracker not connected.\n\n'
+                    'Press "R" to retry connecting\n'
+                    'Press "Q" to quit\n'
+                    'Press "D" to continue in debug mode')
+
+    bg = psychopy.visual.Rect(window, units='norm', width=2, height=2, fillColor=(0.0, 0.0, 0.0))
+    text_stim = psychopy.visual.TextStim(window, warning_text, color=(1.0, 1.0, 1.0))
+
+    bg.draw()
+    text_stim.draw()
+
+    window.flip(clearBuffer=False)
+
+
+def get_connection_failure_response():
+    return psychopy.event.waitKeys(keyList=['r', 'q', 'd'])[0]
+
+
+# A factory function disguised as a class
+def EyeLinker(window, filename, eye):
+    connected, e = try_connection(window)
+
+    if connected:
+        return ConnectedEyeLinker(window, filename, eye)
+    else:
+        display_not_connected_text(window)
+
+    response = get_connection_failure_response()
+
+    while response == 'r':
+        connected, e = try_connection(window)
+        if connected:
+            window.flip()
+            return ConnectedEyeLinker(window, filename, eye)
+        else:
+            print('Could not connect to tracker. Select again.')
+            response = get_connection_failure_response()
+
+    if response == 'q':
+        window.flip()
+        raise e
+    elif response == 'd':
+        window.flip()
+        print('Continuing with mock eyetracking. Eyetracking data will not be saved!')
+        return MockEyeLinker(window, filename, eye)
+
+
+class ConnectedEyeLinker:
+    def __init__(self, window, filename, eye, text_color=None):
         if len(filename) > 12:
             raise ValueError(
                 'EDF filename must be at most 12 characters long including the extension.')
@@ -29,11 +86,15 @@ def __init__(self, window, filename, eye):
         self.resolution = tuple(window.size)
         self.tracker = pl.EyeLink()
         self.genv = PsychoPyCustomDisplay(self.window, self.tracker)
+        self.mock = False
 
-        if all(i >= 0.5 for i in self.window.color):
-            self.text_color = (-1, -1, -1)
+        if text_color is None:
+            if all(i >= 0.5 for i in self.window.color):
+                self.text_color = (-1, -1, -1)
+            else:
+                self.text_color = (1, 1, 1)
         else:
-            self.text_color = (1, 1, 1)
+            self.text_color = text_color
 
     def initialize_graphics(self):
         self.set_offline_mode()
@@ -205,10 +266,10 @@ def drift_correct(self, position=None, setup=1):
         except RuntimeError as e:
             print(e.message)
 
-    def record(self, trial_func):
+    def record(self, to_record_func):
         def wrapped_func():
             self.start_recording()
-            trial_func()
+            to_record_func()
             self.stop_recording()
         return wrapped_func
 
@@ -260,3 +321,43 @@ def send_status(self, status):
     def close_connection(self):
         self.tracker.close()
         pl.closeGraphics()
+
+
+# Creates a mock object to be used if tracker doesn't connect for debug purposes
+method_list = [fn_name for fn_name in dir(ConnectedEyeLinker)
+               if callable(getattr(ConnectedEyeLinker, fn_name)) and not fn_name.startswith("__")]
+
+
+def mock_func(*args, **kwargs):
+    pass
+
+
+class MockEyeLinker:
+    def __init__(self, window, filename, eye, text_color=None):
+        self.window = window
+        self.edf_filename = filename
+        self.edf_open = False
+        self.eye = eye
+        self.resolution = tuple(window.size)
+        self.tracker = None
+        self.genv = None
+        self.gaze_data = (None, None)
+        self.pupil_size = (None, None)
+        self.mock = True
+
+        if text_color is None:
+            if all(i >= 0.5 for i in self.window.color):
+                self.text_color = (-1, -1, -1)
+            else:
+                self.text_color = (1, 1, 1)
+        else:
+            self.text_color = text_color
+
+        for fn_name in method_list:
+            setattr(self, fn_name, mock_func)
+
+        # Decorator must return a function
+        def record(*args, **kwargs):
+            return mock_func
+
+        self.record = record

eyelinker/eyelinker_example.py

@@ -11,6 +11,12 @@
 monitor.setSizePix([1920, 1080])
 
 win = visual.Window([800, 600], units="pix", color=[0, 0, 0], monitor=monitor)
+
+text_stim = visual.TextStim(win, 'Beginning EyeLinker test...')
+text_stim.draw()
+win.flip()
+
+# Will attempt to default to MockEyeLinker if no tracker connected
 tracker = eyelinker.EyeLinker(win, 'test.edf', 'BOTH')
 
 # initialize
@@ -20,6 +26,7 @@
 tracker.send_calibration_settings()
 print('Initalization tests passed...')
 time.sleep(1)
+win.flip()
 
 # most basic functionality
 tracker.display_eyetracking_instructions()

pyplugger/pyplugger.py

@@ -0,0 +1,109 @@
+import socket
+import time
+
+import psychopy.event
+import psychopy.parallel
+import psychopy.visual
+
+
+class PyPlugger:
+    def __init__(self, window, config_file, tcp_ip="100.1.1.3",
+                 tcp_port=6700, parallel_port_address=53328, text_color=None):
+        self.window = window
+        self.config_file = config_file
+        self.tcp_ip = tcp_ip
+        self.tcp_port = tcp_port
+        self.current_mode = None
+        self.socket = None
+
+        psychopy.parallel.setPortAddress(parallel_port_address)
+        psychopy.parallel.setData(0)
+
+        if text_color is None:
+            if all(i >= 0.5 for i in self.window.color):
+                self.text_color = (-1, -1, -1)
+            else:
+                self.text_color = (1, 1, 1)
+        else:
+            self.text_color = text_color
+
+    def initialize_session(self, experiment_name, subject_number, timeout=5):
+        messages = ['1' + self.config_file,
+                    '2' + str(experiment_name),
+                    '3' + str(subject_number),
+                    '4']
+
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.settimeout(timeout)
+        self.socket.connect((self.tcp_ip, self.tcp_port))
+
+        for tcp_message in messages:
+            self.socket.send(tcp_message.encode())
+            time.sleep(1)
+
+    def switch_mode(self, mode, delay=5):
+        self.socket.send(mode.encode())
+        self.current_mode = mode
+        time.sleep(delay)
+
+    def start_recording(self, delay=5):
+        self.socket.send("S".encode())
+        time.sleep(delay)  # Ensure recording has started
+
+    def stop_recording(self, delay=5, exit_mode=False):
+        if exit_mode:
+            cmd = 'X'
+        else:
+            cmd = 'Q'
+
+        self.socket.send(cmd.encode())
+        time.sleep(delay)  # Ensure recording has ended
+
+    @staticmethod
+    def start_event(event):
+        psychopy.parallel.setData(event)
+
+    @staticmethod
+    def end_event():
+        """To be called some time after an event has been sent"""
+        psychopy.parallel.setData(0)
+
+    def display_eeg_instructions(self, eeg_instruction_text=None):
+        self.window.flip()
+
+        if eeg_instruction_text is None:
+            eeg_instruction_text = ('We will be recording EEG in this experiment. '
+                                    'In order to prevent artifacts due to muscle movements, please'
+                                    ' try to avoid moving your head and clenching your jaw.\n\n'
+                                    'Please avoid blinking while performing the task. '
+                                    'Try to blink only in between trials.')
+
+        psychopy.visual.TextStim(self.window, color=self.text_color, units='norm', pos=(0, 0.22),
+                                 height=0.05, text=eeg_instruction_text).draw()
+
+        psychopy.visual.TextStim(self.window, color=self.text_color, units='norm', pos=(0, -0.28),
+                                 height=0.05, text='Press any key to continue.').draw()
+
+        self.window.flip()
+        psychopy.event.waitKeys()
+        self.window.flip()
+
+    def display_interactive_switch_screen(self, require_monitoring=True):
+        switch_text = ('You may switch modes now.\n\n'
+                       'Press "M" for monitor mode.\n'
+                       'Press "I" for impedance mode.\n'
+                       'Press "Q" to continue with the experiment.')
+
+        psychopy.visual.TextStim(self.window, switch_text, color=self.text_color).draw()
+        self.window.flip()
+
+        response = None
+        while response != 'q':
+            response = psychopy.event.waitKeys(keyList=['m', 'i', 'q'])[0]
+            if response != 'q':
+                self.switch_mode(response.upper())
+
+        if self.current_mode != 'M':
+            self.switch_mode('M')
+
+        self.window.flip()

pyplugger/pyplugger_example.py

@@ -0,0 +1,44 @@
+import time
+
+from psychopy import monitors
+from psychopy import visual
+
+import pyplugger
+
+# Necessary psychopy setup
+monitor = monitors.Monitor('test_monitor', width=53, distance=70)
+monitor.setSizePix([1920, 1080])
+
+win = visual.Window([800, 600], units="pix", color=[0, 0, 0], monitor=monitor)
+
+visual.TextStim(win, 'Beginning PyPlugger test...').draw()
+win.flip()
+
+# Initialization
+eeg = pyplugger.PyPlugger(win, config_file='C:\\Users\\AwhVogel\\Desktop\\Colin\\default.xml')
+
+eeg.initialize_session('test_exp', 0)
+print('Initalization tests passed...')
+
+eeg.display_eeg_instructions()
+eeg.display_interactive_switch_screen()  # End in monitoring mode
+
+eeg.start_recording()
+for i in range(5):
+    eeg.start_event(i)
+    time.sleep(0.5)
+    eeg.end_event()
+    time.sleep(0.5)
+
+print('Basic functionality tests passed...')
+
+eeg.display_interactive_switch_screen()  # End in monitoring mode
+
+# Minimum delay testing
+for i in range(5):
+    eeg.start_event(i)
+    time.sleep(0.05)
+    eeg.end_event()
+    time.sleep(0.05)
+eeg.stop_recording(exit_mode=True)
+time.sleep(5)

template/template.py

@@ -381,3 +381,21 @@ def quit_experiment(self):
             self.experiment_window.close()
         print('The experiment has ended.')
         sys.exit(0)
+
+
+class EyeTrackingEEGExperiment(BaseExperiment):
+    def __init__(self, *args, tracker=None, eeg=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.tracker = tracker
+        self.eeg = eeg
+
+    def send_synced_event(self, code, keyword="SYNC", end_eeg_event=False):
+        if keyword is None:
+            message = str(code)
+        else:
+            message = keyword + ' ' + str(code)
+
+        self.eeg.start_event(code)
+        self.tracker.send_message(message)
+        if end_eeg_event:
+            self.eeg.end_eeg_event()