Merge pull request #35 from int-brain-lab/passiveVideoNetworked

Passive video networked

Author: k1o0

iblrig_custom_tasks/_sp_passiveVideo/task.py

@@ -15,7 +15,7 @@
 from pybpodapi.protocol import Bpod
 
 import iblrig.misc
-from iblrig.base_tasks import BpodMixin
+from iblrig.base_tasks import BpodMixin, NetworkSession
 
 _logger = logging.getLogger(f'iblrig.{__name__}')
 
@@ -34,7 +34,7 @@ class MediaStats(vlc.MediaStats):
 
     def fieldnames(self):
         """Return the field names."""
-        return zip(*self._fields_)[0]
+        return list(zip(*self._fields_))[0]
 
     def as_tuple(self):
         """Return all attribute values as a tuple."""
@@ -149,12 +149,12 @@ def get_media_length(self):
             None is returned when no video is loaded.
         """
         if self._media:
-            length = self._media.get_length()
+            length = self._player.get_length()
             if length > -1:
                 return length / 1e3
 
 
-class Session(BpodMixin):
+class Session(BpodMixin, NetworkSession):
     """Play a single video."""
 
     protocol_name = '_sp_passiveVideo'
@@ -181,14 +181,14 @@ def save(self):
             if 20 > self._log_level > 0:
                 stats = self.video.stats
                 stats.to_parquet(self.paths.STATS_FILE_PATH)
-            if self.video._media and self.video._media.get_mrl().endswith(str(self.task_params.VIDEO)):
-                ext = Path(self.task_params.VIDEO).suffix
+            video_src = Path(self.task_params.VIDEO)
+            if self.video._media and self.video._media.get_mrl().endswith(video_src.as_posix()):
+                ext = video_src.suffix
                 video_file_path = self.paths.DATA_FILE_PATH.with_name(f'_sp_video.raw{ext}')
-                _logger.info('Copying %s -> %s', self.task_params.VIDEO, video_file_path)
-                shutil.copy(self.task_params.VIDEO, video_file_path)
+                _logger.info('Copying %s -> %s', video_src, video_file_path)
+                shutil.copy(video_src, video_file_path)
             else:
                 _logger.warning('Video not copied (video most likely was not played)')
-            self. self.video._media.get_mrl()
         self.paths.SESSION_FOLDER.joinpath('transfer_me.flag').touch()
 
     def start_hardware(self):

iblrig_custom_tasks/_sp_passiveVideo/test_sp_videoPassive.py

@@ -1,6 +1,7 @@
 import unittest
-from unittest.mock import Mock
-from iblrig_custom_tasks._sp_passiveVideo.task import Session, Player
+from unittest.mock import Mock, MagicMock, patch
+import pandas as pd
+from iblrig_custom_tasks._sp_passiveVideo.task import Session, Player, MediaStats
 from iblrig.test.base import TaskArgsMixin
 
 
@@ -22,9 +23,48 @@ def test_next_trial(self):
         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()
+
+    @unittest.skip('Integration test')
+    def test_save(self):
+        """Test video presentation.
+
+        This test is an integration test and requires a video file to be present in the path.
+        The Bpod is not used in this test.
+        """
+        self.task_kwargs['hardware_settings']['MAIN_SYNC'] = False
+        task = Session(log_level='DEBUG', **self.task_kwargs)
+        task.task_params.VIDEO = r'C:\Users\User\Downloads\SampleVideo_1280x720_1mb.mp4'
+        task.bpod = MagicMock()
+        with patch.object(task, 'start_mixin_bpod'):
+            task.run()
+        files = list(task.paths.SESSION_FOLDER.rglob('*.*'))
+        expected = [
+            'transfer_me.flag', '_ibl_experiment.description_behavior.yaml',
+            '_iblrig_taskSettings.raw.json', '_ibl_log.info-acquisition.log',
+            '_sp_taskData.raw.pqt', '_sp_video.raw.mp4', '_sp_videoData.stats.pqt']
+        self.assertCountEqual((f.name for f in files), expected)
+        stats = task.video.stats
+        self.assertIsInstance(stats, pd.DataFrame)
+        df = pd.read_parquet(task.paths.STATS_FILE_PATH)
+        self.assertTrue(all(df == stats))
+        df = pd.read_parquet(task.paths.DATA_FILE_PATH)
+        expected = [
+            'intervals_0', 'intervals_1', 'video_runtime',
+            'MediaPlayerPlaying', 'MediaPlayerEndReached']
+        self.assertEqual(len(df), task.task_params.NREPEATS)
+        self.assertEqual(expected, df.columns.tolist())
+
+    def test_MediaStats(self):
+        stats = MediaStats()
+        fields = stats.fieldnames()
+        self.assertIsInstance(fields, tuple)
+        self.assertTrue(all(isinstance(f, str) for f in fields))
+        field = 'demux_bitrate'
+        self.assertIn(field, fields)
+
+        values = stats.as_tuple()
+        self.assertIsInstance(values, tuple)
+        self.assertEqual(len(values), len(fields))
 
 
 if __name__ == '__main__':

projects/_sp_passiveVideo.py

@@ -1,5 +1,6 @@
 import logging
 
+import cv2
 import numpy as np
 import pandas as pd
 import one.alf.io as alfio
@@ -8,11 +9,13 @@
 
 from ibllib.pipes.base_tasks import BehaviourTask
 from ibllib.exceptions import SyncBpodFpgaException
+from ibllib.io.video import get_video_meta
 from ibllib.io.extractors.ephys_fpga import get_protocol_period, get_sync_fronts
 from ibllib.io.raw_daq_loaders import load_timeline_sync_and_chmap
 from ibllib.io.extractors.mesoscope import plot_timeline
 
 _logger = logging.getLogger('ibllib').getChild(__name__)
+_logger.setLevel(logging.DEBUG)
 
 
 class PassiveVideoTimeline(BehaviourTask):
@@ -25,6 +28,7 @@ def signature(self):
         signature = {}
         signature['input_files'] = [
             ('_sp_taskData.raw.*', self.collection, True),  # TODO Create dataset type?
+            ('_sp_video.raw.*', self.collection, False),
             ('_iblrig_taskSettings.raw.*', self.collection, True),
             (f'_{self.sync_namespace}_DAQdata.raw.npy', self.sync_collection, True),
             (f'_{self.sync_namespace}_DAQdata.timestamps.npy', self.sync_collection, True),
@@ -63,7 +67,27 @@ def generate_sync_sequence(seed=1234, ns=3600, res=8):
         finally:
             np.random.set_state(state)
 
-    def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs):
+    def load_sync_sequence_from_video(self, video_file, location='bottom right', size=(5, 5)):
+        cap = cv2.VideoCapture(str(video_file))
+        sequence = []
+        location = location.casefold().split()
+        loc_map = {
+            'top': slice(0, size[1]), 'bottom': slice(-size[1], None),
+            'left': slice(0, size[0]), 'right': slice(-size[0], None)}
+        idx = tuple(loc_map[x] for x in reversed(location))  # h, w
+        success = True
+        while success:
+            success, frame = cap.read()
+            if success:
+                # Find the sync square in the video frame
+                pixel = np.mean(frame[idx])
+                sequence.append(int(pixel > 128))
+        length = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        cap.release()
+        assert len(sequence) == length, 'sequence length does not match video length'
+        return np.array(sequence)
+
+    def extract_frame_times(self, save=True, frame_rate=None, display=False, **kwargs):
         """Extract the Bpod trials data and Timeline acquired signals.
 
         Sync requires three steps:
@@ -76,7 +100,7 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
         save : bool, optional
             Whether to save the video frame times to file, by default True.
         frame_rate : int, optional
-            The frame rate of the video presented, by default 60.
+            The frame rate of the video presented, by default 30.
         display : bool, optional
             When true, plot the aligned frame times. By default False.
 
@@ -97,13 +121,32 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
         SyncBpodFpgaException
             The synchronization of frame times was likely unsuccessful.
         """
+        DEFAULT_FRAME_RATE = 30
         _, (p,), _ = self.input_files[0].find_files(self.session_path)
         # Load raw data
         proc_data = pd.read_parquet(p)
         sync_path = self.session_path / self.sync_collection
         self.timeline = alfio.load_object(sync_path, 'DAQdata', namespace='timeline')
         sync, chmap = load_timeline_sync_and_chmap(sync_path, timeline=self.timeline)
 
+        # Attempt to get the frame rate from the video file if not provided
+        video_file = next(self.session_path.joinpath(self.collection).glob('_sp_video.raw.*'))
+        if video_file.exists():
+            video_meta = get_video_meta(video_file)
+            if frame_rate is not None and frame_rate != video_meta.fps:
+                _logger.warning(
+                    'Frame rate mismatch: %.2f Hz (video) vs %.2f Hz (provided). Using %.2f Hz',
+                    video_meta.fps, frame_rate, video_meta.fps)
+            else:
+                _logger.debug('Video frame rate: %.2f Hz', video_meta.fps)
+            frame_rate = video_meta.fps
+        else:
+            video_meta = None
+            frame_rate = frame_rate or DEFAULT_FRAME_RATE
+            _logger.warning('Video not found. Assumed video frame rate: %.2f Hz', frame_rate)
+        Fs = self.timeline['meta']['daqSampleRate']
+        assert Fs > frame_rate * 1.5, 'DAQ sample rate must be higher than video frame rate'
+
         bpod = get_sync_fronts(sync, chmap['bpod'])
         # Get the spacer times for this protocol
         if any(arg in kwargs for arg in ('tmin', 'tmax')):
@@ -133,7 +176,8 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
         # These durations are longer than video actually played and will be cut down after
         durations = (proc_data['intervals_1'] - proc_data['intervals_0']).values
         max_n_frames = np.max(np.ceil(durations * frame_rate).astype(int))
-        frame_times = np.full((max_n_frames, len(proc_data)), np.nan)
+        n_frames = video_meta.length if video_meta else max_n_frames
+        frame_times = np.full((n_frames, len(proc_data)), np.nan)
 
         sync_sequence = kwargs.get('sync_sequence', self.generate_sync_sequence())
         for i, rep in proc_data.iterrows():
@@ -147,6 +191,9 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
                 end = start + (rep['intervals_1'] - rep['intervals_0'])
             f2ttl = get_sync_fronts(sync, chmap['frame2ttl'])
             ts = f2ttl['times'][np.logical_and(f2ttl['times'] >= start, f2ttl['times'] < end)]
+            if video_meta:
+                _logger.debug('Repeat %i: video duration: %.2fs, f2ttl duration: %.2f',
+                              i, video_meta.duration.seconds, ts[-1] - ts[0])
 
             # video_runtime is the video length reported by VLC.
             # As it was added later, the less accurate media player timestamps may be used if the former is not available
@@ -162,23 +209,24 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
             # Find change points (black <-> white indices)
             x, = np.where(np.abs(np.diff(x)))
             # Include first frame as change point
-            x = np.r_[0, x]
+            x = np.r_[0, x + 1]
             # Synchronize the two by aligning flip times
             DRIFT_THRESHOLD_PPM = 50
-            Fs = self.timeline['meta']['daqSampleRate']
-            fcn, drift = ibldsp.utils.sync_timestamps(sequence_times[x], ts, tbin=1 / Fs, linear=True)
+            fcn, drift = ibldsp.utils.sync_timestamps(sequence_times[x], ts, linear=True)
             # Log any major drift or raise if too large
             if np.abs(drift) > DRIFT_THRESHOLD_PPM * 2 and x.size - ts.size > 100:
-                raise SyncBpodFpgaException(f'sync cluster f*ck: drift = {drift:.2f}, changepoint difference = {x.size - ts.size}')
-            elif drift > DRIFT_THRESHOLD_PPM:
-                _logger.warning('BPOD/FPGA synchronization shows values greater than %.2f ppm',
-                                DRIFT_THRESHOLD_PPM)
+                raise SyncBpodFpgaException(
+                    f'sync cluster f*ck: drift = {drift:.2f}, changepoint difference = {x.size - ts.size}')
+            elif np.abs(drift) > DRIFT_THRESHOLD_PPM:
+                _logger.warning('Frame synchronization shows values greater than %.2g ppm', DRIFT_THRESHOLD_PPM)
+            _logger.debug('Frame synchronization drift: %.2f ppm', drift)
 
             # Get the frame times in timeline time
             frame_times[:len(sequence_times), i] = fcn(sequence_times)
 
         # Trim down to length of repeat with most frames
-        frame_times = frame_times[:np.where(np.all(np.isnan(frame_times), axis=1))[0][0], :]
+        if np.any(empty := np.all(np.isnan(frame_times), axis=1)):
+            frame_times = frame_times[:np.where(empty)[0][0], :]
 
         if display:
             import matplotlib.pyplot as plt
@@ -191,16 +239,46 @@ def extract_frame_times(self, save=True, frame_rate=60, display=False, **kwargs)
             ax[0].title.set_text('frame2ttl')
             cmap = colormaps['plasma']
             for i, times in enumerate(frame_times.T):
+                # Plot the sync sequence and sync'd frame times
                 rgba = cmap(i / frame_times.shape[1])
                 ax[1].plot(times, sync_sequence[:len(times)], c=rgba, label=f'{i}')
+                # Plot the f2ttl values
+                idx = bpod_rep_starts[i]
+                start = bpod['times'][idx]
+                try:
+                    end = bpod['times'][idx + 1]
+                except IndexError:
+                    end = start + (rep['intervals_1'] - rep['intervals_0'])
+                mask = np.logical_and(f2ttl['times'] >= start, f2ttl['times'] < end)
+                squares(f2ttl['times'][mask], f2ttl['polarities'][mask],
+                        yrange=[0, 1], ax=ax[1], linestyle=':', color='k')
             ax[1].title.set_text('aligned sync square sequence')
             ax[1].set_yticks((0, 1))
             ax[1].set_yticklabels([-1, 1])
             plt.legend(markerfirst=False, title='repeat #', loc='upper right', facecolor='white')
+
+            # Check the sync sequence from the video
+            if video_file.exists():
+                observed = self.load_sync_sequence_from_video(video_file)
+                _, ax = plt.subplots(2, 1, sharex=True)
+                ax[0].title.set_text('generated sync square sequence')
+                ax[0].plot(sync_sequence[:observed.size])
+                ax[1].title.set_text('observed sync square sequence')
+                ax[1].plot(observed)
+
+                # resample the f2ttl sequence to the frame times
+                # tts = ts-ts[0]
+                # from scipy import interpolate
+                # interp = interpolate.interp1d(tts, pol, kind = "nearest")
+                # ttts = np.arange(0, tts[-1], 1/frame_rate)
+                # ax[2].plot(interp(ttts))
+                # squares(ts-ts[0], pol, ax=ax[2])
+
             plt.show()
 
         if save:
             filename = self.session_path.joinpath(self.output_collection, '_sp_video.times.npy')
+            np.save(filename, frame_times)
             out_files = [filename]
         else:
             out_files = []

projects/tests/test_passiveVideo.py

@@ -0,0 +1,36 @@
+"""Integration test for _sp_passiveVideo extraction."""
+
+import unittest
+import ibllib.pipes.dynamic_pipeline as dyn
+from one.api import ONE
+
+
+@unittest.skip('Integration test')
+class TestPassiveVideo(unittest.TestCase):
+    """Test _sp_passiveVideo extraction.
+
+    The test session is a little short. The second 30Hz video repeat does not
+    sync very well due to so few pulses. The first 60Hz video repeat is fine.
+    """
+    required_files = ['test/2025-01-27/002', 'SP061/2025-01-27/001']
+
+    def test_extraction(self):
+        one = ONE()
+        session_path = one.cache_dir.joinpath(
+            'cortexlab', 'Subjects', self.required_files[1])
+        tasks = dyn._get_trials_tasks(session_path)
+        tasks = [t for t in tasks.values() if 'passivevideo' in t.name.casefold()]
+        self.assertTrue(any(tasks))
+        # Test
+        # task = tasks[0]  # 60Hz
+        # task = tasks[1]  # 30Hz
+        # SP061
+        task = tasks[0]  # 60Hz
+        ret = task.run(save=True, frame_rate=None)
+        self.assertEqual(0, ret)
+        self.assertEqual(1, len(task.outputs))
+        self.assertTrue(task.outputs[0].exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

pyproject.toml

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