From f7412696dddef7fa22d2c0f8b1afdd77038598d2 Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 13:05:29 +0900 Subject: [PATCH 1/6] Replace the Sequence tree element with a next_action chain. --- .../dynamic_stack_decider/dsd.py | 43 +++----- .../dynamic_stack_decider/parser.py | 32 +++--- .../dynamic_stack_decider/sequence_element.py | 98 ------------------- .../dynamic_stack_decider/tree.py | 65 +++++------- dynamic_stack_decider/test/test_parser.py | 28 +++--- 5 files changed, 70 insertions(+), 196 deletions(-) delete mode 100644 dynamic_stack_decider/dynamic_stack_decider/sequence_element.py diff --git a/dynamic_stack_decider/dynamic_stack_decider/dsd.py b/dynamic_stack_decider/dynamic_stack_decider/dsd.py index b4743b6..0f9acc2 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/dsd.py +++ b/dynamic_stack_decider/dynamic_stack_decider/dsd.py @@ -16,12 +16,10 @@ from dynamic_stack_decider.abstract_stack_element import AbstractStackElement from dynamic_stack_decider.logger import get_logger from dynamic_stack_decider.parser import DsdParser -from dynamic_stack_decider.sequence_element import SequenceElement from dynamic_stack_decider.tree import ( AbstractTreeElement, ActionTreeElement, DecisionTreeElement, - SequenceTreeElement, Tree, ) @@ -188,6 +186,9 @@ def _bind_modules(self, element): element.name in self.actions ), f'Provided element "{element.name}" was not found in registered actions!' element.module = self.actions[element.name] + # Bind the rest of the chain (e.g. the following actions of a sequence) + if element.next_element is not None: + self._bind_modules(element.next_element) elif isinstance(element, DecisionTreeElement): assert ( element.name in self.decisions @@ -195,18 +196,12 @@ def _bind_modules(self, element): element.module = self.decisions[element.name] for child in element.children.values(): self._bind_modules(child) - elif isinstance(element, SequenceTreeElement): - for action in element.action_elements: - self._bind_modules(action) else: raise ValueError(f'Unknown parser tree element type "{type(element)}" for element "{element}"!') def _init_element(self, element: AbstractTreeElement) -> AbstractStackElement: """Initializes the module belonging to the given element.""" - if isinstance(element, SequenceTreeElement): - return SequenceElement(self.blackboard, self, element.action_elements, self._init_element) - else: - return element.module(self.blackboard, self, element.parameters) + return element.module(self.blackboard, self, element.parameters) def set_start_element(self, start_element: AbstractTreeElement): """ @@ -274,12 +269,7 @@ def update(self, reevaluate: bool = True): if reevaluate: # reset flag self.do_not_reevaluate = False - if ( - isinstance(current_instance, AbstractActionElement) - and current_instance.never_reevaluate - or isinstance(current_instance, SequenceElement) - and current_instance.current_action.never_reevaluate - ): + if isinstance(current_instance, AbstractActionElement) and current_instance.never_reevaluate: # Deactivate reevaluation if action had never_reevaluate flag self.set_do_not_reevaluate() # Run the top module @@ -316,15 +306,16 @@ def pop(self): # stop reevaluating self.stack_reevaluate = False else: - if isinstance(self.stack[-1][1], SequenceElement): - # If we are in a sequence, only one action should be popped - if not self.stack[-1][1].in_last_element(): - # We are still in the sequence, therefore we do not want to pop the SequenceElement, - # only a single element of the sequence - # We also do not want to reset do_not_reevaluate because an action in the sequence - # may control the stack beyond its own lifetime but in the sequence element's lifetime - self.stack[-1][1].pop_one().on_pop() - return + current_tree_element = self.stack[-1][0] + if isinstance(current_tree_element, ActionTreeElement) and current_tree_element.next_element is not None: + # This action is part of a chain (e.g. a sequence). Instead of returning to the + # element below, we advance to the next element in the chain. + # We do not reset do_not_reevaluate because an action in the chain may control + # the stack beyond its own lifetime but within the chain's lifetime. + self.stack.pop()[1].on_pop() + next_element = current_tree_element.next_element + self.stack.append((next_element, self._init_element(next_element))) + return # Remove the last element of the stack self.stack.pop()[1].on_pop() @@ -378,11 +369,9 @@ def debug_publish_current_action(self): # Get the top element stack_top = self.stack[-1][1] - # Check if it is an action or a sequence element and retrieve the current action + # Check if it is an action and retrieve it if isinstance(stack_top, AbstractActionElement): current_action = stack_top - elif isinstance(stack_top, SequenceElement): - current_action = stack_top.current_action else: return diff --git a/dynamic_stack_decider/dynamic_stack_decider/parser.py b/dynamic_stack_decider/dynamic_stack_decider/parser.py index bddd4c4..36d5152 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/parser.py +++ b/dynamic_stack_decider/dynamic_stack_decider/parser.py @@ -5,7 +5,7 @@ import yaml from rclpy.node import Node -from dynamic_stack_decider.tree import ActionTreeElement, DecisionTreeElement, SequenceTreeElement, Tree +from dynamic_stack_decider.tree import ActionTreeElement, DecisionTreeElement, Tree class DsdParser: @@ -126,12 +126,11 @@ def parse(self, file_path: str) -> Tree: to_evaluate = [subtree_copy] while len(to_evaluate) > 0: current = to_evaluate.pop(0) - if isinstance(current, SequenceTreeElement): - # For sequence elements, only evaluate the actions - to_evaluate.extend(current.action_elements) - continue - elif isinstance(current, DecisionTreeElement): + if isinstance(current, DecisionTreeElement): to_evaluate.extend(current.children.values()) + elif isinstance(current, ActionTreeElement) and current.next_element is not None: + # Follow the chain of actions (i.e. a sequence) + to_evaluate.append(current.next_element) for name, reference in current.unset_parameters.items(): if reference in parameters: @@ -223,7 +222,7 @@ def _create_tree_element(self, token, parent, lnr): The method derives the type (Action/Decision) and optional parameters from the token :param token: the string describing the element in the dsd description :param parent: the parent element of the new element, None for root - :type parent: Union[DecisionTreeElement, SequenceTreeElement] + :type parent: Union[DecisionTreeElement, ActionTreeElement] :param lnr: Line number of the current line (used for error messages) :type lnr: int :return: a TreeElement containing the information given in token @@ -239,7 +238,7 @@ def _create_tree_element(self, token, parent, lnr): def _create_sequence_element(self, actions, parent, lnr): """ - Create a new sequence element + Create a sequence as a chain of action elements linked via ``next_element`` :param actions: The names of actions in the sequence :type actions: list[str] @@ -247,15 +246,22 @@ def _create_sequence_element(self, actions, parent, lnr): :type parent: AbstractDecisionElement :param lnr: Line number of the current line (used for error messages) :type lnr: int - :return: The sequence element + :return: The head action element of the chain """ - sequence_element = SequenceTreeElement(parent) + head = None + previous = None for action in actions: - element = self._create_tree_element(action, sequence_element, lnr) + # The parent of the head is the given parent, every following action is + # parented to its predecessor in the chain + element = self._create_tree_element(action, parent if previous is None else previous, lnr) if not isinstance(element, ActionTreeElement): raise ParseError("In a sequence, only actions are allowed!") - sequence_element.add_action_element(element) - return sequence_element + if previous is None: + head = element + else: + previous.next_element = element + previous = element + return head class ParseError(AssertionError): diff --git a/dynamic_stack_decider/dynamic_stack_decider/sequence_element.py b/dynamic_stack_decider/dynamic_stack_decider/sequence_element.py deleted file mode 100644 index d653b6c..0000000 --- a/dynamic_stack_decider/dynamic_stack_decider/sequence_element.py +++ /dev/null @@ -1,98 +0,0 @@ -from typing import TYPE_CHECKING, Callable - -from dynamic_stack_decider.abstract_action_element import AbstractActionElement -from dynamic_stack_decider.abstract_stack_element import AbstractStackElement -from dynamic_stack_decider.tree import AbstractTreeElement, ActionTreeElement - -if TYPE_CHECKING: - from dynamic_stack_decider.dsd import DSD - - -class SequenceElement(AbstractStackElement): - """ - A sequence element contains multiple action elements. - The sequence element executes all the actions consecutively. That means that the first action will be performed - until it pops itself off the stack. Then, the second action is performed. When the last action is popped, the - sequence element pops itself too. - This is not an abstract class to inherit from. - """ - - def __init__( - self, - blackboard, - dsd: "DSD", - actions: list[ActionTreeElement], - init_function: Callable[[AbstractTreeElement], AbstractStackElement], - ): - """ - :param blackboard: Shared blackboard for data exchange and code reuse between elements - :param dsd: The stack decider which has this element on its stack. - :param actions: list of of action tree elements / blueprints for the actions - :param init_function: A function that initializes an action element creating a stack element from a tree element - """ - super().__init__(blackboard, dsd, dict()) - # Here we store the 'blueprints' of the actions - self.actions = actions - # We store a reference to the function that initializes the action elements based on the tree - self._init_function = init_function - # Here we store the actual instances of the active action - # The action is only initialized when it is the current action - assert len(actions) > 0, "A sequence element must contain at least one action" - self.current_action: AbstractActionElement = self._init_function(actions[0]) - # Where we are in the sequence - self.current_action_index: int = 0 - - def perform(self, reevaluate=False): - """ - Perform the current action of the sequence. See AbstractStackElement.perform() for more information - - :param reevaluate: Ignored for SequenceElements - """ - # Log the active element - self.publish_debug_data("Active Element", self.current_action.name) - # Pass the perform call to the current action - self.current_action.perform() - # If the action had debug data, publish it - if self.current_action._debug_data: - self.publish_debug_data("Corresponding debug data", self.current_action._debug_data) - - def pop_one(self): - """ - Pop a single element of the sequence - """ - assert not self.in_last_element(), ( - "It is not possible to pop a single element when" "the last element of the sequence is active" - ) - # Save the current action to return it - popped_action = self.current_action - # Increment the index to the next action and initialize it - self.current_action_index += 1 - # We initialize the current action here to avoid the problem described in - # https://github.com/bit-bots/dynamic_stack_decider/issues/107 - self.current_action = self._init_function(self.actions[self.current_action_index]) - # Return the popped action - return popped_action - - def on_pop(self): - """ - This method is called when the sequence is popped from the stack. - This means that the last element of the sequence was also popped, so - """ - self.current_action.on_pop() - - def in_last_element(self): - """Returns if the current element is the last element of the sequence""" - return self.current_action_index == len(self.actions) - 1 - - def repr_dict(self) -> dict: - """ - Represent this stack element as dictionary which is JSON encodable - """ - data = { - "type": "sequence", - "current_action_index": self.current_action_index, - "current_action": self.current_action.repr_dict(), - "debug_data": self._debug_data, - } - self.clear_debug_data() - return data diff --git a/dynamic_stack_decider/dynamic_stack_decider/tree.py b/dynamic_stack_decider/dynamic_stack_decider/tree.py index b47a586..3d7db9d 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/tree.py +++ b/dynamic_stack_decider/dynamic_stack_decider/tree.py @@ -132,51 +132,16 @@ def repr_dict(self) -> dict: return result -class SequenceTreeElement(AbstractTreeElement): - """ - A tree element describing a sequence of actions. Each action - has optional parameters that will be passed to the module on - creation - """ - - def __init__(self, parent): - super().__init__(None, parent) - self.action_elements: list[ActionTreeElement] = [] - - def add_action_element(self, action_element): - """ - Add an action element to the sequence - - :param action_element: The action element to be added - :type action_element: ActionTreeElement - """ - action_element.activation_reason = self.activation_reason - self.action_elements.append(action_element) - - def set_activation_reason(self, reason): - self.activation_reason = reason - for action in self.action_elements: - action.set_activation_reason(reason) - - def __repr__(self): - return "({})".format(", ".join(repr(action) for action in self.action_elements)) - - def repr_dict(self): - """ - Represent this sequence as dictionary which is JSON encodable - """ - # Get the dict from the superclass - result = super().repr_dict() - # Add additional information - result["type"] = "sequence" - result["action_elements"] = [action.repr_dict() for action in self.action_elements] - return result - - class ActionTreeElement(AbstractTreeElement): """ A tree element describing an action. An action has optional - parameters that will be passed to the module on creation + parameters that will be passed to the module on creation. + + An action may point to a ``next_element`` which is activated once the action + pops itself off the stack. This is how sequences (``@A, @B, @C``) are + represented: as a chain of actions linked via ``next_element``. The next + element may be any tree element, so a decision is also allowed to follow an + action. """ def __init__(self, name, parent, parameters=None, unset_parameters=None): @@ -188,9 +153,22 @@ def __init__(self, name, parent, parameters=None, unset_parameters=None): :param unset_parameters: A dictionary of parameters that must be set later """ super().__init__(name, parent, parameters, unset_parameters) - self.in_sequence = False + # The element that is activated when this action pops itself. None if this + # action is the last (or only) element in its chain. + self.next_element: Optional[AbstractTreeElement] = None + + def set_activation_reason(self, reason): + """Set the activation reason for this action and propagate it along the chain""" + self.activation_reason = reason + # All elements of a chain share the activation reason of the head so that a + # reevaluated decision below the chain does not consider its result changed + # while the chain is still running. + if self.next_element is not None: + self.next_element.set_activation_reason(reason) def __repr__(self): + if self.next_element is not None: + return f"@{self.name} ({self.parameters}) -> {repr(self.next_element)}" return f"@{self.name} ({self.parameters})" def __str__(self): @@ -204,4 +182,5 @@ def repr_dict(self) -> dict: result = super().repr_dict() # Add additional information result["type"] = "action" + result["next"] = self.next_element.repr_dict() if self.next_element is not None else None return result diff --git a/dynamic_stack_decider/test/test_parser.py b/dynamic_stack_decider/test/test_parser.py index 2bf7de1..c477914 100644 --- a/dynamic_stack_decider/test/test_parser.py +++ b/dynamic_stack_decider/test/test_parser.py @@ -3,7 +3,7 @@ import pytest from dynamic_stack_decider.parser import DsdParser, Tree -from dynamic_stack_decider.tree import ActionTreeElement, DecisionTreeElement, SequenceTreeElement +from dynamic_stack_decider.tree import ActionTreeElement, DecisionTreeElement @pytest.fixture() @@ -68,15 +68,13 @@ def test_sub_behavior(dsd_tree: Tree): def test_sequence_element(dsd_tree: Tree): - sequence_element = dsd_tree.root_element.get_child("SEQUENCE") - assert isinstance(sequence_element, SequenceTreeElement) - assert len(sequence_element.action_elements) == 2 - first_action = sequence_element.action_elements[0] - assert first_action.name == "FirstAction" + first_action = dsd_tree.root_element.get_child("SEQUENCE") assert isinstance(first_action, ActionTreeElement) - second_action = sequence_element.action_elements[1] - assert second_action.name == "SecondAction" + assert first_action.name == "FirstAction" + second_action = first_action.next_element assert isinstance(second_action, ActionTreeElement) + assert second_action.name == "SecondAction" + assert second_action.next_element is None def test_action_parameters(dsd_tree: Tree): @@ -132,8 +130,8 @@ def test_parameter_subbehavior(dsd_tree: Tree): first_action = parameter_subbehavior_decision.get_child("FIRST") assert first_action.parameters == {"key": "value1"} action_sequence = parameter_subbehavior_decision.get_child("SECOND") - assert action_sequence.action_elements[0].parameters == {"key": "value2"} - assert action_sequence.action_elements[1].parameters == {"key": "value2"} + assert action_sequence.parameters == {"key": "value2"} + assert action_sequence.next_element.parameters == {"key": "value2"} def test_nested_parameter_subbehavior(dsd_tree: Tree): @@ -142,12 +140,12 @@ def test_nested_parameter_subbehavior(dsd_tree: Tree): first_action = inner_subbehavior_decision.get_child("FIRST") assert first_action.parameters == {"key": "nested1"} action_sequence = inner_subbehavior_decision.get_child("SECOND") - assert action_sequence.action_elements[0].parameters == {"key": "nested2"} - assert action_sequence.action_elements[1].parameters == {"key": "nested2"} + assert action_sequence.parameters == {"key": "nested2"} + assert action_sequence.next_element.parameters == {"key": "nested2"} def test_sequence_tree(dsd_tree: Tree): sequence_tree = dsd_tree.root_element.get_child("SEQUENCE_TREE") - assert isinstance(sequence_tree, SequenceTreeElement) - assert sequence_tree.action_elements[0].name == "FirstAction" - assert sequence_tree.action_elements[1].name == "SecondAction" + assert isinstance(sequence_tree, ActionTreeElement) + assert sequence_tree.name == "FirstAction" + assert sequence_tree.next_element.name == "SecondAction" From 640befd753e84f476713bd1fc2ddb907c474d354 Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 13:21:32 +0900 Subject: [PATCH 2/6] Update dsd_follower to deal with the changes to sequences. --- .../dsd_follower.py | 88 +++++++++---------- 1 file changed, 40 insertions(+), 48 deletions(-) diff --git a/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py b/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py index 34623da..fc4ea5c 100644 --- a/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py +++ b/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py @@ -162,10 +162,9 @@ def param_string(params: dict) -> str: assert ( stack_element["type"] == tree_element["type"] ), f"The stack and the tree do not match (element type mismatch: {stack_element['type']} vs. {tree_element['type']})" - if stack_element["type"] != "sequence": - assert ( - stack_element["name"] == tree_element["name"] - ), f"The stack and the tree do not match (element name mismatch: {stack_element['name']} vs. {tree_element['name']})" + assert ( + stack_element["name"] == tree_element["name"] + ), f"The stack and the tree do not match (element name mismatch: {stack_element['name']} vs. {tree_element['name']})" # Initialize parameters of the dot node we are going to create dot_node_params = { @@ -173,21 +172,7 @@ def param_string(params: dict) -> str: } # Go through all possible element types and create the corresponding label and shape - if tree_element["type"] == "sequence": - dot_node_params["shape"] = "box" - - # If we have a sequence we have to create a label for each action and mark the current one (if one is on the stack) - label = ["Sequence:"] - for i, action in enumerate(tree_element["action_elements"]): - # Spaces for indentation - action_label = " " - # Mark current element (if this sequence is on the stack) - if stack_element is not None and i == stack_element["current_action_index"]: - action_label += "--> " - action_label += action["name"] + param_string(action["parameters"]) - label.append(action_label) - dot_node_params["label"] = "\n".join(label) - elif tree_element["type"] == "decision": + if tree_element["type"] == "decision": dot_node_params["shape"] = "ellipse" dot_node_params["label"] = tree_element["name"] + param_string(tree_element["parameters"]) elif tree_element["type"] == "action": @@ -202,6 +187,17 @@ def param_string(params: dict) -> str: # Create node in graph return pydot.Node(**dot_node_params) + @staticmethod + def _stack_matches_tree(stack_element: Optional[dict], tree_element: dict) -> bool: + """ + Check whether a stack element corresponds to a given tree element (same type and name) + """ + return ( + stack_element is not None + and stack_element["type"] == tree_element["type"] + and stack_element["name"] == tree_element["name"] + ) + def _stack_to_dotgraph( self, dot: pydot.Dot, subtree_root: dict, stack_root: Optional[dict] = None, full_tree: bool = False ) -> (pydot.Dot, str): @@ -210,43 +206,51 @@ def _stack_to_dotgraph( :param dot: The dotgraph to modify :param subtree_root: The root node of the subtree which should be added to the graph - :param stack_root: The root node / bottom of the stack which should be added to the graph (this corresponds to subtree_root as they are different representations of the same component). The stack is optional as certain subtrees might not be in contact with the stack. + :param stack_root: The stack element which is currently active in this position of the tree, or None if + this branch of the tree is not on the stack. For a chain of actions (i.e. a sequence) linked via + ``next`` this is the currently active element of the chain, which is not necessarily ``subtree_root``. :param full_tree: If true the whole tree is rendered, otherwise only the current stack and its direct neighbors are rendered :return: The modified dotgraph and the uid of the root node """ - # Sanity check - if stack_root is not None: - assert ( - stack_root["type"] == subtree_root["type"] - ), f"The stack element type and the tree element type do not match ({stack_root['type']} vs. {subtree_root['type']})" - if stack_root["type"] != "sequence": - assert ( - stack_root["name"] == subtree_root["name"] - ), f"The stack element name and the tree element name do not match ({stack_root['name']} vs. {subtree_root['name']})" + # The passed stack element only corresponds to this tree node if their type and name match. For a chain of + # actions (a sequence) the active element might be a later element of the chain, so it does not match here. + node_stack = stack_root if self._stack_matches_tree(stack_root, subtree_root) else None # Generate dot node for the root and mark it as active if it is on the stack - dot_node = DsdFollower._dot_node_from_tree_element(subtree_root, stack_root) + dot_node = DsdFollower._dot_node_from_tree_element(subtree_root, node_stack) # Append this element to graph dot.add_node(dot_node) # Append children to graph if this element is on the stack or if we want to render the whole tree - if "children" in subtree_root and (stack_root is not None or full_tree): + if "children" in subtree_root and (node_stack is not None or full_tree): # Go through all children for activating_result, child in subtree_root["children"].items(): # Get the root of the sub stack if this branch of the tree is the one which is currently on the stack sub_stack_root = None if ( - stack_root is not None - and stack_root["next"] is not None - and stack_root["next"]["activation_reason"] == activating_result + node_stack is not None + and node_stack["next"] is not None + and node_stack["next"]["activation_reason"] == activating_result ): - sub_stack_root = stack_root["next"] + sub_stack_root = node_stack["next"] # Recursively generate dot nodes for the children dot, child_uid = self._stack_to_dotgraph(dot, child, sub_stack_root, full_tree) # Connect the child to the parent element edge = pydot.Edge(src=dot_node.get_name(), dst=child_uid, label=activating_result) dot.add_edge(edge) + + # Append the successor of a chain (e.g. the next action of a sequence) if it exists. + # We render the whole chain if any element of it is on the stack or if the whole tree is requested. + if subtree_root.get("next") is not None and (stack_root is not None or full_tree): + # The active stack element is only passed further down the chain if it does not correspond to this + # node. Otherwise the successor is not (yet) on the stack and is therefore rendered as inactive. + successor_stack_root = None if node_stack is not None else stack_root + + # Recursively generate dot nodes for the successor + dot, next_uid = self._stack_to_dotgraph(dot, subtree_root["next"], successor_stack_root, full_tree) + # Connect the successor to this element + dot.add_edge(pydot.Edge(src=dot_node.get_name(), dst=next_uid)) return dot, dot_node.get_name() def _append_debug_data_to_item( @@ -312,9 +316,6 @@ def to_q_item_model(self): # Start with the root/bottom of the stack stack_element = self.stack - # Store the corresponding tree element - tree_element = self.tree - # Go through all stack elements while stack_element is not None: # Sanity check @@ -328,13 +329,7 @@ def to_q_item_model(self): elem_item.setSelectable(False) # Set the text of the item - if stack_element["type"] == "sequence": - # Get the names of all actions - action_names = [action["name"] for action in tree_element["action_elements"]] - # Join them together and set the text - elem_item.setText("Sequence: " + ", ".join(action_names)) - else: - elem_item.setText(stack_element["name"]) + elem_item.setText(stack_element["name"]) # Add debug data to the item self._append_debug_data_to_item(elem_item, stack_element["debug_data"]) @@ -354,9 +349,6 @@ def to_q_item_model(self): # Go to next element stack_element = stack_element["next"] - # Also select the corresponding tree element if there are any - if "children" in tree_element and stack_element is not None: - tree_element = tree_element["children"][stack_element["activation_reason"]] return model From 78e1ccda0b34c0ad384936122515b91069190798 Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 13:27:07 +0900 Subject: [PATCH 3/6] Fix the incorrect element being highlighted when duplicates names are present --- .../dynamic_stack_decider/dsd.py | 3 +++ .../dynamic_stack_decider/tree.py | 4 ++++ .../dsd_follower.py | 21 ++++++++----------- 3 files changed, 16 insertions(+), 12 deletions(-) diff --git a/dynamic_stack_decider/dynamic_stack_decider/dsd.py b/dynamic_stack_decider/dynamic_stack_decider/dsd.py index 0f9acc2..5e82d89 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/dsd.py +++ b/dynamic_stack_decider/dynamic_stack_decider/dsd.py @@ -343,6 +343,9 @@ def debug_publish_stack(self): for tree_elem, elem_instance in reversed(self.stack): elem_data = elem_instance.repr_dict() elem_data["activation_reason"] = tree_elem.activation_reason + # Attach the id of the corresponding tree element so the visualization can match this + # stack element to the exact tree element (names alone are ambiguous within a chain). + elem_data["id"] = id(tree_elem) elem_data["next"] = data data = elem_data diff --git a/dynamic_stack_decider/dynamic_stack_decider/tree.py b/dynamic_stack_decider/dynamic_stack_decider/tree.py index 3d7db9d..77481ce 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/tree.py +++ b/dynamic_stack_decider/dynamic_stack_decider/tree.py @@ -67,6 +67,10 @@ def repr_dict(self) -> dict: Represent this tree element as dictionary which is JSON encodable """ return { + # A stable identifier for this tree element. It is used to match a stack element to the + # exact tree element it belongs to, which is required because a chain (i.e. a sequence) + # may contain multiple elements with the same name (e.g. @Wait, @SomeAction, @Wait). + "id": id(self), "name": self.name, "parameters": self.parameters, } diff --git a/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py b/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py index fc4ea5c..3a6647f 100644 --- a/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py +++ b/dynamic_stack_decider_visualization/dynamic_stack_decider_visualization/dsd_follower.py @@ -190,13 +190,13 @@ def param_string(params: dict) -> str: @staticmethod def _stack_matches_tree(stack_element: Optional[dict], tree_element: dict) -> bool: """ - Check whether a stack element corresponds to a given tree element (same type and name) + Check whether a stack element corresponds to a given tree element. + + Matching is done via the stable ``id`` of the tree element rather than the name, because a + chain (i.e. a sequence) may contain multiple elements with the same name (e.g. + ``@Wait, @SomeAction, @Wait``) and we need to know exactly which one is active. """ - return ( - stack_element is not None - and stack_element["type"] == tree_element["type"] - and stack_element["name"] == tree_element["name"] - ) + return stack_element is not None and stack_element["id"] == tree_element["id"] def _stack_to_dotgraph( self, dot: pydot.Dot, subtree_root: dict, stack_root: Optional[dict] = None, full_tree: bool = False @@ -243,12 +243,9 @@ def _stack_to_dotgraph( # Append the successor of a chain (e.g. the next action of a sequence) if it exists. # We render the whole chain if any element of it is on the stack or if the whole tree is requested. if subtree_root.get("next") is not None and (stack_root is not None or full_tree): - # The active stack element is only passed further down the chain if it does not correspond to this - # node. Otherwise the successor is not (yet) on the stack and is therefore rendered as inactive. - successor_stack_root = None if node_stack is not None else stack_root - - # Recursively generate dot nodes for the successor - dot, next_uid = self._stack_to_dotgraph(dot, subtree_root["next"], successor_stack_root, full_tree) + # We pass the same active stack element down the whole chain. Only the element it actually + # corresponds to (matched via its id) is rendered as active; all others are rendered inactive. + dot, next_uid = self._stack_to_dotgraph(dot, subtree_root["next"], stack_root, full_tree) # Connect the successor to this element dot.add_edge(pydot.Edge(src=dot_node.get_name(), dst=next_uid)) return dot, dot_node.get_name() From eb3d7804b4f83159e0ddc9d10b8a4bc8e007d24a Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 13:39:36 +0900 Subject: [PATCH 4/6] Allow subtrees and decisions after actions. --- .../dynamic_stack_decider/parser.py | 178 ++++++++++++------ dynamic_stack_decider/test/test.dsd | 4 + dynamic_stack_decider/test/test_parser.py | 32 ++++ 3 files changed, 152 insertions(+), 62 deletions(-) diff --git a/dynamic_stack_decider/dynamic_stack_decider/parser.py b/dynamic_stack_decider/dynamic_stack_decider/parser.py index 36d5152..17fdd42 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/parser.py +++ b/dynamic_stack_decider/dynamic_stack_decider/parser.py @@ -101,56 +101,24 @@ def parse(self, file_path: str) -> Tree: if call.startswith("#"): # A subtree is called here. - subtree_name = call.strip("#") - name, parameters, unset_parameters = self._extract_parameters(subtree_name, lnr) - if name not in subtrees: - raise ParseError(f"Error parsing line {lnr}: {name} not defined") - subtree = subtrees[name] - if (set(parameters.keys()) | set(unset_parameters.keys())) != set(subtree.parameter_list): - raise ParseError( - "Error parsing line {}: Invalid parameters specified.\n" - "Available parameters are {}, specified parameters are {}.".format( - lnr, - set(parameters.keys()) | set(unset_parameters.keys()), - set(subtree.parameter_list), - ) - ) - # The root element of the subtree should be placed in this tree position if current_tree_element is None: # The current subtree is empty, set the subtree as its root element + subtree, _, _ = self._lookup_subtree(call, lnr, subtrees) current_subtree.set_root_element(subtree.root_element) else: - # Copy this subtree - subtree_copy = copy.deepcopy(subtree.root_element) - # Evaluate all unset parameters - to_evaluate = [subtree_copy] - while len(to_evaluate) > 0: - current = to_evaluate.pop(0) - if isinstance(current, DecisionTreeElement): - to_evaluate.extend(current.children.values()) - elif isinstance(current, ActionTreeElement) and current.next_element is not None: - # Follow the chain of actions (i.e. a sequence) - to_evaluate.append(current.next_element) - - for name, reference in current.unset_parameters.items(): - if reference in parameters: - current.parameters[name] = parameters[reference] - elif reference in unset_parameters: - current.unset_parameters[name] = unset_parameters[reference] - else: - raise ParseError( - f"Error evaluating subtree call in line {lnr}: " - f"Unknown reference to {reference}." - ) - - # Append this subtree in the current position + # Copy this subtree and place it in the current tree position + subtree_copy = self._instantiate_subtree(call, lnr, subtrees) current_tree_element.add_child_element(subtree_copy, result) elif re.search(r"\s*,\s*", call): - # A sequence element - actions = re.split(r"\s*,\s*", call) - element = self._create_sequence_element(actions, current_tree_element, lnr) - current_tree_element.add_child_element(element, result) + # A sequence: a chain of actions, optionally ending in a decision or a subtree + tokens = re.split(r"\s*,\s*", call) + head, tail = self._create_chain(tokens, current_tree_element, lnr, subtrees) + current_tree_element.add_child_element(head, result) + if tokens[-1].startswith("$"): + # A decision at the end of a sequence receives its children from the + # following indented lines + current_tree_element = tail elif call.startswith("@"): # An action is called @@ -170,13 +138,17 @@ def parse(self, file_path: str) -> Tree: else: # No arrow, must be the beginning of a new subtree - if re.search(r"\s*,\s*", line): - actions = re.split(r"\s*,\s*", line) - element = self._create_sequence_element(actions, current_tree_element, lnr) + if re.search(r"\s*,\s*", line_content): + tokens = re.split(r"\s*,\s*", line_content) + head, tail = self._create_chain(tokens, current_tree_element, lnr, subtrees) + current_subtree.set_root_element(head) + # A decision at the end of a sequence receives its children from the + # following indented lines + current_tree_element = tail if tokens[-1].startswith("$") else head else: element = self._create_tree_element(line_content, current_tree_element, lnr) - current_subtree.set_root_element(element) - current_tree_element = element + current_subtree.set_root_element(element) + current_tree_element = element last_indent = indent return tree @@ -236,32 +208,114 @@ def _create_tree_element(self, token, parent, lnr): raise ParseError("An element has to start with either $ or @") return element - def _create_sequence_element(self, actions, parent, lnr): + def _create_chain(self, tokens, parent, lnr, subtrees): """ - Create a sequence as a chain of action elements linked via ``next_element`` + Create a sequence as a chain of tree elements linked via ``next_element``. - :param actions: The names of actions in the sequence - :type actions: list[str] - :param parent: The parent element of the sequence - :type parent: AbstractDecisionElement + Every element except the last must be an action, as only actions advance to their successor + by popping themselves. The last element may be an action, a decision, or a subtree call, which + allows a decision or a subtree to follow a sequence of actions. + + :param tokens: The call tokens of the sequence (e.g. ``["@A", "@B", "$D"]``) + :type tokens: list[str] + :param parent: The element the chain is attached to :param lnr: Line number of the current line (used for error messages) :type lnr: int - :return: The head action element of the chain + :param subtrees: The already parsed subtrees, needed to resolve a trailing subtree call + :type subtrees: dict + :return: A tuple of the head element (the child of ``parent``) and the tail element of the chain """ head = None previous = None - for action in actions: - # The parent of the head is the given parent, every following action is - # parented to its predecessor in the chain - element = self._create_tree_element(action, parent if previous is None else previous, lnr) - if not isinstance(element, ActionTreeElement): - raise ParseError("In a sequence, only actions are allowed!") + for i, token in enumerate(tokens): + is_last = i == len(tokens) - 1 + if is_last and token.startswith("#"): + # A subtree may terminate a sequence + element = self._instantiate_subtree(token, lnr, subtrees) + elif is_last: + # The last element may be an action or a decision + element = self._create_tree_element(token, parent, lnr) + else: + # Only actions may be followed by another element in a sequence + if not token.startswith("@"): + raise ParseError( + f"Error parsing line {lnr}: Only actions may be followed by another element " + f"in a sequence, but got '{token}'" + ) + element = self._create_tree_element(token, parent, lnr) + if previous is None: head = element else: previous.next_element = element previous = element - return head + return head, previous + + def _lookup_subtree(self, call, lnr, subtrees): + """ + Look up the subtree referenced by a ``#``-call and validate that its parameters match. + + :param call: The call token, e.g. ``#SubBehavior + key:value`` + :param lnr: Line number of the current line (used for error messages) + :param subtrees: The already parsed subtrees + :return: A tuple of the referenced subtree and the set and unset parameters of the call + """ + name, parameters, unset_parameters = self._extract_parameters(call.strip("#"), lnr) + if name not in subtrees: + raise ParseError(f"Error parsing line {lnr}: {name} not defined") + subtree = subtrees[name] + if (set(parameters.keys()) | set(unset_parameters.keys())) != set(subtree.parameter_list): + raise ParseError( + "Error parsing line {}: Invalid parameters specified.\n" + "Available parameters are {}, specified parameters are {}.".format( + lnr, + set(parameters.keys()) | set(unset_parameters.keys()), + set(subtree.parameter_list), + ) + ) + return subtree, parameters, unset_parameters + + def _instantiate_subtree(self, call, lnr, subtrees): + """ + Create a deep copy of a referenced subtree with its parameters evaluated. + + :param call: The call token, e.g. ``#SubBehavior + key:value`` + :param lnr: Line number of the current line (used for error messages) + :param subtrees: The already parsed subtrees + :return: The root element of the instantiated subtree + """ + subtree, parameters, unset_parameters = self._lookup_subtree(call, lnr, subtrees) + subtree_copy = copy.deepcopy(subtree.root_element) + self._evaluate_subtree_parameters(subtree_copy, parameters, unset_parameters, lnr) + return subtree_copy + + def _evaluate_subtree_parameters(self, root, parameters, unset_parameters, lnr): + """ + Resolve the unset parameters of a copied subtree against the parameters given in the call. + + :param root: The root element of the copied subtree + :param parameters: The set parameters of the subtree call + :param unset_parameters: The unset (referenced) parameters of the subtree call + :param lnr: Line number of the current line (used for error messages) + """ + to_evaluate = [root] + while len(to_evaluate) > 0: + current = to_evaluate.pop(0) + if isinstance(current, DecisionTreeElement): + to_evaluate.extend(current.children.values()) + elif isinstance(current, ActionTreeElement) and current.next_element is not None: + # Follow the chain of actions (i.e. a sequence) + to_evaluate.append(current.next_element) + + for name, reference in current.unset_parameters.items(): + if reference in parameters: + current.parameters[name] = parameters[reference] + elif reference in unset_parameters: + current.unset_parameters[name] = unset_parameters[reference] + else: + raise ParseError( + f"Error evaluating subtree call in line {lnr}: Unknown reference to {reference}." + ) class ParseError(AssertionError): diff --git a/dynamic_stack_decider/test/test.dsd b/dynamic_stack_decider/test/test.dsd index dc80172..ed66943 100644 --- a/dynamic_stack_decider/test/test.dsd +++ b/dynamic_stack_decider/test/test.dsd @@ -46,3 +46,7 @@ $FirstDecision PARAMETER_SUBBEHAVIOR --> #ParameterSubBehavior + key2:value2 + key1:value1 NESTED_PARAMETER_SUBBEHAVIOR --> #NestedParameterSubBehavior + key_nested2:nested2 + key_nested1:nested1 SEQUENCE_TREE --> #SequenceTree + DECISION_AFTER_ACTION --> @FirstAction, $ThirdDecision + FIRST --> @FirstAction + SECOND --> @SecondAction + SUBTREE_AFTER_ACTION --> @FirstAction, #SubBehavior diff --git a/dynamic_stack_decider/test/test_parser.py b/dynamic_stack_decider/test/test_parser.py index c477914..b8a5bc3 100644 --- a/dynamic_stack_decider/test/test_parser.py +++ b/dynamic_stack_decider/test/test_parser.py @@ -35,6 +35,8 @@ def test_possible_results(dsd_tree: Tree): "PARAMETER_SUBBEHAVIOR", "NESTED_PARAMETER_SUBBEHAVIOR", "SEQUENCE_TREE", + "DECISION_AFTER_ACTION", + "SUBTREE_AFTER_ACTION", } @@ -149,3 +151,33 @@ def test_sequence_tree(dsd_tree: Tree): assert isinstance(sequence_tree, ActionTreeElement) assert sequence_tree.name == "FirstAction" assert sequence_tree.next_element.name == "SecondAction" + + +def test_decision_after_action(dsd_tree: Tree): + # An action may be followed by a decision (@FirstAction, $ThirdDecision) + head = dsd_tree.root_element.get_child("DECISION_AFTER_ACTION") + assert isinstance(head, ActionTreeElement) + assert head.name == "FirstAction" + + decision = head.next_element + assert isinstance(decision, DecisionTreeElement) + assert decision.name == "ThirdDecision" + # The decision keeps the activation reason of the chain it belongs to + assert decision.activation_reason == "DECISION_AFTER_ACTION" + # The decision gets its children from the following indented lines + assert decision.get_child("FIRST").name == "FirstAction" + assert decision.get_child("SECOND").name == "SecondAction" + + +def test_subtree_after_action(dsd_tree: Tree): + # An action may be followed by a subtree (@FirstAction, #SubBehavior) + head = dsd_tree.root_element.get_child("SUBTREE_AFTER_ACTION") + assert isinstance(head, ActionTreeElement) + assert head.name == "FirstAction" + + # SubBehavior has ThirdDecision as its root + subtree_root = head.next_element + assert isinstance(subtree_root, DecisionTreeElement) + assert subtree_root.name == "ThirdDecision" + assert subtree_root.get_child("FIRST").name == "FirstAction" + assert subtree_root.get_child("SECOND").name == "SecondAction" From e8afbd02f00b5b697f1ab78abba18cda13387683 Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 15:03:06 +0900 Subject: [PATCH 5/6] Add comprehensive test suite --- .../test/fixture_elements/__init__.py | 9 + .../test/fixture_elements/actions.py | 17 + .../test/fixture_elements/broken.py | 3 + .../test/fixture_elements/decisions.py | 8 + .../test/runtime_elements.py | 246 ++++++ dynamic_stack_decider/test/test_discovery.py | 72 ++ dynamic_stack_decider/test/test_elements.py | 130 +++ .../test/test_parser_errors.py | 118 +++ dynamic_stack_decider/test/test_runtime.py | 744 ++++++++++++++++++ dynamic_stack_decider/test/test_tree.py | 152 ++++ 10 files changed, 1499 insertions(+) create mode 100644 dynamic_stack_decider/test/fixture_elements/__init__.py create mode 100644 dynamic_stack_decider/test/fixture_elements/actions.py create mode 100644 dynamic_stack_decider/test/fixture_elements/broken.py create mode 100644 dynamic_stack_decider/test/fixture_elements/decisions.py create mode 100644 dynamic_stack_decider/test/runtime_elements.py create mode 100644 dynamic_stack_decider/test/test_discovery.py create mode 100644 dynamic_stack_decider/test/test_elements.py create mode 100644 dynamic_stack_decider/test/test_parser_errors.py create mode 100644 dynamic_stack_decider/test/test_runtime.py create mode 100644 dynamic_stack_decider/test/test_tree.py diff --git a/dynamic_stack_decider/test/fixture_elements/__init__.py b/dynamic_stack_decider/test/fixture_elements/__init__.py new file mode 100644 index 0000000..32e56d1 --- /dev/null +++ b/dynamic_stack_decider/test/fixture_elements/__init__.py @@ -0,0 +1,9 @@ +"""Fixture element package used by the discovery tests. Defines one action directly here so the +``__init__.py`` discovery path can be exercised.""" + +from dynamic_stack_decider.abstract_action_element import AbstractActionElement + + +class InitAction(AbstractActionElement): + def perform(self, reevaluate=False): + pass diff --git a/dynamic_stack_decider/test/fixture_elements/actions.py b/dynamic_stack_decider/test/fixture_elements/actions.py new file mode 100644 index 0000000..da50bd6 --- /dev/null +++ b/dynamic_stack_decider/test/fixture_elements/actions.py @@ -0,0 +1,17 @@ +"""Action classes for the discovery tests.""" + +from dynamic_stack_decider.abstract_action_element import AbstractActionElement + + +class FixtureActionA(AbstractActionElement): + def perform(self, reevaluate=False): + pass + + +class FixtureActionB(AbstractActionElement): + def perform(self, reevaluate=False): + pass + + +class NotAnElement: + """A plain class that must be ignored by discovery (it is not a stack element).""" diff --git a/dynamic_stack_decider/test/fixture_elements/broken.py b/dynamic_stack_decider/test/fixture_elements/broken.py new file mode 100644 index 0000000..16220d8 --- /dev/null +++ b/dynamic_stack_decider/test/fixture_elements/broken.py @@ -0,0 +1,3 @@ +"""A module that fails to import. Discovery must skip it and log the error, not crash.""" + +raise RuntimeError("this fixture module cannot be imported") diff --git a/dynamic_stack_decider/test/fixture_elements/decisions.py b/dynamic_stack_decider/test/fixture_elements/decisions.py new file mode 100644 index 0000000..3309a8f --- /dev/null +++ b/dynamic_stack_decider/test/fixture_elements/decisions.py @@ -0,0 +1,8 @@ +"""Decision classes for the discovery tests.""" + +from dynamic_stack_decider.abstract_decision_element import AbstractDecisionElement + + +class FixtureDecision(AbstractDecisionElement): + def perform(self, reevaluate=False): + return "X" diff --git a/dynamic_stack_decider/test/runtime_elements.py b/dynamic_stack_decider/test/runtime_elements.py new file mode 100644 index 0000000..6b4c615 --- /dev/null +++ b/dynamic_stack_decider/test/runtime_elements.py @@ -0,0 +1,246 @@ +""" +Reusable actions, decisions, a blackboard and a DSD builder for the runtime tests. + +The elements are intentionally generic and driven by the blackboard so that a wide range of +behaviors can be expressed purely through small ``.dsd`` snippets in the tests. Every element +records what it does in ``blackboard.events`` so the tests can assert on the exact execution order. +""" + +import os +import tempfile +import textwrap + +from dynamic_stack_decider.abstract_action_element import AbstractActionElement +from dynamic_stack_decider.abstract_decision_element import AbstractDecisionElement +from dynamic_stack_decider.dsd import DSD + + +class Blackboard: + """A minimal blackboard shared between all elements of a DSD.""" + + def __init__(self): + # Ordered log of everything that happened, e.g. ("perform", "Idle") or ("decide", "Guard", True) + self.events: list[tuple] = [] + # Maps a decision name to the result it should currently return + self.results: dict[str, str] = {} + + +# --------------------------------------------------------------------------- +# Actions +# --------------------------------------------------------------------------- +class _LoggingAction(AbstractActionElement): + """Base action that logs its perform and pop calls but otherwise stays on the stack.""" + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + + def on_pop(self): + self.blackboard.events.append(("pop", self.name)) + + +class Idle(_LoggingAction): + """A persistent action that stays on top of the stack.""" + + +class Run(_LoggingAction): + """A second persistent action (distinct name from Idle).""" + + +class Wait(_LoggingAction): + """A third persistent action (distinct name from Idle and Run).""" + + +class _PoppingAction(_LoggingAction): + """Base action that pops itself immediately after performing.""" + + def perform(self, reevaluate=False): + super().perform(reevaluate) + return self.pop() + + +class StepA(_PoppingAction): + """Pops itself immediately. Used to build sequences.""" + + +class StepB(_PoppingAction): + """Pops itself immediately. Used to build sequences.""" + + +class StepC(_PoppingAction): + """Pops itself immediately. Used to build sequences.""" + + +class CountingAction(_LoggingAction): + """Stays on the stack for ``count`` performs (default 1), then pops itself.""" + + def __init__(self, blackboard, dsd, parameters): + super().__init__(blackboard, dsd, parameters) + self._performed = 0 + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + self._performed += 1 + if self._performed >= self.parameters.get("count", 1): + return self.pop() + + +class Raiser(AbstractActionElement): + """Raises an exception in its perform method (to test the DSDs error handling).""" + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + raise ValueError("boom in perform") + + +class Interrupter(_LoggingAction): + """Interrupts the DSD on its first perform, then behaves like a persistent action.""" + + def __init__(self, blackboard, dsd, parameters): + super().__init__(blackboard, dsd, parameters) + self._interrupted = False + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + if not self._interrupted: + self._interrupted = True + self.interrupt() + + +class DoNotReevalOnce(_LoggingAction): + """Requests ``do_not_reevaluate`` on its first perform only, then stays on the stack.""" + + def __init__(self, blackboard, dsd, parameters): + super().__init__(blackboard, dsd, parameters) + self._requested = False + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + if not self._requested: + self._requested = True + self.do_not_reevaluate() + + +class DoNotReevalThenPop(_LoggingAction): + """Requests ``do_not_reevaluate`` and then pops itself (used to check the flag handling on pop).""" + + def perform(self, reevaluate=False): + self.blackboard.events.append(("perform", self.name)) + self.do_not_reevaluate() + return self.pop() + + +# --------------------------------------------------------------------------- +# Decisions +# --------------------------------------------------------------------------- +class _LoggingDecision(AbstractDecisionElement): + """Base decision that logs its perform calls and returns ``blackboard.results[self.name]``.""" + + def perform(self, reevaluate=False): + self.blackboard.events.append(("decide", self.name, reevaluate)) + return self.blackboard.results[self.name] + + +class Chooser(_LoggingDecision): + """A decision that is NOT reevaluated while it is in the middle of the stack.""" + + +class Guard(_LoggingDecision): + """A decision that IS reevaluated while it is in the middle of the stack.""" + + _reevaluate = True + + +class ElseGuard(_LoggingDecision): + """A reevaluating decision, typically used together with an ELSE branch.""" + + _reevaluate = True + + +class ReturnNone(_LoggingDecision): + """A decision whose perform returns None (to test the 'forgot to return' safeguard).""" + + def perform(self, reevaluate=False): + self.blackboard.events.append(("decide", self.name, reevaluate)) + return None + + +class PopOnReeval(_LoggingDecision): + """A reevaluating decision that pops itself while being reevaluated.""" + + _reevaluate = True + + def perform(self, reevaluate=False): + self.blackboard.events.append(("decide", self.name, reevaluate)) + if reevaluate: + self.pop() + return self.blackboard.results[self.name] + + +class InterruptOnReeval(_LoggingDecision): + """A reevaluating decision that interrupts the DSD while being reevaluated.""" + + _reevaluate = True + + def perform(self, reevaluate=False): + self.blackboard.events.append(("decide", self.name, reevaluate)) + if reevaluate: + self.interrupt() + return self.blackboard.results[self.name] + + +ACTIONS = { + cls.__name__: cls + for cls in [ + Idle, + Run, + Wait, + StepA, + StepB, + StepC, + CountingAction, + Raiser, + Interrupter, + DoNotReevalOnce, + DoNotReevalThenPop, + ] +} + +DECISIONS = { + cls.__name__: cls + for cls in [ + Chooser, + Guard, + ElseGuard, + ReturnNone, + PopOnReeval, + InterruptOnReeval, + ] +} + + +def make_dsd(dsd_text, blackboard=None, node=None, debug_topic=None): + """ + Build a DSD from an inline ``.dsd`` description, using the elements defined in this module. + + The description is dedented so that it can be written as a nicely indented multiline string in + the tests. It is then written to a temporary file, parsed and bound. The DSD is returned with the + start element on the stack but is not yet updated, mirroring real usage. + """ + blackboard = blackboard if blackboard is not None else Blackboard() + dsd = DSD(blackboard, debug_topic=debug_topic, node=node) + dsd.actions = dict(ACTIONS) + dsd.decisions = dict(DECISIONS) + + with tempfile.NamedTemporaryFile("w", suffix=".dsd", delete=False) as f: + f.write(textwrap.dedent(dsd_text)) + path = f.name + try: + dsd.load_behavior(path) + finally: + os.unlink(path) + return dsd + + +def stack_names(dsd): + """Return the class names of the instances currently on the stack (bottom to top).""" + return [instance.name for _, instance in dsd.stack] diff --git a/dynamic_stack_decider/test/test_discovery.py b/dynamic_stack_decider/test/test_discovery.py new file mode 100644 index 0000000..cc22dd4 --- /dev/null +++ b/dynamic_stack_decider/test/test_discovery.py @@ -0,0 +1,72 @@ +""" +Tests for element discovery and registration (``discover_elements``, ``register_actions``, +``register_decisions``). These exercise the directory, single-file and ``__init__.py`` code paths, +the type filtering and the resilience against modules that fail to import. +""" + +import os + +import pytest + +from dynamic_stack_decider.abstract_action_element import AbstractActionElement +from dynamic_stack_decider.abstract_decision_element import AbstractDecisionElement +from dynamic_stack_decider.dsd import DSD, discover_elements + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixture_elements") + + +def test_discover_from_directory(): + elements = discover_elements(FIXTURES) + # Classes from submodules and the package __init__ are found + assert "FixtureActionA" in elements + assert "FixtureActionB" in elements + assert "FixtureDecision" in elements + assert "InitAction" in elements + # Plain classes that are not stack elements are ignored + assert "NotAnElement" not in elements + + +def test_discover_from_single_file(): + elements = discover_elements(os.path.join(FIXTURES, "actions.py")) + assert {"FixtureActionA", "FixtureActionB"}.issubset(elements) + # Elements from other modules are not included + assert "FixtureDecision" not in elements + assert "NotAnElement" not in elements + + +def test_discover_from_init_file(monkeypatch): + # The __init__.py path resolves the module name from the parent directory, which requires the + # package's parent (the test directory) to be importable. + monkeypatch.syspath_prepend(os.path.dirname(FIXTURES)) + elements = discover_elements(os.path.join(FIXTURES, "__init__.py")) + assert "InitAction" in elements + + +def test_discover_invalid_path_raises(): + with pytest.raises(ValueError): + discover_elements(os.path.join(FIXTURES, "this_path_does_not_exist")) + + +def test_discover_skips_unimportable_module(): + # broken.py raises on import; discovery must log and continue, still returning the good classes + elements = discover_elements(FIXTURES) + assert "FixtureActionA" in elements + + +def test_register_actions_filters_to_actions(): + dsd = DSD(blackboard=object()) + dsd.register_actions(FIXTURES) + assert "FixtureActionA" in dsd.actions + assert "FixtureActionB" in dsd.actions + # Decisions are filtered out + assert "FixtureDecision" not in dsd.actions + assert all(issubclass(cls, AbstractActionElement) for cls in dsd.actions.values()) + + +def test_register_decisions_filters_to_decisions(): + dsd = DSD(blackboard=object()) + dsd.register_decisions(FIXTURES) + assert "FixtureDecision" in dsd.decisions + # Actions are filtered out + assert "FixtureActionA" not in dsd.decisions + assert all(issubclass(cls, AbstractDecisionElement) for cls in dsd.decisions.values()) diff --git a/dynamic_stack_decider/test/test_elements.py b/dynamic_stack_decider/test/test_elements.py new file mode 100644 index 0000000..6204f85 --- /dev/null +++ b/dynamic_stack_decider/test/test_elements.py @@ -0,0 +1,130 @@ +""" +Unit tests for the element base classes (``AbstractStackElement``, ``AbstractActionElement``, +``AbstractDecisionElement``): the helper methods, debug data handling, the reevaluation flags and +the delegation of ``pop`` / ``interrupt`` / ``do_not_reevaluate`` to the DSD. +""" + +import pytest + +from dynamic_stack_decider.abstract_stack_element import AbstractStackElement + +from .runtime_elements import Blackboard, Chooser, Guard, Idle + + +class FakeDSD: + """Records the calls the elements delegate to the DSD.""" + + def __init__(self): + self.pop_count = 0 + self.interrupt_count = 0 + self.do_not_reevaluate_count = 0 + + def pop(self): + self.pop_count += 1 + + def interrupt(self): + self.interrupt_count += 1 + + def set_do_not_reevaluate(self): + self.do_not_reevaluate_count += 1 + + +def make_action(cls=Idle, parameters=None): + return cls(Blackboard(), FakeDSD(), parameters or {}) + + +def test_name_is_class_name(): + assert make_action().name == "Idle" + + +def test_pop_delegates_to_dsd(): + action = make_action() + action.pop() + assert action._dsd.pop_count == 1 + + +def test_interrupt_delegates_to_dsd(): + action = make_action() + action.interrupt() + assert action._dsd.interrupt_count == 1 + + +def test_do_not_reevaluate_delegates_to_dsd(): + action = make_action() + action.do_not_reevaluate() + assert action._dsd.do_not_reevaluate_count == 1 + + +def test_on_pop_default_is_noop(): + # The default on_pop should simply do nothing and not raise + assert make_action().on_pop() is None + + +def test_publish_debug_data_stores_serializable_data(): + action = make_action() + action.publish_debug_data("label", 42) + assert action._debug_data == {"label": 42} + + +def test_publish_debug_data_wraps_non_serializable_data(): + action = make_action() + + class NotSerializable: + def __str__(self): + return "wrapped" + + action.publish_debug_data("label", NotSerializable()) + assert action._debug_data == {"label": "wrapped"} + + +def test_clear_debug_data(): + action = make_action() + action.publish_debug_data("label", 1) + action.clear_debug_data() + assert action._debug_data == {} + + +def test_action_repr_dict(): + action = make_action() + action.publish_debug_data("k", "v") + assert action.repr_dict() == {"type": "action", "name": "Idle", "debug_data": {"k": "v"}} + + +def test_decision_repr_dict(): + decision = Chooser(Blackboard(), FakeDSD(), {}) + assert decision.repr_dict() == {"type": "decision", "name": "Chooser", "debug_data": {}} + + +def test_never_reevaluate_flag_defaults_false(): + assert make_action().never_reevaluate is False + + +def test_never_reevaluate_flag_from_short_parameter(): + assert make_action(parameters={"r": False}).never_reevaluate is True + + +def test_never_reevaluate_flag_from_long_parameter(): + assert make_action(parameters={"reevaluate": False}).never_reevaluate is True + + +def test_get_reevaluate_default_false(): + assert Chooser(Blackboard(), FakeDSD(), {}).get_reevaluate() is False + + +def test_get_reevaluate_true_when_flag_set(): + assert Guard(Blackboard(), FakeDSD(), {}).get_reevaluate() is True + + +def test_base_perform_raises_not_implemented(): + # The base perform is a safeguard for subclasses that forget to override it + with pytest.raises(NotImplementedError): + AbstractStackElement.perform(make_action()) + + +def test_base_repr_dict(): + # The base implementation is normally overridden; call it directly to cover it + assert AbstractStackElement.repr_dict(make_action()) == { + "type": "abstract", + "name": "Idle", + "debug_data": {}, + } diff --git a/dynamic_stack_decider/test/test_parser_errors.py b/dynamic_stack_decider/test/test_parser_errors.py new file mode 100644 index 0000000..33c66fc --- /dev/null +++ b/dynamic_stack_decider/test/test_parser_errors.py @@ -0,0 +1,118 @@ +""" +Tests for the parser's error handling and less common code paths: malformed indentation, invalid +element/parameter syntax, invalid sequences and subtree references, and node-provided (``%``) +parameters. +""" + +import os +import tempfile + +import pytest + +from dynamic_stack_decider.parser import DsdParser, ParseError + + +def parse(text, node=None): + with tempfile.NamedTemporaryFile("w", suffix=".dsd", delete=False) as f: + f.write(text) + path = f.name + try: + return DsdParser(node).parse(path) + finally: + os.unlink(path) + + +def test_indent_not_multiple_of_four(): + with pytest.raises(ParseError, match="Indent is not a multiple of 4"): + parse("-->T\n$Chooser\n GO --> @FirstAction\n") + + +def test_element_neither_action_nor_decision(): + with pytest.raises(ParseError, match="neither an action nor a decision"): + parse("-->T\n$Chooser\n GO --> FirstAction\n") + + +def test_root_element_must_start_with_marker(): + with pytest.raises(ParseError, match="start with either"): + parse("-->T\nFirstDecision\n") + + +def test_invalid_parameter_list(): + with pytest.raises(ParseError, match="Invalid parameter list"): + parse("-->T\n$Chooser\n GO --> @FirstAction + keywithoutvalue\n") + + +def test_parameter_value_with_space(): + with pytest.raises(ParseError, match="should not contain spaces"): + parse("-->T\n$Chooser\n GO --> @FirstAction + key:val ue\n") + + +def test_decision_not_last_in_sequence(): + with pytest.raises(ParseError, match="Only actions may be followed"): + parse("-->T\n$Chooser\n GO --> @FirstAction, $Guard, @SecondAction\n") + + +def test_subtree_not_defined(): + with pytest.raises(ParseError, match="not defined"): + parse("-->T\n$Chooser\n GO --> #DoesNotExist\n") + + +def test_subtree_invalid_parameters(): + with pytest.raises(ParseError, match="Invalid parameters specified"): + parse( + "#Sub + key1\n" + "@FirstAction + p:*key1\n" + "-->T\n" + "$Chooser\n" + " GO --> #Sub + wrongkey:value\n" + ) + + +def test_subtree_unknown_reference(): + with pytest.raises(ParseError, match="Unknown reference"): + parse( + # The subtree declares parameter 'key1' but the action references the unknown '*missing' + "#Sub + key1\n" + "@FirstAction + p:*missing\n" + "-->T\n" + "$Chooser\n" + " GO --> #Sub + key1:value\n" + ) + + +def test_subtree_used_as_root_of_another_subtree(): + # When a subtree body starts by referencing another subtree, that subtree becomes its root + tree = parse( + "#Inner\n" + "@FirstAction\n" + "#Outer\n" + "FOO --> #Inner\n" + "-->T\n" + "$Chooser\n" + " GO --> #Outer\n" + ) + assert tree.root_element.get_child("GO").name == "FirstAction" + + +class _FakeParameter: + def __init__(self, value): + self.value = value + + +class _FakeNode: + def __init__(self, values): + self._values = values + + def get_parameter(self, name): + return _FakeParameter(self._values[name]) + + +def test_node_provided_parameter(): + node = _FakeNode({"speed": 5}) + tree = parse("-->T\n$Chooser\n GO --> @FirstAction + key:%speed\n", node=node) + assert tree.root_element.get_child("GO").parameters == {"key": 5} + + +def test_node_provided_parameter_without_node_is_none(): + tree = parse("-->T\n$Chooser\n GO --> @FirstAction + key:%speed\n", node=None) + assert tree.root_element.get_child("GO").parameters == {"key": None} diff --git a/dynamic_stack_decider/test/test_runtime.py b/dynamic_stack_decider/test/test_runtime.py new file mode 100644 index 0000000..189b37d --- /dev/null +++ b/dynamic_stack_decider/test/test_runtime.py @@ -0,0 +1,744 @@ +""" +Runtime behavior tests for the DSD execution engine (``dsd.py``). + +These tests drive a DSD through ``update()`` and assert on the resulting stack and execution order. +They aim to cover every runtime code path: pushing/popping, sequences (chains), reevaluation +(including ELSE handling), the ``never_reevaluate`` / ``do_not_reevaluate`` flags, interrupts, +decisions and subtrees following actions, error handling and the debug publishing. +""" + +import json + +import pytest + +from dynamic_stack_decider.abstract_action_element import AbstractActionElement +from dynamic_stack_decider.abstract_decision_element import AbstractDecisionElement + +from .runtime_elements import Blackboard, make_dsd, stack_names + + +class FakePublisher: + """Records published messages and reports a configurable subscription count.""" + + def __init__(self): + self.messages = [] + self.subscription_count = 1 + + def publish(self, msg): + self.messages.append(msg) + + def get_subscription_count(self): + return self.subscription_count + + +class FakeNode: + """A stand-in for an rclpy node that just hands out FakePublishers.""" + + def __init__(self): + self.publishers = {} + + def create_publisher(self, msg_type, topic, qos): + publisher = FakePublisher() + self.publishers[topic] = publisher + return publisher + + +# --------------------------------------------------------------------------- +# Basic decision / action flow +# --------------------------------------------------------------------------- +def test_start_element_on_stack_after_load(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + # After loading, only the start element is on the stack and nothing has been performed yet + assert stack_names(dsd) == ["Chooser"] + assert dsd.blackboard.events == [] + + +def test_decision_pushes_action(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + + assert stack_names(dsd) == ["Chooser", "Idle"] + assert dsd.blackboard.events == [("decide", "Chooser", False), ("perform", "Idle")] + + +def test_persistent_action_stays_on_stack(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + for _ in range(3): + dsd.update() + + assert stack_names(dsd) == ["Chooser", "Idle"] + # The action was performed on every update + assert dsd.blackboard.events.count(("perform", "Idle")) == 3 + + +def test_action_pops_and_decision_redecides(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @StepA + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # StepA popped itself again, so only the decision remains + assert stack_names(dsd) == ["Chooser"] + assert ("pop", "StepA") in dsd.blackboard.events + + dsd.update() + # The decision pushes and runs StepA again + assert dsd.blackboard.events.count(("perform", "StepA")) == 2 + + +def test_nested_decisions(): + dsd = make_dsd( + """ + -->T + $Chooser + DOWN --> $Guard + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "DOWN" + dsd.blackboard.results["Guard"] = "GO" + + dsd.update() + + assert stack_names(dsd) == ["Chooser", "Guard", "Idle"] + + +# --------------------------------------------------------------------------- +# Sequences (chains via next_element) +# --------------------------------------------------------------------------- +def test_sequence_advances_one_action_per_tick(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @StepA, @StepB, @StepC + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # StepA performed and popped, advancing to StepB (not yet performed) + assert stack_names(dsd) == ["Chooser", "StepB"] + + dsd.update() + assert stack_names(dsd) == ["Chooser", "StepC"] + + dsd.update() + # The last action has no successor, so it pops normally and the decision remains + assert stack_names(dsd) == ["Chooser"] + + # Every action was performed and popped exactly once, in order + assert [e for e in dsd.blackboard.events if e[0] == "pop"] == [ + ("pop", "StepA"), + ("pop", "StepB"), + ("pop", "StepC"), + ] + + +def test_sequence_with_multi_tick_action(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @CountingAction + count:2, @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # CountingAction needs two performs before it advances + assert stack_names(dsd) == ["Chooser", "CountingAction"] + + dsd.update() + # After the second perform it popped and advanced to Idle + assert stack_names(dsd) == ["Chooser", "Idle"] + assert dsd.blackboard.events.count(("perform", "CountingAction")) == 2 + + +def test_decision_after_action(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @StepA, $Guard + ON --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + dsd.blackboard.results["Guard"] = "ON" + + dsd.update() + # StepA ran and advanced to the decision Guard + assert stack_names(dsd) == ["Chooser", "Guard"] + + dsd.update() + # Guard now decides and pushes its child + assert stack_names(dsd) == ["Chooser", "Guard", "Idle"] + + +def test_subtree_after_action(): + dsd = make_dsd( + """ + #Sub + $Guard + ON --> @Idle + -->T + $Chooser + GO --> @StepA, #Sub + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + dsd.blackboard.results["Guard"] = "ON" + + dsd.update() + assert stack_names(dsd) == ["Chooser", "Guard"] + + dsd.update() + assert stack_names(dsd) == ["Chooser", "Guard", "Idle"] + + +# --------------------------------------------------------------------------- +# Reevaluation +# --------------------------------------------------------------------------- +def test_reevaluation_rebuilds_stack_on_change(): + dsd = make_dsd( + """ + -->T + $Guard + A --> @Idle + B --> @Run + """ + ) + dsd.blackboard.results["Guard"] = "A" + + dsd.update() + assert stack_names(dsd) == ["Guard", "Idle"] + + # Same result -> no rebuild + dsd.update() + assert stack_names(dsd) == ["Guard", "Idle"] + + # Changed result -> the stack above the decision is dropped and rebuilt + dsd.blackboard.results["Guard"] = "B" + dsd.update() + assert stack_names(dsd) == ["Guard", "Run"] + assert ("pop", "Idle") in dsd.blackboard.events + + +def test_reevaluation_reruns_only_reevaluating_decisions(): + # Guard reevaluates, the nested Chooser does not + dsd = make_dsd( + """ + -->T + $Guard + GO --> $Chooser + A --> @Idle + B --> @Run + """ + ) + dsd.blackboard.results["Guard"] = "GO" + dsd.blackboard.results["Chooser"] = "A" + + dsd.update() + assert stack_names(dsd) == ["Guard", "Chooser", "Idle"] + + # Changing the non-reevaluating Chooser has no effect while it is in the middle of the stack + dsd.blackboard.results["Chooser"] = "B" + dsd.update() + assert stack_names(dsd) == ["Guard", "Chooser", "Idle"] + + +def test_reevaluation_else_branch(): + dsd = make_dsd( + """ + -->T + $ElseGuard + A --> @Idle + ELSE --> @Run + """ + ) + dsd.blackboard.results["ElseGuard"] = "A" + + dsd.update() + assert stack_names(dsd) == ["ElseGuard", "Idle"] + + # An unknown result maps to the ELSE branch -> rebuild into the ELSE child + dsd.blackboard.results["ElseGuard"] = "SOMETHING_ELSE" + dsd.update() + assert stack_names(dsd) == ["ElseGuard", "Run"] + + # Still an ELSE result -> no rebuild (the activation reason stays "ELSE") + dsd.blackboard.results["ElseGuard"] = "YET_ANOTHER" + dsd.update() + assert stack_names(dsd) == ["ElseGuard", "Run"] + assert dsd.blackboard.events.count(("pop", "Run")) == 0 + + # Back to a concrete result -> rebuild into the concrete child + dsd.blackboard.results["ElseGuard"] = "A" + dsd.update() + assert stack_names(dsd) == ["ElseGuard", "Idle"] + + +# --------------------------------------------------------------------------- +# never_reevaluate / do_not_reevaluate +# --------------------------------------------------------------------------- +def test_never_reevaluate_parameter_suppresses_reevaluation(): + dsd = make_dsd( + """ + -->T + $Guard + A --> @Idle + r:false + B --> @Run + """ + ) + dsd.blackboard.results["Guard"] = "A" + + dsd.update() + assert stack_names(dsd) == ["Guard", "Idle"] + + # Because Idle has never_reevaluate set, the guard is not reevaluated and the change is ignored + dsd.blackboard.results["Guard"] = "B" + for _ in range(3): + dsd.update() + assert stack_names(dsd) == ["Guard", "Idle"] + + +def test_do_not_reevaluate_delays_reevaluation_by_one_tick(): + dsd = make_dsd( + """ + -->T + $Guard + A --> @DoNotReevalOnce + B --> @Run + """ + ) + dsd.blackboard.results["Guard"] = "A" + + dsd.update() + assert stack_names(dsd) == ["Guard", "DoNotReevalOnce"] + + # The action requested do_not_reevaluate once, so this update skips reevaluation + dsd.blackboard.results["Guard"] = "B" + dsd.update() + assert stack_names(dsd) == ["Guard", "DoNotReevalOnce"] + + # The flag was reset, so the next update reevaluates and rebuilds + dsd.update() + assert stack_names(dsd) == ["Guard", "Run"] + + +def test_do_not_reevaluate_reset_on_normal_pop(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @DoNotReevalThenPop + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # The action set do_not_reevaluate but then popped (no successor), so the flag is reset + assert dsd.do_not_reevaluate is False + assert stack_names(dsd) == ["Chooser"] + + +def test_do_not_reevaluate_preserved_across_sequence_advance(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @DoNotReevalThenPop, @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # The action advanced to the next chain element and kept do_not_reevaluate set + assert stack_names(dsd) == ["Chooser", "Idle"] + assert dsd.do_not_reevaluate is True + + +# --------------------------------------------------------------------------- +# Interrupts +# --------------------------------------------------------------------------- +def test_external_interrupt_resets_stack(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + assert stack_names(dsd) == ["Chooser", "Idle"] + top_before = dsd.stack[0][1] + + dsd.interrupt() + assert stack_names(dsd) == ["Chooser"] + # A fresh start element instance is created + assert dsd.stack[0][1] is not top_before + + +def test_interrupt_from_within_an_action(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Interrupter + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + # The action interrupted the DSD, clearing the stack back to the start element + assert stack_names(dsd) == ["Chooser"] + + +def test_interrupt_during_reevaluation(): + dsd = make_dsd( + """ + -->T + $InterruptOnReeval + GO --> @Idle + """ + ) + dsd.blackboard.results["InterruptOnReeval"] = "GO" + + dsd.update() + assert stack_names(dsd) == ["InterruptOnReeval", "Idle"] + + # On the next update the root decision is reevaluated and interrupts, resetting the stack + dsd.update() + assert stack_names(dsd) == ["InterruptOnReeval"] + + +# --------------------------------------------------------------------------- +# pop edge cases +# --------------------------------------------------------------------------- +def test_pop_of_only_element_is_noop(): + # A behavior whose root is a self-popping action + dsd = make_dsd( + """ + -->T + @StepA + """ + ) + + dsd.update() + # The action tried to pop itself, but it is the only element, so the stack is unchanged + assert stack_names(dsd) == ["StepA"] + assert ("pop", "StepA") not in dsd.blackboard.events + + +def test_reevaluation_pop_at_root_does_not_shorten_stack(): + # A reevaluating root decision that pops itself during reevaluation. Because it is at the bottom + # of the stack, no element above the reevaluation index is dropped and the update returns early. + dsd = make_dsd( + """ + -->T + $PopOnReeval + GO --> @Idle + """ + ) + dsd.blackboard.results["PopOnReeval"] = "GO" + + dsd.update() + assert stack_names(dsd) == ["PopOnReeval", "Idle"] + + performed_before = dsd.blackboard.events.count(("perform", "Idle")) + dsd.update() + # The stack is untouched and, because reevaluation stopped, the top action was not performed again + assert stack_names(dsd) == ["PopOnReeval", "Idle"] + assert dsd.blackboard.events.count(("perform", "Idle")) == performed_before + assert dsd.stack_reevaluate is False + + +def test_reevaluation_pop_in_the_middle_shortens_stack(): + # A reevaluating decision that is not at the bottom pops itself during reevaluation. This shortens + # the stack down to the reevaluation index. The engine then swallows the resulting internal error + # (this is an unusual edge case); the important contract is that update() does not raise and the + # stack ends up shortened. + dsd = make_dsd( + """ + -->T + $Guard + GO --> $PopOnReeval + X --> @Idle + """ + ) + dsd.blackboard.results["Guard"] = "GO" + dsd.blackboard.results["PopOnReeval"] = "X" + + dsd.update() + assert stack_names(dsd) == ["Guard", "PopOnReeval", "Idle"] + + dsd.update() + # PopOnReeval and its child were dropped during reevaluation + assert stack_names(dsd) == ["Guard"] + assert ("pop", "Idle") in dsd.blackboard.events + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- +def test_exception_in_perform_is_swallowed(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Raiser + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + + # The exception raised inside the action must not propagate out of update() + dsd.update() + assert stack_names(dsd) == ["Chooser", "Raiser"] + assert ("perform", "Raiser") in dsd.blackboard.events + + +def test_unknown_decision_result_is_swallowed(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "UNKNOWN" + + # get_child raises a KeyError which is caught inside update() + dsd.update() + assert stack_names(dsd) == ["Chooser"] + + +def test_decision_returning_none_exits(): + dsd = make_dsd( + """ + -->T + $ReturnNone + GO --> @Idle + """ + ) + + # Returning None from a decision is a programming error and triggers sys.exit + with pytest.raises(SystemExit): + dsd.update() + + +# --------------------------------------------------------------------------- +# _bind_modules error paths +# --------------------------------------------------------------------------- +def test_unknown_action_name_raises_on_load(): + with pytest.raises(AssertionError): + make_dsd( + """ + -->T + $Chooser + GO --> @DoesNotExist + """ + ) + + +def test_unknown_decision_name_raises_on_load(): + with pytest.raises(AssertionError): + make_dsd( + """ + -->T + $DoesNotExist + GO --> @Idle + """ + ) + + +def test_bind_modules_rejects_unknown_element_type(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + + class Bogus: + pass + + with pytest.raises(ValueError): + dsd._bind_modules(Bogus()) + + +# --------------------------------------------------------------------------- +# Misc DSD API +# --------------------------------------------------------------------------- +def test_get_stack_returns_the_stack(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + assert dsd.get_stack() is dsd.stack + + +def test_set_do_not_reevaluate(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + assert dsd.do_not_reevaluate is False + dsd.set_do_not_reevaluate() + assert dsd.do_not_reevaluate is True + + +def test_stack_holds_bound_instances(): + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """ + ) + dsd.blackboard.results["Chooser"] = "GO" + dsd.update() + + root_tree, root_instance = dsd.stack[0] + assert isinstance(root_instance, AbstractDecisionElement) + top_tree, top_instance = dsd.stack[-1] + assert isinstance(top_instance, AbstractActionElement) + # The instances share the blackboard + assert root_instance.blackboard is dsd.blackboard is top_instance.blackboard + + +# --------------------------------------------------------------------------- +# Debug publishing +# --------------------------------------------------------------------------- +def test_debug_tree_published_on_load(): + node = FakeNode() + make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """, + node=node, + debug_topic="dbg", + ) + + tree_messages = node.publishers["dbg/dsd_tree"].messages + assert len(tree_messages) == 1 + tree = json.loads(tree_messages[0].data) + assert tree["type"] == "decision" + assert tree["name"] == "Chooser" + assert "id" in tree + assert tree["children"]["GO"]["type"] == "action" + + +def test_debug_stack_and_current_action_published(): + node = FakeNode() + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """, + node=node, + debug_topic="dbg", + ) + dsd.blackboard.results["Chooser"] = "GO" + + dsd.update() + + stack_messages = node.publishers["dbg/dsd_stack"].messages + assert stack_messages + stack = json.loads(stack_messages[-1].data) + # Bottom of the stack is the decision, above it the action, matched via the stable id + assert stack["type"] == "decision" and "id" in stack + assert stack["next"]["type"] == "action" + assert stack["next"]["name"] == "Idle" + assert stack["next"]["next"] is None + + # The current action is published exactly once (it does not change while Idle stays on top) + action_messages = node.publishers["dbg/dsd_current_action"].messages + dsd.update() + dsd.update() + assert [m.data for m in action_messages] == ["Idle"] + + +def test_debug_stack_not_published_without_subscribers(): + node = FakeNode() + dsd = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """, + node=node, + debug_topic="dbg", + ) + dsd.blackboard.results["Chooser"] = "GO" + node.publishers["dbg/dsd_stack"].subscription_count = 0 + + dsd.update() + + assert node.publishers["dbg/dsd_stack"].messages == [] + + +def test_fresh_blackboard_between_dsds(): + bb1 = Blackboard() + bb2 = Blackboard() + dsd1 = make_dsd( + """ + -->T + $Chooser + GO --> @Idle + """, + blackboard=bb1, + ) + dsd2 = make_dsd( + """ + -->T + $Chooser + GO --> @Run + """, + blackboard=bb2, + ) + dsd1.blackboard.results["Chooser"] = "GO" + dsd2.blackboard.results["Chooser"] = "GO" + + dsd1.update() + assert ("perform", "Idle") in bb1.events + assert bb2.events == [] diff --git a/dynamic_stack_decider/test/test_tree.py b/dynamic_stack_decider/test/test_tree.py new file mode 100644 index 0000000..f129f27 --- /dev/null +++ b/dynamic_stack_decider/test/test_tree.py @@ -0,0 +1,152 @@ +""" +Unit tests for the tree element classes (``tree.py``): child lookup (including ELSE and error cases), +activation-reason propagation along a chain, and the JSON representation (``repr_dict``) including the +stable ``id`` and the ``next`` chain link. +""" + +import pytest + +from dynamic_stack_decider.tree import ActionTreeElement, DecisionTreeElement, Tree + + +# --------------------------------------------------------------------------- +# get_child +# --------------------------------------------------------------------------- +def test_get_child_returns_matching_child(): + d = DecisionTreeElement("D", None) + a = ActionTreeElement("A", d) + d.add_child_element(a, "RESULT") + assert d.get_child("RESULT") is a + assert a.activation_reason == "RESULT" + + +def test_get_child_falls_back_to_else(): + d = DecisionTreeElement("D", None) + a = ActionTreeElement("A", d) + d.add_child_element(a, "ELSE") + assert d.get_child("ANY_UNMAPPED_RESULT") is a + + +def test_get_child_unknown_result_raises_keyerror(): + d = DecisionTreeElement("D", None) + d.add_child_element(ActionTreeElement("A", d), "X") + with pytest.raises(KeyError): + d.get_child("Y") + + +def test_get_child_none_result_exits(): + d = DecisionTreeElement("D", None) + with pytest.raises(SystemExit): + d.get_child(None) + + +def test_action_get_child_is_always_none(): + a = ActionTreeElement("A", None) + assert a.get_child("anything") is None + + +# --------------------------------------------------------------------------- +# Activation reason propagation +# --------------------------------------------------------------------------- +def test_set_activation_reason_propagates_along_chain(): + a = ActionTreeElement("A", None) + b = ActionTreeElement("B", a) + c = ActionTreeElement("C", b) + a.next_element = b + b.next_element = c + + a.set_activation_reason("REASON") + + assert a.activation_reason == "REASON" + assert b.activation_reason == "REASON" + assert c.activation_reason == "REASON" + + +def test_set_activation_reason_does_not_override_decision_children(): + # A chain ending in a decision: the decision gets the chain's reason, but its children keep theirs + a = ActionTreeElement("A", None) + d = DecisionTreeElement("D", a) + child = ActionTreeElement("child", d) + d.add_child_element(child, "C") + a.next_element = d + + a.set_activation_reason("REASON") + + assert d.activation_reason == "REASON" + assert child.activation_reason == "C" + + +def test_add_child_element_propagates_reason_to_chain(): + d = DecisionTreeElement("D", None) + head = ActionTreeElement("head", d) + tail = ActionTreeElement("tail", head) + head.next_element = tail + + d.add_child_element(head, "R") + + assert head.activation_reason == "R" + assert tail.activation_reason == "R" + + +# --------------------------------------------------------------------------- +# repr_dict +# --------------------------------------------------------------------------- +def test_action_repr_dict_contains_id_type_and_next(): + a = ActionTreeElement("A", None, parameters={"k": "v"}) + b = ActionTreeElement("B", a) + a.next_element = b + + r = a.repr_dict() + + assert r["type"] == "action" + assert r["name"] == "A" + assert r["parameters"] == {"k": "v"} + assert "id" in r + assert r["next"]["name"] == "B" + assert r["next"]["next"] is None + + +def test_action_repr_dict_next_is_none_without_successor(): + assert ActionTreeElement("A", None).repr_dict()["next"] is None + + +def test_decision_repr_dict_contains_children_and_id(): + d = DecisionTreeElement("D", None) + d.add_child_element(ActionTreeElement("A", d), "R") + + r = d.repr_dict() + + assert r["type"] == "decision" + assert "id" in r + assert r["children"]["R"]["name"] == "A" + + +def test_repr_dict_id_is_stable_and_unique(): + a1 = ActionTreeElement("A", None) + a2 = ActionTreeElement("A", None) + assert a1.repr_dict()["id"] == a1.repr_dict()["id"] + assert a1.repr_dict()["id"] != a2.repr_dict()["id"] + + +# --------------------------------------------------------------------------- +# Tree and string representations +# --------------------------------------------------------------------------- +def test_tree_repr_dict_delegates_to_root(): + d = DecisionTreeElement("D", None) + d.add_child_element(ActionTreeElement("A", d), "R") + tree = Tree() + tree.set_root_element(d) + + assert tree.repr_dict()["name"] == "D" + assert "D" in repr(tree) + + +def test_str_returns_name(): + assert str(ActionTreeElement("A", None)) == "A" + assert str(DecisionTreeElement("D", None)) == "D" + + +def test_action_repr_includes_successor(): + a = ActionTreeElement("A", None) + a.next_element = ActionTreeElement("B", a) + assert "A" in repr(a) and "B" in repr(a) From 54fad4c305b1d2ff54398320d2a463c625e0f1df Mon Sep 17 00:00:00 2001 From: Cornelius Krupp Date: Thu, 9 Jul 2026 15:41:39 +0900 Subject: [PATCH 6/6] Ruff format --- .../dynamic_stack_decider/dsd.py | 5 +++- .../dynamic_stack_decider/parser.py | 8 +++---- .../test/test_parser_errors.py | 24 +++---------------- 3 files changed, 10 insertions(+), 27 deletions(-) diff --git a/dynamic_stack_decider/dynamic_stack_decider/dsd.py b/dynamic_stack_decider/dynamic_stack_decider/dsd.py index 5e82d89..68eeafa 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/dsd.py +++ b/dynamic_stack_decider/dynamic_stack_decider/dsd.py @@ -307,7 +307,10 @@ def pop(self): self.stack_reevaluate = False else: current_tree_element = self.stack[-1][0] - if isinstance(current_tree_element, ActionTreeElement) and current_tree_element.next_element is not None: + if ( + isinstance(current_tree_element, ActionTreeElement) + and current_tree_element.next_element is not None + ): # This action is part of a chain (e.g. a sequence). Instead of returning to the # element below, we advance to the next element in the chain. # We do not reset do_not_reevaluate because an action in the chain may control diff --git a/dynamic_stack_decider/dynamic_stack_decider/parser.py b/dynamic_stack_decider/dynamic_stack_decider/parser.py index 17fdd42..0b61e43 100644 --- a/dynamic_stack_decider/dynamic_stack_decider/parser.py +++ b/dynamic_stack_decider/dynamic_stack_decider/parser.py @@ -133,7 +133,7 @@ def parse(self, file_path: str) -> Tree: else: raise ParseError( - f"Error parsing line {lnr}: " f"Element {call} is neither an action nor a decision" + f"Error parsing line {lnr}: Element {call} is neither an action nor a decision" ) else: @@ -174,7 +174,7 @@ def _extract_parameters(self, token, lnr): if " " in parameter_value: raise ParseError( - f"Error parsing line {lnr}: Parameter values should not contain spaces. " "Did you forget a comma?" + f"Error parsing line {lnr}: Parameter values should not contain spaces. Did you forget a comma?" ) if parameter_value.startswith("%"): @@ -313,9 +313,7 @@ def _evaluate_subtree_parameters(self, root, parameters, unset_parameters, lnr): elif reference in unset_parameters: current.unset_parameters[name] = unset_parameters[reference] else: - raise ParseError( - f"Error evaluating subtree call in line {lnr}: Unknown reference to {reference}." - ) + raise ParseError(f"Error evaluating subtree call in line {lnr}: Unknown reference to {reference}.") class ParseError(AssertionError): diff --git a/dynamic_stack_decider/test/test_parser_errors.py b/dynamic_stack_decider/test/test_parser_errors.py index 33c66fc..b55c144 100644 --- a/dynamic_stack_decider/test/test_parser_errors.py +++ b/dynamic_stack_decider/test/test_parser_errors.py @@ -59,38 +59,20 @@ def test_subtree_not_defined(): def test_subtree_invalid_parameters(): with pytest.raises(ParseError, match="Invalid parameters specified"): - parse( - "#Sub + key1\n" - "@FirstAction + p:*key1\n" - "-->T\n" - "$Chooser\n" - " GO --> #Sub + wrongkey:value\n" - ) + parse("#Sub + key1\n@FirstAction + p:*key1\n-->T\n$Chooser\n GO --> #Sub + wrongkey:value\n") def test_subtree_unknown_reference(): with pytest.raises(ParseError, match="Unknown reference"): parse( # The subtree declares parameter 'key1' but the action references the unknown '*missing' - "#Sub + key1\n" - "@FirstAction + p:*missing\n" - "-->T\n" - "$Chooser\n" - " GO --> #Sub + key1:value\n" + "#Sub + key1\n@FirstAction + p:*missing\n-->T\n$Chooser\n GO --> #Sub + key1:value\n" ) def test_subtree_used_as_root_of_another_subtree(): # When a subtree body starts by referencing another subtree, that subtree becomes its root - tree = parse( - "#Inner\n" - "@FirstAction\n" - "#Outer\n" - "FOO --> #Inner\n" - "-->T\n" - "$Chooser\n" - " GO --> #Outer\n" - ) + tree = parse("#Inner\n@FirstAction\n#Outer\nFOO --> #Inner\n-->T\n$Chooser\n GO --> #Outer\n") assert tree.root_element.get_child("GO").name == "FirstAction"