bilt-exercise-4

main.py

@@ -1,7 +1,13 @@
 from pydantic import Field
 from speckle_automate import AutomationContext, AutomateBase, execute_automate_function
 from Utilities.helpers import flatten_base, speckle_print
-import random
+
+from rules import RevitRules
+
+
+from Utilities.helpers import flatten_base, speckle_print
+from Utilities.spreadsheet import read_rules_from_spreadsheet
+from Workshop.Exercise_4.rules import apply_rules_to_objects
 
 
 class FunctionInputs(AutomateBase):
@@ -12,16 +18,13 @@ class FunctionInputs(AutomateBase):
     https://docs.pydantic.dev/latest/usage/models/
     """
 
-    # In this exercise, we will add two new input fields to the FunctionInputs class.
-    category: str = Field(
-        title="Revit Category",
-        description="This is the category objects to check.",
-    )
-    property: str = Field(
-        title="Property Name",
-        description="This is the property to check.",
+    # In this exercise, we will move rules to an external source so not to hardcode them.
+    spreadsheet_url: str = Field(
+        title="Spreadsheet URL",
+        description="This is the URL of the spreadsheet to check. It should be a TSV format data source.",
     )
 
+
 def automate_function(
     automate_context: AutomationContext,
     function_inputs: FunctionInputs,
@@ -42,95 +45,20 @@ def automate_function(
     # We can continue to work with a flattened list of objects.
     flat_list_of_objects = list(flatten_base(version_root_object))
 
-    # filter to only include objects that are in the specified category
-    in_category_objects = [
-        speckle_object
-        for speckle_object in flat_list_of_objects
-        if RevitRules.is_category(speckle_object, function_inputs.category)
-    ]
+    # read the rules from the spreadsheet
+    rules = read_rules_from_spreadsheet(function_inputs.spreadsheet_url)
 
-    # check if the property exists on the objects
-    non_property_objects = [
-        obj
-        for obj in in_category_objects
-        if not RevitRules.has_parameter(obj, function_inputs.property)
-    ]
-
-    property_objects = [
-        obj
-        for obj in in_category_objects
-        if RevitRules.has_parameter(obj, function_inputs.property)
-    ]
-
-    # property_objects should be those where while the property is present,
-    # is not an empty string or the default value
-    valid_property_objects = [
-        obj
-        for obj in property_objects
-        if RevitRules.get_parameter_value(obj, function_inputs.property)
-        not in ["", "Default", None]
-    ]
-
-    for obj in valid_property_objects:
-        speckle_print(RevitRules.get_parameter_value(obj, function_inputs.property))
-
-    # invalid_property_objects property_objects not in valid_property_objects
-    invalid_property_objects = [
-        obj for obj in property_objects if obj not in valid_property_objects
-    ]
-
-    # mark all the non-property objects as failed
-
-    (
-        automate_context.attach_error_to_objects(
-            category=f"Missing Property {function_inputs.category} Objects",
-            object_ids=[obj.id for obj in non_property_objects],
-            message=f"This {function_inputs.category} does not have the specified property {function_inputs.property}",
-        )
-        if non_property_objects
-        else None
-    )
-
-    # mark all the invalid property objects as warning
-    (
-        automate_context.attach_warning_to_objects(
-            category=f"Invalid Property {function_inputs.category} Objects",
-            object_ids=[obj.id for obj in invalid_property_objects],
-            message=f"This {function_inputs.category} has the specified property {function_inputs.property} but it is "
-            f"empty or default",
-        )
-        if invalid_property_objects
-        else None
-    )
-
-    # mark all the property objects as successful
-    (
-        automate_context.attach_info_to_objects(
-            category=f"Valid Property {function_inputs.category} Objects",
-            object_ids=[obj.id for obj in property_objects],
-            message=f"This {function_inputs.category} has the specified property {function_inputs.property}",
-        )
-        if property_objects
-        else None
-    )
-
-    if len(non_property_objects) > 0:
-        automate_context.mark_run_failed(
-            "Some objects do not have the specified property."
-        )
-    elif len(invalid_property_objects) > 0:
-        automate_context.mark_run_success(
-            "Some objects have the specified property but it is empty or default.",
-        )
-
-    else:
-        automate_context.mark_run_success(
-            f"All {function_inputs.category} objects have the {function_inputs.property} property."
-        )
+    # apply the rules to the objects
+    apply_rules_to_objects(flat_list_of_objects, rules, automate_context)
 
     # set the automation context view, to the original model / version view
     automate_context.set_context_view()
 
+    # report success
+    automate_context.mark_run_success(
+        f"Successfully applied rules to {len(flat_list_of_objects)} objects."
+    )
+
 
 
 # make sure to call the function with the executor

rules.py

@@ -1,12 +1,16 @@
-from typing import List, Optional, Tuple, Callable, Dict, Any, cast, Union
+from typing import List, Optional, Tuple, Any, cast
+from speckle_automate import AutomationContext, ObjectResultLevel
 from specklepy.objects.base import Base
 from Levenshtein import ratio
+import pandas as pd
 import re
 
+from Utilities.helpers import speckle_print
+
+
 # We're going to define a set of rules that will allow us to filter and
 # process parameters in our Speckle objects. These rules will be encapsulated
-# in a class called `Rules`. We'll also define a set of rules specific to Revit
-# objects in a class called `RevitRules`.
+# in a class called `ParameterRules`.
 
 
 class Rules:
@@ -70,7 +74,7 @@ class Rules:
         Returns:
             bool: True if the object has a display value, False otherwise.
         """
-        # Check if the speckle_object has a display value using the try_get_display_value method
+        # Check for direct displayable state using try_get_display_value
         display_values = Rules.try_get_display_value(speckle_object)
         if display_values and getattr(speckle_object, "id", None) is not None:
             return True
@@ -87,57 +91,6 @@ class Rules:
 
         return False
 
-    # Below are more speculatively defined rules that could be used in a traversal of flat list parsing
-
-    @staticmethod
-    def speckle_type_rule(
-        desired_type: str,
-    ) -> Callable[[Base], bool]:
-        """
-        Rule: Check if a parameter's speckle_type matches the desired type.
-        """
-        return lambda prop: getattr(prop, "speckle_type", None) == desired_type
-
-    @staticmethod
-    def is_speckle_type(prop: Base, desired_type: str) -> bool:
-        """
-        Rule: Check if a parameter's speckle_type matches the desired type.
-        """
-        return getattr(prop, "speckle_type", None) == desired_type
-
-    @staticmethod
-    def has_missing_value(prop: Dict[str, str]) -> bool:
-        """
-        Rule: Missing Value Check.
-
-        The AEC industry often requires all parameters to have meaningful values.
-        This rule checks if a parameter is missing its value, potentially indicating
-        an oversight during data entry or transfer.
-        """
-        return not prop.get("value")
-
-    @staticmethod
-    def has_default_value(prop: Dict[str, str], default="Default") -> bool:
-        """
-        Rule: Default Value Check.
-
-        Default values can sometimes creep into final datasets due to software defaults.
-        This rule identifies parameters that still have their default values, helping
-        to highlight areas where real, meaningful values need to be provided.
-        """
-        return prop.get("value") == default
-
-    @staticmethod
-    def parameter_exists(prop_name: str, parent_object: Dict[str, str]) -> bool:
-        """
-        Rule: Parameter Existence Check.
-
-        For certain critical parameters, their mere presence (or lack thereof) is vital.
-        This rule verifies if a specific parameter exists within an object, allowing
-        teams to ensure that key data points are always present.
-        """
-        return prop_name in parent_object.get("parameters", {})
-
 
 def get_displayable_objects(flat_list_of_objects: List[Base]) -> List[Base]:
     # modify this lambda from before to use the static method from the Checks class
@@ -151,10 +104,42 @@ def get_displayable_objects(flat_list_of_objects: List[Base]) -> List[Base]:
     # and the same logic that could be modified to traverse a tree of objects
 
 
-# Now we're going to define a set of rules that are specific to Revit objects.
+def filter_objects_by_category(
+    speckle_objects: List[Base], category_input: str
+) -> Tuple[List[Base], List[Base]]:
+    """
+    Filters objects by category value and test.
+
+    This function takes a list of Speckle objects, filters out the objects
+    with a matching category value and satisfies the test, and returns
+    both the matching and non-matching objects.
+
+    Args:
+        speckle_objects (List[Base]): The list of Speckle objects to filter.
+        category_input (str): The category value to match against.
+
+    Returns:
+        Tuple[List[Base], List[Base]]: A tuple containing two lists:
+                                        - The first list contains objects with matching category and test.
+                                        - The second list contains objects without matching category or test.
+    """
+    matching_objects = []
+    non_matching_objects = []
+
+    for obj in speckle_objects:
+        if RevitRules.is_category(obj, category_input):
+            matching_objects.append(obj)
+        else:
+            non_matching_objects.append(obj)
+
+    return matching_objects, non_matching_objects
+
+
 class RevitRules:
     @staticmethod
-    def has_parameter(speckle_object: Base, parameter_name: str) -> bool:
+    def has_parameter(
+        speckle_object: Base, parameter_name: str, *_args, **_kwargs
+    ) -> bool:
         """
         Checks if the speckle_object has a Revit parameter with the given name.
 
@@ -170,6 +155,8 @@ class RevitRules:
         Args:
             speckle_object (Base): The Speckle object to check.
             parameter_name (str): The name of the parameter to check for.
+            *_args: Extra positional arguments which are ignored.
+            **_kwargs: Extra keyword arguments which are ignored.
 
         Returns:
             bool: True if the object has the parameter, False otherwise.
@@ -262,6 +249,8 @@ class RevitRules:
             None,
         )
 
+    from typing import Any, Union, List
+
     @staticmethod
     def is_parameter_value(
         speckle_object: Base, parameter_name: str, value_to_match: Any
@@ -281,7 +270,7 @@ class RevitRules:
         return parameter_value == value_to_match
 
     @staticmethod
-    def is_like_parameter_value(
+    def is_parameter_value_like(
         speckle_object: Base,
         parameter_name: str,
         pattern: str,
@@ -313,9 +302,31 @@ class RevitRules:
         else:
             return bool(re.match(pattern, str(parameter_value)))
 
+    @staticmethod
+    def parse_number_from_string(input_string: str):
+        """
+        Attempts to parse an integer or float from a given string.
+
+        Args:
+            input_string (str): The string containing the number to be parsed.
+
+        Returns:
+            int or float: The parsed number, or raises ValueError if parsing is not possible.
+        """
+        try:
+            # First try to convert it to an integer
+            return int(input_string)
+        except ValueError:
+            # If it fails to convert to an integer, try to convert to a float
+            try:
+                return float(input_string)
+            except ValueError:
+                # Raise an error if neither conversion is possible
+                raise ValueError("Input string is not a valid integer or float")
+
     @staticmethod
     def is_parameter_value_greater_than(
-        speckle_object: Base, parameter_name: str, threshold: Union[int, float]
+        speckle_object: Base, parameter_name: str, threshold: str
     ) -> bool:
         """
         Checks if the value of the specified parameter is greater than the given threshold.
@@ -328,18 +339,20 @@ class RevitRules:
         Returns:
             bool: True if the parameter value is greater than the threshold, False otherwise.
         """
+
         parameter_value = RevitRules.get_parameter_value(speckle_object, parameter_name)
         if parameter_value is None:
             return False
+
         if not isinstance(parameter_value, (int, float)):
             raise ValueError(
                 f"Parameter value must be a number, got {type(parameter_value)}"
             )
-        return parameter_value > threshold
+        return parameter_value > RevitRules.parse_number_from_string(threshold)
 
     @staticmethod
     def is_parameter_value_less_than(
-        speckle_object: Base, parameter_name: str, threshold: Union[int, float]
+        speckle_object: Base, parameter_name: str, threshold: str
     ) -> bool:
         """
         Checks if the value of the specified parameter is less than the given threshold.
@@ -359,10 +372,40 @@ class RevitRules:
             raise ValueError(
                 f"Parameter value must be a number, got {type(parameter_value)}"
             )
-        return parameter_value < threshold
+        return parameter_value < RevitRules.parse_number_from_string(threshold)
 
     @staticmethod
     def is_parameter_value_in_range(
+        speckle_object: Base, parameter_name: str, range: str
+    ) -> bool:
+        """
+        Checks if the value of the specified parameter falls within the given range.
+
+        Args:
+            speckle_object (Base): The Speckle object to check.
+            parameter_name (str): The name of the parameter to check.
+            range (str): The range to check against, in the format "min_value, max_value".
+
+        Returns:
+            bool: True if the parameter value falls within the range (inclusive), False otherwise.
+        """
+
+        min_value, max_value = range.split(",")
+        min_value = RevitRules.parse_number_from_string(min_value)
+        max_value = RevitRules.parse_number_from_string(max_value)
+
+        parameter_value = RevitRules.get_parameter_value(speckle_object, parameter_name)
+        if parameter_value is None:
+            return False
+        if not isinstance(parameter_value, (int, float)):
+            raise ValueError(
+                f"Parameter value must be a number, got {type(parameter_value)}"
+            )
+
+        return min_value <= parameter_value <= max_value
+
+    @staticmethod
+    def is_parameter_value_in_range_expanded(
         speckle_object: Base,
         parameter_name: str,
         min_value: Union[int, float],
@@ -413,7 +456,20 @@ class RevitRules:
             bool: True if the parameter value is found in the list, False otherwise.
         """
         parameter_value = RevitRules.get_parameter_value(speckle_object, parameter_name)
-        return parameter_value in value_list
+
+        if isinstance(value_list, str):
+            value_list = [value.strip() for value in value_list.split(",")]
+
+        # parameter_value is effectively Any type, so to find its value in the value_list
+        def is_value_in_list(value: Any, my_list: Any) -> bool:
+            # Ensure that my_list is actually a list
+            if isinstance(my_list, list):
+                return value in my_list or str(value) in my_list
+            else:
+                speckle_print(f"Expected a list, got {type(my_list)} instead.")
+                return False
+
+        return is_value_in_list(parameter_value, value_list)
 
     @staticmethod
     def is_parameter_value_true(speckle_object: Base, parameter_name: str) -> bool:
@@ -498,32 +554,192 @@ class RevitRules:
         return RevitRules.get_parameter_value(speckle_object, "category")
 
 
-def filter_objects_by_category(
-    speckle_objects: List[Base], category_input: str
-) -> Tuple[List[Base], List[Base]]:
-    """
-    Filters objects by category value and test.
+# Mapping of input predicates to the corresponding methods in RevitRules
+input_predicate_mapping = {
+    "exists": "has_parameter",
+    "matches": "is_parameter_value",
+    "greater than": "is_parameter_value_greater_than",
+    "less than": "is_parameter_value_less_than",
+    "in range": "is_parameter_value_in_range",
+    "in list": "is_parameter_value_in_list",
+    "equals": "is_parameter_value",
+    "true": "is_parameter_value_true",
+    "false": "is_parameter_value_false",
+    "is like": "is_parameter_value_like",
+}
 
-    This function takes a list of Speckle objects, filters out the objects
-    with a matching category value and satisfies the test, and returns
-    both the matching and non-matching objects.
+
+def evaluate_condition(speckle_object: Base, condition: pd.Series) -> bool:
+    """
+    Given a Speckle object and a condition, evaluates the condition and returns a boolean value.
+    A condition is a pandas Series object with the following keys:
+    - 'Property Name': The name of the property to evaluate.
+    - 'Predicate': The predicate to use for evaluation.
+    - 'Value': The value to compare against.
 
     Args:
-        speckle_objects (List[Base]): The list of Speckle objects to filter.
-        category_input (str): The category value to match against.
+        speckle_object (Base): The Speckle object to evaluate.
+        condition (pd.Series): The condition to evaluate.
 
     Returns:
-        Tuple[List[Base], List[Base]]: A tuple containing two lists:
-                                        - The first list contains objects with matching category and test.
-                                        - The second list contains objects without matching category or test.
+        bool: The result of the evaluation. True if the condition is met, False otherwise.
     """
-    matching_objects = []
-    non_matching_objects = []
+    property_name = condition["Property Name"]
+    predicate_key = condition["Predicate"]
+    value = condition["Value"]
 
-    for speckle_object in speckle_objects:
-        if RevitRules.is_category(speckle_object, category_input):
-            matching_objects.append(speckle_object)
+    if predicate_key in input_predicate_mapping:
+        method_name = input_predicate_mapping[predicate_key]
+        method = getattr(RevitRules, method_name, None)
+
+        # speckle_print(f"Checking {property_name} {predicate_key} {value}")
+
+        if method:
+            check_answer = method(speckle_object, property_name, value)
+
+            return check_answer
+    return False
+
+
+def process_rule(
+    speckle_objects: List[Base], rule_group: pd.DataFrame
+) -> Tuple[List[Base], List[Base]]:
+    """
+    Processes a set of rules against Speckle objects, returning those that pass and fail.
+    The first rule is used as a filter ('WHERE'), and subsequent rules as conditions ('AND').
+
+    Args:
+        speckle_objects: List of Speckle objects to be processed.
+        rule_group: DataFrame defining the filter and conditions.
+
+    Returns:
+        A tuple of lists containing objects that passed and failed the rule.
+    """
+
+    # Extract the 'WHERE' condition and subsequent 'AND' conditions
+    filter_condition = rule_group.iloc[0]
+    subsequent_conditions = rule_group.iloc[1:]
+
+    # get the last row of the rule_group and get the Message and Report Severity
+    rule_info = rule_group.iloc[-1]
+
+    # Filter objects based on the 'WHERE' condition
+    filtered_objects = [
+        speckle_object
+        for speckle_object in speckle_objects
+        if evaluate_condition(speckle_object, filter_condition)
+    ]
+
+    rule_number = rule_info["Rule Number"]
+
+    speckle_print(
+        f"{ filter_condition['Logic']} {filter_condition['Property Name']} "
+        f"{filter_condition['Predicate']} {filter_condition['Value']}"
+    )
+
+    speckle_print(
+        f"{rule_number}: {len(list(filtered_objects))} objects passed the filter."
+    )
+
+    # Initialize lists for passed and failed objects
+    pass_objects, fail_objects = [], []
+
+    # Evaluate each filtered object against the 'AND' conditions
+    for speckle_object in filtered_objects:
+        if all(
+            evaluate_condition(speckle_object, cond)
+            for _, cond in subsequent_conditions.iterrows()
+        ):
+            pass_objects.append(speckle_object)
         else:
-            non_matching_objects.append(speckle_object)
+            fail_objects.append(speckle_object)
 
-    return matching_objects, non_matching_objects
+    return pass_objects, fail_objects
+
+
+def apply_rules_to_objects(
+    speckle_objects: List[Base],
+    rules_df: pd.DataFrame,
+    automate_context: AutomationContext,
+) -> dict[str, Tuple[List[Base], List[Base]]]:
+    """
+    Applies defined rules to a list of objects and updates the automate context based on the results.
+
+    Args:
+        speckle_objects (List[Base]): The list of objects to which rules are applied.
+        rules_df (pd.DataFrame): The DataFrame containing rule definitions.
+        automate_context (Any): Context manager for attaching rule results.
+    """
+    grouped_rules = rules_df.groupby("Rule Number")
+
+    grouped_results = {}
+
+    for rule_id, rule_group in grouped_rules:
+        rule_id_str = str(rule_id)  # Convert rule_id to string
+
+        # Ensure rule_group has necessary columns
+        if (
+            "Message" not in rule_group.columns
+            or "Report Severity" not in rule_group.columns
+        ):
+            continue  # Or raise an exception if these columns are mandatory
+
+        pass_objects, fail_objects = process_rule(speckle_objects, rule_group)
+
+        attach_results(
+            pass_objects, rule_group.iloc[-1], rule_id_str, automate_context, True
+        )
+        attach_results(
+            fail_objects, rule_group.iloc[-1], rule_id_str, automate_context, False
+        )
+
+        grouped_results[rule_id_str] = (pass_objects, fail_objects)
+
+    # return pass_objects, fail_objects for each rule
+    return grouped_results
+
+
+def attach_results(
+    speckle_objects: List[Base],
+    rule_info: pd.Series,
+    rule_id: str,
+    context: AutomationContext,
+    passed: bool,
+) -> None:
+    """
+    Attaches the results of a rule to the objects in the context.
+
+    Args:
+        speckle_objects (List[Base]): The list of objects to which the rule was applied.
+        rule_info (pd.Series): The information about the rule.
+        rule_id (str): The ID of the rule.
+        context (AutomationContext): The context manager for attaching results.
+        passed (bool): Whether the rule passed or failed.
+    """
+
+    if not speckle_objects:
+        return
+
+    message = f"{rule_info['Message']} - {'Passed' if passed else 'Failed'}"
+    if passed:
+        context.attach_info_to_objects(
+            category=f"Rule {rule_id} Success",
+            object_ids=[speckle_object.id for speckle_object in speckle_objects],
+            message=message,
+        )
+    else:
+
+        speckle_print(rule_info["Report Severity"])
+
+        severity = (
+            ObjectResultLevel.WARNING
+            if rule_info["Report Severity"].capitalize() == "Warning"
+            or rule_info["Report Severity"].capitalize() == "Warn"
+            else ObjectResultLevel.ERROR
+        )
+        context.attach_result_to_objects(
+            category=f"Rule {rule_id} Results",
+            object_ids=[speckle_object.id for speckle_object in speckle_objects],
+            message=message,
+            level=severity,
+        )