read Epc in parallel to increase efficiency

+ adding ".dat" as possible file extension for hdf5 handler and set h5Handler as default

Author: valentin-gauthier-geosiris

energyml-utils/example/attic/compare_inmem_n_stream.py

@@ -39,14 +39,31 @@ def reexport_in_memory(filepath: str, output_folder: Optional[str] = None):
     if output_folder:
         os.makedirs(output_folder, exist_ok=True)
         path_in_memory = f"{output_folder}/{path_in_memory.split('/')[-1]}"
-    epc = Epc.read_file(epc_file_path=filepath, read_rels_from_files=False)
+    epc = Epc.read_file(epc_file_path=filepath, read_rels_from_files=False, recompute_rels=False)
 
     if os.path.exists(path_in_memory):
         os.remove(path_in_memory)
     epc.export_file(path_in_memory)
 
 
-def time_comparison(filepath: str, output_folder: Optional[str] = None, skip_sequential: bool = True):
+def reexport_in_memory_par_read(filepath: str, output_folder: Optional[str] = None):
+    path_in_memory = filepath.replace(".epc", "_in_memory_par_read.epc")
+    if output_folder:
+        os.makedirs(output_folder, exist_ok=True)
+        path_in_memory = f"{output_folder}/{path_in_memory.split('/')[-1]}"
+    epc = Epc.read_file(epc_file_path=filepath, read_rels_from_files=False, read_parallel=True, recompute_rels=False)
+
+    if os.path.exists(path_in_memory):
+        os.remove(path_in_memory)
+    epc.export_file(path_in_memory, parallel=True)
+
+
+def time_comparison(
+    filepath: str,
+    output_folder: Optional[str] = None,
+    skip_sequential_stream: bool = True,
+    skip_parallel_stream: bool = True,
+):
     """Compare performance of different EPC reexport methods."""
     print(f"\n{'=' * 70}")
     print(f"Performance Comparison: {filepath.split('/')[-1]}")
@@ -62,7 +79,15 @@ def time_comparison(filepath: str, output_folder: Optional[str] = None, skip_seq
     results.append(("In-Memory (Epc)", elapsed_inmem))
     print(f"   ✓ Completed in {elapsed_inmem:.3f}s\n")
 
-    if not skip_sequential:
+    # Test 1b: In-Memory with Parallel Read
+    print("⏳ Testing In-Memory EPC processing with Parallel Read...")
+    start = time.perf_counter()
+    reexport_in_memory_par_read(filepath, output_folder)
+    elapsed_inmem_par = time.perf_counter() - start
+    results.append(("In-Memory (Epc) Parallel Read", elapsed_inmem_par))
+    print(f"   ✓ Completed in {elapsed_inmem_par:.3f}s\n")
+
+    if not skip_sequential_stream:
         # Test 2: Streaming Sequential
         print("⏳ Testing Streaming Sequential processing...")
         start = time.perf_counter()
@@ -72,12 +97,13 @@ def time_comparison(filepath: str, output_folder: Optional[str] = None, skip_seq
         print(f"   ✓ Completed in {elapsed_seq:.3f}s\n")
 
     # Test 3: Streaming Parallel
-    print("⏳ Testing Streaming Parallel processing...")
-    start = time.perf_counter()
-    reexport_stream_parallel(filepath, output_folder)
-    elapsed_parallel = time.perf_counter() - start
-    results.append(("Stream Parallel", elapsed_parallel))
-    print(f"   ✓ Completed in {elapsed_parallel:.3f}s\n")
+    if not skip_parallel_stream:
+        print("⏳ Testing Streaming Parallel processing...")
+        start = time.perf_counter()
+        reexport_stream_parallel(filepath, output_folder)
+        elapsed_parallel = time.perf_counter() - start
+        results.append(("Stream Parallel", elapsed_parallel))
+        print(f"   ✓ Completed in {elapsed_parallel:.3f}s\n")
 
     # Calculate speedups
     results_sorted = sorted(results, key=lambda x: x[1])
@@ -119,15 +145,15 @@ def time_comparison(filepath: str, output_folder: Optional[str] = None, skip_seq
 
     update_prop_kind_dict_cache()
 
-    time_comparison(
-        filepath=sys.argv[1] if len(sys.argv) > 1 else "rc/epc/testingPackageCpp22.epc",
-        output_folder="rc/performance_results",
-    )
-
     # time_comparison(
-    #     filepath=sys.argv[1] if len(sys.argv) > 1 else "rc/epc/80wells_surf.epc", output_folder="rc/performance_results"
+    #     filepath=sys.argv[1] if len(sys.argv) > 1 else "rc/epc/testingPackageCpp22.epc",
+    #     output_folder="rc/performance_results",
     # )
 
+    time_comparison(
+        filepath=sys.argv[1] if len(sys.argv) > 1 else "rc/epc/80wells_surf.epc", output_folder="rc/performance_results"
+    )
+
     # time_comparison(
     #     filepath=sys.argv[1] if len(sys.argv) > 1 else "wip/failingData/fix/sample_mini_firp_201_norels_with_media.epc",
     #     output_folder="rc/performance_results",

energyml-utils/src/energyml/utils/data/datasets_io.py

@@ -756,9 +756,11 @@ def _register_default_handlers(self, max_open_files: int) -> None:
         """Register all available handlers based on installed dependencies."""
         # HDF5 Handler
         if __H5PY_MODULE_EXISTS__:
-            self.register_handler([".h5", ".hdf5"], lambda: HDF5ArrayHandler())
+            self.register_handler([".h5", ".hdf5", ".dat"], lambda: HDF5ArrayHandler())  # dat for Galaxy compatibility
         else:
-            self.register_handler([".h5", ".hdf5"], lambda: MockHDF5ArrayHandler())
+            self.register_handler(
+                [".h5", ".hdf5", ".dat"], lambda: MockHDF5ArrayHandler()
+            )  # dat for Galaxy compatibility
 
         # Parquet Handler
         if __PARQUET_MODULE_EXISTS__:
@@ -802,13 +804,18 @@ def get_handler_for_file(self, file_path: str) -> Optional[ExternalArrayHandler]
             file_path: Path to the file
 
         Returns:
-            Handler instance, or None if no handler registered for this extension
+            Handler instance, or h5 handler if extension not found but h5 handler is available and not mock, else None
         """
         ext = os.path.splitext(file_path)[1].lower()
 
         if ext in self._handlers:
             return self._handlers[ext]()
 
+        # search for h5 handler if not mock and return it by default
+        if ".h5" in self._handlers:
+            h = self._handlers[".h5"]()
+            if "mock" not in h.__class__.__name__.lower():
+                return self._handlers[".h5"]()
         return None
 
     def supports_extension(self, extension: str) -> bool:
@@ -973,7 +980,7 @@ def list_arrays(self, source: Union[BytesIO, str, Any]) -> List[str]:
         def can_handle_file(self, file_path: str) -> bool:
             """Check if this handler can process the file."""
             ext = os.path.splitext(file_path)[1].lower()
-            return ext in [".h5", ".hdf5"]
+            return ext in [".h5", ".hdf5", ".dat"]  # dat for Galaxy compatibility
 
 else:
 
@@ -1019,7 +1026,7 @@ def list_arrays(self, source: Union[BytesIO, str, Any]) -> List[str]:
             raise MissingExtraInstallation(extra_name="hdf5")
 
         def can_handle_file(self, file_path: str) -> bool:
-            return os.path.splitext(file_path)[1].lower() in [".h5", ".hdf5"]
+            return os.path.splitext(file_path)[1].lower() in [".h5", ".hdf5", ".dat"]  # dat for Galaxy compatibility
 
 
 # Parquet Handler

energyml-utils/src/energyml/utils/data/mesh.py

@@ -241,7 +241,7 @@ def read_mesh_object(
         ):  # WellboreFrameRep has allready the displacement applied
             # TODO: the displacement should be done in each reader function to manage specific cases
             for s in surfaces:
-                print("CRS : ", s.crs_object.uuid if s.crs_object is not None else "None")
+                logging.debug(f"CRS : {s.crs_object.uuid if s.crs_object is not None else 'None'}")
                 crs_displacement(s.point_list, s.crs_object)
         return surfaces
     else:
@@ -833,7 +833,11 @@ def read_wellbore_trajectory_representation(
 
     # Get CRS from trajectory geometry if available
     try:
-        crs = workspace.get_object(get_obj_uri(get_object_attribute(energyml_object, "geometry.LocalCrs")))
+        crs_attr = get_object_attribute(energyml_object, "geometry.LocalCrs")
+        if crs_attr is not None:
+            crs = workspace.get_object(get_obj_uri(crs_attr))
+        else:
+            raise ObjectNotFoundNotError("LocalCrs attribute not found in trajectory geometry")
     except Exception:
         logging.debug("Could not get CRS from trajectory geometry")