bugfix for epcStream id use

Author: valentin-gauthier-geosiris

energyml-utils/example/attic/arrays_test.py

@@ -365,6 +365,8 @@ def read_pointset(
     # meshes = []
     meshes = read_mesh_object(energyml_object=pointset, workspace=epc)
 
+    print(epc.get_obj_rels(pointset))
+
     # logging.debug("=" * 40)
     # print_tuple_list(search_attribute_matching_name_with_path(pointset, r"NodePatch.[\d]+.Geometry.Points"))
     # logging.debug("=" * 40)

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

@@ -275,8 +275,8 @@ def read_point_representation(
     ) + search_attribute_matching_name_with_path(  # resqml 2.0.1
         energyml_object, r"NodePatchGeometry.[\d]+.Points"
     )
-    logging.debug(f"Found {len(patches_geom)} patches for point representation")
-    logging.debug(f"\t=> {patches_geom}")
+    # logging.debug(f"Found {len(patches_geom)} patches for point representation")
+    # logging.debug(f"\t=> {patches_geom}")
 
     for points_path_in_obj, points_obj in patches_geom:
         points = read_array(

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

@@ -67,6 +67,7 @@
     get_file_folder,
     make_path_relative_to_other_file,
     make_path_relative_to_filepath_list,
+    as_identifier,
 )
 
 
@@ -150,7 +151,8 @@ def remove(self, obj: Any) -> None:
     def get_by_identifier(self, identifier: Union[str, Uri]) -> Optional[Any]:
         """Get object by identifier (O(1) lookup)."""
         # Try identifier lookup first
-        obj = self._by_identifier.get(str(identifier))
+        # obj = self._by_identifier.get(str(identifier))
+        obj = self._by_identifier.get(as_identifier(identifier))
         if obj is not None:
             return obj
 

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

@@ -58,12 +58,10 @@
 )
 from energyml.utils.uri import Uri, create_uri_from_content_type_or_qualified_type
 from energyml.utils.constants import (
-    CORE_PROPERTIES_FOLDER_NAME,
     EPCRelsRelationshipType,
     EpcExportVersion,
     MimeType,
     OptimizedRegex,
-    file_extension_to_mime_type,
     date_to_datetime,
 )
 from energyml.utils.epc_utils import (
@@ -72,6 +70,7 @@
     gen_core_props_path,
     make_path_relative_to_filepath_list,
     make_path_relative_to_other_file,
+    as_identifier,
 )
 
 from energyml.utils.introspection import (
@@ -1623,7 +1622,7 @@ def put_object(self, obj: Any, dataspace: Optional[str] = None) -> Optional[str]
         identifier = uri.as_identifier()
         existing_metadata = self._metadata_mgr.get_metadata(identifier)
         file_path = gen_energyml_object_path(obj, self._metadata_mgr._export_version)
-        is_update = existing_metadata is not None
+        # is_update = existing_metadata is not None
 
         # Write object data and metadata to EPC
         try:
@@ -2082,38 +2081,23 @@ def _update_access_order(self, identifier: str) -> None:
     def _id_from_uri_or_identifier(
         self, identifier: Union[str, Uri, Any], get_first_if_simple_uuid: bool = True
     ) -> Optional[str]:
-        if identifier is None:
-            return None
-        elif isinstance(identifier, str):
-            if OptimizedRegex.UUID.fullmatch(identifier) is not None:
-                if not get_first_if_simple_uuid:
-                    logging.warning(
-                        f"Identifier {identifier} is a simple UUID, but get_first_if_simple_uuid is False, cannot resolve to full identifier"
-                    )
-                    return None
-                # If it's a simple UUID, we need to find the corresponding identifier from metadata
-                t_metadata_identifiers = self._metadata_mgr.get_uuid_identifiers(identifier)
-                if t_metadata_identifiers is not None and len(t_metadata_identifiers) > 0:
-                    return t_metadata_identifiers[
-                        0
-                    ]  # If multiple metadata entries for the same UUID, we take the first one (this should not happen in a well-formed EPC file)
-                else:
-                    logging.warning(f"No metadata found for UUID {identifier}, cannot get relationships")
-                    return None
+        try:
+            return as_identifier(identifier)
+        except Exception:
+            if not get_first_if_simple_uuid:
+                logging.warning(
+                    f"Identifier {identifier} is a simple UUID, but get_first_if_simple_uuid is False, cannot resolve to full identifier"
+                )
+                return None
+            # If it's a simple UUID, we need to find the corresponding identifier from metadata
+            t_metadata_identifiers = self._metadata_mgr.get_uuid_identifiers(identifier)
+            if t_metadata_identifiers is not None and len(t_metadata_identifiers) > 0:
+                return t_metadata_identifiers[
+                    0
+                ]  # If multiple metadata entries for the same UUID, we take the first one (this should not happen in a well-formed EPC file)
             else:
-                return identifier
-        elif isinstance(identifier, Uri):
-            return identifier.as_identifier()
-        elif isinstance(identifier, ResourceMetadata):
-            return self._id_from_uri_or_identifier(identifier.identifier)
-        elif isinstance(identifier, EpcObjectMetadata):
-            return self._id_from_uri_or_identifier(identifier.uri)
-        else:
-            # Try to get URI from object
-            obj_uri = get_obj_uri(obj=identifier, dataspace=None)
-            if obj_uri is not None:
-                return obj_uri.as_identifier()
-            return str(identifier)
+                logging.warning(f"No metadata found for UUID {identifier}, cannot get relationships")
+            return None
 
     def _rebuild_all_rels_sequential(self, clean_first: bool = True) -> Dict[str, int]:
         """

energyml-utils/src/energyml/utils/epc_utils.py

@@ -7,7 +7,6 @@
 import logging
 import os
 import os
-import re
 from typing import Optional, Set, Tuple, Union, Any, List, Dict, Callable
 from pathlib import Path
 import zipfile
@@ -25,6 +24,8 @@
     Override,
 )
 
+from energyml.utils.exception import NotEnoughInformationError
+
 from energyml.utils.constants import (
     CORE_PROPERTIES_FOLDER_NAME,
     EPCRelsRelationshipType,
@@ -65,7 +66,7 @@
 from energyml.utils.manager import get_class_pkg
 from energyml.utils.serialization import read_energyml_xml_str, serialize_xml, read_energyml_json_str
 from energyml.utils.uri import Uri, parse_uri
-
+from energyml.utils.storage_interface import ResourceMetadata
 
 #     ____  ___  ________  __
 #    / __ \/   |/_  __/ / / /
@@ -285,6 +286,32 @@ def make_path_relative_to_filepath_list(paths: List[str], ref_path: Optional[Uni
 # /_/  /_/___//____/\____/
 
 
+def as_identifier(identifier: Union[str, Uri, Any]) -> Optional[str]:
+    if identifier is None:
+        return None
+    elif isinstance(identifier, str):
+        if identifier.startswith("eml:///"):
+            return as_identifier(parse_uri(identifier))
+        if OptimizedRegex.UUID.fullmatch(identifier) is not None:
+            raise NotEnoughInformationError(
+                "Simple uuid is not enough to be used as an identifier, please provide a full URI or an object with a valid URI or identifier that contains the version : 'UUID.VERSION' even if VERSION can be an empty string"
+            )
+        else:
+            return identifier
+    elif isinstance(identifier, Uri):
+        return identifier.as_identifier()
+    elif isinstance(identifier, ResourceMetadata):
+        return as_identifier(identifier.identifier)
+    elif hasattr(identifier, "uri"):  # EpcObjectMetadata
+        return as_identifier(identifier.uri)
+    else:
+        # Try to get URI from object
+        obj_uri = get_obj_uri(obj=identifier, dataspace=None)
+        if obj_uri is not None:
+            return obj_uri.as_identifier()
+        return str(identifier)
+
+
 def create_external_relationship(path: str, _id: Optional[str] = None) -> Relationship:
     return Relationship(
         target=path,

energyml-utils/src/energyml/utils/exception.py

@@ -48,6 +48,13 @@ def __init__(self, msg):
         super().__init__(msg)
 
 
+class NotEnoughInformationError(Exception):
+    """Exception for not enough information to perform an operation"""
+
+    def __init__(self, msg):
+        super().__init__(msg)
+
+
 # EPC Validation Exceptions