Store libVLC stats in debug mode

Author: k1o0

iblrig_custom_tasks/_sp_passiveVideo/__init__.py

@@ -6,6 +6,7 @@
 import time
 from pathlib import Path
 from collections import defaultdict
+from functools import partial
 import logging
 import warnings
 
@@ -27,6 +28,18 @@
         'git+https://github.com/int-brain-lab/project_extraction.git"', RuntimeWarning)
 
 
+class MediaStats(vlc.MediaStats):
+    """A class to store media stats."""
+
+    def fieldnames(self):
+        """Return the field names."""
+        return zip(*self._fields_)[0]
+
+    def as_tuple(self):
+        """Return all attribute values as a tuple."""
+        return tuple(map(partial(getattr, self), self.fieldnames()))
+
+
 class Player:
     """A VLC player."""
     def __init__(self, rate=1):
@@ -35,6 +48,8 @@ def __init__(self, rate=1):
         self._player.set_fullscreen(True)
         self._player.set_rate(rate)
         self._media = None
+        self._media_stats = MediaStats()
+        self._stats = []
         self.events = defaultdict(list)
         em = self._player.event_manager()
         for event in (vlc.EventType.MediaPlayerPlaying, vlc.EventType.MediaPlayerEndReached):
@@ -46,6 +61,27 @@ def _record_event(self, event):
         # Have to convert to str as object pointer may change
         self.events[str(event.type).split('.')[-1]].append(time.time())
 
+    def update_media_stats(self):
+        """Update media stats.
+
+        Returns
+        -------
+        bool
+            True if the stats have changed since the last update.
+        """
+        if not vlc.libvlc_media_get_stats(self._player.get_media(), self._media_stats):
+            return False
+        stats = tuple((time.time(), *self._media_stats.as_tuple()))
+        if not any(self._stats) or stats[1:] != self._stats[-1][1:]:
+            self._stats.append(stats)
+            return True
+        return False
+
+    @property
+    def stats(self):
+        """Return media stats."""
+        return pd.DataFrame(self._stats, columns=['time', *self._media_stats.fieldnames()])
+
     def play(self, path):
         """Play a video.
 
@@ -112,8 +148,10 @@ def __init__(self, **kwargs):
         if self.hardware_settings.get('MAIN_SYNC', False):
             raise NotImplementedError('Recording frame2ttl on Bpod not yet implemented')
         self.paths.DATA_FILE_PATH = self.paths.DATA_FILE_PATH.with_name('_sp_taskData.raw.pqt')
+        self.paths.STATS_FILE_PATH = self.paths.DATA_FILE_PATH.with_name('_sp_videoData.stats.pqt')
         self.video = None
         self.trial_num = -1
+        self._log_level = logging.getLevelNamesMapping()[kwargs.get('log_level', 'INFO')]
         columns = ['intervals_0', 'intervals_1']
         self.data = pd.DataFrame(pd.NA, index=range(self.task_params.NREPEATS), columns=columns)
 
@@ -122,10 +160,13 @@ def save(self):
         if self.video:
             data = pd.concat([self.data, pd.DataFrame.from_dict(self.video.events)], axis=1)
             data.to_parquet(self.paths.DATA_FILE_PATH)
+            if 20 > self._log_level > 0:
+                stats = self.video.stats
+                stats.to_parquet(self.paths.STATS_FILE_PATH)
         self.paths.SESSION_FOLDER.joinpath('transfer_me.flag').touch()
 
     def start_hardware(self):
-        self.start_mixin_bpod()  # used for protocol spacer only
+        self.start_mixin_bpod()
         self.video = Player()
 
     def next_trial(self):
@@ -150,15 +191,15 @@ def _set_bpod_out(self, val):
 
     def _run(self):
         """This is the method that runs the video."""
-        # make the bpod send spacer signals to the main sync clock for protocol discovery
-        self.send_spacers()
         for rep in range(self.task_params.NREPEATS):  # Main loop
             self.next_trial()
             self._set_bpod_out(True)
             # TODO c.f. MediaListPlayerPlayed event
             while not self.video.is_started:
                 ...  # takes time to actually start playback
             while self.video.is_playing or (end_time := self.video.get_ended_time(rep)) is None:
+                if 20 > self._log_level > 0:
+                    self.video.update_media_stats()
                 time.sleep(0.05)
             # trial finishes when playback finishes
             self._set_bpod_out(False)

iblrig_custom_tasks/_sp_passiveVideo/task.py

@@ -0,0 +1,31 @@
+import unittest
+from unittest.mock import Mock
+from iblrig_custom_tasks._sp_passiveVideo.task import Session, Player
+from iblrig.test.base import TaskArgsMixin
+
+
+class TestPassiveVideo(TaskArgsMixin, unittest.TestCase):
+
+    def setUp(self):
+        self.get_task_kwargs()
+
+    def test_next_trial(self):
+        self.assertRaises(NotImplementedError, Session, **self.task_kwargs)
+        self.task_kwargs['hardware_settings']['MAIN_SYNC'] = False
+        task = Session(log_level='DEBUG', **self.task_kwargs)
+        task.video = Mock(auto_spec=Player)
+        task.task_params.VIDEO = r'C:\Users\Work\Downloads\ONE\perlin-xyscale2-tscale50-comb08-5min.mp4'
+        task.task_params.VIDEO = r'C:\Users\Work\Downloads\SampleVideo_1280x720_1mb.mp4'
+        task.next_trial()
+        task.video.play.assert_called_once_with(task.task_params.VIDEO)
+        task.video.replay.assert_not_called()
+        task.video.reset_mock()
+        task.next_trial()
+        task.video.replay.assert_called_once()
+        # task.bpod = MagicMock()
+        # with patch.object(task, 'start_mixin_bpod'):
+        #     task.run()
+
+
+if __name__ == '__main__':
+    unittest.main()

iblrig_custom_tasks/_sp_passiveVideo/test_sp_videoPassive.py

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "project_extraction"
-version         = "0.4.2"
+version         = "0.5.0"
 description     = "Custom extractors for satellite tasks"
 dynamic         = [ "readme" ]
 keywords        = [ "IBL", "neuro-science" ]