Add missing files

constants.py

@@ -0,0 +1,7 @@
+TIME_ALIASES = ['time', 'timestamp']
+LEFT_X_ALIASES = ['l por x [px]']
+LEFT_Y_ALIASES = ['l por y [px]']
+RIGHT_X_ALIASES = ['r por x [px]']
+RIGHT_Y_ALIASES = ['r por y [px]']
+X_ALIASES = ['por x [px]', 'x']
+Y_ALIASES = ['por y [px]', 'y']

detectors.py

@@ -0,0 +1,267 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of PyGaze - the open-source toolbox for eye tracking
+#
+#	PyGazeAnalyser is a Python module for easily analysing eye-tracking data
+#	Copyright (C) 2014  Edwin S. Dalmaijer
+#
+#	This program is free software: you can redistribute it and/or modify
+#	it under the terms of the GNU General Public License as published by
+#	the Free Software Foundation, either version 3 of the License, or
+#	(at your option) any later version.
+#
+#	This program is distributed in the hope that it will be useful,
+#	but WITHOUT ANY WARRANTY; without even the implied warranty of
+#	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#	GNU General Public License for more details.
+#
+#	You should have received a copy of the GNU General Public License
+#	along with this program.  If not, see <http://www.gnu.org/licenses/>
+
+# EyeTribe Reader
+#
+# Reads files as produced by PyTribe (https://github.com/esdalmaijer/PyTribe),
+# and performs a very crude fixation and blink detection: every sample that
+# is invalid (usually coded '0.0') is considered to be part of a blink, and
+# every sample in which the gaze movement velocity is below a threshold is
+# considered to be part of a fixation. For optimal event detection, it would be
+# better to use a different algorithm, e.g.:
+# Nystrom, M., & Holmqvist, K. (2010). An adaptive algorithm for fixation,
+# saccade, and glissade detection in eyetracking data. Behavior Research
+# Methods, 42, 188-204. doi:10.3758/BRM.42.1.188
+#
+# (C) Edwin Dalmaijer, 2014
+# [email protected]
+#
+# version 1 (01-Jul-2014)
+
+__author__ = "Edwin Dalmaijer"
+
+import numpy
+
+
+def blink_detection(x, y, time, missing=0.0, minlen=10):
+    """Detects blinks, defined as a period of missing data that lasts for at
+	least a minimal amount of samples
+	
+	arguments
+
+	x		-	numpy array of x positions
+	y		-	numpy array of y positions
+	time		-	numpy array of EyeTribe timestamps
+
+	keyword arguments
+
+	missing	-	value to be used for missing data (default = 0.0)
+	minlen	-	integer indicating the minimal amount of consecutive
+				missing samples
+	
+	returns
+	Sblk, Eblk
+				Sblk	-	list of lists, each containing [starttime]
+				Eblk	-	list of lists, each containing [starttime, endtime, duration]
+	"""
+
+    # empty list to contain data
+    Sblk = []
+    Eblk = []
+
+    # check where the missing samples are
+    mx = numpy.array(x == missing, dtype=int)
+    my = numpy.array(y == missing, dtype=int)
+    miss = numpy.array((mx + my) == 2, dtype=int)
+
+    # check where the starts and ends are (+1 to counteract shift to left)
+    diff = numpy.diff(miss)
+    starts = numpy.where(diff == 1)[0] + 1
+    ends = numpy.where(diff == -1)[0] + 1
+
+    # compile blink starts and ends
+    for i in range(len(starts)):
+        # get starting index
+        s = starts[i]
+        # get ending index
+        if i < len(ends):
+            e = ends[i]
+        elif len(ends) > 0:
+            e = ends[-1]
+        else:
+            e = -1
+        # append only if the duration in samples is equal to or greater than
+        # the minimal duration
+        if e - s >= minlen:
+            # add starting time
+            Sblk.append([time[s]])
+            # add ending time
+            Eblk.append([time[s], time[e], time[e] - time[s]])
+
+    return Sblk, Eblk
+
+
+def remove_missing(x, y, time, missing):
+    mx = numpy.array(x == missing, dtype=int)
+    my = numpy.array(y == missing, dtype=int)
+    x = x[(mx + my) != 2]
+    y = y[(mx + my) != 2]
+    time = time[(mx + my) != 2]
+    return x, y, time
+
+
+def fixation_detection(x, y, time, missing=0.0, maxdist=25, mindur=50):
+    """Detects fixations, defined as consecutive samples with an inter-sample
+	distance of less than a set amount of pixels (disregarding missing data)
+	
+	arguments
+
+	x		-	numpy array of x positions
+	y		-	numpy array of y positions
+	time		-	numpy array of EyeTribe timestamps
+
+	keyword arguments
+
+	missing	-	value to be used for missing data (default = 0.0)
+	maxdist	-	maximal inter sample distance in pixels (default = 25)
+	mindur	-	minimal duration of a fixation in milliseconds; detected
+				fixation cadidates will be disregarded if they are below
+				this duration (default = 100)
+	
+	returns
+	Sfix, Efix
+				Sfix	-	list of lists, each containing [starttime]
+				Efix	-	list of lists, each containing [starttime, endtime, duration, endx, endy]
+	"""
+
+    x, y, time = remove_missing(x, y, time, missing)
+
+    # empty list to contain data
+    Sfix = []
+    Efix = []
+
+    # loop through all coordinates
+    si = 0
+    fixstart = False
+    for i in range(1, len(x)):
+        # calculate Euclidean distance from the current fixation coordinate
+        # to the next coordinate
+        squared_distance = ((x[si] - x[i]) ** 2 + (y[si] - y[i]) ** 2)
+        dist = 0.0
+        if squared_distance > 0:
+            dist = squared_distance ** 0.5
+        # check if the next coordinate is below maximal distance
+        if dist <= maxdist and not fixstart:
+            # start a new fixation
+            si = 0 + i
+            fixstart = True
+            Sfix.append([time[i]])
+        elif dist > maxdist and fixstart:
+            # end the current fixation
+            fixstart = False
+            # only store the fixation if the duration is ok
+            if time[i - 1] - Sfix[-1][0] >= mindur:
+                Efix.append([Sfix[-1][0], time[i - 1], time[i - 1] - Sfix[-1][0], x[si], y[si]])
+            # delete the last fixation start if it was too short
+            else:
+                Sfix.pop(-1)
+            si = 0 + i
+        elif not fixstart:
+            si += 1
+    # add last fixation end (we can lose it if dist > maxdist is false for the last point)
+    if len(Sfix) > len(Efix):
+        Efix.append([Sfix[-1][0], time[len(x) - 1], time[len(x) - 1] - Sfix[-1][0], x[si], y[si]])
+    return Sfix, Efix
+
+
+def saccade_detection(x, y, time, missing=0.0, minlen=5, maxvel=40, maxacc=340):
+    """Detects saccades, defined as consecutive samples with an inter-sample
+	velocity of over a velocity threshold or an acceleration threshold
+	
+	arguments
+
+	x		-	numpy array of x positions
+	y		-	numpy array of y positions
+	time		-	numpy array of tracker timestamps in milliseconds
+
+	keyword arguments
+
+	missing	-	value to be used for missing data (default = 0.0)
+	minlen	-	minimal length of saccades in milliseconds; all detected
+				saccades with len(sac) < minlen will be ignored
+				(default = 5)
+	maxvel	-	velocity threshold in pixels/second (default = 40)
+	maxacc	-	acceleration threshold in pixels / second**2
+				(default = 340)
+	
+	returns
+	Ssac, Esac
+			Ssac	-	list of lists, each containing [starttime]
+			Esac	-	list of lists, each containing [starttime, endtime, duration, startx, starty, endx, endy]
+	"""
+    x, y, time = remove_missing(x, y, time, missing)
+
+    # CONTAINERS
+    Ssac = []
+    Esac = []
+
+    # INTER-SAMPLE MEASURES
+    # the distance between samples is the square root of the sum
+    # of the squared horizontal and vertical interdistances
+    intdist = (numpy.diff(x) ** 2 + numpy.diff(y) ** 2) ** 0.5
+    # get inter-sample times
+    inttime = numpy.diff(time)
+    # recalculate inter-sample times to seconds
+    inttime = inttime / 1000.0
+
+    # VELOCITY AND ACCELERATION
+    # the velocity between samples is the inter-sample distance
+    # divided by the inter-sample time
+    vel = intdist / inttime
+    # the acceleration is the sample-to-sample difference in
+    # eye movement velocity
+    acc = numpy.diff(vel)
+
+    # SACCADE START AND END
+    t0i = 0
+    stop = False
+    while not stop:
+        # saccade start (t1) is when the velocity or acceleration
+        # surpass threshold, saccade end (t2) is when both return
+        # under threshold
+
+        # detect saccade starts
+        sacstarts = numpy.where((vel[1 + t0i:] > maxvel).astype(int) + (acc[t0i:] > maxacc).astype(int) >= 1)[0]
+        if len(sacstarts) > 0:
+            # timestamp for starting position
+            t1i = t0i + sacstarts[0] + 1
+            if t1i >= len(time) - 1:
+                t1i = len(time) - 2
+            t1 = time[t1i]
+
+            # add to saccade starts
+            Ssac.append([t1])
+
+            # detect saccade endings
+            sacends = numpy.where((vel[1 + t1i:] < maxvel).astype(int) + (acc[t1i:] < maxacc).astype(int) == 2)[0]
+            if len(sacends) > 0:
+                # timestamp for ending position
+                t2i = sacends[0] + 1 + t1i + 2
+                if t2i >= len(time):
+                    t2i = len(time) - 1
+                t2 = time[t2i]
+                dur = t2 - t1
+
+                # ignore saccades that did not last long enough
+                if dur >= minlen:
+                    # add to saccade ends
+                    Esac.append([t1, t2, dur, x[t1i], y[t1i], x[t2i], y[t2i]])
+                else:
+                    # remove last saccade start on too low duration
+                    Ssac.pop(-1)
+
+                # update t0i
+                t0i = 0 + t2i
+            else:
+                stop = True
+        else:
+            stop = True
+
+    return Ssac, Esac

eyetrack2saccade.py

@@ -0,0 +1,67 @@
+import pandas as pd
+import numpy as np
+from .constants import TIME_ALIASES, LEFT_X_ALIASES, LEFT_Y_ALIASES, RIGHT_X_ALIASES, RIGHT_Y_ALIASES, X_ALIASES, \
+    Y_ALIASES
+from .detectors import saccade_detection
+from typing import Union
+
+class Eye2SacExtractor:
+    data: pd.DataFrame = None
+    x: np.array = None
+    y: np.array = None
+    time: np.array = None
+
+    def _load_data(self, file_path: str):
+        if file_path.endswith('.csv'):
+            return pd.read_csv(file_path, sep=',')
+        elif file_path.endswith('.txt'):
+            return pd.read_csv(file_path, sep='\t')
+        else:
+            raise ValueError('File format not supported. Please provide a csv or txt file.')
+
+    def _clean_data(self):
+        self.data.dropna(inplace=True)
+
+    def _map_relevant_data(self):
+        # convert column names to lowercase
+        self.data.columns = self.data.columns.str.lower()
+
+        # map and extract relevant data
+        try:
+            self.time = self.data[self.data.columns.intersection(TIME_ALIASES)].to_numpy().flatten()
+
+            if self.data.columns.intersection(X_ALIASES).any() and self.data.columns.intersection(Y_ALIASES).any():
+                self.x = self.data[self.data.columns.intersection(X_ALIASES)].to_numpy().flatten()
+                self.y = self.data[self.data.columns.intersection(Y_ALIASES)].to_numpy().flatten()
+            else:
+                left_x = self.data[self.data.columns.intersection(LEFT_X_ALIASES)].to_numpy().flatten()
+                left_y = self.data[self.data.columns.intersection(LEFT_Y_ALIASES)].to_numpy().flatten()
+                right_x = self.data[self.data.columns.intersection(RIGHT_X_ALIASES)].to_numpy().flatten()
+                right_y = self.data[self.data.columns.intersection(RIGHT_Y_ALIASES)].to_numpy().flatten()
+                self._preprocess(left_x, left_y, right_x, right_y)
+
+        except KeyError:
+            raise ValueError('Required data columns are missing or not in the correct naming format.')
+
+    def _preprocess(self, left_x: np.array, left_y: np.array, right_x: np.array, right_y: np.array):
+        # combine left and right eye data into average value
+        self.x = np.mean([left_x, right_x], axis=0)
+        self.y = np.mean([left_y, right_y], axis=0)
+
+    def extract_features(self, data: Union[pd.DataFrame, str]):
+        if isinstance(data, pd.DataFrame):
+            self.data = data
+        elif isinstance(data, str):
+            self.data = self._load_data(data)
+        else:
+            raise ValueError('Data must be a pandas DataFrame or a file path to a csv or txt file.')
+        self._clean_data()
+        self._map_relevant_data()
+
+        return self._extract_features()
+
+
+    def _extract_features(self, missing: float = 0.0, minlen: int = 5, maxvel: int = 40, maxacc: int = 340) -> pd.DataFrame :
+        _, esac = saccade_detection(self.x, self.y, self.time, missing=missing, minlen=minlen, maxvel=maxvel, maxacc=maxacc)
+        esac_df = pd.DataFrame(esac, columns=['starttime', 'endtime', 'duration', 'startx', 'starty', 'endx', 'endy'])
+        return esac_df

eyetrack_2_saccade_pipeline.py

@@ -1,5 +1,5 @@
 from transformers import Pipeline
-from src.feature_extraction import Eye2SacExtractor
+from .eyetrack2saccade import Eye2SacExtractor
 
 class Eye2SacPipeline(Pipeline):
 
@@ -7,11 +7,9 @@ class Eye2SacPipeline(Pipeline):
     
     def _sanitize_parameters(self, **kwargs):
         preprocess_kwargs = {}
-        if "maybe_arg" in kwargs:
-            preprocess_kwargs["maybe_arg"] = kwargs["maybe_arg"]
         return preprocess_kwargs, {}, {}
 
-    def preprocess(self, inputs, maybe_arg=2):
+    def preprocess(self, inputs):
         return self.eye2SacExtractor.extract_features(inputs)
 
     def _forward(self, model_inputs):