Add operator to conditional links

src/ewokscore/graph/analysis.py

@@ -1,12 +1,21 @@
 from collections import defaultdict
+from typing import Any
 from typing import Dict
 from typing import Iterator
+from typing import List
 from typing import Set
+from typing import Tuple
 
 import networkx
 
 from ..inittask import get_task_class
 from ..node import NodeIdType
+from .conditions import COMPLEMENTARY_CONDITION_OPERATORS
+from .conditions import DEFAULT_CONDITION_OPERATOR
+from .conditions import condition_value_for_set
+from .conditions import condition_values_equal
+from .conditions import is_condition_collection_value
+from .conditions import normalize_condition
 
 
 def graph_is_cyclic(graph: networkx.DiGraph) -> bool:
@@ -283,26 +292,87 @@ def node_condition_values(
     condition_values = defaultdict(set)
     for target_id in node_successors(graph, source_id, link_has_conditions=True):
         for condition in graph[source_id][target_id]["conditions"]:
-            varname = condition["source_output"]
-            value = condition["value"]
-            condition_values[varname].add(value)
+            varname, _, value = normalize_condition(condition)
+            condition_values[varname].add(condition_value_for_set(value))
     return condition_values
 
 
+def _node_condition_predicates(
+    graph: networkx.DiGraph, source_id: NodeIdType
+) -> Dict[str, List[Tuple[str, Any]]]:
+    condition_predicates = defaultdict(list)
+    for target_id in node_successors(graph, source_id, link_has_conditions=True):
+        for condition in graph[source_id][target_id]["conditions"]:
+            varname, operator_name, value = normalize_condition(condition)
+            condition_predicates[varname].append((operator_name, value))
+    return condition_predicates
+
+
+def _has_condition_predicate(
+    predicates: List[Tuple[str, Any]], operator_name: str, value: Any
+) -> bool:
+    return any(
+        other_operator_name == operator_name
+        and condition_values_equal(other_value, value)
+        for other_operator_name, other_value in predicates
+    )
+
+
+def _condition_predicates_are_covered(predicates: List[Tuple[str, Any]]) -> bool:
+    for operator_name, value in predicates:
+        complement_operator = COMPLEMENTARY_CONDITION_OPERATORS.get(operator_name)
+        if complement_operator is None:
+            continue
+        if _has_condition_predicate(predicates, complement_operator, value):
+            return True
+
+    return any(
+        operator_name == "not_in"
+        and _not_in_predicate_is_covered(predicates, value)
+        for operator_name, value in predicates
+    ) or _boolean_condition_values_are_covered(predicates)
+
+
+def _boolean_condition_values_are_covered(
+    predicates: List[Tuple[str, Any]]
+) -> bool:
+    return _has_condition_predicate(
+        predicates, DEFAULT_CONDITION_OPERATOR, True
+    ) and _has_condition_predicate(predicates, DEFAULT_CONDITION_OPERATOR, False)
+
+
+def _not_in_predicate_is_covered(
+    predicates: List[Tuple[str, Any]], values: Any
+) -> bool:
+    if not is_condition_collection_value(values):
+        return False
+    return all(_value_is_covered(predicates, value) for value in values)
+
+
+def _value_is_covered(predicates: List[Tuple[str, Any]], value: Any) -> bool:
+    for operator_name, other_value in predicates:
+        if operator_name == DEFAULT_CONDITION_OPERATOR and condition_values_equal(
+            value, other_value
+        ):
+            return True
+        if operator_name == "in" and _collection_contains_value(other_value, value):
+            return True
+    return False
+
+
+def _collection_contains_value(values: Any, value: Any) -> bool:
+    if not is_condition_collection_value(values):
+        return False
+    return any(condition_values_equal(value, candidate) for candidate in values)
+
+
 def node_has_noncovered_conditions(
     graph: networkx.DiGraph, source_id: NodeIdType
 ) -> bool:
-    conditions_else_value = graph.nodes[source_id].get("conditions_else_value", None)
-    complements = {
-        True: {False, conditions_else_value},
-        False: {True, conditions_else_value},
-    }
-    condition_values = node_condition_values(graph, source_id)
-    for values in condition_values.values():
-        for value in values:
-            cvalue = complements.get(value, conditions_else_value)
-            if cvalue not in values:
-                return True
+    condition_predicates = _node_condition_predicates(graph, source_id)
+    for predicates in condition_predicates.values():
+        if not _condition_predicates_are_covered(predicates):
+            return True
     return False
 
 

src/ewokscore/graph/compare.py

@@ -3,6 +3,8 @@
 
 import networkx
 
+from .conditions import DEFAULT_CONDITION_OPERATOR
+
 
 def graphs_are_equal(graph1: networkx.DiGraph, graph2: networkx.DiGraph) -> bool:
     if not _attrs_are_equal(graph1.graph, graph2.graph):
@@ -37,8 +39,8 @@ def _attrs_are_equal(attrs1: dict, attrs2: dict) -> bool:
             value1 = _items_to_dict(value1, ["source_output", "target_input"])
             value2 = _items_to_dict(value2, ["source_output", "target_input"])
         elif name == "conditions":
-            value1 = _items_to_dict(value1, ["source_output"])
-            value2 = _items_to_dict(value2, ["source_output"])
+            value1 = _conditions_to_dict(value1)
+            value2 = _conditions_to_dict(value2)
         if value1 != value2:
             return False
     return True
@@ -49,3 +51,12 @@ def _items_to_dict(items: List[dict], keys: Sequence[str]) -> dict:
     for item in items:
         result[tuple(item[k] for k in keys)] = item
     return result
+
+
+def _conditions_to_dict(conditions: List[dict]) -> dict:
+    result = dict()
+    for condition in conditions:
+        condition = dict(condition)
+        condition["operator"] = condition.get("operator", DEFAULT_CONDITION_OPERATOR)
+        result[(condition["source_output"], condition["operator"])] = condition
+    return result

src/ewokscore/graph/conditions.py

@@ -0,0 +1,63 @@
+import operator
+from typing import Any
+from typing import Tuple
+
+from ..node import NodeIdType
+
+
+def _in(actual: Any, expected_values: Any) -> bool:
+    return actual in expected_values
+
+
+def _not_in(actual: Any, expected_values: Any) -> bool:
+    return actual not in expected_values
+
+
+CONDITION_OPERATORS = {
+    "eq": operator.eq,
+    "ne": operator.ne,
+    "lt": operator.lt,
+    "le": operator.le,
+    "gt": operator.gt,
+    "ge": operator.ge,
+    "in": _in,
+    "not_in": _not_in,
+}
+
+CONDITION_OPERATOR_NAMES = tuple(CONDITION_OPERATORS)
+CONDITION_COLLECTION_OPERATOR_NAMES = ("in", "not_in")
+DEFAULT_CONDITION_OPERATOR = "eq"
+COMPLEMENTARY_CONDITION_OPERATORS = {
+    "eq": "ne",
+    "ne": "eq",
+    "lt": "ge",
+    "ge": "lt",
+    "le": "gt",
+    "gt": "le",
+    "in": "not_in",
+    "not_in": "in",
+}
+
+
+def normalize_condition(condition: dict) -> Tuple[NodeIdType, str, Any]:
+    return (
+        condition["source_output"],
+        condition.get("operator", DEFAULT_CONDITION_OPERATOR),
+        condition["value"],
+    )
+
+
+def is_condition_collection_value(value: Any) -> bool:
+    return isinstance(value, (list, tuple, set))
+
+
+def condition_values_equal(value1: Any, value2: Any) -> bool:
+    return value1 == value2
+
+
+def condition_value_for_set(value: Any) -> Any:
+    if isinstance(value, list):
+        return tuple(value)
+    if isinstance(value, set):
+        return frozenset(value)
+    return value

src/ewokscore/graph/schema/model.py

@@ -34,6 +34,7 @@ class EwoksDataMapping(BaseModel):
 
 class EwoksCondition(BaseModel):
     source_output: NodeId
+    operator: Literal["eq", "ne", "lt", "le", "gt", "ge", "in", "not_in"] = "eq"
     value: Any
 
 

src/ewokscore/graph/schema/update.py

@@ -1,5 +1,10 @@
+from collections import defaultdict
+
 import networkx
 
+from ..conditions import DEFAULT_CONDITION_OPERATOR
+from ..conditions import condition_values_equal
+
 
 def v0_update(graph: networkx.DiGraph) -> None:
     """Outdated version"""
@@ -24,3 +29,40 @@ def from_v1_1_to_v1_2(graph: networkx.DiGraph) -> None:
 def from_v1_2_to_v1_3(graph: networkx.DiGraph) -> None:
     """v1.3 adds `workflow_input_schema` and `workflow_output_schema`"""
     graph.graph["schema_version"] = "1.3"
+    """This version does not have conditional link operators."""
+    _migrate_conditions_else_value(graph)
+    graph.graph["schema_version"] = "1.2"
+
+
+def _migrate_conditions_else_value(graph: networkx.DiGraph) -> None:
+    for source_id, node_attrs in graph.nodes.items():
+        if "conditions_else_value" not in node_attrs:
+            continue
+
+        conditions_else_value = node_attrs.get("conditions_else_value", None)
+        covered_values = defaultdict(list)
+        else_conditions = []
+
+        for target_id in graph.successors(source_id):
+            for condition in graph[source_id][target_id].get("conditions", list()):
+                if (
+                    condition.get("operator", DEFAULT_CONDITION_OPERATOR)
+                    != DEFAULT_CONDITION_OPERATOR
+                ):
+                    continue
+                source_output = condition["source_output"]
+                if condition_values_equal(condition["value"], conditions_else_value):
+                    else_conditions.append(condition)
+                else:
+                    condition["operator"] = DEFAULT_CONDITION_OPERATOR
+                    values = covered_values[source_output]
+                    if not any(
+                        condition_values_equal(condition["value"], value)
+                        for value in values
+                    ):
+                        values.append(condition["value"])
+
+        for condition in else_conditions:
+            source_output = condition["source_output"]
+            condition["operator"] = "not_in"
+            condition["value"] = list(covered_values.get(source_output, list()))

src/ewokscore/graph/validate.py

@@ -3,6 +3,10 @@
 
 from ..inittask import validate_task_executable
 from .analysis import required_predecessors
+from .conditions import CONDITION_COLLECTION_OPERATOR_NAMES
+from .conditions import CONDITION_OPERATOR_NAMES
+from .conditions import DEFAULT_CONDITION_OPERATOR
+from .conditions import is_condition_collection_value
 from .schema import LATEST_VERSION
 from .schema import normalize_schema_version
 from .schema import update_graph_schema
@@ -50,3 +54,16 @@ def _validate_links(graph: networkx.DiGraph) -> None:
             raise ValueError(err_msg.format("map_all_data", "data_mapping"))
         if linkattrs.get("on_error") and linkattrs.get("conditions"):
             raise ValueError(err_msg.format("on_error", "conditions"))
+        for condition in linkattrs.get("conditions", list()):
+            operator = condition.get("operator", DEFAULT_CONDITION_OPERATOR)
+            if operator not in CONDITION_OPERATOR_NAMES:
+                raise ValueError(
+                    f"Link {source}->{target}: unsupported condition operator {repr(operator)}"
+                )
+            if (
+                operator in CONDITION_COLLECTION_OPERATOR_NAMES
+                and not is_condition_collection_value(condition.get("value"))
+            ):
+                raise ValueError(
+                    f"Link {source}->{target}: condition operator {repr(operator)} requires a non-string iterable value"
+                )