Skip to content

Commit

Permalink
[WIP]
Browse files Browse the repository at this point in the history
  • Loading branch information
@agebhar1
agebhar1 committed Sep 26, 2023
1 parent c8b380d commit 660358d
Show file tree
Hide file tree
Showing 3 changed files with 933 additions and 165 deletions.
6 changes: 6 additions & 0 deletions graph/planner/direct_acyclic_graph.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,9 @@ def _topsort(self):
result.reverse()
self._topological_order = result

def has_node(self, node) -> bool:
return node in self._g

def nodes(self) -> set[str]:
return {item[0] for item in self._g.items()}

Expand All @@ -90,6 +93,9 @@ def topological_order(self) -> list[str]:
def inverse(self) -> dict[str, list[AdjacentEdge]]:
return self._g_inverse

def dependent_edge_types(self, node) -> set[EdgeType]:
return {adjacent[1] for adjacent in self._g_inverse[node]}

def childs(
self,
node: str,
Expand Down
269 changes: 145 additions & 124 deletions graph/planner/planner.py
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,19 @@
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 # TODO set?
transitive_action: None | str = None # TODO set?
state: set[str] = field(default_factory=set)
transitive_state: set[str] = field(default_factory=set)
dependent_edge_types: set[EdgeType] = field(default_factory=set)


class Planner:
_state: DirectAcyclicGraph
_target: DirectAcyclicGraph
Expand All @@ -24,141 +35,151 @@ def __init__(
self._unhealthy = unhealthy or []
self._selected = selected

def apply(self) -> list[str]:
def apply(self) -> tuple[dict[str, NodeState], list[str]]:
nodes_state = self._state.nodes()
nodes_target = self._target.nodes()

nodes_add = set()
nodes_delete = set()
nodes_with_edge_type_tool_delete = set()
modified_transitive = set()
unhealthy_transitive = set()

if self._selected == "*":
nodes_add = {node for node in nodes_target - nodes_state}
nodes_delete = {node for node in nodes_state - nodes_target}

for node in self._modified:
modified_transitive = modified_transitive.union(
self._state.childs_transitive(
node, accept=lambda adjacent: adjacent[1] != EdgeType.TOOL
).union({node})
)
for node in self._unhealthy:
unhealthy_transitive = unhealthy_transitive.union(
self._state.childs_transitive(
node, accept=lambda adjacent: adjacent[1] != EdgeType.TOOL
).union({node})
)
else:
# partial apply mode
for item in self._selected:
if item.startswith("-"):
# delete case
node = str(item[1:])
has_parent_with_edge_type_tool = len(self._state.parents(node)) > 0
if has_parent_with_edge_type_tool:
parents_edge_type_tool_transitive = (
self._state.parents_transitive(
node,
accept=lambda adjacent: adjacent[1] == EdgeType.TOOL
and adjacent[0] in self._unhealthy,
)
)
unhealthy_transitive = unhealthy_transitive.union(
parents_edge_type_tool_transitive
)

childs_transitive = self._state.childs_transitive(
node,
accept=lambda adjacent: adjacent[1] != EdgeType.TOOL,
).union({node})

nodes_with_edge_type_tool_delete = (
nodes_with_edge_type_tool_delete.union(childs_transitive)
)

else:
# use state (graph) instead of target, because target will be mostly ignored in this mode
childs_transitive = self._state.childs_transitive(
node,
accept=lambda adjacent: adjacent[1] != EdgeType.TOOL,
).union({node})

nodes_delete = nodes_delete.union(childs_transitive)

else:
node = item

is_node_modified = node in self._modified
is_node_unhealthy = node in self._unhealthy

if (
is_node_modified or is_node_unhealthy
): # and node in node_states <- implicit
childs_transitive = self._state.childs_transitive(
node,
accept=lambda adjacent: adjacent[1] != EdgeType.TOOL,
).union({node})

key = "modified"
if is_node_modified:
modified_transitive = modified_transitive.union(
childs_transitive
)
if is_node_unhealthy and not is_node_modified:
key = "unhealthy"
unhealthy_transitive = unhealthy_transitive.union(
childs_transitive
)

elif node in nodes_target:
node_with_parents_transitive = self._target.parents_transitive(
node
).union({node})
nodes_add = node_with_parents_transitive.difference(nodes_state)

for node_add in nodes_add:
parent_transitive = self._target.parents_transitive(
node_add,
accept=lambda adjacent: (
adjacent[0] in self._unhealthy
),
)
unhealthy_transitive = unhealthy_transitive.union(
parent_transitive
)

node_action = {node: None for node in nodes_state.union(nodes_target)}

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 = {
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()
),
dependent_edge_types=(
self._state.dependent_edge_types(node)
if self._state.has_node(node)
else set()
).union(
self._target.dependent_edge_types(node)
if self._target.has_node(node)
else set()
),
)
for node in (nodes_state.union(nodes_target))
}

# add required nodes by parent dependency
for node in nodes_add:
node_action[node] = "add"
parent_transitive = self._target.parents_transitive(node)
for node_add_transitive in parent_transitive:
if node_add_transitive not in nodes_state:
nodes[node_add_transitive].transitive_action = "add"

# delete child nodes
for node in nodes_delete:
node_action[node] = "delete"

for node in modified_transitive:
node_action[node] = "modified"

for node in unhealthy_transitive:
node_action[node] = "unhealthy"
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[node_transitive_delete].transitive_action = "delete"

# propagate 'modified'/'unhealthy' state
unhealthy_adjacent_tool_edge_nodes = set()
for key, value in nodes.items():
node = key
state = value.state
# TODO test if reachable/required from user selected nodes
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[node_transitive_state].transitive_state = nodes[
node_transitive_state
].transitive_state.union({"unhealthy"})

# collect nodes ('unhealthy')' w/ EdgeType.TOOL in reverse graph ~> incoming edge(s)
if EdgeType.TOOL in value.dependent_edge_types:
unhealthy_adjacent_tool_edge_nodes.add(node)

if "modified" in state and value.user_selected:
# propagate 'modified' 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[node_transitive_state].transitive_state = nodes[
node_transitive_state
].transitive_state.union({"modified"})

operations = []
# recover unhealthy nodes (required by dependent 'active' nodes)
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}")

# TODO re-run w/ updated unhealthy list

nodes_re_add_modified_or_unhealthy = (
[]
) # optimization, not evaluate modified/unhealthy again
# 'delete' run
# TODO postpone nodes wich are required as tool
for node in self._state.topological_order():
if node_action[node] in ["delete", "modified", "unhealthy"]:
value = nodes[node]

modified = (
"modified" in value.state
and value.user_selected
or "modified" in value.transitive_state
)
unhealthy = (
"unhealthy" in value.state and value.user_selected
) or "unhealthy" in value.transitive_state
delete = (
"delete" == value.primary_action or "delete" == value.transitive_action
)

if not delete and (modified or unhealthy):
operations.append(f"-{node}")
nodes_re_add_modified_or_unhealthy.append(node)

if delete:
operations.append(f"-{node}")

# 'add' run
for node in reversed(self._target.topological_order()):
if node_action[node] in ["add", "modified", "unhealthy"]:
operations.append(f"+{node}")
value = nodes[node]

# handle node deletion of nodes w/ edges of type TOOL
node_action = {node: None for node in nodes_state.union(nodes_target)}
for node in nodes_with_edge_type_tool_delete:
node_action[node] = "delete"
add = "add" == value.primary_action or "add" == value.transitive_action
if add or node in nodes_re_add_modified_or_unhealthy:
operations.append(f"+{node}")

for node in self._state.topological_order():
if node_action[node] in ["delete"]:
operations.append(f"-{node}")
# TODO delete postponed nodes required as tool but selected for 'delete'

return operations
return nodes, operations
Loading

0 comments on commit 660358d

Please sign in to comment.