user/acwikla/refactor_position_and_tests (#197)

src/datamigration/nwb/components/position/lf_position.py

@@ -0,0 +1,5 @@
+class LfPosition:
+
+    def __init__(self, position_data, timestamps):
+        self.position_data = position_data
+        self.timestamps = timestamps

src/datamigration/nwb/components/position/lf_position_builder.py

@@ -0,0 +1,8 @@
+from src.datamigration.nwb.components.position.lf_position import LfPosition
+
+
+class LfPositionBuilder:
+
+    @staticmethod
+    def build(position_data, timestamps):
+        return LfPosition(position_data, timestamps)

src/datamigration/nwb/components/position/position_extractor.py → src/datamigration/nwb/components/position/lf_position_extractor.py

@@ -3,14 +3,17 @@
 from src.datamigration.nwb.components.iterator.multi_thread_timestamp_iterator import MultiThreadTimestampIterator
 from src.datamigration.nwb.components.position.pos_data_manager import PosDataManager
 from src.datamigration.nwb.components.position.pos_timestamp_manager import PosTimestampManager
+from src.datamigration.tools.validate_input_parameters import validate_input_parameters
 
 
-class PositionExtractor:
+class LfPositionExtractor:
     def __init__(self, datasets):
         self.datasets = datasets
         self.all_pos, self.continuous_time = self.__extract_data()
 
     def __extract_data(self):
+        validate_input_parameters(__name__, self.datasets)
+
         all_pos = []
         continuous_time = []
         for dataset in self.datasets:

src/datamigration/nwb/components/position/lf_position_manager.py

@@ -0,0 +1,15 @@
+from src.datamigration.nwb.components.position.lf_position_builder import LfPositionBuilder
+from src.datamigration.nwb.components.position.position_creator import PositionCreator
+from src.datamigration.nwb.components.position.lf_position_extractor import LfPositionExtractor
+
+
+class LfPositionManager:
+    def __init__(self, datasets):
+        self.lf_position_extractor = LfPositionExtractor(datasets)
+        self.lf_position_builder = LfPositionBuilder()
+
+    def get_lf_position(self):
+        position_data = self.lf_position_extractor.get_position()
+        timestamps = self.lf_position_extractor.get_timestamps()
+        return self.lf_position_builder.build(position_data, timestamps)
+

src/datamigration/nwb/components/position/position_creator.py

@@ -1,15 +1,20 @@
 from pynwb.behavior import Position
 
+from src.datamigration.tools.validate_input_parameters import validate_input_parameters
+
 
 class PositionCreator:
 
     @staticmethod
-    def create_position(position_data, timestamps):
+    def create(lf_position):
+        validate_input_parameters(__name__, lf_position)
+        validate_input_parameters(__name__, lf_position.position_data, lf_position.timestamps)
+
         position = Position()
         position.create_spatial_series(
             name='series',
-            data=position_data,
+            data=lf_position.position_data,
             reference_frame='Description defining what the zero-position is',
-            timestamps=timestamps
+            timestamps=lf_position.timestamps
         )
         return position

src/datamigration/nwb_file_builder.py

@@ -20,17 +20,18 @@
 from src.datamigration.nwb.components.dio.dio_files import DioFiles
 from src.datamigration.nwb.components.dio.dio_injector import DioInjector
 from src.datamigration.nwb.components.dio.dio_manager import DioManager
+from src.datamigration.nwb.components.electrode_group.electrode_group_injector import ElectrodeGroupInjector
 from src.datamigration.nwb.components.electrode_group.fl_electrode_group_creator import FlElectrodeGroupCreator
 from src.datamigration.nwb.components.electrode_group.lf_electrode_group_manager import FlElectrodeGroupManager
-from src.datamigration.nwb.components.electrode_group.electrode_group_injector import ElectrodeGroupInjector
 from src.datamigration.nwb.components.electrodes.electrode_creator import ElectrodesCreator
-from src.datamigration.nwb.components.electrodes.lf_electrode_manager import LfElectrodeManager
 from src.datamigration.nwb.components.electrodes.electrode_extension_creator import ElectrodeExtensionCreator
 from src.datamigration.nwb.components.electrodes.electrode_extension_injector import ElectrodeExtensionInjector
+from src.datamigration.nwb.components.electrodes.lf_electrode_manager import LfElectrodeManager
 from src.datamigration.nwb.components.mda.electrical_series_creator import ElectricalSeriesCreator
-from src.datamigration.nwb.components.mda.mda_injector import MdaInjector
 from src.datamigration.nwb.components.mda.lf_mda_manager import LfMdaManager
-from src.datamigration.nwb.components.position.position_builder import PositionBuilder
+from src.datamigration.nwb.components.mda.mda_injector import MdaInjector
+from src.datamigration.nwb.components.position.lf_position_manager import LfPositionManager
+from src.datamigration.nwb.components.position.position_creator import PositionCreator
 from src.datamigration.nwb.components.processing_module.processing_module_creator import ProcessingModuleCreator
 from src.datamigration.nwb.components.task.task_builder import TaskBuilder
 
@@ -79,9 +80,11 @@ def __init__(self,
         self.pm_creator = ProcessingModuleCreator('behavior', 'Contains all behavior-related data')
 
         self.task_builder = TaskBuilder(self.metadata)
-        self.position_builder = PositionBuilder(self.datasets)
-        self.lf_apparatus_manager = LfApparatusManager(self.metadata['apparatus']['data'])
 
+        self.lf_position_manager = LfPositionManager(self.datasets)
+        self.position_creator = PositionCreator()
+
+        self.lf_apparatus_manager = LfApparatusManager(self.metadata['apparatus']['data'])
 
         self.lf_probe_manager = LfProbeManager(self.probes, self.metadata['electrode groups'])
         self.device_injector = DeviceInjector()
@@ -173,7 +176,9 @@ def __build_and_inject_processing_module(self, nwb_content):
         self.pm_creator.insert(task)
 
         logger.info('Position: Building')
-        position = self.position_builder.build()
+        lf_position = self.lf_position_manager.get_lf_position()
+        logger.info('Position: Creating')
+        position = self.position_creator.create(lf_position)
         logger.info('Position: Injecting into ProcessingModule')
         self.pm_creator.insert(position)
 

src/test/datamigration/electrodes/test_electrodeExtensionCreator.py

@@ -6,8 +6,6 @@
 from src.datamigration.exceptions.none_param_in_init_exception import NoneParamInInitException
 from src.datamigration.header.module.header import Header
 from src.datamigration.nwb.components.electrodes.electrode_extension_creator import ElectrodeExtensionCreator
-from src.datamigration.nwb.components.electrodes.electrode_metadata_extension_creator import \
-    ElectrodesMetadataExtensionCreator
 
 path = os.path.dirname(os.path.abspath(__file__))
 

src/test/datamigration/electrodes/test_electrodeExtensionInjector.py

@@ -4,18 +4,15 @@
 
 from dateutil.tz import tzlocal
 from hdmf.common import VectorData
-from pynwb import NWBFile
-
 from ndx_franklab_novela.fl_electrode_group import FLElectrodeGroup
-from testfixtures import should_raise
+from pynwb import NWBFile
 
-from src.datamigration.exceptions.none_param_in_init_exception import NoneParamInInitException
 from src.datamigration.exceptions.not_compatible_metadata import NotCompatibleMetadata
 from src.datamigration.nwb.components.electrodes.electrode_creator import ElectrodesCreator
-from src.datamigration.nwb.components.electrodes.lf_electrode_manager import LfElectrodeManager
+from src.datamigration.nwb.components.electrodes.electrode_extension_injector import ElectrodeExtensionInjector
 from src.datamigration.nwb.components.electrodes.electrode_metadata_extension_creator import \
     ElectrodesMetadataExtensionCreator
-from src.datamigration.nwb.components.electrodes.electrode_extension_injector import ElectrodeExtensionInjector
+from src.datamigration.nwb.components.electrodes.lf_electrode_manager import LfElectrodeManager
 
 
 class TestElectrodeExtensionInjector(unittest.TestCase):

src/test/e2etests/test_positionExtractor.py

@@ -0,0 +1,46 @@
+import os
+import unittest
+
+from testfixtures import should_raise
+
+from src.datamigration.exceptions.none_param_in_init_exception import NoneParamInInitException
+from src.datamigration.nwb.components.position.lf_position_manager import LfPositionManager
+from src.datamigration.nwb.components.position.position_creator import PositionCreator
+from src.datamigration.tools.file_scanner import Dataset
+
+path = os.path.dirname(os.path.abspath(__file__))
+
+
+@unittest.skip("test requires continuoustime.dat file and can't be used on travis")
+class TestPositionExtraction(unittest.TestCase):
+
+    @staticmethod
+    def create_test_dataset():
+        dataset = Dataset('test_dataset')
+        dataset.add_data_to_dataset(path + '/../datamigration/res/pos_test/', 'pos')
+        dataset.add_data_to_dataset(path + '/../test_data/beans/preprocessing/20190718/20190718_beans_01_s1.time/',
+                                    'time')
+        return dataset
+
+    def test_position_extractor_reading_data_successfully(self):
+        dataset = self.create_test_dataset()
+
+        lf_position_manager = LfPositionManager(datasets=[dataset, ])
+        position_creator = PositionCreator()
+
+        lf_position = lf_position_manager.get_lf_position()
+        position = position_creator.create(lf_position)
+
+        self.assertIsNotNone(position)
+        self.assertEqual([32658, 4], position['Fields'].data.shape,
+                         'Shape should be [32658, 4]')
+        self.assertEqual((32658,), position['Fields'].timestamps.shape,
+                         'Shape should be (32658,)')
+
+    @should_raise(NoneParamInInitException)
+    def test_position_extractor_fails_reading_data_due_to_None_datasets(self):
+        lf_position_manager = LfPositionManager(datasets=None)
+        position_creator = PositionCreator()
+
+        lf_position = lf_position_manager.get_lf_position()
+        position_creator.create(lf_position)