Clean up compare_reference_attributes CHECK function and adjust optimizer schema accordingly

Author: cwang39403

esdlvalidator/validation/functions/check_attributes_not_null_or_valid.py

@@ -73,7 +73,7 @@ def get_function_definition(self):
                         "type": "string",
                         "description": "Dot-separated reference path (e.g., costInformation.investmentCosts)",
                     },
-                    # To locate the target reference entity when reference is of EOrderedSet type.
+                    # To locate the target reference entity when reference is of EOrderedSet type (e.g., port).
                     "ref_list_filter": {
                         "type": "object",
                         "required": ["is_type"],
@@ -159,7 +159,7 @@ def execute(self):
 
         return CheckResult(True)
 
-    def _resolve_reference(self, entity: any, ref: dict, results: list):
+    def _resolve_reference(self, entity: object, ref: dict, results: list):
         if not ref:
             return entity
 

esdlvalidator/validation/functions/check_reference_attributes_comparison.py

@@ -1,7 +1,7 @@
 import jsonschema
 from jsonschema.exceptions import ValidationError
 
-from esdlvalidator.core.esdl import utils as esdlUtils
+from esdlvalidator.core.esdl import utils
 
 from esdlvalidator.validation.functions import utils
 from esdlvalidator.validation.functions.function import (
@@ -15,194 +15,227 @@
 
 from pyecore.ecore import EOrderedSet
 
-# TODO: check ports are connected to 1 hydraulically coupled network 
 
 @FunctionFactory.register(FunctionType.CHECK, "compare_reference_attributes")
 class CompareRefAttributes(FunctionCheck):
+    """
+    Compares two attributes from either the input entity or its nested references.
+
+    Supported operators:
+    - greater_than
+    - less_than
+    - equal
+
+    Each side (left/right) may specify a reference path and filter to locate the target entity.
+    """
 
     def get_function_definition(self):
         return FunctionDefinition(
             "compare_reference_attributes",
-            "TODO",
+            "Compare two attributes from the located references (or from the entity itself) using a specified operator.",
             [
                 ArgDefinition(
-                    "ref",
-                    "TODO",
+                    "left",
+                    "Reference and attribute to compare on the left side.",
                     True,
                 ),
                 ArgDefinition(
                     "operator",
-                    "TODO",
-                    True,
-                ),
-                ArgDefinition(
-                    "left",
-                    "TODO",
+                    "Comparison operator",
                     True,
                 ),
                 ArgDefinition(
                     "right",
-                    "TODO",
+                    "Reference and attribute to compare on the right side",
                     True,
                 ),
-                ArgDefinition("resultMsgJSON", "Display output in JSON format", False),
+                ArgDefinition("resultMsgJSON", "If True, returns results in structured JSON format.", False),
             ],
         )
 
+    NOT_FOUND = "Not found"
+    UNSET = "Unset"
     # Supported operator types. Can be extended in future if required.
     OPERATOR_ENUM = ["greater_than", "equal", "less_than"]
 
     args_schema = {
         "type": "object",
-        "required": ["ref", "operator", "left", "right"],
+        "required": ["left", "operator", "right"],
         "properties": {
-            "ref": {"type": "string"},
+            "left": {"$ref": "#/$defs/refAttribute"},
             "operator": {"type": "string", "enum": OPERATOR_ENUM},
-            "left": {
+            "right": {"$ref": "#/$defs/refAttribute"},
+            "resultMsgJSON": {"type": "boolean"},
+        },
+        "$defs": {
+            "refAttribute": {
                 "type": "object",
-                "required": ["ref_filter", "attribute"],
+                "required": ["attribute"],
                 "properties": {
-                    "ref_filter": {
+                    # If ref is not specified, it refers to the input entity.
+                    "ref": {
                         "type": "object",
-                        "required": ["is_type", "match"],
-                        "properties": {"is_type": {"type": "string"}, "match": {"type": "object"}},
+                        "description": "Optional reference filter to locate neseted reference entity.",
+                        "required": ["path"],
+                        "properties": {
+                            "path": {
+                                "type": "string",
+                                "description": "Dot-separated reference path (e.g., costInformation.investmentCosts)",
+                            },
+                            # To locate the target reference entity when reference is of EOrderedSet type (e.g., port).
+                            "ref_list_filter": {
+                                "type": "object",
+                                "required": ["is_type"],
+                                "properties": {
+                                    "is_type": {"type": "string"},
+                                    "match": {
+                                        "type": "object",
+                                        "description": "Optional attribute filters",
+                                    },
+                                },
+                                "additionalProperties": False,
+                            },
+                        },
+                        "additionalProperties": False,
                     },
-                    "attribute": {"type": "string"},
-                },
-            },
-            "right": {
-                "type": "object",
-                "required": ["ref_filter", "attribute"],
-                "properties": {
-                    "ref_filter": {
-                        "type": "object",
-                        "required": ["is_type", "match"],
-                        "properties": {"is_type": {"type": "string"}, "match": {"type": "object"}},
+                    "attribute": {
+                        "type": "string",
+                        "description": "Attribute name to extract from the (nested reference) entity",
                     },
-                    "attribute": {"type": "string"},
                 },
-            },
-            "resultMsgJSON": {"type": "boolean"},
+                "additionalProperties": False,
+            }
         },
     }
 
-    NOT_FOUND = "Not found"
-    UNSET = "Unset"
-
     def execute(self):
         entity = self.value
         args_dict = self.args
-
         results = []
 
         try:
             jsonschema.validate(instance=args_dict, schema=self.args_schema)
         except ValidationError as e:
             raise ValueError(f"Schema validation failed at {list(e.path)}: {e.message}")
 
-        ref = utils.get_attribute(self.args, "ref")
         operator = utils.get_attribute(self.args, "operator")
-        # TODO: rename to ref_args?
-        left = utils.get_attribute(self.args, "left")
-        right = utils.get_attribute(self.args, "right")
+        left_args = utils.get_attribute(self.args, "left")
+        right_args = utils.get_attribute(self.args, "right")
         resultMsgJSON = utils.get_attribute(self.args, "resultMsgJSON")
 
-        # Check if the retrieved asset references are valid
-        references = utils.get_attr_or_ref_attr(entity, ref)
-        if references == self.NOT_FOUND:
-            raise ValueError(f"[{ref}] not found.")
-        if references == self.UNSET:
-            results.append(f"[{ref}] should be defined, but is unset.")
-        elif not isinstance(references, EOrderedSet):
-            raise TypeError(f"Expect [{ref}] to be of EOrderedSet type, found {type(references)}")
-        else:
-            left_entity = self.get_ref(references, left["ref_filter"])
-            right_entity = self.get_ref(references, right["ref_filter"])
-
-            for entity_ref, side_args in [(left_entity, left), (right_entity, right)]:
-                if entity_ref is None:
-                    match_str = ", ".join(f"[{k}] = '{v}'" for k, v in side_args["ref_filter"]["match"].items())
-                    results.append(
-                        f"[{ref}] should contain a [{side_args['ref_filter']['is_type']}] ({match_str}), but not found."
-                    )
-
-            if left_entity and right_entity:
-                left_val = self.get_attribute_value(left_entity, left, results)
-                right_val = self.get_attribute_value(right_entity, right, results)
-
-                if left_val is not None and right_val is not None:
-                    try:
-                        left_num = float(left_val)
-                        right_num = float(right_val)
-                    except (TypeError, ValueError):
-                        # TODO: message
-                        results.append(f"Cannot compare non-numeric values: left='{left_val}', right='{right_val}'")
-                    else:
-                        if operator == "greater_than" and left_num <= right_num:
-                            results.append(
-                                f"[{left['attribute']}] (vale: {left_num}) of [{left['ref_filter']['is_type']}] should be greater than [{right['attribute']}] (value: {right_val}) of [{right['ref_filter']['is_type']}]."
-                            )
-                        elif operator == "less_than" and left_num >= right_num:
-                            results.append(
-                                f"[{left['attribute']}] (vale: {left_num}) of [{left['ref_filter']['is_type']}] should be less than [{right['attribute']}] (value: {right_val}) of [{right['ref_filter']['is_type']}]."
-                            )
-                        elif operator == "equal" and left_num != right_num:
-                            results.append(
-                                f"[{left['attribute']}] (vale: {left_num}) of [{left['ref_filter']['is_type']}] should be equal to [{right['attribute']}] (value: {right_val}) of [{right['ref_filter']['is_type']}]."
-                            )
-
-        if len(results) > 0:
-            if resultMsgJSON:
-                msg = {"offending_asset": utils.get_attribute(entity, "id"), "message": results}
-                return CheckResult(False, msg)
+        left_entity = self._resolve_reference(entity, left_args.get("ref"), results)
+        right_entity = self._resolve_reference(entity, right_args.get("ref"), results)
+
+        left_val = self._get_attribute_value(left_entity, left_args, results)
+        right_val = self._get_attribute_value(right_entity, right_args, results)
+
+        if left_val is not None and right_val is not None:
+            results.extend(self._compare_values(left_val, right_val, operator, left_args, right_args))
+
+        if results:
+            msg = (
+                {"offending_asset": utils.get_attribute(entity, "id"), "message": results} if resultMsgJSON else results
+            )
+            return CheckResult(False, msg)
+        return CheckResult(True)
+
+    def _resolve_reference(self, entity: object, ref: dict, results: list):
+        """Resolves the target entity from a reference path and optional filter."""
+
+        if not ref:
+            return entity
+
+        ref_path = ref["path"]
+        reference = utils.get_attr_or_ref_attr(entity, ref_path)
+
+        if reference == self.NOT_FOUND:
+            raise ValueError(f"[{ref_path}] not found.")
+        if reference == self.UNSET:
+            results.append(f"[{ref_path}] should be defined, but is unset.")
+            return None
+
+        if isinstance(reference, EOrderedSet):
+            if len(reference) > 1 and ref.get("ref_list_filter"):
+                entity_to_check = utils.get_ref(reference, ref["ref_list_filter"])
             else:
-                return CheckResult(False, results)
-        else:
-            return CheckResult(True)
-
-    def get_ref(self, references: EOrderedSet, ref_filter: dict):
-        """
-        Filters a set of ESDL references to find the first entity matching the given type and attribute criteria.
-
-        Args:
-            references: A collection of ESDL entities.
-            ref_filter: A dictionary with keys:
-                - 'is_type': the expected ESDL type as a string.
-                - 'match': a dictionary of attribute names and expected values.
-
-        Returns:
-            The first matching entity, or None if no match is found.
-        """
-        ref_type = ref_filter["is_type"]
-        match_dict = ref_filter["match"]
-        esdlType = esdlUtils.get_esdl_class_from_string(ref_type)[0]
-
-        for entity in references:
-            if not isinstance(entity, esdlType):
-                continue
-            if all(utils.get_attr_or_ref_attr(entity, key) == value for key, value in match_dict.items()):
-                return entity
-
-        return None
-
-    def get_attribute_value(self, entity, args: dict, results):
-        filter = args["ref_filter"]
-        attr_path = args["attribute"]
+                entity_to_check = reference[0]
+
+            if entity_to_check is None:
+                match_str = ", ".join(f"[{k}] = '{v}'" for k, v in ref["ref_list_filter"].get("match", {}).items())
+                results.append(
+                    f"[{ref_path}] should contain a [{ref['ref_list_filter']['is_type']}] ({match_str}), but not found."
+                )
+            return entity_to_check
+
+        return reference
+
+    def _get_attribute_value(self, entity: object, attr_args: dict, results: list):
+        """Retrieves the attribute value from the given entity and logs unset or missing cases."""
+
+        attr_path = attr_args["attribute"]
         value = utils.get_attr_or_ref_attr(entity, attr_path)
 
         if value == self.NOT_FOUND:
-            raise ValueError(f"[{attr_path}] not found.")
+            raise ValueError(f"{self._format_attr_context(attr_args)} not found.")
         if value == self.UNSET:
-            # TODO: rename variable
-            # msg_head = f"[{attr_path}]"
-
-            match_str = ", ".join(f"{k}={v}" for k, v in filter["match"].items())
-            msg_head = f"[{filter['is_type']}] ({match_str})"
-            # if len(attr_path.split(".")) > 1:
-            #     parent_path = attr_path.split(".")[-2]
-            #     parent_value = utils.get_attr_or_ref_attr(entity, parent_path)
-            #     if parent_path and parent_value.name:
-            #         msg_head = f"[{attr_path.split('.')[-1]}] of [{parent_path}] ({parent_value.name})"
-            results.append(f"{msg_head} should have [{attr_path}] defined, but is unset.")
+            results.append(f"{self._format_attr_context(attr_args)} should be defined, but is unset.")
             return None
         return value
+
+    def _format_attr_context(self, attr_args: dict):
+        """Generates a context-aware label for an attribute used in error messages."""
+
+        attr = attr_args["attribute"]
+        ref = attr_args.get("ref")
+
+        if not ref:
+            return f"[{attr}]"
+
+        ref_type = ref.get("ref_list_filter", {}).get("is_type")
+        match = ref.get("ref_list_filter", {}).get("match", {})
+
+        match_str = ""
+        if match:
+            match_str = " with " + ", ".join(f"[{k}] = '{v}'" for k, v in match.items())
+
+        return f"[{attr}] from [{ref_type}]{match_str}"
+
+    def _compare_values(self, left_val, right_val, operator, left_args, right_args):
+        """Compares two numeric values using the specified operator and returns result messages."""
+
+        results = []
+        try:
+            left_num = float(left_val)
+            right_num = float(right_val)
+        except (TypeError, ValueError):
+            results.append(f"Cannot compare non-numeric values: left='{left_val}', right='{right_val}'")
+            return results
+
+        left_desc = self._format_side_description(left_args, left_num)
+        right_desc = self._format_side_description(right_args, right_num)
+
+        if operator == "greater_than" and left_num <= right_num:
+            results.append(f"{left_desc} should be greater than {right_desc}.")
+        elif operator == "less_than" and left_num >= right_num:
+            results.append(f"{left_desc} should be less than {right_desc}.")
+        elif operator == "equal" and left_num != right_num:
+            results.append(f"{left_desc} should be equal to {right_desc}.")
+        return results
+
+    def _format_side_description(self, attr_args: dict, value: float):
+        """Formats a descriptive string for an attribute value based on its reference context."""
+
+        attr = attr_args["attribute"]
+        ref = attr_args.get("ref")
+
+        if not ref:
+            return f"[{attr}] (value: {value})"
+
+        ref_type = ref.get("ref_list_filter", {}).get("is_type")
+        match = ref.get("ref_list_filter", {}).get("match", {})
+
+        match_str = ""
+        if match:
+            match_str = " with " + ", ".join(f"[{k}] = '{v}'" for k, v in match.items())
+
+        return f"[{attr}] (value: {value}) from [{ref_type}]{match_str}"