path: root/src/argaze/TobiiGlassesPro2/TobiiController.py

#!/usr/bin/env python

import datetime
import uuid

from argaze.TobiiGlassesPro2 import TobiiNetworkInterface, TobiiData, TobiiVideo

TOBII_DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S+%f'
TOBII_DATETIME_FORMAT_HUMREAD = '%d/%m/%Y %H:%M:%S'

DEFAULT_PROJECT_NAME = 'DefaultProject'
DEFAULT_PARTICIPANT_NAME = 'DefaultParticipant'
DEFAULT_RECORD_NAME = 'DefaultRecord'

class TobiiController(TobiiNetworkInterface.TobiiNetworkInterface):
	"""Handle Tobii glasses Pro 2 device using network interface."""

	project_name = None
	"""Project name."""

	project_id = None
	"""Project identifier."""

	participant_name = None
	"""Participant name."""

	participant_id = None
	"""Participant identifier."""

	calibration_id = None
	"""Calibration identifier."""

	def __init__(self, ip_address, project_name = DEFAULT_PROJECT_NAME, participant_name = DEFAULT_PARTICIPANT_NAME):
		"""Create a project, a participant and start calibration."""

		super().__init__(ip_address)

		# bind to project or create one if it doesn't exist
		self.project_name = project_name
		self.project_id = self.set_project(self.project_name)

		# bind to participant or create one if it doesn't exist
		self.participant_name = participant_name
		self.participant_id = self.set_participant(self.project_id, self.participant_name)

		self.__recording_index = 0

		self.__data_stream = None
		self.__video_stream = None

		super().wait_for_status('/api/system/status', 'sys_status', ['ok']) == 'ok'

	def __get_current_datetime(self, timeformat=TOBII_DATETIME_FORMAT):
		return datetime.datetime.now().replace(microsecond=0).strftime(timeformat)

	# STREAMING FEATURES

	def enable_data_stream(self):
		"""Enable Tobii Glasses Pro 2 data streaming."""

		if self.__data_stream == None:
			self.__data_stream = TobiiData.TobiiDataStream(self)

		return self.__data_stream

	def enable_video_stream(self):
		"""Enable Tobii Glasses Pro 2 video camera streaming."""

		if self.__video_stream == None:
			self.__video_stream = TobiiVideo.TobiiVideoStream(self)
			
		return self.__video_stream

	def start_streaming(self):

		if self.__data_stream != None:
			self.__data_stream.open()

		if self.__video_stream != None:
			 self.__video_stream.open()

	def stop_streaming(self):

		if self.__data_stream != None:
			self.__data_stream.close()

		if self.__video_stream != None:
			 self.__video_stream.close()

	# PROJECT FEATURES

	def set_project(self, project_name = DEFAULT_PROJECT_NAME):
		"""Bind to a project or create one if it doesn't exist."""

		project_id = self.get_project_id(project_name)

		if project_id is None:

			data = {
				'pr_info' : {
					'CreationDate': self.__get_current_datetime(timeformat=TOBII_DATETIME_FORMAT_HUMREAD),
					'EagleId':  str(uuid.uuid5(uuid.NAMESPACE_DNS, project_name)),
					'Name': project_name
				},
				'pr_created': self.__get_current_datetime()
			}

			json_data = super().post_request('/api/projects', data)

			return json_data['pr_id']

		else:
			
			return project_id

	def get_project_id(self, project_name):

		project_id = None
		projects = super().get_request('/api/projects')

		for project in projects:

			try:
				if project['pr_info']['Name'] == project_name:
					project_id = project['pr_id']
			except:
				pass

		return project_id

	def get_projects(self):
		return super().get_request('/api/projects')

	# PARTICIPANT FEATURES

	def set_participant(self, project_id, participant_name = DEFAULT_PARTICIPANT_NAME, participant_notes = ''):
		"""Bind to a participant or create one if it doesn't exist."""

		participant_id = self.get_participant_id(participant_name)
		
		if participant_id is None:

			data = {
				'pa_project': project_id,
				'pa_info': { 
					'EagleId': str(uuid.uuid5(uuid.NAMESPACE_DNS, self.participant_name)),
					'Name': self.participant_name,
					'Notes': participant_notes
				},
				'pa_created': self.__get_current_datetime()
			}

			json_data = super().post_request('/api/participants', data)

			return json_data['pa_id']

		else:

			return participant_id

	def get_participant_id(self, participant_name):

		participant_id = None
		participants = super().get_request('/api/participants')

		for participant in participants:

			try:
				if participant['pa_info']['Name'] == participant_name:
					participant_id = participant['pa_id']

			except:
				pass

		return participant_id

	def get_participants(self):
		return super().get_request('/api/participants')

	# CALIBRATION

	def calibrate(self):
		"""Start Tobii glasses calibration for current project and participant."""

		input('Position Tobbi glasses calibration target then presse \'Enter\' to start calibration.')

		data = {
			'ca_project': self.project_id, 
			'ca_type': 'default',
			'ca_participant': self.participant_id,
			'ca_created': self.__get_current_datetime()
		}

		json_data = super().post_request('/api/calibrations', data)

		self.calibration_id = json_data['ca_id']

		super().post_request('/api/calibrations/' + self.calibration_id + '/start')

		status = super().wait_for_status('/api/calibrations/' + self.calibration_id + '/status', 'ca_state', ['calibrating', 'calibrated', 'stale', 'uncalibrated', 'failed'])

		if status == 'uncalibrated' or status == 'stale' or status == 'failed':
			raise Error(f'Tobii calibration {self.calibration_id} {status}')

	# RECORDING FEATURES

	def __wait_for_recording_status(self, recording_id, status_array = ['init', 'starting', 'recording', 'pausing', 'paused', 'stopping', 'stopped', 'done', 'stale', 'failed']):
		return super().wait_for_status('/api/recordings/' + recording_id + '/status', 'rec_state', status_array)

	def create_recording(self, participant_name, recording_notes = ''):

		participant_id = self.get_participant_id(participant_name)

		if participant_id is None:
			raise NameError(f'{participant_name} participant doesn\'t exist')

		self.__recording_index += 1
		recording_name = f'Recording_{self.__recording_index}'

		data = {
			'rec_participant': participant_id,
			'rec_info': {
				'EagleId': str(uuid.uuid5(uuid.NAMESPACE_DNS, participant_name)),
				'Name': recording_name,
				'Notes': recording_notes
			},
			'rec_created': self.__get_current_datetime()
		}

		json_data = super().post_request('/api/recordings', data)

		return json_data['rec_id']

	def start_recording(self, recording_id):
		"""Start recording on the Tobii interface's SD Card."""
		
		super().post_request('/api/recordings/' + recording_id + '/start')
		return self.__wait_for_recording_status(recording_id, ['recording']) == 'recording'

	def stop_recording(self, recording_id):
		"""Stop recording on the Tobii interface's SD Card."""

		super().post_request('/api/recordings/' + recording_id + '/stop')
		return self.__wait_for_recording_status(recording_id, ['done']) == "done"

	def pause_recording(self, recording_id):
		"""Pause recording on the Tobii interface's SD Card."""

		super().post_request('/api/recordings/' + recording_id + '/pause')
		return self.__wait_for_recording_status(recording_id, ['paused']) == "paused"

	def __get_recording_status(self):
		return self.get_status()['sys_recording']

	def get_current_recording_id(self):
		return self.__get_recording_status()['rec_id']

	def is_recording(self):

		rec_status = self.__get_recording_status()

		if rec_status != {}:
			if rec_status['rec_state'] == "recording":
				return True

		return False

	def get_recordings(self):
		return super().get_request('/api/recordings')

	# MISC

	def eject_sd(self):
		super().get_request('/api/eject')

	def get_battery_info(self):
		return ( "Battery info = [ Level: %.2f %% - Remaining Time: %.2f s ]" % (float(self.get_battery_level()), float(self.get_battery_remaining_time())) )

	def get_battery_level(self):
		return self.get_battery_status()['level']

	def get_battery_remaining_time(self):
		return self.get_battery_status()['remaining_time']

	def get_battery_status(self):
		return self.get_status()['sys_battery']

	def get_et_freq(self):
		return self.get_configuration()['sys_et_freq']

	def get_et_frequencies(self):
		return self.get_status()['sys_et']['frequencies']

	def identify(self):
		super().get_request('/api/identify')

	def get_address(self):
		return self.address

	def get_configuration(self):
		return super().get_request('/api/system/conf')

	def get_status(self):
		return super().get_request('/api/system/status')

	def get_storage_info(self):
		return ( "Storage info = [ Remaining Time: %.2f s ]" % float(self.get_battery_remaining_time()) )

	def get_storage_remaining_time(self):
		return self.get_storage_status()['remaining_time']

	def get_storage_status(self):
		return self.get_status()['sys_storage']

	def get_video_freq(self):
		return self.get_configuration()['sys_sc_fps']

	def send_custom_event(self, event_type, event_tag = ''):
		data = {'type': event_type, 'tag': event_tag}
		super().post_request('/api/events', data, wait_for_response=False)

	def send_experimental_var(self, variable_name, variable_value):
		self.send_custom_event('#%s#' % variable_name, variable_value)

	def send_experimental_vars(self, variable_names_list, variable_values_list):
		self.send_custom_event('@%s@' % str(variable_names_list), str(variable_values_list))

	def send_tobiipro_event(self, event_type, event_value):
		self.send_custom_event('JsonEvent', "{'event_type': '%s','event_value': '%s'}" % (event_type, event_value))

	def set_et_freq_50(self):
		data = {'sys_et_freq': 50}
		json_data = super().post_request('/api/system/conf', data)

	def set_et_freq_100(self):
		"""May not be available. Check get_et_frequencies() first."""
		data = {'sys_et_freq': 100}
		json_data = super().post_request('/api/system/conf', data)

	def set_et_indoor_preset(self):
		data = {'sys_sc_preset': 'Indoor'}
		json_data = super().post_request('/api/system/conf', data)

	def set_et_outdoor_preset(self):
		data = {'sys_ec_preset': 'ClearWeather'}
		json_data = super().post_request('/api/system/conf', data)

	def set_video_auto_preset(self):
		data = {'sys_sc_preset': 'Auto'}
		json_data = super().post_request('/api/system/conf', data)

	def set_video_gaze_preset(self):
		data = {'sys_sc_preset': 'GazeBasedExposure'}
		json_data = super().post_request('/api/system/conf', data)

	def set_video_freq_25(self):
		data = {'sys_sc_fps': 25}
		json_data = super().post_request('/api/system/conf/', data)

	def set_video_freq_50(self):
		data = {'sys_sc_fps': 50}
		json_data = super().post_request('/api/system/conf/', data)