added test for photometry copier

iblrig/test/test_transfers.py

@@ -18,6 +18,11 @@
 from iblrig.test.base import TASK_KWARGS
 from iblrig.transfer_experiments import BehaviorCopier, EphysCopier, SessionCopier, VideoCopier
 from iblrig_tasks._iblrig_tasks_trainingChoiceWorld.task import Session
+import iblrig.neurophotometrics
+
+import numpy as np
+import pandas as pd
+import pandera
 
 
 def _create_behavior_session(ntrials=None, hard_crash=False, kwargs=None):
@@ -52,8 +57,8 @@ def _create_behavior_session(ntrials=None, hard_crash=False, kwargs=None):
     return session
 
 
-class TestIntegrationTransferExperiments(unittest.TestCase):
-    """This test emulates the `transfer_data` command as run on the rig."""
+class TestIntegrationTransferExperimentsBase(unittest.TestCase):
+    """this base class copier testing"""
 
     def setUp(self):
         self.iblrig_settings = iblrig.path_helper.load_pydantic_yaml(
@@ -82,6 +87,83 @@ def side_effect(self, *args, filename=None, **kwargs):
         else:
             return self.iblrig_settings
 
+
+class TestIntegrationTransferExperimentsPhotometry(TestIntegrationTransferExperimentsBase):
+    """for testing the photometry"""
+
+    def create_fake_data(self):
+        datestr = datetime.now().strftime('%Y-%m-%d')
+        timestr = datetime.now().strftime('T%H%M%S')
+        folder_neurophotometrics = self.iblrig_settings['iblrig_local_data_path'].joinpath('neurophotometrics', datestr, timestr)
+        folder_neurophotometrics.mkdir(exist_ok=True, parents=True)
+
+        cols_dtypes = dict(
+            ChannelName=str, Channel='int8', AlwaysTrue='bool', SystemTimestamp='float64', ComputerTimestamp='float64'
+        )
+        cols = list(cols_dtypes.keys())
+        digital_inputs_df = pd.DataFrame(np.random.randn(10, len(cols)), columns=cols)
+        for col, dtype in cols_dtypes.items():
+            digital_inputs_df[col] = digital_inputs_df[col].astype(dtype)
+
+        schema_digital_inputs = pandera.DataFrameSchema(
+            columns=dict(
+                ChannelName=pandera.Column(str, coerce=True),
+                Channel=pandera.Column(pandera.Int8, coerce=True),
+                AlwaysTrue=pandera.Column(bool, coerce=True),
+                SystemTimestamp=pandera.Column(pandera.Float64),
+                ComputerTimestamp=pandera.Column(pandera.Float64),
+            )
+        )
+        digital_inputs_df = schema_digital_inputs.validate(digital_inputs_df)
+        digital_inputs_df.to_csv(folder_neurophotometrics / 'digital_inputs.csv', index=False, header=False)
+
+        # raw
+        schema_raw_data = pandera.DataFrameSchema(
+            columns=dict(
+                FrameCounter=pandera.Column(pandera.Int64),
+                SystemTimestamp=pandera.Column(pandera.Float64),
+                LedState=pandera.Column(pandera.Int16, coerce=True),
+                ComputerTimestamp=pandera.Column(pandera.Float64),
+                Region00=pandera.Column(pandera.Float64),  # hard coding regions here
+                Region01=pandera.Column(pandera.Float64),
+            )
+        )
+        cols_dtypes = dict(
+            FrameCounter='int64',
+            SystemTimestamp='float64',
+            LedState='int16',
+            ComputerTimestamp='float64',
+            Region00='float64',
+            Region01='float64',
+        )
+
+        cols = list(cols_dtypes.keys())
+        raw_photometry_df = pd.DataFrame(np.random.randn(10, len(cols)), columns=cols)
+        for col, dtype in cols_dtypes.items():
+            raw_photometry_df[col] = raw_photometry_df[col].astype(dtype)
+
+        raw_photometry_df = schema_raw_data.validate(raw_photometry_df)
+        raw_photometry_df.to_csv(folder_neurophotometrics / 'raw_photometry.csv', index=False)
+
+    def test_copier(self):
+        session = _create_behavior_session(ntrials=50, kwargs=self.session_kwargs)
+        self.create_fake_data()
+
+        # the workaround to find the settings.yaml
+        with mock.patch('iblrig.path_helper._load_settings_yaml') as mocker:
+            mocker.side_effect = self.side_effect
+            # the actual code to test
+            iblrig.neurophotometrics.init_neurophotometrics_subject(
+                session_stub=session.paths['SESSION_FOLDER'],
+                rois=['Region00', 'Region01'],
+                locations=['VTA', 'SNc'],
+            )
+            iblrig.neurophotometrics.copy_photometry_subject(session.paths['SESSION_FOLDER'])
+
+
+class TestIntegrationTransferExperiments(TestIntegrationTransferExperimentsBase):
+    """This test emulates the `transfer_data` command as run on the rig."""
+
     def test_behavior_copy_complete_session(self):
         """
         Here there are 2 cases, one is about a complete session, the other is about a session that crashed