diff options
Diffstat (limited to 'src')
26 files changed, 239 insertions, 153 deletions
diff --git a/src/argaze/ArFeatures.py b/src/argaze/ArFeatures.py index bb6a71b..1585780 100644 --- a/src/argaze/ArFeatures.py +++ b/src/argaze/ArFeatures.py @@ -106,7 +106,6 @@ class ArLayer(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # Init parent classes DataFeatures.SharedObject.__init__(self) - DataFeatures.PipelineStepObject.__init__(self) # Init private attributes self.__aoi_scene = None @@ -461,7 +460,6 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): # Init parent classes DataFeatures.SharedObject.__init__(self) - DataFeatures.PipelineStepObject.__init__(self) # Init private attributes self.__size = (1, 1) @@ -728,6 +726,20 @@ class ArFrame(DataFeatures.SharedObject, DataFeatures.PipelineStepObject): return d + @DataFeatures.PipelineStepEnter + def __enter__(self): + + for name, layer in self._layers.items(): + + layer.__enter__() + + @DataFeatures.PipelineStepExit + def __exit__(self, exception_type, exception_value, exception_traceback): + + for name, layer in self._layers.items(): + + layer.__exit__(exception_type, exception_value, exception_traceback) + @DataFeatures.PipelineStepMethod def look(self, timestamped_gaze_position: GazeFeatures.GazePosition = GazeFeatures.GazePosition()) -> Iterator[Union[object, type, dict]]: """ @@ -920,9 +932,6 @@ class ArScene(DataFeatures.PipelineStepObject): def __init__(self, **kwargs): """Initialize ArScene""" - # Init parent classes - super().__init__() - # Init private attributes self._layers = {} self.__frames = {} @@ -1116,7 +1125,7 @@ class ArCamera(ArFrame): def __init__(self, **kwargs): """Initialize ArCamera.""" - # Init parent class + # Init ArFrame class super().__init__() # Init private attributes @@ -1366,8 +1375,6 @@ class ArContext(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - DataFeatures.PipelineStepObject.__init__(self) - # Init private attributes self.__pipeline = None self.__exceptions = DataFeatures.TimestampedExceptions() @@ -1424,7 +1431,7 @@ class ArContext(DataFeatures.PipelineStepObject): return self - @DataFeatures.PipelineStepEnter + @DataFeatures.PipelineStepExit def __exit__(self, exception_type, exception_value, exception_traceback): """Exit from ArContext.""" pass diff --git a/src/argaze/ArUcoMarkers/ArUcoCamera.py b/src/argaze/ArUcoMarkers/ArUcoCamera.py index e84d71a..1371a38 100644 --- a/src/argaze/ArUcoMarkers/ArUcoCamera.py +++ b/src/argaze/ArUcoMarkers/ArUcoCamera.py @@ -47,7 +47,7 @@ class ArUcoCamera(ArFeatures.ArCamera): def __init__(self, **kwargs): """Initialize ArUcoCamera""" - # Init parent class + # Init ArCamera class super().__init__() # Init private attribute diff --git a/src/argaze/ArUcoMarkers/ArUcoDetector.py b/src/argaze/ArUcoMarkers/ArUcoDetector.py index b84030c..f135c1d 100644 --- a/src/argaze/ArUcoMarkers/ArUcoDetector.py +++ b/src/argaze/ArUcoMarkers/ArUcoDetector.py @@ -131,9 +131,6 @@ class ArUcoDetector(DataFeatures.PipelineStepObject): def __init__(self, **kwargs): """Initialize ArUcoDetector.""" - # Init parent class - super().__init__() - # Init private attributes self.__dictionary = None self.__optic_parameters = None @@ -327,15 +324,12 @@ class ArUcoDetector(DataFeatures.PipelineStepObject): return self.__board_corners -class Observer(DataFeatures.PipelineStepObserver): +class Observer(): """Define ArUcoDetector observer to count how many times detection succeeded and how many times markers are detected.""" - @DataFeatures.PipelineStepInit - def __init__(self, **kwargs): + def __init__(self): """Initialize marker detection metrics.""" - DataFeatures.PipelineStepObserver.__init__(self) - self.__try_count = 0 self.__success_count = 0 self.__detected_ids = [] diff --git a/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py b/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py index 36960f3..642b1d0 100644 --- a/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py +++ b/src/argaze/ArUcoMarkers/ArUcoMarkersGroup.py @@ -85,9 +85,6 @@ class ArUcoMarkersGroup(DataFeatures.PipelineStepObject): def __init__(self, **kwargs): """Initialize ArUcoMarkersGroup""" - # Init parent classes - super().__init__() - # Init private attributes self.__dictionary = None self.__places = {} diff --git a/src/argaze/ArUcoMarkers/ArUcoScene.py b/src/argaze/ArUcoMarkers/ArUcoScene.py index 078c9e2..dbad14d 100644 --- a/src/argaze/ArUcoMarkers/ArUcoScene.py +++ b/src/argaze/ArUcoMarkers/ArUcoScene.py @@ -35,7 +35,7 @@ class ArUcoScene(ArFeatures.ArScene): def __init__(self, **kwargs): """Initialize ArUcoScene""" - # Init parent classes + # Init ArScene classes super().__init__() # Init private attribute diff --git a/src/argaze/AreaOfInterest/AOIFeatures.py b/src/argaze/AreaOfInterest/AOIFeatures.py index a65400d..680397b 100644 --- a/src/argaze/AreaOfInterest/AOIFeatures.py +++ b/src/argaze/AreaOfInterest/AOIFeatures.py @@ -548,8 +548,6 @@ class Heatmap(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() - # Init private attributes self.__size = (1, 1) self.__buffer = 0 diff --git a/src/argaze/DataFeatures.py b/src/argaze/DataFeatures.py index 48a5b08..90501a3 100644 --- a/src/argaze/DataFeatures.py +++ b/src/argaze/DataFeatures.py @@ -627,9 +627,15 @@ def PipelineStepInit(method): Parameters: kwargs: any arguments defined by PipelineStepMethodInit. """ + + # Init class attributes method(self, **kwargs) - PipelineStepObject.update_attributes(self, kwargs) + # Init pipeline step object attributes + PipelineStepObject.__init__(self) + + # Update all attributes + self.update_attributes(kwargs) return wrapper @@ -639,21 +645,11 @@ def PipelineStepEnter(method): def wrapper(self): """Wrap pipeline step __enter__ method to call super, observers and children __enter__ method.""" - PipelineStepObject.__enter__(self) + logging.debug('%s.__enter__', type(self).__name__) method(self) - # Start children pipeline step objects - for child in self.children: - - child.__enter__() - - # Start observers - for observer in self.observers: - - observer.__enter__() - - return self + return PipelineStepObject.__enter__(self) return wrapper @@ -663,17 +659,9 @@ def PipelineStepExit(method): def wrapper(self, *args): """Wrap pipeline step __exit__ method to call super, observers and children __exit__ method.""" - PipelineStepObject.__exit__(self, *args) + logging.debug('%s.__exit__', type(self).__name__) - # Stop observers - for observer in self.observers: - - observer.__exit__(*args) - - # Stop children pipeline step objects - for child in self.children: - - child.__exit__(*args) + PipelineStepObject.__exit__(self, *args) method(self, *args) @@ -764,7 +752,7 @@ class PipelineStepObject(): Define class to assess pipeline step methods execution time and observe them. """ - def __init__(self, **kwargs): + def __init__(self): """Initialize PipelineStepObject.""" logging.debug('%s.__init__', type(self).__name__) @@ -777,20 +765,39 @@ class PipelineStepObject(): # Parent attribute will be setup later by parent it self self.__parent = None - # Update attributes - self.update_attributes(kwargs) - def __enter__(self): """Define default method to enter into pipeline step object context.""" - logging.debug('%s.__enter__', type(self).__name__) + # Start children pipeline step objects + for child in self.children: + + # DEBUG + print('ENTERING CHILD', type(child)) + + child.__enter__() + + # Start observers + for observer in self.observers: + + # DEBUG + print('ENTERING OBSERVER', type(observer)) + + observer.__enter__() return self def __exit__(self, exception_type, exception_value, exception_traceback): """Define default method to exit from pipeline step object context.""" - logging.debug('PipelineStepObject.__exit__') + # Stop observers + for observer in self.observers: + + observer.__exit__(exception_type, exception_value, exception_traceback) + + # Stop children pipeline step objects + for child in self.children: + + child.__exit__(exception_type, exception_value, exception_traceback) def update_attributes(self, object_data: dict): """Update pipeline step object attributes with dictionary.""" @@ -1060,28 +1067,3 @@ def PipelineStepMethod(method): return result return wrapper - -class PipelineStepObserver(): - """Define abstract class to observe pipeline step object use. - - !!! note - To subscribe to a method call, the inherited class simply needs to define 'on_<method_name>' functions with timestamp, object and traceback argument. - """ - - def __enter__(self): - """ - Define abstract __enter__ method to use observer as a context. - - !!! warning - This method is called provided that the observed PipelineStepObject is created as a context using a with statement. - """ - return self - - def __exit__(self, exception_type, exception_value, exception_traceback): - """ - Define abstract __exit__ method to use observer as a context. - - !!! warning - This method is called provided that the observed PipelineStepObject is created as a context using a with statement. - """ - pass diff --git a/src/argaze/GazeAnalysis/Basic.py b/src/argaze/GazeAnalysis/Basic.py index 74426de..063ee2b 100644 --- a/src/argaze/GazeAnalysis/Basic.py +++ b/src/argaze/GazeAnalysis/Basic.py @@ -27,6 +27,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init ScanPathAnalyzer class super().__init__() self.__path_duration = 0 @@ -72,6 +73,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__path_duration = 0 diff --git a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py index 48ecbda..9e2aa77 100644 --- a/src/argaze/GazeAnalysis/DeviationCircleCoverage.py +++ b/src/argaze/GazeAnalysis/DeviationCircleCoverage.py @@ -31,6 +31,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIMatcher class super().__init__() self.__coverage_threshold = 0 diff --git a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py index 3fd9613..0864b18 100644 --- a/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/DispersionThresholdIdentification.py @@ -115,6 +115,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init GazeMovementIdentifier class super().__init__() self.__deviation_max_threshold = 0 diff --git a/src/argaze/GazeAnalysis/Entropy.py b/src/argaze/GazeAnalysis/Entropy.py index c1cddd6..a73901e 100644 --- a/src/argaze/GazeAnalysis/Entropy.py +++ b/src/argaze/GazeAnalysis/Entropy.py @@ -35,6 +35,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__transition_matrix_analyzer = None diff --git a/src/argaze/GazeAnalysis/ExploreExploitRatio.py b/src/argaze/GazeAnalysis/ExploreExploitRatio.py index 44addd3..3b2d53b 100644 --- a/src/argaze/GazeAnalysis/ExploreExploitRatio.py +++ b/src/argaze/GazeAnalysis/ExploreExploitRatio.py @@ -33,6 +33,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init ScanPathAnalyzer class super().__init__() self.__short_fixation_duration_threshold = 0. diff --git a/src/argaze/GazeAnalysis/FocusPointInside.py b/src/argaze/GazeAnalysis/FocusPointInside.py index 3626e22..dbcb438 100644 --- a/src/argaze/GazeAnalysis/FocusPointInside.py +++ b/src/argaze/GazeAnalysis/FocusPointInside.py @@ -31,6 +31,7 @@ class AOIMatcher(GazeFeatures.AOIMatcher): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIMatcher class super().__init__() self.__reset() diff --git a/src/argaze/GazeAnalysis/KCoefficient.py b/src/argaze/GazeAnalysis/KCoefficient.py index 9bed17c..7e3caab 100644 --- a/src/argaze/GazeAnalysis/KCoefficient.py +++ b/src/argaze/GazeAnalysis/KCoefficient.py @@ -33,6 +33,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init ScanPathAnalyzer class super().__init__() self.__K = 0 @@ -90,6 +91,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__K = 0 diff --git a/src/argaze/GazeAnalysis/LempelZivComplexity.py b/src/argaze/GazeAnalysis/LempelZivComplexity.py index fc50991..810dbba 100644 --- a/src/argaze/GazeAnalysis/LempelZivComplexity.py +++ b/src/argaze/GazeAnalysis/LempelZivComplexity.py @@ -34,6 +34,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__lempel_ziv_complexity = 0 diff --git a/src/argaze/GazeAnalysis/LinearRegression.py b/src/argaze/GazeAnalysis/LinearRegression.py index c2f532a..00fd649 100644 --- a/src/argaze/GazeAnalysis/LinearRegression.py +++ b/src/argaze/GazeAnalysis/LinearRegression.py @@ -34,6 +34,7 @@ class GazePositionCalibrator(GazeFeatures.GazePositionCalibrator): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init GazePositionCalibrator class super().__init__() self.__linear_regression = LinearRegression() diff --git a/src/argaze/GazeAnalysis/NGram.py b/src/argaze/GazeAnalysis/NGram.py index ac5a0dd..ca60734 100644 --- a/src/argaze/GazeAnalysis/NGram.py +++ b/src/argaze/GazeAnalysis/NGram.py @@ -31,6 +31,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__n_min = 2 diff --git a/src/argaze/GazeAnalysis/NearestNeighborIndex.py b/src/argaze/GazeAnalysis/NearestNeighborIndex.py index 615643e..a577eba 100644 --- a/src/argaze/GazeAnalysis/NearestNeighborIndex.py +++ b/src/argaze/GazeAnalysis/NearestNeighborIndex.py @@ -34,6 +34,7 @@ class ScanPathAnalyzer(GazeFeatures.ScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init ScanPathAnalyzer class super().__init__() self.__size = (0, 0) diff --git a/src/argaze/GazeAnalysis/TransitionMatrix.py b/src/argaze/GazeAnalysis/TransitionMatrix.py index 16cb56e..dd5cf87 100644 --- a/src/argaze/GazeAnalysis/TransitionMatrix.py +++ b/src/argaze/GazeAnalysis/TransitionMatrix.py @@ -34,6 +34,7 @@ class AOIScanPathAnalyzer(GazeFeatures.AOIScanPathAnalyzer): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init AOIScanPathAnalyzer class super().__init__() self.__transition_matrix_probabilities = pandas.DataFrame() diff --git a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py index f881132..a0aab68 100644 --- a/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py +++ b/src/argaze/GazeAnalysis/VelocityThresholdIdentification.py @@ -114,6 +114,7 @@ class GazeMovementIdentifier(GazeFeatures.GazeMovementIdentifier): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): + # Init GazeMovementIdentifier class super().__init__() self.__velocity_max_threshold = 0 diff --git a/src/argaze/GazeFeatures.py b/src/argaze/GazeFeatures.py index bb2fb5b..bd7970b 100644 --- a/src/argaze/GazeFeatures.py +++ b/src/argaze/GazeFeatures.py @@ -290,7 +290,7 @@ class GazePositionCalibrator(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() + pass def store(self, observed_gaze_position: GazePosition, expected_gaze_position: GazePosition): """Store observed and expected gaze positions. @@ -538,7 +538,7 @@ class GazeMovementIdentifier(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() + pass @DataFeatures.PipelineStepMethod def identify(self, timestamped_gaze_position: GazePosition, terminate:bool=False) -> GazeMovement: @@ -813,8 +813,6 @@ class ScanPathAnalyzer(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() - self.__analysis = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property) and value.fset is None] def analysis(self) -> DataFeatures.DataDictionary: @@ -834,8 +832,6 @@ class AOIMatcher(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() - self.__exclude = [] @property @@ -1182,8 +1178,6 @@ class AOIScanPathAnalyzer(DataFeatures.PipelineStepObject): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - super().__init__() - self.__analysis = [name for (name, value) in self.__class__.__dict__.items() if isinstance(value, property) and value.fset is None] def analysis(self) -> DataFeatures.DataDictionary: diff --git a/src/argaze/utils/UtilsFeatures.py b/src/argaze/utils/UtilsFeatures.py index 26a63ea..75deca8 100644 --- a/src/argaze/utils/UtilsFeatures.py +++ b/src/argaze/utils/UtilsFeatures.py @@ -16,9 +16,14 @@ __credits__ = [] __copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)" __license__ = "GPLv3" +import os +import pathlib import time import types import traceback + +from argaze import DataFeatures + import numpy import cv2 @@ -162,39 +167,75 @@ def tuple_to_string(t: tuple, separator: str = ", ") -> str: return separator.join(f'\"{e}\"' for e in t) -class FileWriter(): - """Write data into a file line by line. +class FileWriter(DataFeatures.PipelineStepObject): + """Write data into a file line by line.""" - Parameters: - path: File path where to write data. - header: String or tuple to write first. - separator: String used to separate elements during tuple to string conversion. - """ + @DataFeatures.PipelineStepInit + def __init__(self, **kwargs): + + # Init private attributes + self.__path = None + self.__separator = ',' + self.__header = None + + @property + def path(self) -> str: + """File path where to write data.""" + return self.__path + + @path.setter + def path(self, path: str): - def __init__(self, path: str, header: str|tuple, separator: str = ','): - """Check that folder structure exist and create file then, write header line.""" + self.__path = pathlib.Path(path) - import os - import pathlib + @property + def separator(self) -> str: + """String used to separate elements during tuple to string conversion.""" + return self.__separator + + @separator.setter + def separator(self, separator: str): + + self.__separator = separator + + @property + def header(self) -> str|tuple: + """String or tuple to write first.""" + return self.__header + + @header.setter + def header(self, header: str|tuple): + + self.__header = header + + @DataFeatures.PipelineStepEnter + def __enter__(self): + """Check that folder structure exist and open file then, write header line.""" + + # DEBUG + print('FileWriter.__enter__') - self.path = pathlib.Path(path) - self.separator = separator + if not os.path.exists(self.__path.parent.absolute()): - if not os.path.exists(self.path.parent.absolute()): - os.makedirs(self.path.parent.absolute()) + os.makedirs(self.__path.parent.absolute()) # Open file - self._file = open(self.path, 'w', encoding='utf-8', buffering=1) + self.__file = open(self.__path, 'w', encoding='utf-8', buffering=1) # Write header if required - if header is not None: + if self.__header is not None: # Format list or tuple element into quoted strings - if not isinstance(header, str): + if not isinstance(self.__header, str): - header = tuple_to_string(header, self.separator) + self.__header = tuple_to_string(self.__header, self.__separator) - print(header, file=self._file, flush=True) + print(self.__header, file=self.__file, flush=True) + + @DataFeatures.PipelineStepExit + def __exit__(self, exception_type, exception_value, exception_traceback): + """Close file.""" + self.__file.close() def write(self, log: str|tuple): """Write log as a new line into file. @@ -206,28 +247,13 @@ class FileWriter(): # Format list or tuple element into quoted strings if not isinstance(log, str): - log = tuple_to_string(log, self.separator) + log = tuple_to_string(log, self.__separator) # Write into file - print(log, file=self._file, flush=True) - - def __del__(self): - """Close file.""" + print(log, file=self.__file, flush=True) - 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. +class VideoWriter(DataFeatures.PipelineStepObject): + """Open ffmpeg application as sub-process. FFmpeg input PIPE: RAW images in BGR color format FFmpeg output MP4 file encoded with HEVC codec. @@ -242,25 +268,75 @@ class VideoWriter(): -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 + @DataFeatures.PipelineStepInit + def __init__(self, **kwargs): + + # Init private attributes + self.__path = None + self.__width = 320 + self.__height = 240 + self.__fps = 25 + + @property + def path(self) -> str: + """File path where to write images.""" + return self.__path + + @path.setter + def path(self, path: str): + + self.__path = pathlib.Path(path) + + @property + def width(self) -> int: + """Video horizontal resolution.""" + return self.__width + + @width.setter + def width(self, width: int): self.__width = width + + @property + def height(self) -> int: + """Video vertical resolution.""" + return self.__height + + @height.setter + def height(self, height: int): + 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) + @property + def fps(self) -> int: + """frame per second.""" + return self.__fps + + @fps.setter + def fps(self, fps: int): - def write(self, image: numpy.array): - """Write raw video frame to input stream of ffmpeg sub-process.""" + self.__fps = fps - # Resize image to adapt to video resolution - output = cv2.resize(image, dsize=(self.__width, self.__height), interpolation=cv2.INTER_LINEAR) + @DataFeatures.PipelineStepEnter + def __enter__(self): + """Check that folder structure exist then, open ffmpeg subprocess.""" + + import subprocess as sp + import shlex - self.__process.stdin.write(output.tobytes()) + # DEBUG + print('VideoWriter.__enter__') - def __del__(self): + if not os.path.exists(self.__path.parent.absolute()): + + os.makedirs(self.__path.parent.absolute()) + + self.__process = sp.Popen(shlex.split(f'ffmpeg -hide_banner -loglevel error -y -s {self.__width}x{self.__height} -pixel_format bgr24 -f rawvideo -r {self.__fps} -i pipe: -vcodec libx265 -x265-params log-level=error -pix_fmt yuv420p -crf 24 {self.__path}'), stdin=sp.PIPE) + + @DataFeatures.PipelineStepExit + def __exit__(self, exception_type, exception_value, exception_traceback): # Close and flush stdin self.__process.stdin.close() @@ -272,6 +348,14 @@ class VideoWriter(): # Note: We don't have to terminate the sub-process (after process.wait(), the sub-process is supposed to be closed). self.__process.terminate() + 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 PrintCallStack(method): """Define a decorator to print call stack until the decorated method.""" diff --git a/src/argaze/utils/contexts/OpenCV.py b/src/argaze/utils/contexts/OpenCV.py index 5a35fba..a65d40e 100644 --- a/src/argaze/utils/contexts/OpenCV.py +++ b/src/argaze/utils/contexts/OpenCV.py @@ -31,7 +31,7 @@ class Window(ArFeatures.ArContext): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - # Init parent classes + # Init ArContext classe super().__init__() @DataFeatures.PipelineStepEnter diff --git a/src/argaze/utils/contexts/TobiiProGlasses2.py b/src/argaze/utils/contexts/TobiiProGlasses2.py index 6b7236b..7830036 100644 --- a/src/argaze/utils/contexts/TobiiProGlasses2.py +++ b/src/argaze/utils/contexts/TobiiProGlasses2.py @@ -312,7 +312,7 @@ class LiveStream(ArFeatures.ArContext): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - # Init parent classes + # Init ArContext classe super().__init__() # Init private attributes @@ -1197,7 +1197,7 @@ class PostProcessing(ArFeatures.ArContext): @DataFeatures.PipelineStepInit def __init__(self, **kwargs): - # Init parent classes + # Init ArContext classe super().__init__() # Init private attributes diff --git a/src/argaze/utils/demo/gaze_analysis_pipeline.json b/src/argaze/utils/demo/gaze_analysis_pipeline.json index 0c81be3..0d1062b 100644 --- a/src/argaze/utils/demo/gaze_analysis_pipeline.json +++ b/src/argaze/utils/demo/gaze_analysis_pipeline.json @@ -1,7 +1,7 @@ { "argaze.ArFeatures.ArFrame": { "name": "GrayRectangle", - "size": [640, 383], + "size": [1920, 1149], "background": "frame_background.jpg", "gaze_movement_identifier": { "argaze.GazeAnalysis.DispersionThresholdIdentification.GazeMovementIdentifier": { @@ -24,7 +24,7 @@ } }, "heatmap": { - "size": [32, 24] + "size": [128, 96] }, "layers": { "demo_layer": { @@ -50,8 +50,7 @@ }, "observers": { "loggers.AOIScanPathAnalysisLogger": { - "path": "_export/logs/aoi_scan_path_metrics.csv", - "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "LZC"] + "path": "_export/logs/aoi_scan_path_metrics.csv" } } } @@ -117,12 +116,10 @@ }, "observers": { "loggers.FixationLogger": { - "path": "_export/logs/fixations.csv", - "header": ["Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI"] + "path": "_export/logs/fixations.csv" }, "loggers.ScanPathAnalysisLogger": { - "path": "_export/logs/scan_path_metrics.csv", - "header": ["Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR"] + "path": "_export/logs/scan_path_metrics.csv" }, "loggers.VideoRecorder": { "path": "_export/logs/video.mp4", diff --git a/src/argaze/utils/demo/loggers.py b/src/argaze/utils/demo/loggers.py index 5f1986e..071e20b 100644 --- a/src/argaze/utils/demo/loggers.py +++ b/src/argaze/utils/demo/loggers.py @@ -20,7 +20,13 @@ from argaze import DataFeatures, GazeFeatures from argaze.GazeAnalysis import * from argaze.utils import UtilsFeatures -class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): +class FixationLogger(UtilsFeatures.FileWriter): + + def __init__(self, **kwargs): + + super().__init__(**kwargs) + + self.header = "Timestamp (ms)", "Focus (px)", "Duration (ms)", "AOI" def on_look(self, timestamp, frame, exception): """Log frame fixations.""" @@ -37,7 +43,13 @@ class FixationLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter self.write(log) -class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): +class ScanPathAnalysisLogger(UtilsFeatures.FileWriter): + + def __init__(self, **kwargs): + + super().__init__(**kwargs) + + self.header = "Timestamp (ms)", "Duration (ms)", "Step", "K", "NNI", "XXR" def on_look(self, timestamp, frame, exception): """Log frame scan path metrics.""" @@ -57,14 +69,20 @@ class ScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.Fi self.write(log) -class VideoRecorder(DataFeatures.PipelineStepObserver, UtilsFeatures.VideoWriter): +class VideoRecorder(UtilsFeatures.VideoWriter): + + def on_look(self, timestamp, frame, exception): + """Write frame image.""" + + self.write(frame.image()) + +class AOIScanPathAnalysisLogger(UtilsFeatures.FileWriter): - def on_look(self, timestamp, frame, exception): - """Write frame image.""" + def __init__(self, **kwargs): - self.write(frame.image()) + super().__init__(**kwargs) -class AOIScanPathAnalysisLogger(DataFeatures.PipelineStepObserver, UtilsFeatures.FileWriter): + self.header = "Timestamp (ms)", "Duration (ms)", "Step", "K", "LZC" def on_look(self, timestamp, layer, exception): """Log layer aoi scan path metrics""" |