bugfix fo h5 file path when uri of file is given as object uri (occurs for etp support)

Author: valentin-gauthier-geosiris

energyml-utils/example/attic/arrays_test.py

@@ -383,7 +383,8 @@ def read_wellbore_frame_repr_demo_jfr_02_26(
     epc_path: str = r"rc/epc/out-galaxy-12-pts.epc",
     well_uuid: str = "cfad9cb6-99fe-4172-b560-d2feca75dd9f",
 ) -> List[AbstractMesh]:
-    epc = Epc.read_file(f"{epc_path}", read_rels_from_files=False, recompute_rels=False)
+    # epc = Epc.read_file(f"{epc_path}", read_rels_from_files=False, recompute_rels=False)
+    epc = EpcStreamReader(f"{epc_path}", rels_update_mode=RelsUpdateMode.MANUAL)
 
     frame_repr = epc.get_object_by_uuid(well_uuid)[0]
     # print(frame_repr)
@@ -471,7 +472,7 @@ def test_read_write_array(h5_path):
     # meshes = read_representation_set_representation()
     # meshes = read_trset()
     # meshes = read_pointset()
-    # meshes = read_wellbore_frame_repr_demo_jfr_02_26()
+    meshes = read_wellbore_frame_repr_demo_jfr_02_26()
 
     print(f"Number of meshes read: {len(meshes)}")
 
@@ -494,4 +495,4 @@ def test_read_write_array(h5_path):
                     raise e
 
     # read_props_and_cbt()
-    test_read_write_array("test_array_rw.h5")
+    # test_read_write_array("test_array_rw.h5")

energyml-utils/example/attic/test_list_object.py

@@ -4,17 +4,25 @@
 
 def list_epc_classical(epc_file):
     """List contents of an EPC file."""
-    epc = EpcStreamReader(epc_file, rels_update_mode=RelsUpdateMode.MANUAL)
 
-    time_start = datetime.now()
-    # for obj in epc.list_objects():
-    #     print(f"Object: {obj}")
-    print(len(epc.list_objects(object_type="eml23.DataobjectCollection")))
+    if not isinstance(epc_file, list):
+        epc_file = [epc_file]
 
-    for obj in epc.list_objects(object_type="eml23.DataobjectCollection"):
-        print(f"DataobjectCollection: {obj}")
-    time_end = datetime.now()
-    print(f"Time taken: {time_end - time_start}")
+    for f in epc_file:
+        print(f"Processing EPC file: {f}")
+        epc = EpcStreamReader(f, rels_update_mode=RelsUpdateMode.MANUAL)
+
+        time_start = datetime.now()
+        # for obj in epc.list_objects():
+        #     print(f"Object: {obj}")
+        print(len(epc.list_objects(object_type="resqml22.BoundaryFeature")))
+
+        for obj in sorted(epc.list_objects(object_type="resqml22.BoundaryFeature"), key=lambda o: o.title):
+            print(f"BoundaryFeature: {obj}")
+        for obj in sorted(epc.list_objects(object_type="resqml22.RockVolumeFeature"), key=lambda o: o.title):
+            print(f"RockVolumeFeature: {obj}")
+        time_end = datetime.now()
+        print(f"Time taken: {time_end - time_start}")
 
 
 # def list_epc_fast(epc_file):
@@ -48,12 +56,16 @@ def list_epc_classical(epc_file):
 
 
 if __name__ == "__main__":
-    epc_file = "D:/Geosiris/Clients/BRGM/git/pointset-extraction/rc/output/full-local/full-local.epc"
+    epc_file = [
+        "D:/Geosiris/Clients/BRGM/git/pointset-extraction/rc/output/full-local/full-local.epc",
+        "D:/Geosiris/Clients/BRGM/git/csv-to-energyml/rc/output/full-local/result-out-local-egis-full.epc",
+    ]
+    # epc_file = "D:/Geosiris/Clients/BRGM/git/pointset-extraction/rc/output/full-local/full-local.epc"
     print("Listing EPC contents (classical method):")
     list_epc_classical(epc_file)
 
-    # print("\nListing EPC contents (fast method):")
+    # print("Listing EPC contents (fast method):")
     # list_epc_fast(epc_file)
 
-    # print("\nListing EPC contents (sequential method):")
+    # print("Listing EPC contents (sequential method):")
     # list_epc_seq(epc_file)

energyml-utils/src/energyml/utils/epc.py

@@ -1305,13 +1305,13 @@ def get_h5_file_paths(self, obj_or_id: Optional[Any] = None) -> List[str]:
 
         h5_paths = set(make_path_relative_to_filepath_list(list(h5_paths), self.epc_file_path))
 
-        if len(h5_paths) == 0:
-            # Collect all .h5 files in the EPC file's folder
-            epc_folder = self.get_epc_file_folder()
-            if epc_folder is not None and os.path.isdir(epc_folder):
-                for fname in os.listdir(epc_folder):
-                    if fname.lower().endswith(".h5"):
-                        h5_paths.add(os.path.join(epc_folder, fname))
+        # if len(h5_paths) == 0:
+        # Collect all .h5 files in the EPC file's folder
+        epc_folder = self.get_epc_file_folder()
+        if epc_folder is not None and os.path.isdir(epc_folder):
+            for fname in os.listdir(epc_folder):
+                if fname.lower().endswith(".h5"):
+                    h5_paths.add(os.path.join(epc_folder, fname))
 
         return list(h5_paths)
 

energyml-utils/src/energyml/utils/epc_stream.py

@@ -1332,7 +1332,9 @@ def get_statistics(self) -> EpcStreamingStats:
         """Get current statistics about the EPC streaming operations."""
         return self.stats
 
-    def get_h5_file_paths(self, obj: Union[str, Uri, Any], make_path_absolute_from_epc_path: bool = True) -> List[str]:
+    def get_h5_file_paths(
+        self, obj_or_id: Union[str, Uri, Any] = None, make_path_absolute_from_epc_path: bool = True
+    ) -> List[str]:
         """
         Get all HDF5 file paths referenced in the EPC file (from rels to external resources).
         Optimized to avoid loading the object when identifier/URI is provided.
@@ -1347,7 +1349,7 @@ def get_h5_file_paths(self, obj: Union[str, Uri, Any], make_path_absolute_from_e
 
         rels_path = None
 
-        _id = self._id_from_uri_or_identifier(identifier=obj, get_first_if_simple_uuid=True)
+        _id = self._id_from_uri_or_identifier(identifier=obj_or_id, get_first_if_simple_uuid=True)
         if _id is not None:
             rels_path = self._metadata_mgr.gen_rels_path_from_identifier(_id)
 
@@ -1372,13 +1374,13 @@ def get_h5_file_paths(self, obj: Union[str, Uri, Any], make_path_absolute_from_e
         if make_path_absolute_from_epc_path:
             h5_paths = set(make_path_relative_to_filepath_list(list(h5_paths), self.epc_file_path))
 
-        if len(h5_paths) == 0:
-            # Collect all .h5 files in the EPC file's folder
-            epc_folder = get_file_folder(self.epc_file_path)
-            if epc_folder is not None and os.path.isdir(epc_folder):
-                for fname in os.listdir(epc_folder):
-                    if fname.lower().endswith(".h5"):
-                        h5_paths.add(os.path.join(epc_folder, fname))
+        # if len(h5_paths) == 0:
+        # Collect all .h5 files in the EPC file's folder
+        epc_folder = get_file_folder(self.epc_file_path)
+        if epc_folder is not None and os.path.isdir(epc_folder):
+            for fname in os.listdir(epc_folder):
+                if fname.lower().endswith(".h5"):
+                    h5_paths.add(os.path.join(epc_folder, fname))
 
         return list(h5_paths)