diff --git a/src/ewokscore/graph/analysis.py b/src/ewokscore/graph/analysis.py index 63a00c5..f7b6398 100644 --- a/src/ewokscore/graph/analysis.py +++ b/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 diff --git a/src/ewokscore/graph/compare.py b/src/ewokscore/graph/compare.py index 1e8a612..2c3ba67 100644 --- a/src/ewokscore/graph/compare.py +++ b/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 diff --git a/src/ewokscore/graph/conditions.py b/src/ewokscore/graph/conditions.py new file mode 100644 index 0000000..a29ca76 --- /dev/null +++ b/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 diff --git a/src/ewokscore/graph/schema/model.py b/src/ewokscore/graph/schema/model.py index b49ab05..21e31b0 100644 --- a/src/ewokscore/graph/schema/model.py +++ b/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 diff --git a/src/ewokscore/graph/schema/update.py b/src/ewokscore/graph/schema/update.py index 1d886c7..94dca0b 100644 --- a/src/ewokscore/graph/schema/update.py +++ b/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())) diff --git a/src/ewokscore/graph/validate.py b/src/ewokscore/graph/validate.py index 53ceb6e..f9ebb1e 100644 --- a/src/ewokscore/graph/validate.py +++ b/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" + )