[ScanFactory] adapt scan factory to hdf5

tomoscan/esrf/hdf5scan.py

@@ -104,11 +104,17 @@ class HDF5TomoScan(TomoScanBase):
                 self.master_file = scan
                 scan = os.path.dirname(scan)
             else:
+                self.master_file = self.get_master_file(scan)
+
                 self.master_file = os.path.join(scan, os.path.basename(scan))
-                if os.path.exists(self.master_file + '.hdf5'):
+                if os.path.exists(self.master_file + '.nx'):
+                    self.master_file = self.master_file + '.nx'
+                elif os.path.exists(self.master_file + '.hdf5'):
                     self.master_file = self.master_file + '.hdf5'
-                else:
+                elif os.path.exists(self.master_file + '.h5'):
                     self.master_file = self.master_file + '.h5'
+                else:
+                    self.master_file = self.master_file + '.nx'
         else:
             self.master_file = None
 
@@ -143,6 +149,22 @@ class HDF5TomoScan(TomoScanBase):
         self._rotation_angles = None
         self._distance = None
 
+    @staticmethod
+    def get_master_file(scan_path):
+        if os.path.isfile(scan_path):
+            master_file = scan_path
+        else:
+            master_file = os.path.join(scan_path, os.path.basename(scan_path))
+            if os.path.exists(master_file + '.nx'):
+                master_file = master_file + '.nx'
+            elif os.path.exists(master_file + '.hdf5'):
+                master_file = master_file + '.hdf5'
+            elif os.path.exists(master_file + '.h5'):
+                master_file = master_file + '.h5'
+            else:
+                master_file = master_file + '.nx'
+        return master_file
+
     @docstring(TomoScanBase.clear_caches)
     def clear_caches(self) -> None:
         self._projections = None
@@ -186,6 +208,9 @@ class HDF5TomoScan(TomoScanBase):
         res = []
         res_buf = []
 
+        if not os.path.isfile(file_path):
+            raise ValueError('given file path should be a file')
+
         with h5py.File(file_path, 'r') as h5f:
             for root_node in h5f.keys():
                 node = h5f[root_node]
@@ -212,12 +237,14 @@ class HDF5TomoScan(TomoScanBase):
 
     @docstring(TomoScanBase.is_tomoscan_dir)
     @staticmethod
-    def is_tomoscan_dir(directory: str, **kwargs):
-        master_file_base = os.path.join(directory, os.path.basename(directory))
-        if os.path.exists(master_file_base + '.hdf5'):
-            return True
+    def is_tomoscan_dir(directory: str, **kwargs) -> bool:
+        if os.path.isfile(directory):
+            master_file = directory
         else:
-            return os.path.exists(master_file_base + '.h5')
+            master_file = HDF5TomoScan.get_master_file(scan_path=directory)
+        if master_file:
+            entries = HDF5TomoScan._get_valid_entries(file_path=master_file)
+            return len(entries) > 0
 
     @docstring(TomoScanBase.is_abort)
     def is_abort(self, **kwargs):

tomoscan/scanfactory.py

@@ -41,11 +41,10 @@ class ScanFactory:
     """
 
     @staticmethod
-    def create_scan_object(scan_path):
+    def create_scan_object(scan_path: str) -> TomoScanBase:
         """
         
-        :param scan_path: path to the scan directory or file
-        :rtype: TextIOWrapper
+        :param str scan_path: path to the scan directory or file
         :return: ScanBase instance fitting the scan folder or scan path
         :rtype: TomoScanBase
         """
@@ -59,7 +58,30 @@ class ScanFactory:
         raise ValueError('%s is not a valid scan path' % scan_path)
 
     @staticmethod
-    def create_scan_object_frm_dict(_dict):
+    def create_scan_objects(scan_path: str) -> tuple:
+        """
+
+        :param str scan_path: path to the scan directory or file
+        :return: all possible instances of TomoScanBase contained in the given
+                 path
+        :rtype: tuple
+        """
+        scan_path = scan_path.rstrip(os.path.sep)
+        if EDFTomoScan.is_tomoscan_dir(scan_path):
+            return (EDFTomoScan(scan=scan_path), )
+        elif HDF5TomoScan.is_tomoscan_dir(scan_path):
+            scans = []
+            master_file = HDF5TomoScan.get_master_file(scan_path=scan_path)
+            entries = HDF5TomoScan._get_valid_entries(master_file)
+            for entry in entries:
+                scans.append(HDF5TomoScan(scan=scan_path, entry=entry,
+                                          index=None))
+            return tuple(scans)
+
+        raise ValueError('%s is not a valid scan path' % scan_path)
+
+    @staticmethod
+    def create_scan_object_frm_dict(_dict: dict) -> TomoScanBase:
         """
         Create a TomoScanBase instance from a dictionary. It should contains
         the TomoScanBase._DICT_TYPE_KEY key at least.
@@ -78,7 +100,7 @@ class ScanFactory:
                              'is not managed')
 
     @staticmethod
-    def is_tomoscan_dir(scan_path):
+    def is_tomoscan_dir(scan_path: str) -> bool:
         """
         
         :param str scan_path: path to the scan directory or file
@@ -91,7 +113,7 @@ class ScanFactory:
         )
 
     @staticmethod
-    def create_from_json(desc):
+    def create_from_json(desc: dict) -> TomoScanBase:
         """Create a ScanBase instance from a json description"""
         data = json.load(desc)
 

tomoscan/test/test_factory.py

@@ -29,6 +29,7 @@ __date__ = "24/01/2017"
 
 
 import unittest
+import os
 from ..esrf.edfscan import EDFTomoScan
 from ..esrf.hdf5scan import HDF5TomoScan
 from ..scanfactory import ScanFactory
@@ -58,54 +59,35 @@ class TestScanFactory(unittest.TestCase):
     def test_one_nx(self):
         """Can we create a TomoScanBase from a .nx master file containing
         one acquisition"""
-        scan_dir = UtilsTest.getH5Dataset('frm_edftomomill_oneentry.nx')
-        scan = ScanFactory.create_scan_object(scan_dir)
+        file_name = 'frm_edftomomill_oneentry.nx'
+        master_file = UtilsTest.getH5Dataset(file_name)
+        scan = ScanFactory.create_scan_object(master_file)
         self.assertTrue(isinstance(scan, HDF5TomoScan))
-        self.assertEqual(scan.path, DataUrl(scan_dir, data_path='entry',
-                                            scheme='tomoscan'))
+        self.assertEqual(scan.path, os.path.dirname(master_file))
+        self.assertEqual(scan.master_file, master_file)
+        self.assertEqual(scan.entry, '/entry')
 
-    # def test_one_two_nx(self):
-    #     """Can we create a TomoScanBase from a .nx master file containing
-    #     two acquisitions"""
-    #     scan_dir = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
-    #     scan = ScanFactory.create_scan_object(scan_dir)
-    #     self.assertTrue(isinstance(scan, HDF5TomoScan))
-    #     self.assertEqual(scan.path, DataUrl(scan_dir, data_path='entry0000',
-    #                                         scheme='tomoscan'))
-    #
-    # def test_one_frm_two_nx_with_index(self):
-    #     """Can we create a TomoScanBase from a .nx master file containing
-    #     two acquisitions. Pick the one at given index"""
-    #     scan_dir = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
-    #     scan = ScanFactory.create_scan_object(scan_dir)
-    #     self.assertTrue(isinstance(scan, HDF5TomoScan))
-    #     self.assertEqual(scan.path, DataUrl(scan_dir, data_path='entry0000',
-    #                                         scheme='tomoscan'))
-    #
-    # def test_one_frm_two_nx_with_entry_name(self):
-    #     """Can we create a TomoScanBase from a .nx master file containing
-    #     two acquisitions. Pick the one with the given name"""
-    #     scan_dir = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
-    #     scan = ScanFactory.create_scan_object(scan_dir)
-    #     self.assertTrue(isinstance(scan, HDF5TomoScan))
-    #     self.assertEqual(scan.path, DataUrl(scan_dir, data_path='entry0000',
-    #                                         scheme='tomoscan'))
+    def test_one_two_nx(self):
+        """Can we create a TomoScanBase from a .nx master file containing
+        two acquisitions"""
+        master_file = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
+        scan = ScanFactory.create_scan_object(master_file)
+        self.assertTrue(isinstance(scan, HDF5TomoScan))
+        self.assertEqual(scan.path, os.path.dirname(master_file))
+        self.assertEqual(scan.master_file, master_file)
+        self.assertEqual(scan.entry, '/entry0000')
 
-    #
-    # def test_two_nx(self):
-    #     """Can we create two TomoScanBase from a .nx master file containing
-    #     two acquisitions using the ScanFactory"""
-    #     scan_dir = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
-    #     scans = ScanFactory.create_scan_objects(scan_dir)
-    #     scan_0000, scan_0001 = scans
-    #     self.assertTrue(isinstance(scan_0000, EDFTomoScan))
-    #     self.assertTrue(isinstance(scan_0001, EDFTomoScan))
-    #     self.assertEqual(scan_0000.path, DataUrl(scan_dir,
-    #                                              data_path='entry0000',
-    #                                              scheme='tomoscan'))
-    #     self.assertEqual(scan_0001.path, DataUrl(scan_dir,
-    #                                              data_path='entry0001',
-    #                                              scheme='tomoscan'))
+    def test_two_nx(self):
+        """Can we create two TomoScanBase from a .nx master file containing
+        two acquisitions using the ScanFactory"""
+        master_file = UtilsTest.getH5Dataset('frm_edftomomill_twoentries.nx')
+        scans = ScanFactory.create_scan_objects(master_file)
+        self.assertEqual(len(scans), 2)
+        for scan, scan_entry in zip(scans, ('/entry0000', '/entry0001')):
+            self.assertTrue(isinstance(scan, HDF5TomoScan))
+            self.assertEqual(scan.path, os.path.dirname(master_file))
+            self.assertEqual(scan.master_file, master_file)
+            self.assertEqual(scan.entry, scan_entry)
 
 
 def suite():