Dev 05 25 (#13)

* adding type tips in random data generation if type not found

* bugfix for h5 dataset reading

* including json property dict

Author: valentin-gauthier-geosiris

energyml-utils/example/main.py

@@ -7,6 +7,7 @@
 from energyml.eml.v2_3.commonv2 import *
 from energyml.eml.v2_3.commonv2 import AbstractObject
 from energyml.resqml.v2_0_1.resqmlv2 import DoubleHdf5Array
+from energyml.resqml.v2_0_1.resqmlv2 import TriangulatedSetRepresentation as Tr20
 from energyml.resqml.v2_2.resqmlv2 import (
     TriangulatedSetRepresentation,
     FaultInterpretation,
@@ -484,4 +485,11 @@ def test_dor_conversion():
     # class_field()
     # test_get_projected_uom()
     # test_crs()
-    test_dor_conversion()
+    # test_dor_conversion()
+    print(get_obj_uri(tr, "coucou"))
+
+    tr201 = Tr20(
+        citation=tr_cit,
+        uuid=gen_uuid(),
+    )
+    print(get_obj_uri(tr201, "coucou"))

energyml-utils/example/tools.py

@@ -8,11 +8,13 @@
 
 from src.energyml.utils.constants import get_property_kind_dict_path_as_xml
 from src.energyml.utils.data.datasets_io import CSVFileReader, HDF5FileWriter, ParquetFileWriter, DATFileReader
-from src.energyml.utils.data.mesh import MeshFileFormat, export_multiple_data
+from src.energyml.utils.data.mesh import MeshFileFormat, export_multiple_data, export_obj, read_mesh_object
 from src.energyml.utils.epc import Epc, gen_energyml_object_path
 from src.energyml.utils.introspection import (
     get_class_from_simple_name,
+    get_module_name_and_type_from_content_or_qualified_type,
     random_value_from_class,
+    search_class_in_module_from_partial_name,
     set_attribute_from_path,
     get_object_attribute,
     get_qualified_type_from_class,
@@ -294,6 +296,14 @@ def generate_data():
     except NameError:
         obj_class = get_class_from_qualified_type(args.type)
 
+    if obj_class is None:
+        print("Class not found, please check the type name.")
+        print("Possible types are :")
+        module_name, object_type = get_module_name_and_type_from_content_or_qualified_type(args.type)
+        for cn in search_class_in_module_from_partial_name(module_name, object_type):
+            print(f" - {cn.__name__}")
+        return
+
     obj = random_value_from_class(obj_class)
     if args.file_format.lower() == "xml":
         print(serialize_xml(obj))
@@ -524,3 +534,26 @@ def describe_as_csv():
             out.write("\n")
 
     print("Finished")
+
+
+# def export_wavefront():
+#     parser = argparse.ArgumentParser()
+#     parser.add_argument("--epc", "-f", type=str, help="Epc file path")
+#     parser.add_argument("--output", "-o", type=str, help="Output folder path")
+#     parser.add_argument("--uuid", "-u", type=str, help="The uuids of representations to extract", nargs="+")
+
+#     args = parser.parse_args()
+
+#     epc = Epc.read_file(args.epc)
+#     for uuid in args.uuid:
+#         obj = epc.get_object_by_uuid(uuid)[0]
+
+#         mesh = read_mesh_object(
+#             energyml_object=obj,
+#             workspace=epc,
+#         )
+
+#         if obj is not None:
+#             fname = gen_energyml_object_path(obj)
+#             with open(os.path.join(args.output, fname + ".obj"), "w") as f:
+#                 export_obj(mesh_list=mesh, out=f)  # Assuming the object can be serialized to XML

energyml-utils/pyproject.toml

@@ -38,6 +38,10 @@ keywords = ["energyml", "resqml", "witsml", "prodml"]
 packages = [
     { include = "energyml", from = "src" },
 ]
+include = [
+    "src/energyml/utils/rc/*.xml",
+    "src/energyml/utils/rc/*.json"
+]
 #exclude = [
 #    "src/energyml/main.py"
 #]

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

@@ -52,6 +52,23 @@
 # HDF5
 if __H5PY_MODULE_EXISTS__:
 
+    def h5_list_datasets(h5_file_path):
+        """
+        List all datasets in an HDF5 file.
+        :param h5_file_path: Path to the HDF5 file
+        :return: List of dataset names in the HDF5 file
+        """
+        res = []
+        with h5py.File(h5_file_path, "r") as f:
+            # Function to print the names of all datasets
+            def list_datasets(name, obj):
+                if isinstance(obj, h5py.Dataset):  # Check if the object is a dataset
+                    res.append(name)
+
+            # Visit all items in the HDF5 file and apply the list function
+            f.visititems(list_datasets)
+        return res
+
     @dataclass
     class HDF5FileReader(DatasetReader):
         def read_array(self, source: Union[BytesIO, str], path_in_external_file: str) -> Optional[List[Any]]:
@@ -481,10 +498,8 @@ def get_external_file_path_from_external_path(
 
 def get_external_file_path_possibilities_from_folder(file_raw_path: str, folder_path: str) -> List[str]:
     external_path_respect = file_raw_path
-    external_path_rematch = (
-        f"{folder_path + '/' if folder_path is not None and len(folder_path) else ''}{os.path.basename(file_raw_path)}"
-    )
-    external_path_no_folder = f"{os.path.basename(file_raw_path)}"
+    external_path_rematch = f"{folder_path + '/' if folder_path is not None and len(folder_path) else ''}{os.path.basename(file_raw_path or '')}"
+    external_path_no_folder = f"{os.path.basename(file_raw_path)}" if file_raw_path is not None else ""
 
     return [
         external_path_respect,
@@ -542,7 +557,8 @@ def read_external_dataset_array(
                 break  # stop after the first read success
             except MissingExtraInstallation as mei:
                 raise mei
-            except OSError as e:
+            except Exception as e:
+                logging.debug(f"Failed to read external file {s} for {path_in_obj} with path {path_in_external} : {e}")
                 pass
         if not succeed:
             raise Exception(f"Failed to read external file. Paths tried : {sources}")

energyml-utils/src/energyml/utils/introspection.py

@@ -97,6 +97,24 @@ def find_class_in_module(module_name, class_name):
     return None
 
 
+def search_class_in_module_from_partial_name(module_name: str, class_partial_name: str) -> Optional[List[type]]:
+    """
+    Search a class in a module using a partial name.
+    :param module_name: The name of the module to search in.
+    :param class_partial_name: The partial name of the class to search for.
+    :return: A list of classes that match the partial name.
+
+    """
+    try:
+        module = import_module(module_name)
+        classes = get_module_classes_from_name(module_name)
+        matching_classes = [cls for cls_name, cls in classes if class_partial_name.lower() in cls_name.lower()]
+        return matching_classes
+    except ImportError as e:
+        logging.error(f"Module '{module_name}' not found: {e}")
+    return None
+
+
 def get_class_methods(cls: Union[type, Any]) -> List[str]:
     """
     Returns the list of the methods names for a specific class.
@@ -196,44 +214,32 @@ def get_energyml_class_in_related_dev_pkg(cls: type):
     return res
 
 
-def get_class_from_qualified_type(qualified_type: str) -> Optional[type]:
-    return get_class_from_content_type(qualified_type)
-
-
-def get_class_from_content_type(content_type: str) -> Optional[type]:
+def get_module_name_and_type_from_content_or_qualified_type(cqt: str) -> Tuple[str, str]:
     """
-    Return a :class:`type` object matching with the content-type :param:`content_type`.
-    :param content_type:
-    :return:
+    Return a tuple (module_name, type) from a content-type or qualified-type string.
     """
     ct = None
     try:
-        ct = parse_content_type(content_type)
+        ct = parse_content_type(cqt)
     except AttributeError:
         pass
     if ct is None:
         try:
-            ct = parse_qualified_type(content_type)
+            ct = parse_qualified_type(cqt)
         except AttributeError:
             pass
 
     domain = ct.group("domain")
     if domain is None:
         # logging.debug(f"\tdomain {domain} xmlDomain {ct.group('xmlDomain')} ")
         domain = "opc"
+
     if domain == "opc":
         xml_domain = ct.group("xmlDomain")
         if "." in xml_domain:
             xml_domain = xml_domain[xml_domain.rindex(".") + 1 :]
-        # Don't know what to do with http://schemas.f2i-consulting.com/package/2014/metadata/extended-core-properties
-        # if "extended" in xml_domain:
-        #     xml_domain = xml_domain.replace("extended", "")
-        #     if xml_domain.startswith("-"):
-        #         xml_domain = xml_domain[1:]
-        #
         opc_type = pascal_case(xml_domain).replace("-", "")
-        # logging.debug("\tenergyml.opc.opc." + opc_type)
-        return get_class_from_name("energyml.opc.opc." + opc_type)
+        return ("energyml.opc.opc", opc_type)
     else:
         domain = ct.group("domain")
         obj_type = ct.group("type")
@@ -246,10 +252,21 @@ def get_class_from_content_type(content_type: str) -> Optional[type]:
         if domain.lower() == "resqml" and version_num.startswith("2_0"):
             version_num = "2_0_1"
 
-        # logging.debug(get_module_name(domain, version_num)
-        #     + "."
-        #     + obj_type)
-        return get_class_from_name(get_module_name(domain, version_num) + "." + obj_type)
+        return (get_module_name(domain, version_num), obj_type)
+
+
+def get_class_from_qualified_type(qualified_type: str) -> Optional[type]:
+    return get_class_from_content_type(qualified_type)
+
+
+def get_class_from_content_type(content_type: str) -> Optional[type]:
+    """
+    Return a :class:`type` object matching with the content-type :param:`content_type`.
+    :param content_type:
+    :return:
+    """
+    module_name, object_type = get_module_name_and_type_from_content_or_qualified_type(content_type)
+    return get_class_from_name(module_name + "." + object_type)
 
 
 def get_module_name(domain: str, domain_version: str):
@@ -1078,7 +1095,7 @@ def get_obj_pkg_pkgv_type_uuid_version(
 
     if ct is not None:
         ct_match = parse_content_type(ct)
-        logging.debug("ct : ", ct_match)
+        logging.debug("ct : %S", ct_match)
         if ct_match is not None:
             pkg = ct_match.group("domain")
             pkg_v = ct_match.group("domainVersion")
@@ -1087,7 +1104,7 @@ def get_obj_pkg_pkgv_type_uuid_version(
         try:
             qt = get_object_attribute_no_verif(obj, "qualified_type")
             qt_match = parse_qualified_type(qt)
-            logging.debug("qt : ", qt, obj.__dict__, qt_match)
+            logging.debug("qt : %s %s", qt, obj.__dict__, qt_match)
             if qt_match is not None:
                 pkg = qt_match.group("domain")
                 pkg_v = qt_match.group("domainVersion")
@@ -1392,11 +1409,14 @@ def random_value_from_class(cls: type):
     if not is_primitive(cls):
         # import_related_module(cls.__module__)
         energyml_module_context = get_related_energyml_modules_name(cls)
-    return _random_value_from_class(
-        cls=cls,
-        energyml_module_context=energyml_module_context,
-        attribute_name=None,
-    )
+    if cls is not None:
+        return _random_value_from_class(
+            cls=cls,
+            energyml_module_context=energyml_module_context,
+            attribute_name=None,
+        )
+    else:
+        return None
 
 
 def _random_value_from_class(