diff --git a/docs/CHANGELOG.rst b/docs/CHANGELOG.rst index f96c53dd..927f7a63 100644 --- a/docs/CHANGELOG.rst +++ b/docs/CHANGELOG.rst @@ -53,6 +53,9 @@ Runtime Changes: is generally a bad idea, and should be implemented on a case-by-case basis with care to properly end the trial and log any desired trial data before starting calibration. +* The :meth:`~klibs.KLEyeTracking.EyeTracker.get_event_queue` method for the + ``EyeTracker`` class now returns :obj:`~klibs.KLEyeTracking.EyeEvent` + objects instead of tracker-specific event types. API Changes: @@ -96,7 +99,12 @@ API Changes: used and complicated the API considerably. * Removed the unused ``unit`` parameter from the ``now()`` method in the :class:`~klibs.KLEyeTracking.EyeTracker` class. - +* Added a new API :mod:`klibs.KLEyeTracking.events`, defining a clean and + tracker-agnostic way of parsing and extracting information from eye events + from the :class:`~klibs.KLEyeTracking.EyeTracker` event queue. +* Added a new method :meth:`~klibs.KLEyeTracking.EyeTracker.get_sample` to + retrieve the most recent :obj:`~klibs.KLEyeTracking.GazeSample` event from + the tracker. Fixed Bugs: diff --git a/docs/source/api/KLEyeTracking/events.rst b/docs/source/api/KLEyeTracking/events.rst new file mode 100644 index 00000000..8b6154e9 --- /dev/null +++ b/docs/source/api/KLEyeTracking/events.rst @@ -0,0 +1,11 @@ +Eye Events +========== + +During eye tracking, a number of different eye events can occur. A participant +might make a saccade from one location to another, fixate a stimulus for a given +duration, or blink in a critical period. To facilitate working with these events +in your code, klibs converts different eye events into Pythonic objects that can +be easily processed and inspected. + +.. automodule:: klibs.KLEyeTracking.events + :members: diff --git a/docs/source/api/KLEyeTracking/eyelink.rst b/docs/source/api/KLEyeTracking/eyelink.rst index aa74ad46..2f217279 100755 --- a/docs/source/api/KLEyeTracking/eyelink.rst +++ b/docs/source/api/KLEyeTracking/eyelink.rst @@ -1,6 +1,21 @@ -KLEyeLink -========= +EyeLink +======= -.. automodule:: klibs.KLEyeTracking.KLEyeLink - :undoc-members: - :members: +The EyeLink module allows for the use of SR Research EyeLink eye trackers to +be used with the klibs runtime. + +If pylink (``sr-research-pylink``) is installed in your project's Python +environment and the ``eye_tracker_availiable`` parameter is set to True, an +instance of this class will be provided as the ``self.el`` attribute of the +Experiment object. + +EyeLink trackers record eye movement and gaze data for each session in EDF data +files. The EDF recording for each session is automatically transferred to the +project's ``ExpAssets/EDF`` subfolder when the task exits (unless the task +crashes with an exception). + +To avoid duplication with the :class:`~klibs.KLEyeTracking.EyeTracker` docs, +only methods with notable EyeLink-specific details are documented below: + +.. autoclass:: klibs.KLEyeTracking.KLEyeLink.EyeLink + :members: calibrate, write diff --git a/docs/source/api/KLEyeTracking/trylink.rst b/docs/source/api/KLEyeTracking/trylink.rst index c418158c..67a61c2e 100755 --- a/docs/source/api/KLEyeTracking/trylink.rst +++ b/docs/source/api/KLEyeTracking/trylink.rst @@ -1,6 +1,23 @@ -KLTryLink -========= +TryLink +======= -.. automodule:: klibs.KLEyeTracking.KLTryLink - :undoc-members: - :members: +The TryLink module provides an :class:`~klibs.KLEyeTracking.EyeTracker` +implementation using mouse movements to simulate eye movements. For testing and +developing eye tracking experiments without access to real hardware. + +In addition to simulating gaze position using the cursor position, TryLink also +generates simulated fixation and saccade events. Blink events are also simulated +when the mouse button is held down if ``dm_click_to_blink`` is enabled in the +project's params file. + +TryLink simulation is used by default unless the ``sr-research-pylink`` package +is installed, or if ``eye_tracking`` is True but ``eye_tracker_availiable`` is +False in the project's params file. In addition, TryLink simulation can also be +forced by adding the flag ``-ELx`` to the ``klibs run`` command when launching +an experiment. + +To avoid duplication with the :class:`~klibs.KLEyeTracking.EyeTracker` docs, +only methods with notable TryLink-specific details are documented below: + +.. autoclass:: klibs.KLEyeTracking.TryLink + :members: calibrate, drift_correct, gaze, write diff --git a/docs/source/api/eyetracking.rst b/docs/source/api/eyetracking.rst index ff307dab..3fe0a894 100755 --- a/docs/source/api/eyetracking.rst +++ b/docs/source/api/eyetracking.rst @@ -1,5 +1,5 @@ `klibs.KLEyeTracking` - Working with eye trackers -************************************************** +************************************************* The KLEyeTracking module provides a simple unified interface for working with different kinds of eye trackers in your experiments. All supported eye trackers make use of the same common interface @@ -9,7 +9,7 @@ with the interface should work on any of the different types of eye trackers tha Currently, KLibs has built-in support for two kinds of eye tracker: SR Research EyeLink trackers, and a simulated eye tracker that uses mouse movements as a stand-in for gaze and gaze events, allowing you to write and run eye-tracking experiments without a physical eye tracker -present. You can view more detailed documentation for each of these tracker types here: +present. Relevant details about each tracker type are provided here: .. toctree:: :maxdepth: 1 @@ -50,10 +50,79 @@ Additionally, if you want to disable automatic starting and stopping of the eye recording manually, you can set the parameter ``manual_eyelink_recording`` to True in your project's ``params.py`` file. +Understanding The Event Queue +----------------------------- + +During eye tracking, a number of different eye events can occur. A participant +might make a saccade from one location to another, fixate a stimulus for a given +duration, or blink in a critical period. When these events occur, they are +logged and placed in a queue so that they can be retrieved and inspected for +events of interest (e.g. a saccade away from fixation). + +The eye event queue is retrieved using the EyeTracker's +:meth:`~klibs.KLEyeTracking.EyeTracker.get_event_queue` method, which returns a +list of :obj:`~klibs.KLEyeTracking.events.EyeEvent` objects. A table of the +different event types and their corresponding type constants is provided below: + ++---------------------+---------------------------------------------------+ +| Event Type | Event Class | ++=====================+===================================================+ +| ``GAZE_SAMPLE`` | :obj:`~klibs.KLEyeTracking.events.GazeSample` | ++---------------------+---------------------------------------------------+ +| ``FIXATION_START`` | :obj:`~klibs.KLEyeTracking.events.FixationStart` | ++---------------------+---------------------------------------------------+ +| ``FIXATION_END`` | :obj:`~klibs.KLEyeTracking.events.FixationEnd` | ++---------------------+---------------------------------------------------+ +| ``FIXATION_UPDATE`` | :obj:`~klibs.KLEyeTracking.events.FixationUpdate` | ++---------------------+---------------------------------------------------+ +| ``SACCADE_START`` | :obj:`~klibs.KLEyeTracking.events.SaccadeStart` | ++---------------------+---------------------------------------------------+ +| ``SACCADE_END`` | :obj:`~klibs.KLEyeTracking.events.SaccadeEnd` | ++---------------------+---------------------------------------------------+ +| ``BLINK_START`` | :obj:`~klibs.KLEyeTracking.events.BlinkStart` | ++---------------------+---------------------------------------------------+ +| ``BLINK_END`` | :obj:`~klibs.KLEyeTracking.events.BlinkEnd` | ++---------------------+---------------------------------------------------+ + +To avoid problems with event processing when performing multiple tests in a +short interval (e.g. checking for fixations in multiple boundaries every +iteration of a loop), you should use this method to retrieve the eye tracker +event queue once at the start and then pass it to every function that inspects +eye events. + +For example, the following code will not reliably detect saccades to the second +boundary: + +.. code-block:: python + + while self.evm.before('cue_on'): + draw_stimuli() + if self.el.saccade_to_boundary('boundary1'): + print('saccade to boundary 1!') + if self.el.saccade_to_boundary('boundary2'): + print('saccade to boundary 2!') + flip() # refresh screen + +This is because the first call to :meth:`saccade_to_boundary` fetches and clears +the eye event queue, meaning that the second call is only processing the +saccades that have happened in the tiny interval since the first call. To fix +this, you should fetch the queue once per loop: + +.. code-block:: python + + while self.evm.before('cue_on'): + draw_stimuli() + el_q = self.el.get_event_queue() # fetch queue once per loop + if self.el.saccade_to_boundary('boundary1', event_queue=el_q): + print('saccade to boundary 1!') + if self.el.saccade_to_boundary('boundary2', event_queue=el_q): + print('saccade to boundary 2!') + flip() # refresh screen + + The EyeTracker API =================== -.. automodule:: klibs.KLEyeTracking.KLEyeTracker - :undoc-members: +.. autoclass:: klibs.KLEyeTracking.EyeTracker :members: \ No newline at end of file diff --git a/klibs/KLEnvironment.py b/klibs/KLEnvironment.py index 8197f9f1..0964c546 100755 --- a/klibs/KLEnvironment.py +++ b/klibs/KLEnvironment.py @@ -56,7 +56,7 @@ def db(self): @property def el(self): - """:obj:`~klibs.KLEyeTracking.KLEyeTracker`: If 'P.eye_tracking' is True, this is the + """:obj:`~klibs.KLEyeTracking.EyeTracker`: If 'P.eye_tracking' is True, this is the EyeTracker instance for the KLibs runtime environment. Otherwise, None. """ diff --git a/klibs/KLExperiment.py b/klibs/KLExperiment.py index 9f91920e..c0f29b10 100755 --- a/klibs/KLExperiment.py +++ b/klibs/KLExperiment.py @@ -22,6 +22,8 @@ def __init__(self): from klibs.KLEventInterface import EventManager from klibs.KLResponseCollectors import ResponseCollector from klibs.KLTrialFactory import TrialFactory + from klibs.KLEyeTracking import EyeTracker + from klibs.KLEnvironment import el super(Experiment, self).__init__() @@ -33,6 +35,7 @@ def __init__(self): self.rc = ResponseCollector() # add default response collector self.database = self.db # use database from env self._evm = EventManager() + self._el = el if el else EyeTracker() self._exp_factors, self._exp_structure = self._get_exp_structure() self.trial_factory = TrialFactory(self._exp_factors) @@ -505,6 +508,20 @@ def participant_info(self): return out + @property + def el(self): + """:obj:`~klibs.KLEyeTracking.EyeTracker`: The eye tracker instance for the task. + + Raises: + RuntimeError: If ``eye_tracking`` is False in the project's params.py file. + + """ + if not P.eye_tracking: + e = "Eye tracking must be enabled in params.py before accessing self.el." + raise RuntimeError(e) + return self._el + + @property def evm(self): """:obj:`~klibs.KLEventInterface.EventManager`: The trial event sequencer for diff --git a/klibs/KLEyeTracking/KLEyeLink.py b/klibs/KLEyeTracking/KLEyeLink.py index 4ac2c64a..8c0c9d73 100755 --- a/klibs/KLEyeTracking/KLEyeLink.py +++ b/klibs/KLEyeTracking/KLEyeLink.py @@ -8,14 +8,12 @@ from klibs.KLExceptions import EyeTrackerError from klibs.KLConstants import (LEFT_EYE, RIGHT_EYE, BINOCULAR, EL_NO_EYES, FIXATION_START, FIXATION_UPDATE, FIXATION_END, GAZE_SAMPLE, - SACCADE_START, SACCADE_END, BLINK_START, BLINK_END, - EL_GAZE_START, EL_GAZE_END, EL_GAZE_AVG, EL_TIME_START, EL_TIME_END, - EL_ALL_EVENTS, EL_TRUE, EL_FALSE, - TK_S, TK_MS) + SACCADE_START, SACCADE_END, BLINK_START, BLINK_END, + EL_ALL_EVENTS, EL_TRUE, EL_FALSE) from klibs import P from klibs.KLInternal import now, hide_stderr from klibs.KLUserInterface import show_cursor, hide_cursor -from klibs.KLEyeTracking.KLEyeTracker import EyeTracker +from klibs.KLEyeTracking.KLEyeTracker import EyeTracker, EventParser if PYLINK_AVAILABLE: with hide_stderr(macos_only=True): @@ -31,6 +29,8 @@ # easy way to do that in pure Python (and ping requires admin sometimes?). EyeLink # communication itself is via UDP. +NO_DATA = 0 +MISSING_DATA = -32768 EYE_NAME_MAP = { RIGHT_EYE: 'right', @@ -70,23 +70,101 @@ def _get_eyelink_version(s): return s.replace("EYELINK", "EyeLink").strip() -class EyeLink(EyeTracker): - """A connection to an SR Research EyeLink eye tracker, providing a friendly interface to the - pylink API along with a pretty setup/calibration display. - - If pylink is installed in your project's Python environment and ``P.eye_tracker_availiable`` - is set to True in a project that makes use of eye tracking, an instance of this class will be - provided as the attribute ``self.el`` of the Experiment object. +class EyeLinkEventParser(EventParser): + """A parser defining how to extract different data from EyeLink events. + + """ + def get_type(self, e): + """Retrieves the type of a given event (e.g. ``BLINK_START``). + """ + return e.getType() + + def eye(self, e): + """Retrieves the eye for a given event. + """ + return e.getEye() + + def timestamp(self, e): + """Gets the timestamp (in msec) for a gaze sample. + """ + return e.getTime() + + def left_gaze(self, e): + """Gets the left eye (x, y) gaze coordinates for a gaze sample. + """ + eye = e.getLeftEye() + if eye: + x, y = eye.getGaze() + if int(x) != MISSING_DATA: + return (x, y) + return None + + def right_gaze(self, e): + """Gets the right eye (x, y) gaze coordinates for a gaze sample. + """ + eye = e.getRightEye() + if eye: + x, y = eye.getGaze() + if int(x) != MISSING_DATA: + return (x, y) + return None + + def left_pupil_size(self, e): + """Gets the left eye pupil size (arbitrary units) for a gaze sample. + """ + eye = e.getLeftEye() + if eye: + ps = eye.getPupilSize() + if int(ps) != MISSING_DATA: + return ps + return None + + def right_pupil_size(self, e): + """Gets the right eye pupil size (arbitrary units) for a gaze sample. + """ + eye = e.getRightEye() + if eye: + ps = eye.getPupilSize() + if int(ps) != MISSING_DATA: + return ps + return None + + def start_time(self, e): + """Gets the timestamp (in msec) for the start of the event. + """ + return e.getStartTime() + + def end_time(self, e): + """Gets the timestamp (in msec) for the end of the event. + """ + return e.getEndTime() - For more general and comprehensive documentation, see the :class:`~.EyeTracker` class. + def start_gaze(self, e): + """Gets the (x, y) gaze coordinates at the start of the event. + """ + return e.getStartGaze() + + def end_gaze(self, e): + """Gets the (x, y) gaze coordinates at the end of the event. + """ + return e.getEndGaze() - Attributes: - version (str): The model name and software version of the eye tracker. - initialized (bool): A flag indicating whether :meth:`setup` has been run successfully. - local_start_time (float): The time at which the tracker last started recording, according - to the local computer's clock. - tracker_start_time (float): The time at which the tracker last started recording, according - to the eye tracker's internal clock. + def avg_gaze(self, e): + """Gets the average (x, y) gaze coordinates for the event. + """ + return e.getAverageGaze() + + def duration(self, e): + """Gets the total duration (in msec) for the event. + """ + # Pylink docs say to add 4 seconds for real duration, but mentions + # EyeLink II when it says this. Is this stil correct for newer trackers? + return (e.getEndTime() - e.getStartTime()) + 4 + + + +class EyeLink(EyeTracker): + """A connection to an SR Research EyeLink eye tracker. """ def __init__(self): @@ -96,6 +174,7 @@ def __init__(self): self._recording = False self._unresolved_exceptions = 0 self._quitting = False + self._parser = EyeLinkEventParser() self.version = None self.initialized = False @@ -130,19 +209,25 @@ def _check_connection(self): raise RuntimeError(e) - def setup(self): - """Initalizes the EyeLink for the first time and enters setup/calibration mode. - - Called automatically after demographics collection during the KLibs runtime unless - ``P.manual_eyelink_setup`` is True, in which case it must be called manually - before the eye tracker is first used in the experiment. - - """ - EyeTracker.setup(self) # so it shows up in the docs - - def calibrate(self): - """Enters the calibration and setup mode for the EyeLink. + """Enters calibration and setup mode for the EyeLink. + + For EyeLink trackers, this will open a setup screen where you can perform camera + setup, calibrate the tracker, and validate an existing calibration: + + +--------------+-------------------------------+ + | Key | Function | + +==============+===============================+ + | Enter/Return | Toggles camera setup display | + +--------------+-------------------------------+ + | C | Starts calibration | + +--------------+-------------------------------+ + | V | Starts calibration validation | + +--------------+-------------------------------+ + | O | Gaze at fixation onset | + +--------------+-------------------------------+ + | Esc | Exits tracker setup | + +--------------+-------------------------------+ """ self._el.doTrackerSetup() @@ -216,58 +301,64 @@ def shut_down(self, incomplete=False): return self._el.close() - def get_event_queue(self, include=[], exclude=[]): - """Fetches and returns the EyeLink's event queue, emptying it in the process. + def _get_tracker_events(self, events, samples=False): + """Internal method for retrieving eye events from the tracker. Args: - include (:obj:`List`, optional): A list specifying the types of eye events to fetch - from the event queue. Includes all eye event types by default, unless they are - explicitly excluded. - exclude (:obj:`List`, optional): A list specifying the types of eye events to exclude - from the returned queue. Defaults to an empty list (i.e. no events excluded.) + events (list): A list of the desired event types to retrieve. + samples (bool): Whether gaze samples should be returned in addition to other + event types. Returns: - A :obj:`List` of eye events. + list: A list of tracker-specific events. """ - - if len(include): - valid_events = set(include) - elif len(exclude): - valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]).difference(exclude) - else: - valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]) - - samples = int(GAZE_SAMPLE in valid_events) - events = int(len(valid_events.intersection(EL_ALL_EVENTS)) > 0) - + # NOTE: Currently input, button, and message events are simply discarded queue = [] - if self._el.getDataCount(samples, events) != 0: # i.e. if data available - last_sample = None + get_samples = int(samples) + get_events = int(len(events.intersection(EL_ALL_EVENTS)) > 0) + + # If any data waiting, process it all + any_samples = self._el.getDataCount(1, 0) + if self._el.getDataCount(get_samples, get_events) != 0: + last_event = None while True: - d_type = self._el.getNextData() - if d_type == 0: + e_type = self._el.getNextData() + if e_type == NO_DATA: break - elif d_type not in valid_events: + elif e_type not in valid_events: continue - data = self._el.getFloatData() - # once the same sample has been sent twice, gtfo - if data == last_sample: + event = self._el.getFloatData() + # If same event sent twice, stop immediately + if event == last_event: break - else: - last_sample = data + last_sample = data queue.append(data) - if samples == True and len(queue) == 0: # if no samples from getNextData, fetch latest - newest_sample = self._el.getNewestSample() - queue = [newest_sample] if newest_sample != None else [] + # If no gaze samples from getNextData, append latest sample + if samples and not any_samples: + s = self._el.getNewestSample() + if s != None: + queue.append(s) return queue + def _get_latest_sample(self): + """Internal method for retrieving the most recent gaze sample from the tracker. + + Returns: + A tracker-specific gaze sample event. + + """ + s = self._el.getNewestSample() + return s if s != None else None + + def clear_queue(self): - """Clears the event queue for the eye tracker. Any uninspected eye events will be - discarded. + """Clears the event queue for the eye tracker. + + Any uninspected eye events will be discarded. """ self._el.resetData() @@ -347,9 +438,9 @@ def gaze(self, return_integers=True, binocular_mode=RIGHT_EYE): rx, ry = sample.getRightEye().getGaze() lx, ly = sample.getLeftEye().getGaze() # if either eye is missing, use good eye instead of averaging - if int(lx) == -32768: + if int(lx) == MISSING_DATA: gaze_pos = (rx, ry) - elif int(rx) == -32768: + elif int(rx) == MISSING_DATA: gaze_pos = (lx, ly) else: gaze_pos = ( (rx+lx)/2, (ry+ly)/2 ) @@ -373,157 +464,31 @@ def now(self): def write(self, message): - """Writes a message to the EyeLink EDF file. Unicode characters are supported. - - Args: - message (str): The message to write to the eye tracker's data file. - - """ - self._el.sendMessage(message) - - - def get_event_type(self, e): - """Fetches the numeric code corresponding to an eye event's type (e.g end saccade, start - fixation). + """Writes a message to the EyeLink EDF file. - Args: - e: An eye event from the tracker. + Any messages set to the EyeLink will be included as message events in the + recorded data, which can be used as markup to identify the onsets of important + trial events (e.g. 'target_on') in the data for later analysis. - Returns: - The numeric type of the event (e.g. ``EL_END_SACCADE``, ``EL_START_FIXATION``). - - """ - return e.getType() - + It is recommended to include the current block and trial number in markup + messages to make it easy to connect eye tracking data back to the trial data:: - def get_event_gaze(self, event, inspect): - """Retrieves the gaze coordinates from an eye event or gaze sample. - - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to return the gaze - coordinates from. - inspect: A flag indicating the type of gaze to retrieve: the gaze at the start of - the event (``EL_GAZE_START``), the gaze at the end of the event (``EL_GAZE_END``), - or the event's average gaze (``EL_GAZE_AVG``). Note that not all eye events have - all three gaze attributes. Has no effect on gaze samples. + self.el.write("cue_on b{0} t{1}".format(P.block_number, P.trial_number)) - Returns: - The (x, y) gaze coordinates for the event. - - Raises: - EyeTrackerError: If asked to inspect the end gaze or average gaze for an eye event that - lacks that attribute (e.g. ``EL_GAZE_AVG`` for a ``SACCADE_END`` event). - - """ - if event.getType() == GAZE_SAMPLE: - if event.isLeftSample(): - coords = event.getLeftEye().getGaze() - elif event.isRightSample(): - coords = event.getRightEye().getGaze() - elif event.isBinocular(): - coords = event.getRightEye().getGaze() - else: - return None - else: - try: - if inspect == EL_GAZE_START: - coords = event.getStartGaze() - elif inspect == EL_GAZE_END: - coords = event.getEndGaze() - elif inspect == EL_GAZE_AVG: - coords = event.getAverageGaze() - except AttributeError: - typename = self.get_event_name(event.type) - err = "Cannot inspect {0} for {1} events." - raise EyeTrackerError(err.format(inspect, typename)) - - return coords - - - def get_event_timestamp(self, event, report): - """Gets the timestamp from an eye event or gaze sample. - - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to return the timestamp - from. - report: A flag indicating whether to report the start time (``EL_TIME_START``) or - end time (``EL_TIME_END``) of the eye event. Has no effect on gaze samples, - which do not have separate start/end times. + Unicode characters are supported. - Returns: - The timestamp for the start or end of the event. - - Raises: - EyeTrackerError: If asked to report the end timestamp for an eye event that only has - a start timestamp (e.g. ``FIXATION_START``). - - """ - if event.getType() == GAZE_SAMPLE: - return event.getTime() - else: - try: - return event.getStartTime() if report == EL_TIME_START else event.getEndTime() - except AttributeError: - typename = self.get_event_name(event.type) - err = "Cannot report {0} for {1} events." - raise EyeTrackerError(err.format(report, typename)) - - - def get_event_info(self, event): - """Returns all available info about an eye event or gaze sample in the form of a - :obj:`dict`. - - Usage:: - - q = self.el.get_event_queue() # fetch all unprocessed eye events - for event in q: - info = self.el.get_event_info(event) - if info['type'] == SACCADE_END: - end_gaze = info[EL_GAZE_END] - end_time = info[EL_TIME_END] - - See the table in :meth:`within_boundary` for a list of the attributes available for each - eye event type. - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to collect the attributes of. - - Returns: - A :obj:`dict` containing the available attributes of the event, such as start/end time, - start/end/average gaze, and type name. + message (str): The message to write to the eye tracker's data file. """ - info = {} - info['type'] = event.getType() - if info['type'] == GAZE_SAMPLE: - gaze = self.get_event_gaze(event, EL_GAZE_START) - time = self.get_event_timestamp(event, EL_TIME_START) - info[EL_TIME_START] = time - info[EL_TIME_END] = time - info[EL_GAZE_START] = gaze - info[EL_GAZE_END] = gaze - - elif info['type'] in [BLINK_START, BLINK_END]: - info[EL_TIME_START] = event.getStartTime() - if info['type'] == BLINK_END: - info[EL_TIME_END] = event.getEndTime() - - else: - info[EL_TIME_START] = event.getStartTime() - info[EL_GAZE_START] = event.getStartGaze() - if info['type'] in [SACCADE_END, FIXATION_END, FIXATION_UPDATE]: - info[EL_TIME_END] = event.getEndTime() - info[EL_GAZE_END] = event.getEndGaze() - if info['type'] != SACCADE_END: - info[EL_GAZE_AVG] = event.getAverageGaze() - - return info + self._el.sendMessage(message) @property def eye(self): - """str or None: The eye(s) currently being recorded by the tracker. Can be 'left', 'right', - 'both', or None (if no eye is currently available). + """str or None: The eye(s) currently being recorded by the tracker. + + Can be 'left', 'right', 'both', or None (if no eye is currently available). """ return EYE_NAME_MAP[self._el.eyeAvailable()] diff --git a/klibs/KLEyeTracking/KLEyeTracker.py b/klibs/KLEyeTracking/KLEyeTracker.py index 6128e20c..552b4c0b 100644 --- a/klibs/KLEyeTracking/KLEyeTracker.py +++ b/klibs/KLEyeTracking/KLEyeTracker.py @@ -4,37 +4,167 @@ from klibs.KLExceptions import EyeTrackerError from klibs.KLConstants import (LEFT_EYE, RIGHT_EYE, BINOCULAR, FIXATION_START, FIXATION_UPDATE, FIXATION_END, GAZE_SAMPLE, - SACCADE_START, SACCADE_END, BLINK_START, BLINK_END, + SACCADE_START, SACCADE_END, BLINK_START, BLINK_END, EL_GAZE_START, EL_GAZE_END, EL_GAZE_AVG, EL_TIME_START, EL_TIME_END, - EL_ALL_EVENTS, EL_TRUE, EL_FALSE, - TK_S, TK_MS) + EL_ALL_EVENTS) from klibs import P from klibs.KLUtilities import iterable, pretty_list from klibs.KLBoundary import BoundarySet from klibs.KLGraphics import blit, fill, flip from klibs.KLGraphics.KLDraw import drift_correct_target +from klibs.KLEyeTracking.events import ( + GazeSample, FixationStart, FixationEnd, FixationUpdate, + SaccadeStart, SaccadeEnd, BlinkStart, BlinkEnd +) + # TODO: Add way of specifying calibration type (e.g. 'HV9') in params +class EventParser(): + """A base class defining how to parse eye events from a given backend. + + This class serves as an adapter between the hardware-specific events generated + by a given eye tracker and the platform-agnostic eye events generated by klibs. + + """ + def get_type(self, e): + """Retrieves the type of a given event (e.g. ``BLINK_START``). + + Returns: + int: The type code for the event. + + """ + raise NotImplementedError() + + def eye(self, e): + """Retrieves the eye for a given event. + + Returns: + int: ``LEFT_EYE``, ``RIGHT_EYE``, or ``BINOCULAR``. + + """ + raise NotImplementedError() + + def timestamp(self, e): + """Gets the timestamp for a gaze sample. + + Returns: + float: The timestamp (in msec) for the gaze sample. + + """ + raise NotImplementedError() + + def left_gaze(self, e): + """Gets the left eye (x, y) gaze coordinates for a gaze sample. + + Returns: + tuple or None: The (x, y) gaze coodinates (in pixels) of the left eye for + the current sample (or None if the left eye is missing or not tracked). + + """ + raise NotImplementedError() + + def right_gaze(self, e): + """Gets the right eye (x, y) gaze coordinates for a gaze sample. + + Returns: + tuple or None: The (x, y) gaze coodinates (in pixels) of the right eye for + the current sample (or None if the right eye is missing or not tracked). + + """ + raise NotImplementedError() + + def left_pupil_size(self, e): + """Gets the left eye pupil size for a gaze sample. + + Pupil size units should be considered arbitrary and tracker-specific. + + Returns: + float or None: The pupil size (in arbitrary units) of the left eye for + the current sample (or None if the left eye is missing or not tracked). + + """ + raise NotImplementedError() + + def right_pupil_size(self, e): + """Gets the right eye pupil size for a gaze sample. + + Pupil size units should be considered arbitrary and tracker-specific. + + Returns: + float or None: The pupil size (in arbitrary units) of the right eye for + the current sample (or None if the right eye is missing or not tracked). + + """ + raise NotImplementedError() + + def start_time(self, e): + """Gets the timestamp (in msec) for the start of the event. + + Returns: + float: The timestamp (in msec) of the event onset. + + """ + raise NotImplementedError() + + def end_time(self, e): + """Gets the timestamp (in msec) for the end of the event. + + Returns: + float: The timestamp (in msec) of the end of the event. + + """ + raise NotImplementedError() + + def start_gaze(self, e): + """Gets the (x, y) gaze coordinates at the start of the event. + + Returns: + tuple: The (x, y) gaze coordinates (in pixels) at the start of the event. + + """ + raise NotImplementedError() + + def end_gaze(self, e): + """Gets the (x, y) gaze coordinates at the end of the event. + + Returns: + tuple: The (x, y) gaze coordinates (in pixels) at the end of the event. + + """ + raise NotImplementedError() + + def avg_gaze(self, e): + """Gets the average (x, y) gaze coordinates for the event. + + Returns: + tuple: The (x, y) gaze coordinates (in pixels) of the average gaze during + the event. + + """ + raise NotImplementedError() + + def duration(self, e): + """Gets the total duration (in msec) for the event. + + Returns: + float: The total duration (in msec) of the event. + + """ + raise NotImplementedError() + + class EyeTracker(): - """A base eye tracker class, laying out the essential attributes and methods required for an - eye tracker to be used with KLibs. - - Primarily used for defining a common and unified API for both SR Research EyeLink eye trackers - (via pylink) and eye tracker emulation via the mouse cursor, but hopefully flexible enough to - allow for supporting additional eye tracker brands/models in the future. + """A platform-independent interface for working with eye trackers. Attributes: version (str): The model name and software version of the eye tracker. - initialized (bool): A flag indicating whether :meth:`setup` has been run successfully. - local_start_time (float): The time at which the tracker last started recording, according - to the local computer's clock. - tracker_start_time (float): The time at which the tracker last started recording, according - to the eye tracker's internal clock. + initialized (bool): A flag indicating whether :meth:`setup` has been run. + tracker_start_time (float): The time (in milliseconds) at which the tracker + last started recording, according to the eye tracker's internal clock. """ - def __init__(self): self.version = None self.initialized = False @@ -45,6 +175,7 @@ def __init__(self): self._unresolved_exceptions = 0 self._quitting = False self._el = None + self._parser = None self._event_names = { GAZE_SAMPLE: 'gaze position', BLINK_START: 'blink start', @@ -56,104 +187,42 @@ def __init__(self): FIXATION_UPDATE: 'fixation update' } self._event_defaults = { - GAZE_SAMPLE: [None, None], - SACCADE_START: [EL_TIME_START, EL_GAZE_START], - SACCADE_END: [EL_TIME_END, EL_GAZE_END], - FIXATION_START: [EL_TIME_START, EL_GAZE_START], - FIXATION_END: [EL_TIME_END, EL_GAZE_AVG], - FIXATION_UPDATE: [EL_TIME_END, EL_GAZE_AVG] + GAZE_SAMPLE: None, + SACCADE_START: EL_GAZE_START, + SACCADE_END: EL_GAZE_END, + FIXATION_START: EL_GAZE_START, + FIXATION_END: EL_GAZE_AVG, + FIXATION_UPDATE: EL_GAZE_AVG, } - def _within_boundary(self, label, event, report, inspect): - """Checks whether an event is within a boundary. For internal use. + def _get_tracker_events(self, events, samples=False): + """Internal method for retrieving eye events from the tracker. + + This must be implemented for all EyeTracker types. Args: - label (str): The label of the boundary to check if the event is in, added using - add_boundary(). - event: The eye event (e.g. saccade, fixation, gaze sample) to test against the - given boundary. - report: A flag indicating whether to report the start time (``EL_TIME_START``) or - end time (``EL_TIME_END``) of the eye event. - inspect: A flag indicating which gaze attribute of the eye event should be checked - against the boundary: the gaze at the start of the event (``EL_GAZE_START``), - the gaze at the end of the event (``EL_GAZE_END``), or the event's average gaze - (``EL_GAZE_AVG``). Note that not all eye events have all three gaze attributes. + events (list): A list of the desired event types to retrieve. + samples (bool): Whether gaze samples should be returned in addition to other + event types. Returns: - The timestamp of the start or end of the event (see the 'report' argument) if the - inspected region was within the specified boundary, otherwise False. + list: A list of tracker-specific events. """ - timestamp = self.get_event_timestamp(event, report) - gaze_coords = self.get_event_gaze(event, inspect) + raise NotImplementedError() - result = self.boundaries.within_boundary(label, gaze_coords) - return timestamp if result else False + def _get_latest_sample(self): + """Internal method for retrieving the most recent gaze sample from the tracker. - def _exited_boundary(self, label, event, report): - """Checks whether a saccade event exited a given boundary. For internal use. - - Args: - label (str): The label of the boundary to check if the event is in, added using - add_boundary(). - event: The saccade-end event to test against the given boundary. - report: A flag indicating whether to report the start time (``EL_TIME_START``) or - end time (``EL_TIME_END``) of the saccade. + This must be implemented for all EyeTracker types. Returns: - The timestamp of the start or end of the event (see the 'report' argument) if the - saccade exited the given boundary, otherwise False. + A tracker-specific gaze sample event. """ - e_type = self.get_event_type(event) - if e_type != SACCADE_END: - typename = self.get_event_name(e_type) - err = "Only saccade end events can be used for boundary exit tests ({0} given)" - raise EyeTrackerError(err.format(typename)) - - timestamp = self.get_event_timestamp(event, report) - start_gaze = self.get_event_gaze(event, EL_GAZE_START) - end_gaze = self.get_event_gaze(event, EL_GAZE_END) - - start_inside = self.boundaries.within_boundary(label, start_gaze) - end_inside = self.boundaries.within_boundary(label, end_gaze) - return timestamp if (start_inside and not end_inside) else False - - - def _saccade_in_direction(self, doi, event, report): - """Checks whether a saccade event occurred in a given direction. For internal use. - - Args: - doi (:obj:`List` of str): The names of the direction(s) of interest to watch for - saccades in. Both a vertical ('up' or 'down') or horizontal ('left' or 'right') - direction of interest can be specified. - event: The saccade-end event to check the direction of. - report: A flag indicating whether to report the start time (``EL_TIME_START``) or - end time (``EL_TIME_END``) of the saccade. - - Returns: - The timestamp of the start or end of the event (see the 'report' argument) if the - saccade direction matched all the given directions of interest, otherwise False. - - """ - e_type = self.get_event_type(event) - if e_type != SACCADE_END: - typename = self.get_event_name(e_type) - err = "Only saccade end events can be used for saccade direction tests ({0} given)" - raise EyeTrackerError(err.format(typename)) - - sxp, syp = self.get_event_gaze(event, EL_GAZE_START) - exp, eyp = self.get_event_gaze(event, EL_GAZE_END) - sacc_direction = [None, None] - sacc_direction[0] = "right" if (exp - sxp) > 0 else "left" - sacc_direction[1] = "down" if (eyp - syp) > 0 else "up" - timestamp = self.get_event_timestamp(event, report) - - # Check if the direction(s) of interest match the direction of the saccade - result = all(direction in sacc_direction for direction in doi) - return timestamp if result else False + raise NotImplementedError() def _setup(self): @@ -177,8 +246,9 @@ def setup(self): """ from klibs.KLEnvironment import db self._setup() + P.eye_tracker_name = self.version if P.demographics_collected and 'session_info' in db.table_schemas: - db.update('session_info', {'eyetracker': self.version}) + db.update('session_info', {'eyetracker': self.version}) self.calibrate() self.initialized = True @@ -220,52 +290,79 @@ def stop(self): def shut_down(self, incomplete=False): - """Terminates recording and disconnects from the eye tracker, putting it into standby mode. + """Terminates recording and disconnects from the eye tracker. + Should also transfer any data files from the current session to the KLibs computer from the tracker (e.g. EDF files). Called automatically whenever KLibs exits. For internal use only. Args: - incomplete (bool, optional): Whether the full session was completed before the function - was called. If True, any tracker data files will be written to an 'incomplete' - subfolder of the eye tracker data directory ('ExpAssets/EDF'). Defaults to False. + incomplete (bool, optional): Whether the full session was completed before + this method was called. If True, any tracker data files will be written + to an 'incomplete' subfolder of the EDF directory. Defaults to False. """ pass + def get_sample(self): + """Returns the most recent gaze sample from the tracker. + + The returned sample can be inspected for attributes such as timestamp, + left/right eye gaze, current pupil size, and more:: + + s = self.el.get_sample() + stime = s.timestamp + # If both eyes present, get difference in pupil size + if s.left and s.right: + ps_left = s.left.pupil_size + ps_right = s.right.pupil_size + ps_diff = abs(ps_left - ps_right) + + See the :obj:`~klibs.KLEyeTracking.GazeSample` documentation for a full list + of attributes. + + Returns: + :obj:`~klibs.KLEyeTracking.GazeSample`: The most recent gaze sample. + + """ + if not self.recording: + err = "Cannot retrieve eye data when eye tracker is not recording." + raise RuntimeError(err) + e = self._get_latest_sample() + if not e: + return None + return GazeSample(e, self._parser) + + def get_event_queue(self, include=[], exclude=[]): - """Fetches and returns the eye tracker's event queue, emptying it in the process. + """Retrieves and empties the eye tracker's event queue. - To avoid problems with event processing when performing multiple tests in a short interval - (e.g. checking for fixations in multiple boundaries every iteration of a loop), you should - use this method to retrieve the eye tracker event queue once at the start and then pass it - to every function that inspects eye events. - - For example, the following code will not reliably detect saccades to the second boundary:: - - while self.evm.before('cue_on'): - draw_stimuli() - if self.el.saccade_to_boundary('boundary1'): - print('saccade to boundary 1!') - if self.el.saccade_to_boundary('boundary2'): - print('saccade to boundary 2!') - flip() # refresh screen - - This is because the first call to :meth:`saccade_to_boundary` fetches and clears the eye - event queue, meaning that the second call is only processing the saccades that have - happened in the tiny interval since the first call. To fix this, you should fetch the - queue once per loop:: - - while self.evm.before('cue_on'): - draw_stimuli() - el_q = self.el.get_event_queue() # fetch queue once per loop - if self.el.saccade_to_boundary('boundary1', event_queue=el_q): - print('saccade to boundary 1!') - if self.el.saccade_to_boundary('boundary2', event_queue=el_q): - print('saccade to boundary 2!') - flip() # refresh screen + The different types of events returned by this method are listed in the + :class:`~klibs.KLEyeTracking.EyeEvent` documentation along with their + corresponding type codes. You can inspect these events directly or pass them + to any of the convenience methods provided by the EyeTracker class:: + + events = self.el.get_event_queue() + + # Check for saccades from fixation + if self.el.saccade_from_boundary('fixation', events): + self.show_error_message(self.looked_away_msg) + err = 'looked_away' + + # Check for any blinks + for e in events: + if e.type in [BLINK_START, BLINK_END]: + self.show_error_message(self.blinked_msg) + err = 'blink' + break + + To avoid problems with event processing when performing multiple tests in a + short interval (e.g. checking for fixations in multiple boundaries every + iteration of a loop), you should use this method to retrieve the eye tracker + event queue once at the start and then pass it to every function that inspects + eye events. Args: include (:obj:`List`, optional): A list specifying the types of eye events to fetch @@ -275,15 +372,44 @@ def get_event_queue(self, include=[], exclude=[]): from the returned queue. Defaults to an empty list (i.e. no events excluded.) Returns: - A :obj:`List` of eye events. - - """ - pass + list: A list of :obj:`~klibs.KLEyeTracking.EyeEvent` objects. + + """ + if len(include): + valid_events = set(include) + elif len(exclude): + valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]).difference(exclude) + else: + valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]) + samples = GAZE_SAMPLE in valid_events + + # Wrap tracker-specific eye events with pythonic EyeEvent objects + queue = [] + for e in self._get_tracker_events(valid_events, samples): + if e.type == BLINK_START: + queue.append(BlinkStart(e, self._parser)) + elif e.type == BLINK_END: + queue.append(BlinkEnd(e, self._parser)) + elif e.type == SACCADE_START: + queue.append(SaccadeStart(e, self._parser)) + elif e.type == SACCADE_END: + queue.append(SaccadeEnd(e, self._parser)) + elif e.type == FIXATION_START: + queue.append(FixationStart(e, self._parser)) + elif e.type == FIXATION_UPDATE: + queue.append(FixationUpdate(e, self._parser)) + elif e.type == FIXATION_END: + queue.append(FixationEnd(e, self._parser)) + elif e.type == GAZE_SAMPLE: + queue.append(GazeSample(e, self._parser)) + + return queue def clear_queue(self): - """Clears the event queue for the eye tracker. Any uninspected eye events will be - discarded. + """Clears the event queue for the eye tracker. + + Any uninspected eye events will be discarded. """ pass @@ -377,13 +503,10 @@ def within_boundary(self, label, valid_events, event_queue=None): timestamp = None for e in event_queue: - if e == None or self.get_event_type(e) not in valid_events: + if e == None or e.type not in valid_events: continue - e_type = self.get_event_type(e) - _report = self._event_defaults[e_type][0] - _inspect = self._event_defaults[e_type][1] - timestamp = self._within_boundary(label, e, _report, _inspect) - if timestamp: + gaze = self.get_event_gaze(e, self._event_defaults[e.type]) + if gaze and self.boundaries.within_boundary(label, gaze): return True return False @@ -438,15 +561,13 @@ def saccade_to_boundary(self, label, event_queue=None): """ if not event_queue: event_queue = self.get_event_queue([SACCADE_END]) - if not len(event_queue): - return False for e in event_queue: - if e == None or self.get_event_type(e) != SACCADE_END: + if e == None or e.type != SACCADE_END: continue - started_within = self._within_boundary(label, e, EL_TIME_START, EL_GAZE_START) - ended_within = self._within_boundary(label, e, EL_TIME_END, EL_GAZE_END) - if ended_within and not started_within: + starts_in = self.boundaries.within_boundary(label, e.start_gaze) + ends_in = self.boundaries.within_boundary(label, e.end_gaze) + if ends_in and not starts_in: return True return False @@ -470,14 +591,13 @@ def saccade_from_boundary(self, label, event_queue=None): """ if not event_queue: event_queue = self.get_event_queue([SACCADE_END]) - if not len(event_queue): - return False for e in event_queue: - if e == None or self.get_event_type(e) != SACCADE_END: + if e == None or e.type != SACCADE_END: continue - timestamp = self._exited_boundary(label, e, EL_TIME_START) - if timestamp: + starts_in = self.boundaries.within_boundary(label, e.start_gaze) + ends_in = self.boundaries.within_boundary(label, e.end_gaze) + if starts_in and not ends_in: return True return False @@ -526,14 +646,18 @@ def saccade_in_direction(self, doi, event_queue=None): if not event_queue: event_queue = self.get_event_queue([SACCADE_END]) - if not len(event_queue): - return False for e in event_queue: - if e == None or self.get_event_type(e) != SACCADE_END: + if e == None or e.type != SACCADE_END: continue - timestamp = self._saccade_in_direction(doi, e, EL_TIME_START) - if timestamp: + sxp, syp = e.start_gaze + exp, eyp = e.end_gaze + directions = [ + "right" if (exp - sxp) > 0 else "left", + "down" if (eyp - syp) > 0 else "up" + ] + # Check if the direction(s) of interest match the direction of the saccade + if all(d in directions for d in doi): return True return False @@ -606,7 +730,7 @@ def _drift_correct(self, loc, target_callback): def gaze(self, return_integers=True, binocular_mode=RIGHT_EYE): - """Fetches the (x,y) coordinates of the participant's current gaze on the screen. + """Fetches the (x,y) coordinates of the participant's gaze on the screen. Args: return_integers (bool, optional): Whether to return the gaze coordinates as integers @@ -644,8 +768,10 @@ def write(self, message): def get_event_type(self, e): - """Fetches the numeric code corresponding to an eye event's type (e.g end saccade, start - fixation). + """Fetches the numeric code corresponding to an eye event's type. + + Deprecated, use the ``type`` attribute of :obj:`~klibs.KLEyeTracking.EyeEvent` + objects instead. Args: e: An eye event from the tracker. @@ -654,7 +780,7 @@ def get_event_type(self, e): The numeric type of the event (e.g. ``SACCADE_END``, ``FIXATION_START``). """ - pass + return e.type def get_event_name(self, event): @@ -697,7 +823,22 @@ def get_event_gaze(self, event, inspect): lacks that attribute (e.g. ``EL_GAZE_AVG`` for a ``SACCADE_END`` event). """ - pass + if event.type == GAZE_SAMPLE: + return event.right.gaze if event.right else event.left.gaze + else: + try: + if inspect == EL_GAZE_START: + gaze = event.start_gaze + elif inspect == EL_GAZE_END: + gaze = event.end_gaze + elif inspect == EL_GAZE_AVG: + gaze = event.average_gaze + except AttributeError: + typename = self.get_event_name(event.type) + err = "Cannot inspect {0} for {1} events." + raise EyeTrackerError(err.format(inspect, typename)) + + return gaze def get_event_timestamp(self, event, report): @@ -718,25 +859,22 @@ def get_event_timestamp(self, event, report): a start timestamp (e.g. ``FIXATION_START``). """ - pass + if event.type == GAZE_SAMPLE: + return event.timestamp + else: + try: + return event.start_time if report == EL_TIME_START else event.end_time + except AttributeError: + typename = self.get_event_name(event.type) + err = "Cannot report {0} for {1} events." + raise EyeTrackerError(err.format(report, typename)) def get_event_info(self, event): - """Returns all available info about an eye event or gaze sample in the form of a - :obj:`dict`. - - Usage:: - - q = self.el.get_event_queue() # fetch all unprocessed eye events - for event in q: - info = self.el.get_event_info(event) - if info['type'] == SACCADE_END: - end_gaze = info[EL_GAZE_END] - end_time = info[EL_TIME_END] + """Returns info about an eye event or gaze sample in the form of a dict. - - See the table in :meth:`within_boundary` for a list of the attributes available for each - eye event type. + Deprecated, the attributes of :obj:`~klibs.KLEyeTracking.EyeEvent` objects + should be used instead. Args: event: The eye event (e.g. saccade, fixation, gaze sample) to collect the attributes of. @@ -746,7 +884,27 @@ def get_event_info(self, event): start/end/average gaze, and type name. """ - pass + info = {} + info['type'] = event.type + if event.type == GAZE_SAMPLE: + info[EL_TIME_START] = event.timestamp + info[EL_TIME_END] = event.timestamp + info[EL_GAZE_START] = event.gaze + info[EL_GAZE_END] = event.gaze + elif event.type in [BLINK_START, BLINK_END]: + info[EL_TIME_START] = event.start_time + if event.type == BLINK_END: + info[EL_TIME_END] = event.end_time + else: + info[EL_TIME_START] = event.start_time + info[EL_GAZE_START] = event.start_gaze + if event.type in [SACCADE_END, FIXATION_END, FIXATION_UPDATE]: + info[EL_TIME_END] = event.end_time + info[EL_GAZE_END] = event.end_gaze + if event.type != SACCADE_END: + info[EL_GAZE_AVG] = event.average_gaze + + return info @property diff --git a/klibs/KLEyeTracking/KLTryLink.py b/klibs/KLEyeTracking/KLTryLink.py index 9e5550cf..7eeb03eb 100755 --- a/klibs/KLEyeTracking/KLTryLink.py +++ b/klibs/KLEyeTracking/KLTryLink.py @@ -1,13 +1,12 @@ # -*- coding: utf-8 -*- __author__ = 'Jonathan Mulle & Austin Hurst' +from time import monotonic_ns + from klibs.KLExceptions import EyeTrackerError from klibs.KLConstants import (LEFT_EYE, RIGHT_EYE, BINOCULAR, FIXATION_START, FIXATION_UPDATE, FIXATION_END, GAZE_SAMPLE, - SACCADE_START, SACCADE_END, BLINK_START, BLINK_END, - EL_GAZE_START, EL_GAZE_END, EL_GAZE_AVG, EL_TIME_START, EL_TIME_END, - EL_ALL_EVENTS, EL_TRUE, EL_FALSE, - TK_S, TK_MS) + SACCADE_START, SACCADE_END, BLINK_START, BLINK_END) from klibs import P from klibs.KLUtilities import now, line_segment_len, px_to_deg from klibs.KLBoundary import CircleBoundary @@ -15,7 +14,7 @@ from klibs.KLUserInterface import ( ui_request, mouse_pos, mouse_clicked, key_pressed, show_cursor, hide_cursor ) -from klibs.KLEyeTracking.KLEyeTracker import EyeTracker +from klibs.KLEyeTracking.KLEyeTracker import EyeTracker, EventParser from klibs.KLEyeTracking.utils import GazeEventGenerator from sdl2 import SDL_GetTicks, SDL_Delay @@ -39,28 +38,80 @@ def __init__(self, time, gaze): self.gaze = gaze -class TryLink(EyeTracker): - """A simulated eye tracker, using the position and movements of the mouse cursor as a stand-in - for gaze position and saccades/fixations. Used for developing, testing, and demoing experiments - that require an eye tracker without the need to have eye tracking hardware present. +class TryLinkEventParser(EventParser): + """A parser defining how to extract data from simulated tracker events. + + """ + def get_type(self, e): + """Retrieves the type of a given event (e.g. ``BLINK_START``). + """ + return e.type - TryLink simulation is used by default if the 'pylink' Python package is not installed, or if - ``P.eye_tracking`` is True but ``P.eye_tracker_availiable`` is False. In addition, TryLink - simulation can be forced by adding the flag ``-ELx`` to the ``klibs run`` command when - launching an experiment. + def eye(self, e): + """Retrieves the eye for a given event. + """ + return RIGHT_EYE - For more general and comprehensive documentation, see the :class:`~.EyeTracker` class. + def timestamp(self, e): + """Gets the timestamp (in msec) for a gaze sample. + """ + return e.time - Attributes: - version (str): The model name and software version of the eye tracker. - initialized (bool): A flag indicating whether :meth:`setup` has been run successfully. - local_start_time (float): The time at which the tracker last started recording, according - to the local computer's clock. - tracker_start_time (float): The time at which the tracker last started recording, according - to the eye tracker's internal clock. + def left_gaze(self, e): + """Gets the left eye (x, y) gaze coordinates for a gaze sample. + """ + return None - """ + def right_gaze(self, e): + """Gets the right eye (x, y) gaze coordinates for a gaze sample. + """ + return e.gaze + + def left_pupil_size(self, e): + """Gets the left eye pupil size (arbitrary units) for a gaze sample. + """ + return None + + def right_pupil_size(self, e): + """Gets the right eye pupil size (arbitrary units) for a gaze sample. + """ + return 2000 if e.gaze else None + + def start_time(self, e): + """Gets the timestamp (in msec) for the start of the event. + """ + return e.start_time + + def end_time(self, e): + """Gets the timestamp (in msec) for the end of the event. + """ + return e.end_time + + def start_gaze(self, e): + """Gets the (x, y) gaze coordinates at the start of the event. + """ + return e.start_gaze + + def end_gaze(self, e): + """Gets the (x, y) gaze coordinates at the end of the event. + """ + return e.end_gaze + def avg_gaze(self, e): + """Gets the average (x, y) gaze coordinates for the event. + """ + return e.avg_gaze + + def duration(self, e): + """Gets the total duration (in msec) for the event. + """ + return e.end_time - e.start_time + + +class TryLink(EyeTracker): + """A mouse-based simulated eye tracker for task development and testing. + + """ def __init__(self): EyeTracker.__init__(self) self.version = None @@ -69,6 +120,7 @@ def __init__(self): self.tracker_start_time = None self._recording = False self._generator = GazeEventGenerator() + self._parser = TryLinkEventParser() def _setup(self): @@ -78,12 +130,10 @@ def _setup(self): self.version = "TryLink 0.1a" - def setup(self): - EyeTracker.setup(self) # so it shows up in the docs - - def calibrate(self): - """Enters the calibration and setup mode for the eye tracker. Does nothing in TryLink mode. + """Enters the calibration and setup mode for the eye tracker. + + Does nothing in TryLink mode. """ pass @@ -121,42 +171,31 @@ def stop(self): def shut_down(self, incomplete=False): - """Terminates recording and disconnects from the eye tracker, putting it into standby mode. - Should also transfer any data files from the current session to the KLibs computer - from the tracker (e.g. EDF files). Does nothing when using TryLink simulation. + """Terminates recording and disconnects from the eye tracker. Called automatically whenever KLibs exits. For internal use only. Args: - incomplete (bool, optional): Whether the full session was completed before the function - was called. Has no effect in TryLink simulation mode. Defaults to False. + incomplete (bool, optional): Whether the full session was completed. + Defaults to False. """ self.stop() return 0 - def get_event_queue(self, include=[], exclude=[]): - """Fetches and returns the eye tracker's event queue, emptying it in the process. - - In TryLink simulation mode, this uses some fancy voodoo with mouse cursor movement to - produce eye events, allowing you to write and test your eye tracking code on any computer - without needing physical access to an eye tracker itself. + def _get_tracker_events(self, events, samples=False): + """Internal method for retrieving eye events from the tracker. + + Args: + events (list): A list of the desired event types to retrieve. + samples (bool): Whether gaze samples should be returned in addition to other + event types. Returns: - A :obj:`List` of simulated eye events. + list: A list of tracker-specific events. """ - if len(include): - valid_events = set(include) - elif len(exclude): - valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]).difference(exclude) - else: - valid_events = set(EL_ALL_EVENTS + [GAZE_SAMPLE]) - - samples = GAZE_SAMPLE in valid_events - events = len(valid_events.intersection(EL_ALL_EVENTS)) > 0 - timestamp = self.now() x, y = self.gaze() mouse_clicked = mouse_button_state().any_pressed @@ -166,21 +205,35 @@ def get_event_queue(self, include=[], exclude=[]): x, y = (None, None) queue = [] - if samples and not blink: - # NOTE: Check whether tracker sends gaze samples during blinks - queue.append(MouseSample(timestamp, (x, y))) - if events: - self._generator.update(x, y, timestamp) - for e in self._generator.get_events(): - if e.type in valid_events: - queue.append(e) + self._generator.update(x, y, timestamp) + for e in self._generator.get_events(): + if e.type in events: + queue.append(e) + if samples: + gaze = None if blink else (x, y) + queue.append(MouseSample(timestamp, gaze)) return queue + def _get_latest_sample(self): + """Internal method for retrieving the most recent gaze sample from the tracker. + + Returns: + A tracker-specific gaze sample event. + + """ + timestamp = self.now() + mouse_clicked = mouse_button_state().any_pressed + blink = P.dm_click_to_blink and mouse_clicked + gaze = None if blink else self.gaze() + return MouseSample(timestamp, gaze) + + def clear_queue(self): - """Clears the event queue for emulated eye events. Any uninspected eye events will be - discarded. + """Clears the event queue for emulated eye events. + + Any uninspected eye events will be discarded. """ self._generator.reset() @@ -225,174 +278,78 @@ def _drift_correct(self, loc, target_callback): return round(px_to_deg(dist_px), 2) - def gaze(self, return_integers=True, binocular_mode=RIGHT_EYE): - """Fetches the (x,y) coordinates of the participant's current gaze on the screen. In - TryLink mode, this fetches the current coordinates of the mouse cursor. - - Args: - return_integers (bool, optional): Whether to return the gaze coordinates as integers - or floats. Defaults to True (integers). - binocular_mode (int, optional): Tells the function which gaze coordinates to return - for binocular samples. Can be any of ``RIGHT_EYE`` (returns right eye gaze), - ``LEFT_EYE`` (returns left eye gaze), or ``BINOCULAR`` (returns average gaze - of both eyes). Defaults to ``RIGHT_EYE``. In TryLink mode, this has no effect. + def drift_correct(self, location=None, target=None, fill_color=None, draw_target=True): + """Interactively checks the accuracy of the tracker's calibration. - Returns: - A :obj:`Tuple` containing the (x,y) pixel coordinates of the participant's gaze. - - """ - gaze_pos = mouse_pos() - return gaze_pos if return_integers else tuple(float(p) for p in gaze_pos) - + In TryLink mode, drift correct targets are presented on screen exactly as + they are with real hardware. To perform a drift correct in TryLink mode, move + the mouse cursor to the midpoint of the drift correct target and either press + the space bar or click the mouse. - def now(self): - """Fetches the current time according to the tracker's internal clock. + Args: + location (tuple, optional): The (x, y) pixel coordinates at which to draw + the drift correct target. Defaults to the center of the screen. + target (optional): A :obj:`Drawbject` or other stimulus texture to use as + the drift correct target. Defaults to a circular target. + fill_color (tuple, optional): The fill colour to use for the background of + the drift correct screen. Defaults to ``P.default_fill_color``. + draw_target (bool, optional): If True, this method will draw and present the + drift correct target when called. If False it will not draw the target, + allowing the use of custom drift correct screens. Defaults to True. Returns: - float: The current tracker time (in milliseconds). + float: The magnitude of the measured drift (in degrees of visual angle). """ - return float(SDL_GetTicks()) + return EyeTracker.drift_correct(self, location, target, fill_color, draw_target) - def write(self, message): - """Writes a message to the eye tracker's data file. In TryLink mode, this method - does nothing. + def gaze(self, return_integers=True, binocular_mode=RIGHT_EYE): + """Fetches the (x,y) coordinates of the participant's gaze on the screen. - Args: - message (str): The message to write to the eye tracker's data file. - - """ - pass - + In TryLink mode, this fetches the current coordinates of the mouse cursor. - def get_event_type(self, e): - """Fetches the numeric code corresponding to an eye event's type (e.g end saccade, start - fixation). - Args: - e: An eye event from the tracker. + return_integers (bool, optional): Whether to return the gaze coordinates as + integers or floats. Defaults to True (integers). + binocular_mode (int, optional): Tells the function which gaze coordinates + to return for binocular samples. In TryLink mode, this has no effect. Returns: - The numeric type of the event (e.g. ``EL_END_SACCADE``, ``EL_START_FIXATION``). + A :obj:`Tuple` containing the (x,y) pixel coordinates of the participant's gaze. """ - return e.type + gaze_pos = mouse_pos() + return gaze_pos if return_integers else tuple(float(p) for p in gaze_pos) - def get_event_gaze(self, event, inspect): - """Retrieves the gaze coordinates from an eye event or gaze sample. + def now(self): + """Fetches the current time according to the tracker's internal clock. - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to return the gaze - coordinates from. - inspect: A flag indicating the type of gaze to retrieve: the gaze at the start of - the event (``EL_GAZE_START``), the gaze at the end of the event (``EL_GAZE_END``), - or the event's average gaze (``EL_GAZE_AVG``). Note that not all eye events have - all three gaze attributes. Has no effect on gaze samples. - Returns: - The (x, y) gaze coordinates for the event. - - Raises: - EyeTrackerError: If asked to inspect the end gaze or average gaze for an eye event that - lacks that attribute (e.g. ``EL_GAZE_AVG`` for a ``SACCADE_END`` event). + float: The current tracker time (in milliseconds). """ - if event.type == GAZE_SAMPLE: - coords = event.gaze - else: - try: - if inspect == EL_GAZE_START: - coords = event.start_gaze - elif inspect == EL_GAZE_END: - coords = event.end_gaze - elif inspect == EL_GAZE_AVG: - coords = event.avg_gaze - except AttributeError: - typename = self.get_event_name(event.type) - err = "Cannot inspect {0} for {1} events." - raise EyeTrackerError(err.format(inspect, typename)) - - return coords - - - def get_event_timestamp(self, event, report): - """Gets the timestamp from an eye event or gaze sample. - - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to return the timestamp - from. - report: A flag indicating whether to report the start time (``EL_TIME_START``) or - end time (``EL_TIME_END``) of the eye event. Has no effect on gaze samples, - which do not have separate start/end times. - - Returns: - The timestamp for the start or end of the event. + return float(monotonic_ns() / 1000000) - Raises: - EyeTrackerError: If asked to report the end timestamp for an eye event that only has - a start timestamp (e.g. ``FIXATION_START``). - """ - if event.type == GAZE_SAMPLE: - return event.time - else: - try: - return event.start_time if report == EL_TIME_START else event.end_time - except AttributeError: - typename = self.get_event_name(event.type) - err = "Cannot report {0} for {1} events." - raise EyeTrackerError(err.format(report, typename)) - - - def get_event_info(self, event): - """Returns all available info about an eye event or gaze sample in the form of a - :obj:`dict`. + def write(self, message): + """Writes a message to the eye tracker's data file. - Usage:: - - q = self.el.get_event_queue() # fetch all unprocessed eye events - for event in q: - info = self.el.get_event_info(event) - if info['type'] == SACCADE_END: - end_gaze = info[EL_GAZE_END] - end_time = info[EL_TIME_END] + Does nothing in TryLink mode. - See the table in :meth:`within_boundary` for a list of the attributes available for each - eye event type. - Args: - event: The eye event (e.g. saccade, fixation, gaze sample) to collect the attributes of. - - Returns: - A :obj:`dict` containing the available attributes of the event, such as start/end time, - start/end/average gaze, and type name. + message (str): The message to write to the eye tracker's data file. """ - info = {} - info['type'] = event.type - if event.type == GAZE_SAMPLE: - info[EL_TIME_START] = event.time - info[EL_TIME_END] = event.time - info[EL_GAZE_START] = event.gaze - info[EL_GAZE_END] = event.gaze - else: - info[EL_TIME_START] = event.start_time - info[EL_GAZE_START] = event.start_gaze - if event.type in [SACCADE_END, FIXATION_END, FIXATION_UPDATE]: - info[EL_TIME_END] = event.end_time - info[EL_GAZE_END] = event.end_gaze - if event.type != SACCADE_END: - info[EL_GAZE_AVG] = event.avg_gaze - - return info + pass @property def eye(self): - """str or None: The eye(s) currently being recorded by the tracker. Can be 'left', 'right', - 'both', or None (if no eye is currently available). Always the right eye in TryLink mode. + """str: The eye(s) currently being recorded by the tracker. + + Always 'right' in TryLink mode. """ return 'right' diff --git a/klibs/KLEyeTracking/__init__.py b/klibs/KLEyeTracking/__init__.py index 787a3d83..032660c3 100755 --- a/klibs/KLEyeTracking/__init__.py +++ b/klibs/KLEyeTracking/__init__.py @@ -5,6 +5,7 @@ from klibs.KLInternal import package_available from .KLEyeTracker import EyeTracker from .KLTryLink import TryLink +from .events import EyeEvent, GazeSample PYLINK_AVAILABLE = package_available('pylink') diff --git a/klibs/KLEyeTracking/events.py b/klibs/KLEyeTracking/events.py new file mode 100644 index 00000000..abc15bca --- /dev/null +++ b/klibs/KLEyeTracking/events.py @@ -0,0 +1,402 @@ + +from klibs.KLConstants import ( + BLINK_START, BLINK_END, SACCADE_START, SACCADE_END, + FIXATION_START, FIXATION_END, FIXATION_UPDATE, + LEFT_EYE, RIGHT_EYE, BINOCULAR +) +from klibs.KLUtilities import line_segment_len as linear_dist + + +class EyeEvent(): + """An eye event generated by an eye tracker. + + EyeEvents provide a friendly way of working with different types of eye events + that can occur during an eye tracking experiment, allowing easy access to + attributes such as the start and end times of events, the average gaze position + of fixations, the amplitudes of saccades, the pupil size for a gaze sample, + and more. + + +---------------------+------------------------+ + | Event Type | Event Class | + +=====================+========================+ + | ``GAZE_SAMPLE`` | :obj:`~GazeSample` | + +---------------------+------------------------+ + | ``FIXATION_START`` | :obj:`~FixationStart` | + +---------------------+------------------------+ + | ``FIXATION_END`` | :obj:`~FixationEnd` | + +---------------------+------------------------+ + | ``FIXATION_UPDATE`` | :obj:`~FixationUpdate` | + +---------------------+------------------------+ + | ``SACCADE_START`` | :obj:`~SaccadeStart` | + +---------------------+------------------------+ + | ``SACCADE_END`` | :obj:`~SaccadeEnd` | + +---------------------+------------------------+ + | ``BLINK_START`` | :obj:`~BlinkStart` | + +---------------------+------------------------+ + | ``BLINK_END`` | :obj:`~BlinkEnd` | + +---------------------+------------------------+ + + EyeEvents are Pythonic wrappers for tracker-specific event types, making it easy + to write code that works seamlessly with multiple tracker backends. To access any + attributes or methods not wrapped by the EyeEvent API, you can access the + underlying raw tracker event using through the ``self.raw`` attribute:: + + for e in self.el.get_event_queue(GAZE_SAMPLE): + # If using EyeLink hardware, get button box state from gaze samples + if self.el.eyelink: + buttons = e.raw.getButtons() + + """ + def __init__(self, raw, parser): + self.raw = raw + self.parser = parser + + @property + def type(self): + """int: The event type.""" + return self.parser.get_type(self.raw) + + +class EyeSample(): + """Contains the gaze and pupil size information for a given eye. + + Attributes: + gaze (tuple): The (x, y) gaze coordinates (in pixels) of the eye + for the sample. + pupil_size (float): The pupil size of the eye (in arbitrary tracker-defined + units) for the sample. + + """ + def __init__(self, gaze, ps): + self.gaze = gaze + self.pupil_size = ps + + +class GazeSample(EyeEvent): + """An event representing a single gaze sample from an eye tracker. + + Gaze samples can be inspected for information about the gaze, pupil size, + and tracker status at the time the sample was recorded:: + + for e in self.el.get_event_queue(GAZE_SAMPLE): + if e.left and e.right: + gaze_l = e.left.gaze + gaze_r = e.right.gaze + eye_dist = line_segment_len(gaze_l, gaze_r) + + """ + @property + def type(self): + """int: The event type (``klibs.GAZE_SAMPLE``)""" + return self.parser.get_type(self.raw) + + @property + def timestamp(self): + """float: The timestamp (in msec) of the sample.""" + return self.parser.timestamp(self.raw) + + @property + def gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) for the current sample. + + If recording binocular data, this is the average gaze of both eyes unless + data from either eye is missing. To access the gaze for a specific eye, use the + :attr:`~left` and :attr:`~right` attributes. + + """ + eye = self.parser.eye(self.raw) + if eye == RIGHT_EYE: + return self.parser.right_gaze(self.raw) + elif eye == LEFT_EYE: + return self.parser.left_gaze(self.raw) + elif eye == BINOCULAR: + gaze_l = self.parser.left_gaze(self.raw) + gaze_r = self.parser.right_gaze(self.raw) + if gaze_l and gaze_r: + return ((gaze_l[0] + gaze_r[0]) / 2.0, (gaze_l[1] + gaze_r[1]) / 2.0) + elif gaze_r: + return gaze_r + elif gaze_l: + return gaze_l + return None + + @property + def pupil_size(self): + """float: The pupil size (in arbitrary units) for the current sample. + + If recording binocular data, this is the average pupil size of both eyes unless + data from either eye is missing. To access the pupil size for a specific eye, + use the :attr:`~left` and :attr:`~right` attributes. + + """ + eye = self.parser.eye(self.raw) + if eye == RIGHT_EYE: + return self.parser.right_pupil_size(self.raw) + elif eye == LEFT_EYE: + return self.parser.left_pupil_size(self.raw) + elif eye == BINOCULAR: + psl = self.parser.left_pupil_size(self.raw) + psr = self.parser.right_pupil_size(self.raw) + if psl and psr: + return (psl + psr) / 2.0 + elif psr: + return psr + elif psl: + return psl + return None + + @property + def left(self): + """:obj:`~EyeSample` or None: The left eye data for the sample. + + If the left eye is missing or not being recorded, this will return None. + + """ + if self.parser.eye(self.raw) == RIGHT_EYE: + return None + gaze = self.parser.left_gaze(self.raw) + ps = self.parser.left_pupil_size(self.raw) + return EyeSample(gaze, ps) + + @property + def right(self): + """:obj:`~EyeSample` or None: The right eye data for the sample. + + If the right eye is missing or not being recorded, this will return None. + + """ + if self.parser.eye(self.raw) == LEFT_EYE: + return None + gaze = self.parser.right_gaze(self.raw) + ps = self.parser.right_pupil_size(self.raw) + return EyeSample(gaze, ps) + + @property + def binocular(self): + """bool: Indicates whether the sample has binocular data.""" + return self.parser.eye(self.raw) == BINOCULAR + + + +class FixationStart(EyeEvent): + """An eye event representing the start of a fixation. + + """ + @property + def type(self): + """int: The event type (``klibs.FIXATION_START``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the fixation.""" + return self.parser.start_time(self.raw) + + @property + def start_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) at the start of the fixation. + + """ + return self.parser.start_gaze(self.raw) + + +class FixationUpdate(EyeEvent): + """An eye event representing a fixation update. + + Fixation updates occur at regular intervals (usually 50 ms) during a fixation, + allowing the attributes of a fixation to be inspected while it is still in + progress. + + """ + @property + def type(self): + """int: The event type (``klibs.FIXATION_UPDATE``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the fixation.""" + return self.parser.start_time(self.raw) + + @property + def start_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) at the start of the fixation. + + """ + return self.parser.start_gaze(self.raw) + + @property + def end_time(self): + """float: The timestamp (in msec) of the last fixation update.""" + return self.parser.end_time(self.raw) + + @property + def end_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) at the end of the last update. + + """ + return self.parser.end_gaze(self.raw) + + @property + def average_gaze(self): + """tuple: The average (x, y) gaze coordinates (in pixels) for the fixation. + + """ + return self.parser.avg_gaze(self.raw) + + @property + def duration(self): + """float: The total duration (in msec) of the fixation so far.""" + return self.parser.duration(self.raw) + + +class FixationEnd(EyeEvent): + """An eye event representing the end of a fixation. + + """ + @property + def type(self): + """int: The event type (``klibs.FIXATION_END``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the fixation.""" + return self.parser.start_time(self.raw) + + @property + def start_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) at the start of the fixation. + + """ + return self.parser.start_gaze(self.raw) + + @property + def end_time(self): + """float: The timestamp (in msec) of the end of the fixation.""" + return self.parser.end_time(self.raw) + + @property + def end_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) at the end of the fixation. + + """ + return self.parser.end_gaze(self.raw) + + @property + def average_gaze(self): + """tuple: The average (x, y) gaze coordinates (in pixels) for the fixation. + + """ + return self.parser.avg_gaze(self.raw) + + @property + def duration(self): + """float: The total duration (in msec) of the fixation.""" + return self.parser.duration(self.raw) + + +class SaccadeStart(EyeEvent): + """An eye event representing the start of a saccade. + + """ + @property + def type(self): + """int: The event type (``klibs.SACCADE_START``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the saccade.""" + return self.parser.start_time(self.raw) + + @property + def start_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) of the start of the saccade. + + """ + return self.parser.start_gaze(self.raw) + + +class SaccadeEnd(EyeEvent): + """An eye event representing the end of a saccade. + + """ + @property + def type(self): + """int: The event type (``klibs.SACCADE_END``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the saccade.""" + return self.parser.start_time(self.raw) + + @property + def start_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) of the start of the saccade. + + """ + return self.parser.start_gaze(self.raw) + + @property + def end_time(self): + """float: The timestamp (in msec) of the end of the saccade.""" + return self.parser.end_time(self.raw) + + @property + def end_gaze(self): + """tuple: The (x, y) gaze coordinates (in pixels) of the end of the saccade. + + """ + return self.parser.end_gaze(self.raw) + + @property + def duration(self): + """float: The total duration (in msec) of the saccade.""" + return self.parser.duration(self.raw) + + @property + def amplitude(self): + """float: The overall amplitude (in pixels) of the saccade.""" + return linear_dist(self.start_gaze, self.end_gaze) + + +class BlinkStart(EyeEvent): + """An eye event representing the start of a blink. + + """ + @property + def type(self): + """int: The event type (``klibs.BLINK_START``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the blink.""" + return self.parser.start_time(self.raw) + + +class BlinkEnd(EyeEvent): + """An eye event representing the end of a blink. + + """ + @property + def type(self): + """int: The event type (``klibs.BLINK_END``)""" + return self.parser.get_type(self.raw) + + @property + def start_time(self): + """float: The timestamp (in msec) of the start of the blink.""" + return self.parser.start_time(self.raw) + + @property + def end_time(self): + """float: The timestamp (in msec) of the end of the blink.""" + return self.parser.end_time(self.raw) + + @property + def duration(self): + """float: The total duration (in msec) of the blink.""" + return self.parser.duration(self.raw) diff --git a/klibs/KLEyeTracking/utils.py b/klibs/KLEyeTracking/utils.py index 22fd4bf7..af33a93c 100644 --- a/klibs/KLEyeTracking/utils.py +++ b/klibs/KLEyeTracking/utils.py @@ -70,23 +70,23 @@ class GazeEventTemplate(): """ def __init__(self, start_time, start_x, start_y): - self.__start = start_time - self.__x_samples = [start_x] - self.__y_samples = [start_y] + self._start = start_time + self._x_samples = [start_x] + self._y_samples = [start_y] self._last_sample_time = start_time self.last_update = start_time def add_sample(self, x, y, timestamp): - self.__x_samples.append(x) - self.__y_samples.append(y) + self._x_samples.append(x) + self._y_samples.append(y) self._last_sample_time = timestamp def vector_change(self, x, y): - if len(self.__x_samples) < 2: + if len(self._x_samples) < 2: return 0.0 - a1 = (self.__x_samples[0], self.__y_samples[0]) - a2 = (self.__x_samples[1], self.__y_samples[1]) - b1 = (self.__x_samples[-1], self.__y_samples[-1]) + a1 = (self._x_samples[0], self._y_samples[0]) + a2 = (self._x_samples[1], self._y_samples[1]) + b1 = (self._x_samples[-1], self._y_samples[-1]) b2 = (x, y) diff = atan2(b2[1]-b1[1], b2[0]-b1[0]) - atan2(a2[1]-a1[1], a2[0]-a1[0]) return (degrees(diff) + 180) % 360 - 180 @@ -96,13 +96,13 @@ def get_velocity(self, x, y, timestamp): return abs(dist / (timestamp - self._last_sample_time)) def dispersion(self, x, y): - dx = max(self.__x_samples + [x]) - min(self.__x_samples + [x]) - dy = max(self.__y_samples + [y]) - min(self.__y_samples + [y]) + dx = max(self._x_samples + [x]) - min(self._x_samples + [x]) + dy = max(self._y_samples + [y]) - min(self._y_samples + [y]) return dx + dy @property def start_time(self): - return self.__start + return self._start @property def end_time(self): @@ -110,15 +110,15 @@ def end_time(self): @property def start_gaze(self): - return (self.__x_samples[0], self.__y_samples[0]) + return (self._x_samples[0], self._y_samples[0]) @property def end_gaze(self): - return (self.__x_samples[-1], self.__y_samples[-1]) + return (self._x_samples[-1], self._y_samples[-1]) @property def avg_gaze(self): - return (mean(self.__x_samples), mean(self.__y_samples)) + return (mean(self._x_samples), mean(self._y_samples)) class GazeEventGenerator(): diff --git a/klibs/KLParams.py b/klibs/KLParams.py index 7114fa71..e748f3d4 100755 --- a/klibs/KLParams.py +++ b/klibs/KLParams.py @@ -61,6 +61,7 @@ # Eye Tracking Settings eye_tracking = False eye_tracker_available = False +eye_tracker_name = "NA" manual_eyelink_setup = False manual_eyelink_recording = False show_gaze_dot = False # overridden by dm_show_gaze_dot if in development mode diff --git a/klibs/KLRuntimeInfo.py b/klibs/KLRuntimeInfo.py index e16c59b7..6749532d 100644 --- a/klibs/KLRuntimeInfo.py +++ b/klibs/KLRuntimeInfo.py @@ -193,8 +193,7 @@ def runtime_info_init(): info['condition'] = P.condition if P.eye_tracking: - from klibs.KLEnvironment import el - info['eyetracker'] = el.version if el.initialized else 'NA' + info['eyetracker'] = P.eye_tracker_name info['el_velocity_thresh'] = P.saccadic_velocity_threshold info['el_accel_thresh'] = P.saccadic_acceleration_threshold info['el_motion_thresh'] = P.saccadic_motion_threshold diff --git a/klibs/tests/test_KLEyeTracking.py b/klibs/tests/test_KLEyeTracking.py index d09fbed2..0e176d70 100644 --- a/klibs/tests/test_KLEyeTracking.py +++ b/klibs/tests/test_KLEyeTracking.py @@ -3,7 +3,7 @@ from klibs.KLConstants import * from klibs.KLBoundary import RectangleBoundary from klibs.KLEyeTracking.utils import GazeEventGenerator -from klibs.KLEyeTracking.KLTryLink import TryLink +from klibs.KLEyeTracking import TryLink, GazeSample class CustomTryLink(TryLink): @@ -168,6 +168,56 @@ def test_get_event_queue(self): start_fix = e assert sample assert start_fix + # Test the generation of EyeEvents + tst.add_samples([ + (402, 298, 40), (None, None, 50), (405, 304, 78) + ]) + events = [] + while len(tst.samples): + # Needs to be called once per sample + events += tst.get_event_queue([FIXATION_END, BLINK_START, BLINK_END]) + event_log = 0 + for e in events: + if e.type == FIXATION_END: + assert e.start_time == 0 + assert e.end_time == 40 + assert e.start_gaze == (400, 300) + assert e.end_gaze == (402, 298) + assert e.average_gaze == (401, 299) + event_log += 1 + elif e.type == BLINK_START: + assert e.start_time == 50 + event_log += 10 + elif e.type == BLINK_END: + assert e.start_time == 50 + assert e.end_time == 78 + assert e.duration == 28 + event_log += 100 + assert event_log == 111 + # Test gaze sample attributes + tst.add_samples([(404, 305, 80)]) + events = tst.get_event_queue([GAZE_SAMPLE]) + assert events[0].type == GAZE_SAMPLE + assert events[0].timestamp == 80 + assert events[0].gaze == (404, 305) + assert events[0].pupil_size == 2000 + assert not events[0].binocular + assert not events[0].left + assert events[0].right + assert events[0].right.gaze == (404, 305) + assert events[0].right.pupil_size == 2000 + + def test_get_sample(self): + tst = CustomTryLink() + # Start recording and check for gaze sample + start fixation + tst.start(1) + tst.add_samples([(400, 300, 15)]) + s = tst.get_sample() + assert isinstance(s, GazeSample) + assert s.timestamp == 15 + assert s.gaze == (400, 300) + assert s.right + assert not s.left def test_within_boundary(self): bounds = RectangleBoundary('test', (500, 400), (600, 500)) @@ -249,3 +299,25 @@ def test_saccade_from_boundary(self): events = tst.get_event_queue() assert tst.saccade_from_boundary('fixation', event_queue=events) + def test_saccade_in_direction(self): + tst = CustomTryLink() + # Start recording and make sure no saccade detected + tst.start(1) + tst.add_samples([(400, 300, 0)]) + events = tst.get_event_queue() + assert len(events) > 0 + # Saccade from boundary and re-test + tst.add_samples([ + (401, 301, 10), (420, 320, 20), (550, 440, 30), (551, 439, 40) + ]) + while len(tst.samples): + # Needs to be called once per sample + events = tst.get_event_queue() + assert tst.saccade_in_direction('down', events) + assert tst.saccade_in_direction('right', events) + assert not tst.saccade_in_direction('up', events) + assert not tst.saccade_in_direction('left', events) + assert tst.saccade_in_direction(['down', 'right'], events) + assert not tst.saccade_in_direction(['down', 'left'], events) + assert not tst.saccade_in_direction(['up', 'right'], events) + assert not tst.saccade_in_direction(['up', 'left'], events)