path: root/src/argaze/utils/contexts/File.py

"""Define eye tracking data file context"""

"""
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 <https://www.gnu.org/licenses/>.
"""

__author__ = "Théo de la Hogue"
__credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"

import logging
import time
import math
import threading
import ast

import pandas
import numpy

from argaze import ArFeatures, DataFeatures, GazeFeatures
from argaze.utils import UtilsFeatures


class CSV(ArFeatures.DataPlaybackContext):

	@DataFeatures.PipelineStepInit
	def __init__(self, **kwargs):

		# Init ArContext class
		super().__init__()

		# Init private attributes
		self.__path = None
		self.__separator = ','

		self.__timestamp_column = None
		self.__x_column = None
		self.__y_column = None
		self.__xy_column = None
		self.__left_eye_x_column = None
		self.__left_eye_y_column = None
		self.__left_eye_validity_column = None
		self.__right_eye_x_column = None
		self.__right_eye_y_column = None
		self.__right_eye_validity_column = None

		self.__rescale_to_pipeline_size = False

		self.__start = math.nan
		self.__end = math.nan
		self.__duration = 0.
		self.__progression = 0.

	@property
	def path(self) -> str:
		"""Path to data file."""
		return self.__path

	@path.setter
	def path(self, path: str):

		self.__path = path

	@property
	def separator(self) -> str:
		"""Value delimiter character"""
		return self.__separator

	@separator.setter
	def separator(self, separator: str):

		self.__separator = separator

	@property
	def timestamp_column(self) -> str:
		"""Timestamp column name in file. 
		!!! warning
		    This column name is mandatory.
		"""
		return self.__timestamp_column

	@timestamp_column.setter
	def timestamp_column(self, timestamp_column: str):

		self.__timestamp_column = timestamp_column

	@property
	def x_column(self) -> str:
		"""X coordinate column name.
		!!! note
		    This column name is required in the case where gaze position coordinates are splitted in two separated columns.
		"""
		return self.__x_column

	@x_column.setter
	def x_column(self, x_column: str):

		self.__x_column = x_column

	@property
	def y_column(self) -> str:
		"""Y coordinate column name.
		!!! note
		    This column name is required in the case where gaze position coordinates are splitted in two separated columns.
		"""
		return self.__y_column

	@y_column.setter
	def y_column(self, y_column: str):

		self.__y_column = y_column

	@property
	def xy_column(self) -> str:
		"""X and Y coordinate column name.
		!!! note
		    This column name is required in the case where gaze position coordinates are joined as a list in one single column.
		"""
		return self.__xy_column

	@xy_column.setter
	def xy_column(self, xy_column: str):

		self.__xy_column = xy_column

	@property
	def left_eye_x_column(self) -> str:
		"""X coordinate for left eye column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__left_eye_x_column

	@left_eye_x_column.setter
	def left_eye_x_column(self, left_eye_x_column: str):

		self.__left_eye_x_column = left_eye_x_column

	@property
	def left_eye_y_column(self) -> str:
		"""Y coordinate for left eye column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__left_eye_y_column

	@left_eye_y_column.setter
	def left_eye_y_column(self, left_eye_y_column: str):

		self.__left_eye_y_column = left_eye_y_column

	@property
	def left_eye_validity_column(self) -> str:
		"""Validity of left eye coordinate column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__left_eye_validity_column

	@left_eye_validity_column.setter
	def left_eye_validity_column(self, left_eye_validity_column: str):

		self.__left_eye_validity_column = left_eye_validity_column

	@property
	def right_eye_x_column(self) -> str:
		"""X coordinate for right eye column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__right_eye_x_column

	@right_eye_x_column.setter
	def right_eye_x_column(self, right_eye_x_column: str):

		self.__right_eye_x_column = right_eye_x_column

	@property
	def right_eye_y_column(self) -> str:
		"""Y coordinate for right eye column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__right_eye_y_column

	@right_eye_y_column.setter
	def right_eye_y_column(self, right_eye_y_column: str):

		self.__right_eye_y_column = right_eye_y_column

	@property
	def right_eye_validity_column(self) -> str:
		"""Validity of right eye coordinate column name.
		!!! note
		    This column name is required in the case where gaze position coordinates and validity are given for each eye in six separated columns.
		"""
		return self.__right_eye_validity_column

	@right_eye_validity_column.setter
	def right_eye_validity_column(self, right_eye_validity_column: str):

		self.__right_eye_validity_column = right_eye_validity_column

	@property
	def rescale_to_pipeline_size(self) -> bool:
		"""Rescale gaze positions to pipeline size."""
		return self.__rescale_to_pipeline_size

	@rescale_to_pipeline_size.setter 
	def rescale_to_pipeline_size(self, rescale: bool):

		self.__rescale_to_pipeline_size = rescale

	@property
	def start(self) -> int|float:
		"""Start reading timestamp."""
		return self.__start

	@start.setter
	def start(self, start: int|float):

		self.__start = start

	@property
	def end(self) -> int|float:
		"""End reading timestamp."""
		return self.__end

	@end.setter
	def end(self, end: int|float):

		self.__end = end

	@property
	def duration(self) -> int|float:
		"""Get data duration."""

		return self.__duration

	@property
	def progression(self) -> float:
		"""Get data processing progression between 0 and 1."""

		return self.__progression

	@DataFeatures.PipelineStepEnter
	def __enter__(self):

		logging.info('CSV file context starts...')

		# Select data to load from CSV file
		data_columns = []
		data_types = {}
		data_converters = {}

		# Select timestamp column
		if self.__timestamp_column:

			data_columns.append(self.__timestamp_column)
			data_types[self.__timestamp_column] = numpy.float64

		else:

			logging.error('Missing timestamp column name')

		# Case where gaze position coordinates are splitted in two separated columns
		if self.__x_column and self.__y_column:

			data_columns.append(self.__x_column)
			data_types[self.__x_column] = numpy.float64

			data_columns.append(self.__y_column)
			data_types[self.__y_column] = numpy.float64

		# Case where gaze position coordinates are joined as a list in one single column
		elif self.__xy_column:

			data_columns.append(self.__xy_column)
			data_converters[self.__xy_column] = ast.literal_eval

		# Case where gaze position coordinates and validity are given for each eye in six separated columns.
		elif self.__left_eye_x_column and self.__left_eye_y_column and self.__left_eye_validity_column and self.__right_eye_x_column and self.__right_eye_y_column and self.__right_eye_validity_column:

			data_columns.append(self.__left_eye_x_column)
			data_types[self.__left_eye_x_column] = numpy.float64

			data_columns.append(self.__left_eye_y_column)
			data_types[self.__left_eye_y_column] = numpy.float64

			data_columns.append(self.__left_eye_validity_column)
			data_types[self.__left_eye_validity_column] = numpy.bool_

			data_columns.append(self.__right_eye_x_column)
			data_types[self.__right_eye_x_column] = numpy.float64

			data_columns.append(self.__right_eye_y_column)
			data_types[self.__right_eye_y_column] = numpy.float64

			data_columns.append(self.__right_eye_validity_column)
			data_types[self.__right_eye_validity_column] = numpy.bool_

		dataframe = pandas.read_csv(self.__path, delimiter = self.__separator, low_memory = False, usecols=data_columns, dtype=data_types, converters=data_converters)

		# Optionnaly select a time range
		if not math.isnan(self.start):

			dataframe = dataframe.loc[(dataframe[self.__timestamp_column] >= self.start)]

		if not math.isnan(self.end):

			dataframe = dataframe.loc[(dataframe[self.__timestamp_column] <= self.end)]

		# Transform dataframe to a get timestamp, x, y, validity columns

		# First, rename timestamp column
		dataframe.rename(columns={self.__timestamp_column: 'timestamp'}, inplace=True)

		# Case where gaze position coordinates are splitted in two separated columns
		if self.__x_column and self.__y_column:

			dataframe.rename(columns={self.__x_column: 'x',self.__y_column: 'y'}, inplace=True)

		# Case where gaze position coordinates are joined as a list in one single column
		elif self.__xy_column:

			dataframe[['x','y']] = pandas.DataFrame(dataframe[self.__xy_column].to_list(), columns=['x','y'])

		# Case where gaze position coordinates and validity are given for each eye in six separated columns.
		elif self.__left_eye_x_column and self.__left_eye_y_column and self.__left_eye_validity_column and self.__right_eye_x_column and self.__right_eye_y_column and self.__right_eye_validity_column:

			# Edit gaze position X coordinate as the mean of left and right eyes positions if both are valid else select the valid one
			dataframe['x'] = numpy.where(
				dataframe[self.__left_eye_validity_column] & dataframe[self.__right_eye_validity_column],
				(dataframe[self.__left_eye_x_column] + dataframe[self.__right_eye_x_column]) / 2,
				numpy.where(
					dataframe[self.__left_eye_validity_column],
					dataframe[self.__left_eye_x_column],
					numpy.where(
						dataframe[self.__right_eye_validity_column],
						dataframe[self.__right_eye_x_column],
						None
					)
				)
			)

			# Edit gaze position Y coordinate as the mean of left and right eyes positions if both are valid else select the valid one
			dataframe['y'] = numpy.where(
				dataframe[self.__left_eye_validity_column] & dataframe[self.__right_eye_validity_column],
				(dataframe[self.__left_eye_y_column] + dataframe[self.__right_eye_y_column]) / 2,
				numpy.where(
					dataframe[self.__left_eye_validity_column],
					dataframe[self.__left_eye_y_column],
					numpy.where(
						dataframe[self.__right_eye_validity_column],
						dataframe[self.__right_eye_y_column],
						None
					)
				)
			)

		# Mark unvalid gaze positions
		dataframe[['validity']] = pandas.DataFrame(dataframe['x'].notna() & dataframe['y'].notna())
		unvalid_count = len(dataframe[dataframe['validity'] == False])

		if unvalid_count:
			logging.info('%i unvalid gaze positions have been detected', unvalid_count)

		# Scale valid gaze positions to pipeline size
		if self.__rescale_to_pipeline_size:

			dataframe.loc[dataframe['validity'], 'x'] *= self.pipeline.size[0]
			dataframe.loc[dataframe['validity'], 'y'] *= self.pipeline.size[1]

			logging.info('gaze positions have been scaled to pipeline size')

		# Open reading thread
		self.__reading_thread = threading.Thread(target=self.__read, kwargs={'dataframe': dataframe})

		logging.debug('> starting reading thread...')
		self.__reading_thread.start()

	@DataFeatures.PipelineStepExit
	def __exit__(self, exception_type, exception_value, exception_traceback):

		logging.info('CSV file context stops...')

		# Close data stream
		self.stop()

		# Stop reading thread
		threading.Thread.join(self.__reading_thread)

	def __read(self, dataframe: pandas.DataFrame):
		"""Read and process gaze positions from dataframe."""

		# Setup start and end dates
		start_ts = dataframe.iloc[0].timestamp if math.isnan(self.start) else self.start
		end_ts = dataframe.iloc[-1].timestamp if math.isnan(self.end) else self.end

		self.__duration = end_ts - start_ts
		self.__progression = 0.

		logging.info('Reading %i gaze positions from %f to %f', len(dataframe), start_ts, end_ts)

		for index, row in dataframe.iterrows():

			# Stop reading
			if not self.is_running():

				break

			# Pause reading
			while self.is_paused() and self.is_running():

				time.sleep(0.1)

			# Process gaze position
			self._process_gaze_position(x = row['x'], y = row['y'], timestamp = row['timestamp'])

			# Update progression
			self.__progression = (row['timestamp'] - start_ts) / self.__duration