Adding demo_loggers.py file.

1 files changed, 84 insertions, 0 deletions

diff --git a/src/argaze/utils/demo_data/demo_loggers.py b/src/argaze/utils/demo_data/demo_loggers.py
new file mode 100644
index 0000000..5f1986e
--- /dev/null
+++ b/src/argaze/utils/demo_data/demo_loggers.py
@@ -0,0 +1,84 @@
+""" 
+
+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/>.
+"""
+
+__author__ = "Théo de la Hogue"
+__credits__ = []
+__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
+__license__ = "GPLv3"
+
+from argaze import DataFeatures, GazeFeatures
+from argaze.GazeAnalysis import *
+from argaze.utils import UtilsFeatures
+
+class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter):
+
+	def on_look(self, timestamp, frame, exception):
+		"""Log frame fixations."""
+
+		# Log fixations
+		if GazeFeatures.is_fixation(frame.last_gaze_movement()) and frame.last_gaze_movement().is_finished():
+
+			log = (
+				timestamp, 
+				frame.last_gaze_movement().focus, 
+				frame.last_gaze_movement().duration, 
+				frame.layers['demo_layer'].last_looked_aoi_name()
+			)
+
+			self.write(log)
+
+class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter):
+
+	def on_look(self, timestamp, frame, exception):
+		"""Log frame scan path metrics."""
+
+		if frame.is_analysis_available():
+
+			analysis = frame.analysis()
+
+			log = (
+				timestamp, 
+				analysis[Basic.ScanPathAnalyzer].path_duration, 
+				analysis[Basic.ScanPathAnalyzer].steps_number, 
+				analysis[KCoefficient.ScanPathAnalyzer].K, 
+				analysis[NearestNeighborIndex.ScanPathAnalyzer].nearest_neighbor_index, 
+				analysis[ExploreExploitRatio.ScanPathAnalyzer].explore_exploit_ratio
+			)
+
+			self.write(log)
+
+class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter):
+
+		def on_look(self, timestamp, frame, exception):
+			"""Write frame image."""
+
+			self.write(frame.image())
+
+class AOIScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter):
+
+	def on_look(self, timestamp, layer, exception):
+		"""Log layer aoi scan path metrics"""
+
+		if layer.is_analysis_available():
+
+			analysis = layer.analysis()
+
+			log = (
+				timestamp,
+				analysis[Basic.AOIScanPathAnalyzer].path_duration, 
+				analysis[Basic.AOIScanPathAnalyzer].steps_number, 
+				analysis[KCoefficient.AOIScanPathAnalyzer].K, 
+				analysis[LempelZivComplexity.AOIScanPathAnalyzer].lempel_ziv_complexity
+			)
+
+			self.write(log)