aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/argaze/utils/UtilsFeatures.py72
-rw-r--r--src/argaze/utils/demo_data/demo_frame_logger.py17
2 files changed, 78 insertions, 11 deletions
diff --git a/src/argaze/utils/UtilsFeatures.py b/src/argaze/utils/UtilsFeatures.py
index 0cb6d77..76d5f59 100644
--- a/src/argaze/utils/UtilsFeatures.py
+++ b/src/argaze/utils/UtilsFeatures.py
@@ -10,6 +10,8 @@ __license__ = "BSD"
from typing import Tuple
import time
import types
+import numpy
+import cv2
def printProgressBar (iteration:int, total:int, prefix:str = '', suffix:str = '', decimals:int = 1, length:int = 100, fill:str = '█', printEnd:str = "\r"):
"""
@@ -152,7 +154,7 @@ def tuple_to_string(t: tuple, separator: str = ", ") -> str:
return separator.join(f'\"{e}\"' for e in t)
class FileWriter():
- """Write into a file line by line.
+ """Write data into a file line by line.
Parameters:
path: File path where to write data.
@@ -185,11 +187,6 @@ class FileWriter():
print(header, file=self._file, flush=True)
- def __del__(self):
- """Close file."""
-
- self._file.close()
-
def write(self, log: str|tuple):
"""Write log as a new line into file.
@@ -203,4 +200,65 @@ class FileWriter():
log = tuple_to_string(log, self.separator)
# Write into file
- print(log, file=self._file, flush=True) \ No newline at end of file
+ print(log, file=self._file, flush=True)
+
+ def __del__(self):
+ """Close file."""
+
+ self._file.close()
+
+class VideoWriter():
+ """Write images into a file using ffmpeg.
+
+ Parameters:
+ path: File path where to write images.
+ width: video horizontal resolution.
+ height: video vertical resolution.
+ fps: frame per second.
+ """
+
+ def __init__(self, path: str, width: int, height: int, fps: int):
+ """Open ffmpeg application as sub-process.
+ FFmpeg input PIPE: RAW images in BGR color format
+ FFmpeg output MP4 file encoded with HEVC codec.
+
+ Arguments list:
+ -y Overwrite output file without asking
+ -s {width}x{height} Input resolution width x height (1344x756)
+ -pixel_format bgr24 Input frame color format is BGR with 8 bits per color component
+ -f rawvideo Input format: raw video
+ -r {fps} Frame rate: fps
+ -i pipe: ffmpeg input is a PIPE
+ -vcodec libx265 Video codec: H.265 (HEVC)
+ -pix_fmt yuv420p Output video color space YUV420 (saving space compared to YUV444)
+ -crf 24 Constant quality encoding (lower value for higher quality and larger output file).
+ {output_filename} Output file name: output_filename (output.mp4)
+ """
+
+ import subprocess as sp
+ import shlex
+
+ self.__width = width
+ self.__height = height
+
+ self.__process = sp.Popen(shlex.split(f'ffmpeg -hide_banner -loglevel error -y -s {width}x{height} -pixel_format bgr24 -f rawvideo -r {fps} -i pipe: -vcodec libx265 -x265-params log-level=error -pix_fmt yuv420p -crf 24 {path}'), stdin=sp.PIPE)
+
+ def write(self, image: numpy.array):
+ """Write raw video frame to input stream of ffmpeg sub-process."""
+
+ # Resize image to adapt to video resolution
+ output = cv2.resize(image, dsize=(self.__width, self.__height), interpolation=cv2.INTER_LINEAR)
+
+ self.__process.stdin.write(output.tobytes())
+
+ def __del__(self):
+
+ # Close and flush stdin
+ self.__process.stdin.close()
+
+ # Wait for sub-process to finish
+ self.__process.wait()
+
+ # Terminate the sub-process
+ # Note: We don't have to terminate the sub-process (after process.wait(), the sub-process is supposed to be closed).
+ self.__process.terminate() \ No newline at end of file
diff --git a/src/argaze/utils/demo_data/demo_frame_logger.py b/src/argaze/utils/demo_data/demo_frame_logger.py
index 1c8046a..3448492 100644
--- a/src/argaze/utils/demo_data/demo_frame_logger.py
+++ b/src/argaze/utils/demo_data/demo_frame_logger.py
@@ -13,7 +13,7 @@ from argaze.utils import UtilsFeatures
class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter):
def on_look(self, timestamp, frame):
- """Log fixations"""
+ """Log fixations."""
# Log fixations
if GazeFeatures.is_fixation(frame.last_gaze_movement) and frame.last_gaze_movement.finished:
@@ -30,7 +30,7 @@ class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter
class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter):
def on_look(self, timestamp, frame):
- """Log scan path metrics"""
+ """Log scan path metrics."""
if frame.analysis_available:
@@ -45,8 +45,17 @@ class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.Fi
self.write(log)
+class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter):
+
+ def on_look(self, timestamp, frame):
+ """Write frame image."""
+
+ self.write(frame.image())
+
# Export loggers instances to register them as pipeline step object observers
__observers__ = {
"Fixation logger": FixationLogger(path="_export/logs/fixations.csv", header="Timestamp (ms), Focus (px), Duration (ms), AOI"),
- "Scan path analysis logger": ScanPathAnalysisLogger(path="_export/logs/scan_path_metrics.csv", header="Timestamp (ms), Duration (ms), Step, K, NNI, XXR")
- } \ No newline at end of file
+ "Scan path analysis logger": ScanPathAnalysisLogger(path="_export/logs/scan_path_metrics.csv", header="Timestamp (ms), Duration (ms), Step, K, NNI, XXR"),
+ "Video recorder": VideoRecorder(path="_export/logs/video.mp4", width=1920, height=1080, fps=15)
+ }
+