Validation (#14)

* access to property kind by uuid and better validation

* bugfix : obj_ prefix is now printed for resqml201 objects

Author: valentin-gauthier-geosiris

energyml-utils/README.md

@@ -40,7 +40,7 @@ This package supports read/write in xml/json the following packages :
 - WITSMl : 2.0, 2.1
 - PRODML : 2.0, 2.2
 
-/!\\ By default, these package are not installed and are published independently.
+/!\\ By default, these packages are not installed and are published independently.
 You can install only the versions you need by adding the following lines in the .toml file : 
 ```toml
 energyml-common2-0 = "^1.12.0"

energyml-utils/example/main.py

@@ -486,10 +486,13 @@ def test_dor_conversion():
     # test_get_projected_uom()
     # test_crs()
     # test_dor_conversion()
-    print(get_obj_uri(tr, "coucou"))
+    # print(get_obj_uri(tr, "coucou"))
 
     tr201 = Tr20(
         citation=tr_cit,
         uuid=gen_uuid(),
     )
-    print(get_obj_uri(tr201, "coucou"))
+    # print(get_obj_uri(tr201, "coucou"))
+
+    print(get_usable_class(tr))
+    print(get_usable_class(tr201))

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

@@ -52,7 +52,7 @@
 # HDF5
 if __H5PY_MODULE_EXISTS__:
 
-    def h5_list_datasets(h5_file_path):
+    def h5_list_datasets(h5_file_path: Union[BytesIO, str]) -> List[str]:
         """
         List all datasets in an HDF5 file.
         :param h5_file_path: Path to the HDF5 file
@@ -93,28 +93,31 @@ def extract_h5_datasets(
             :param h5_datasets_paths:
             :return:
             """
+            if h5_datasets_paths is None:
+                h5_datasets_paths = h5_list_datasets(input_h5)
             if len(h5_datasets_paths) > 0:
-                with h5py.File(output_h5, "w") as f_dest:
+                with h5py.File(output_h5, "a") as f_dest:
                     with h5py.File(input_h5, "r") as f_src:
                         for dataset in h5_datasets_paths:
                             f_dest.create_dataset(dataset, data=f_src[dataset])
 
     @dataclass
     class HDF5FileWriter:
+
         def write_array(
             self,
             target: Union[str, BytesIO, bytes],
-            array: Union[list, np.array],
+            array: Union[list, np.ndarray],
             path_in_external_file: str,
-            dtype: Optional = None,
+            dtype: Optional[np.dtype] = None,
         ):
             if isinstance(array, list):
                 array = np.asarray(array)
             print("writing array", target)
             with h5py.File(target, "a") as f:
                 # print(array.dtype, h5py.string_dtype(), array.dtype == 'O')
                 # print("\t", dtype or (h5py.string_dtype() if array.dtype == '0' else array.dtype))
-                if array.dtype == "O":
+                if isinstance(array, np.ndarray) and array.dtype == "O":
                     array = np.asarray([s.encode() if isinstance(s, str) else s for s in array])
                     np.void(array)
                 dset = f.create_dataset(path_in_external_file, array.shape, dtype or array.dtype)

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

@@ -627,6 +627,21 @@ def update_prop_kind_dict_cache():
         __CACHE_PROP_KIND_DICT__[prop["Uuid"]] = read_energyml_json_str(json.dumps(prop))[0]
 
 
+def get_property_kind_by_uuid(uuid: str) -> Optional[Any]:
+    """
+    Get a property kind by its uuid.
+    :param uuid: the uuid of the property kind
+    :return: the property kind or None if not found
+    """
+    if len(__CACHE_PROP_KIND_DICT__) == 0:
+        # update the cache to check if it is a
+        try:
+            update_prop_kind_dict_cache()
+        except FileNotFoundError as e:
+            logging.error(f"Failed to parse propertykind dict {e}")
+    return __CACHE_PROP_KIND_DICT__.get(uuid, None)
+
+
 def as_dor(obj_or_identifier: Any, dor_qualified_type: str = "eml23.DataObjectReference"):
     """
     Create an DOR from an object to target the latter.

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

@@ -1167,6 +1167,45 @@ def get_direct_dor_list(obj: Any) -> List[Any]:
     return search_attribute_matching_type(obj, "DataObjectreference")
 
 
+def get_obj_usable_class(o: Any) -> Optional[type]:
+    """Used only for resqml201 that has classes Obj_TriangulatedSetRepresentation and TriangulatedSetRepresentation for example.
+    This function will return Obj_TriangulatedSetRepresentation class
+    """
+
+    if o is not None:
+        if not isinstance(o, type):
+            o = type(o)
+        if isinstance(o, type):
+            if o.__bases__ is not None:
+                for bc in o.__bases__:
+                    # print(bc)
+                    if bc.__name__.lower() == f"obj{get_obj_type(o).lower()}":
+                        return bc
+        return o if isinstance(o, type) else None
+    return None
+
+
+def as_obj_prefixed_class_if_possible(o: Any) -> Any:
+    """Used only for resqml201 that has classes Obj_TriangulatedSetRepresentation and TriangulatedSetRepresentation for example.
+    This function will return an instance of Obj_TriangulatedSetRepresentation if possible
+    """
+    if o is not None:
+        if not isinstance(o, type):
+            o_type = type(o)
+            if o_type.__bases__ is not None:
+                for bc in o_type.__bases__:
+                    # print(bc)
+                    if bc.__name__.lower() == f"obj{get_obj_type(o_type).lower()}":
+                        try:
+                            return bc(**o.__dict__)
+                        except Exception as e:
+                            logging.error(f"Failed to convert {o} to {bc}")
+                            logging.error(e)
+                            return o
+        return o
+    return None
+
+
 def get_data_object_type(cls: Union[type, Any], print_dev_version=True, nb_max_version_digits=2):
     return get_class_pkg(cls) + "." + get_class_pkg_version(cls, print_dev_version, nb_max_version_digits)
 
@@ -1515,3 +1554,81 @@ def _random_value_from_class(
 
     logging.error(f"@_random_value_from_class Not supported object type generation {cls}")
     return None
+
+
+if __name__ == "__main__":
+
+    from energyml.eml.v2_3.commonv2 import *
+    from energyml.eml.v2_0.commonv2 import Citation as Cit201
+    from energyml.resqml.v2_0_1.resqmlv2 import TriangulatedSetRepresentation as Tr20, ObjTriangulatedSetRepresentation
+    from energyml.resqml.v2_2.resqmlv2 import (
+        TriangulatedSetRepresentation,
+        FaultInterpretation,
+    )
+    from .serialization import *
+
+    fi_cit = Citation(
+        title="An interpretation",
+        originator="Valentin",
+        creation=epoch_to_date(epoch()),
+        editor="test",
+        format="Geosiris",
+        last_update=epoch_to_date(epoch()),
+    )
+
+    fi = FaultInterpretation(
+        citation=fi_cit,
+        uuid=gen_uuid(),
+        object_version="0",
+    )
+
+    tr_cit = Citation(
+        title="--",
+        # title="test title",
+        originator="Valentin",
+        creation=epoch_to_date(epoch()),
+        editor="test",
+        format="Geosiris",
+        last_update=epoch_to_date(epoch()),
+    )
+
+    tr_cit201 = Cit201(
+        title="--",
+        # title="test title",
+        originator="Valentin",
+        # creation=str(epoch_to_date(epoch()))
+        editor="test",
+        format="Geosiris",
+        # last_update=str(epoch_to_date(epoch())),
+    )
+    dor = DataObjectReference(
+        uuid=fi.uuid,
+        title="a DOR title",
+        object_version="0",
+        qualified_type="a wrong qualified type",
+    )
+    tr = TriangulatedSetRepresentation(
+        citation=tr_cit,
+        uuid=gen_uuid(),
+        represented_object=dor,
+    )
+
+    tr201 = Tr20(
+        citation=tr_cit201,
+        uuid=gen_uuid(),
+    )
+    tr201_bis = ObjTriangulatedSetRepresentation(
+        citation=tr_cit201,
+        uuid=gen_uuid(),
+    )
+    # print(get_obj_uri(tr201, "coucou"))
+
+    print(get_obj_usable_class(tr))
+    print(get_obj_usable_class(tr201))
+
+    print(serialize_xml(tr201_bis))
+    print(serialize_xml(tr201))
+    print(serialize_json(tr201))
+    print(serialize_xml(as_obj_prefixed_class_if_possible(tr201)))
+    print("--> ", serialize_json(tr))
+    # print(serialize_xml((get_usable_class(tr201))(tr201)))

energyml-utils/src/energyml/utils/serialization.py

@@ -25,6 +25,7 @@
 
 from .exception import UnknownTypeFromQualifiedType, NotParsableType
 from .introspection import (
+    as_obj_prefixed_class_if_possible,
     get_class_from_name,
     get_energyml_class_in_related_dev_pkg,
     get_class_from_content_type,
@@ -254,6 +255,7 @@ def read_energyml_json_file(
 
 
 def serialize_xml(obj) -> str:
+    obj = as_obj_prefixed_class_if_possible(obj)
     context = XmlContext(
         # element_name_generator=text.camel_case,
         # attribute_name_generator=text.kebab_case
@@ -264,6 +266,7 @@ def serialize_xml(obj) -> str:
 
 
 def serialize_json(obj, json_version: JSON_VERSION = JSON_VERSION.OSDU_OFFICIAL) -> str:
+    obj = as_obj_prefixed_class_if_possible(obj)
     if json_version == JSON_VERSION.XSDATA:
         context = XmlContext(
             # element_name_generator=text.camel_case,