validation improvement

Author: valentin-gauthier-geosiris

energyml-utils/example/tools.py

@@ -6,6 +6,8 @@
 import pathlib
 from typing import Optional, List, Dict, Any
 
+from src.energyml.utils.validation import validate_epc
+
 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, export_obj, read_mesh_object
@@ -536,6 +538,86 @@ def describe_as_csv():
     print("Finished")
 
 
+def validate_files():
+    parser = argparse.ArgumentParser()
+    # parser.add_argument("--folder", type=str, help="Input folder")
+    parser.add_argument("--file", "-f", type=str, help="Input file (json or xml or epc)")
+
+    args = parser.parse_args()
+
+    objects = []
+
+    if not os.path.exists(args.file):
+        print(f"File {args.file} does not exist.")
+        return
+    elif not os.path.isdir(args.file) and not args.file.lower().endswith((".json", ".xml", ".epc")):
+        print(f"File {args.file} is not a valid input file (should be a folder or a json/xml/epc file).")
+        return
+    elif os.path.isdir(args.file):
+        for filename in os.listdir(args.file):
+            f = os.path.join(args.file, filename)
+            if os.path.isfile(f):
+                if f.endswith(".json"):
+                    with open(f, "rb") as file:
+                        f_content = file.read()
+                        try:
+                            objs = read_energyml_json_bytes(f_content, JSON_VERSION.OSDU_OFFICIAL)
+                            objects.extend(objs)
+                        except Exception as e:
+                            print(f"File {filename} is NOT a valid EnergyML JSON file: {e}")
+                elif f.endswith(".xml"):
+                    with open(f, "rb") as file:
+                        f_content = file.read()
+                        try:
+                            obj = read_energyml_xml_bytes(f_content)
+                            objects.append(obj)
+                        except Exception as e:
+                            print(f"File {filename} is NOT a valid EnergyML XML file: {e}")
+                elif f.endswith(".epc"):
+                    try:
+                        epc = Epc.read_file(f)
+                        if epc is not None:
+                            objects.extend(epc.energyml_objects)
+                        else:
+                            print(f"File {filename} is NOT a valid EnergyML EPC file: Empty EPC")
+                    except Exception as e:
+                        print(f"File {filename} is NOT a valid EnergyML EPC file: {e}")
+    elif os.path.isfile(args.file):
+        f = args.file
+        filename = os.path.basename(f)
+        if f.endswith(".json"):
+            with open(f, "rb") as file:
+                f_content = file.read()
+                try:
+                    objs = read_energyml_json_bytes(f_content, JSON_VERSION.OSDU_OFFICIAL)
+                    objects.extend(objs)
+                except Exception as e:
+                    print(f"File {filename} is NOT a valid EnergyML JSON file: {e}")
+        elif f.endswith(".xml"):
+            with open(f, "rb") as file:
+                f_content = file.read()
+                try:
+                    obj = read_energyml_xml_bytes(f_content)
+                    objects.append(obj)
+                except Exception as e:
+                    print(f"File {filename} is NOT a valid EnergyML XML file: {e}")
+        elif f.endswith(".epc"):
+            try:
+                epc = Epc.read_file(f)
+                if epc is not None:
+                    objects.extend(epc.energyml_objects)
+                else:
+                    print(f"File {filename} is NOT a valid EnergyML EPC file: Empty EPC")
+            except Exception as e:
+                print(f"File {filename} is NOT a valid EnergyML EPC file: {e}")
+
+    epc = Epc()
+    epc.energyml_objects = objects
+
+    for err in validate_epc(epc):
+        print(err.toJson())
+
+
 # def export_wavefront():
 #     parser = argparse.ArgumentParser()
 #     parser.add_argument("--epc", "-f", type=str, help="Epc file path")

energyml-utils/pyproject.toml

@@ -133,4 +133,5 @@ generate_data = "example.tools:generate_data"
 xml_to_json = "example.tools:xml_to_json"
 json_to_xml = "example.tools:json_to_xml"
 json_to_epc = "example.tools:json_to_epc"
-describe_as_csv = "example.tools:describe_as_csv"
+describe_as_csv = "example.tools:describe_as_csv"
+validate = "example.tools:validate_files"

energyml-utils/src/energyml/utils/validation.py

@@ -3,7 +3,8 @@
 import re
 from dataclasses import dataclass, field, Field
 from enum import Enum
-from typing import Any, List
+import traceback
+from typing import Any, List, Optional
 
 from .epc import (
     get_obj_identifier,
@@ -12,6 +13,7 @@
 from .introspection import (
     get_class_fields,
     get_object_attribute,
+    is_primitive,
     search_attribute_matching_type_with_path,
     get_object_attribute_no_verif,
     get_object_attribute_rgx,
@@ -20,9 +22,11 @@
     get_obj_version,
     get_content_type_from_class,
     get_qualified_type_from_class,
-    is_enum, get_object_uri,
+    is_enum,
+    get_object_uri,
 )
 
+
 class ErrorType(Enum):
     CRITICAL = "critical"
     DEBUG = "debug"
@@ -52,13 +56,16 @@ class ValidationObjectError(ValidationError):
 
     target_obj: Any = field(default=None)
 
-    attribute_dot_path: str = field(default=None)
+    attribute_dot_path: Optional[str] = field(default=None)
 
     def __str__(self):
         return f"{ValidationError.__str__(self)}\n\t{get_obj_identifier(self.target_obj)} : '{self.attribute_dot_path}'"
 
     def toJson(self):
-        return super().toJson() | {"target_obj": str(get_object_uri(self.target_obj)), "attribute_dot_path": self.attribute_dot_path}
+        return super().toJson() | {
+            "target_obj": str(get_object_uri(self.target_obj)),
+            "attribute_dot_path": self.attribute_dot_path,
+        }
 
 
 @dataclass
@@ -69,7 +76,7 @@ def __str__(self):
 
 @dataclass
 class MissingEntityError(ValidationObjectError):
-    missing_uuid: str = field(default=None)
+    missing_uuid: Optional[str] = field(default=None)
 
     def __str__(self):
         return f"{ValidationError.__str__(self)}\n\tMissing entity in {get_obj_identifier(self.target_obj)} at path '{self.attribute_dot_path}'. Missing entity uuid: {self.missing_uuid}"
@@ -122,8 +129,8 @@ def dor_validation(energyml_objects: List[Any]) -> List[ValidationError]:
                             error_type=ErrorType.CRITICAL,
                             target_obj=obj,
                             attribute_dot_path=dor_path,
-                            missing_uuid=dor_uuid
-                            # msg=f"[DOR ERR] has wrong information. Unkown object with uuid '{dor_uuid}'",
+                            missing_uuid=dor_uuid,
+                            msg=f"[DOR ERR] has wrong information. Unknown object with uuid '{dor_uuid}'",
                         )
                     )
                 else:
@@ -133,7 +140,7 @@ def dor_validation(energyml_objects: List[Any]) -> List[ValidationError]:
                             error_type=ErrorType.CRITICAL,
                             target_obj=obj,
                             attribute_dot_path=dor_path,
-                            msg=f"[DOR ERR] has wrong information. Unkown object version '{dor_version}'. "
+                            msg=f"[DOR ERR] has wrong information. Unknown object version '{dor_version}'. "
                             f"Version must be one of {accessible_version}",
                         )
                     )
@@ -201,6 +208,8 @@ def _patterns_validation(obj: Any, root_obj: Any, current_attribute_dot_path: st
     """
     error_list = []
 
+    if is_primitive(obj):
+        return error_list
     if isinstance(obj, list):
         cpt = 0
         for val in obj:
@@ -235,82 +244,105 @@ def validate_attribute(value: Any, root_obj: Any, att_field: Field, path: str) -
                     attribute_dot_path=path,
                 )
             )
+    elif isinstance(att_field, str):
+        errs.append(
+            ValidationObjectError(
+                error_type=ErrorType.WARNING,
+                target_obj=root_obj,
+                attribute_dot_path=path,
+                msg=f"Attribute '{att_field}' is a string but got value '{value}'",
+            )
+        )
     elif not is_enum(value):  # sometimes enums values fails the validation
-        min_length = att_field.metadata.get("min_length", None)
-        max_length = att_field.metadata.get("max_length", None)
-        pattern = att_field.metadata.get("pattern", None)
-        min_occurs = att_field.metadata.get("min_occurs", None)
-        min_inclusive = att_field.metadata.get("min_inclusive", None)
-        # white_space
-
-        if max_length is not None:
-            length = len(value)
-            if length > max_length:
-                errs.append(
-                    ValidationObjectError(
-                        error_type=ErrorType.CRITICAL,
-                        target_obj=root_obj,
-                        attribute_dot_path=path,
-                        msg=f"Max length was {max_length} but found {length}",
-                    )
-                )
-
-        if min_length is not None:
-            length = len(value)
-            if length < min_length:
-                errs.append(
-                    ValidationObjectError(
-                        error_type=ErrorType.CRITICAL,
-                        target_obj=root_obj,
-                        attribute_dot_path=path,
-                        msg=f"Max length was {min_length} but found {length}",
+        try:
+            min_length = att_field.metadata.get("min_length", None)
+            max_length = att_field.metadata.get("max_length", None)
+            pattern = att_field.metadata.get("pattern", None)
+            min_occurs = att_field.metadata.get("min_occurs", None)
+            min_inclusive = att_field.metadata.get("min_inclusive", None)
+            # white_space
+
+            if max_length is not None:
+                length = len(value)
+                if length > max_length:
+                    errs.append(
+                        ValidationObjectError(
+                            msg=f"Max length was {max_length} but found {length}",
+                            error_type=ErrorType.CRITICAL,
+                            target_obj=root_obj,
+                            attribute_dot_path=path,
+                        )
                     )
-                )
 
-        if min_occurs is not None:
-            if isinstance(value, list) and min_occurs > len(value):
-                errs.append(
-                    ValidationObjectError(
-                        error_type=ErrorType.CRITICAL,
-                        target_obj=root_obj,
-                        attribute_dot_path=path,
-                        msg=f"Min occurs was {min_occurs} but found {len(value)}",
+            if min_length is not None:
+                length = len(value)
+                if length < min_length:
+                    errs.append(
+                        ValidationObjectError(
+                            msg=f"Max length was {min_length} but found {length}",
+                            error_type=ErrorType.CRITICAL,
+                            target_obj=root_obj,
+                            attribute_dot_path=path,
+                        )
                     )
-                )
 
-        if min_inclusive is not None:
-            potential_err = ValidationObjectError(
-                error_type=ErrorType.CRITICAL,
-                target_obj=root_obj,
-                attribute_dot_path=path,
-                msg=f"Min occurs was {min_inclusive} but found {value}",
-            )
-            if isinstance(value, list):
-                for val in value:
-                    if (isinstance(val, str) and len(val) > min_inclusive) or (
-                        (isinstance(val, int) or isinstance(val, float)) and val > min_inclusive
-                    ):
-                        errs.append(potential_err)
-
-        if pattern is not None:
-            if not isinstance(value, list):
-                testing_value_list = [value]
-            else:
-                testing_value_list = value
-
-            for v in testing_value_list:
-                if is_enum(v):
-                    v = v.value
-                if re.match(pattern, v) is None:
+            if min_occurs is not None:
+                if isinstance(value, list) and min_occurs > len(value):
                     errs.append(
                         ValidationObjectError(
+                            msg=f"Min occurs was {min_occurs} but found {len(value)}",
                             error_type=ErrorType.CRITICAL,
                             target_obj=root_obj,
                             attribute_dot_path=path,
-                            msg=f"Pattern error. Value '{v}' was supposed to respect pattern '{pattern}'",
                         )
                     )
 
+            if min_inclusive is not None:
+                potential_err = ValidationObjectError(
+                    msg=f"Min occurs was {min_inclusive} but found {value}",
+                    error_type=ErrorType.CRITICAL,
+                    target_obj=root_obj,
+                    attribute_dot_path=path,
+                )
+                if isinstance(value, list):
+                    for val in value:
+                        if (isinstance(val, str) and len(val) > min_inclusive) or (
+                            (isinstance(val, int) or isinstance(val, float)) and val > min_inclusive
+                        ):
+                            errs.append(potential_err)
+
+            if pattern is not None:
+                if not isinstance(value, list):
+                    testing_value_list = [value]
+                else:
+                    testing_value_list = value
+
+                for v in testing_value_list:
+                    if is_enum(v):
+                        v = v.value
+                    if re.match(pattern, v) is None:
+                        errs.append(
+                            ValidationObjectError(
+                                msg=f"Pattern error. Value '{v}' was supposed to respect pattern '{pattern}'",
+                                error_type=ErrorType.CRITICAL,
+                                target_obj=root_obj,
+                                attribute_dot_path=path,
+                            )
+                        )
+        except Exception as e:
+            print(f"Error while validating attribute '{att_field}' with value '{value}': {str(e)} for {path}")
+            errs.append(
+                ValidationObjectError(
+                    msg=f"Error while validating attribute '{att_field}' with value '{value}': {str(e)}",
+                    error_type=ErrorType.CRITICAL,
+                    target_obj=root_obj,
+                    attribute_dot_path=path,
+                )
+            )
+            traceback.print_exc()
+            exit(0)
+            return errs
+
     return errs + _patterns_validation(
         obj=value,
         root_obj=root_obj,