[WIP]

graph/planner/planner.py

@@ -1,8 +1,18 @@
-from typing import Union, Tuple, List
+from dataclasses import dataclass, field
+from typing import Union
 
 from direct_acyclic_graph import DirectAcyclicGraph, EdgeType
 
 
+@dataclass
+class NodeState:
+    user_selected: bool = True
+    primary_action: None | str = None
+    transitive_action: None | str = None
+    state: set[str] = field(default_factory=set)
+    transitive_state: set[str] = field(default_factory=set)
+
+
 class Planner:
     _state: DirectAcyclicGraph
     _target: DirectAcyclicGraph
@@ -24,6 +34,88 @@ def __init__(
         self._unhealthy = unhealthy or []
         self._selected = selected
 
+    def propagate(self) -> tuple[dict[str, NodeState], list[str]]:
+        nodes_state = self._state.nodes()
+        nodes_target = self._target.nodes()
+
+        nodes_add = (nodes_target - nodes_state).intersection(
+            (nodes_target - nodes_state)
+            if self._selected == "*"
+            else {node for node in self._selected if not node.startswith("-")}
+        )
+        nodes_delete = (
+            (nodes_state - nodes_target)
+            if self._selected == "*"
+            else {node[1:] for node in self._selected if node.startswith("-")}
+        )
+        nodes_selected = (
+            nodes_state.union(nodes_target)
+            if self._selected == "*"
+            else {
+                node if not node.startswith("-") else node[1:]
+                for node in self._selected
+            }
+        )
+
+        nodes_state = {
+            node: NodeState(
+                user_selected=node in nodes_selected,
+                primary_action=(
+                    "add"
+                    if node in nodes_add
+                    else "delete"
+                    if node in nodes_delete
+                    else None
+                ),
+                state=({"modified"} if node in self._modified else set()).union(
+                    {"unhealthy"} if node in self._unhealthy else set()
+                ),
+            )
+            for node in (nodes_state.union(nodes_target))
+        }
+
+        for node in nodes_delete:
+            childs_transitive = self._state.childs_transitive(
+                node,
+                accept=lambda adjacent: adjacent[1] != EdgeType.TOOL,
+            )
+            for node_transitive_delete in childs_transitive:
+                # if nodes[node_transitive_delete].primary_action != "delete":
+                nodes_state[node_transitive_delete].transitive_action = "delete"
+
+        unhealthy_adjacent_tool_edge_nodes = set()
+        for key, value in nodes_state.items():
+            node = key
+            state = value.state
+            if "unhealthy" in state:
+                # propagate 'unhealthy' state to children != EdgeType.TOOL
+                childs_transitive = self._state.childs_transitive(
+                    node,
+                    accept=lambda adjacent: adjacent[1] != EdgeType.TOOL,
+                )
+                for node_transitive_state in childs_transitive:
+                    nodes_state[node_transitive_state].transitive_state = nodes_state[
+                        node_transitive_state
+                    ].transitive_state.union({"unhealthy"})
+
+                # collect nodes ('unhealthy')' w/ EdgeType.TOOL in reverse graph ~> incoming edge(s)
+                state_inverse = self._state.inverse()
+                for edge in state_inverse[node]:
+                    if edge[1] == EdgeType.TOOL:
+                        unhealthy_adjacent_tool_edge_nodes.add(node)
+                        break
+
+        operations = []
+        if len(unhealthy_adjacent_tool_edge_nodes) > 0:
+            for node in reversed(self._state.topological_order()):
+                if node in unhealthy_adjacent_tool_edge_nodes:
+                    operations.append(f"-{node}")
+            for node in self._state.topological_order():
+                if node in unhealthy_adjacent_tool_edge_nodes:
+                    operations.append(f"+{node}")
+
+        return nodes_state, operations
+
     def apply(self) -> list[str]:
         nodes_state = self._state.nodes()
         nodes_target = self._target.nodes()