Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .github/workflows/pre-commit.yml
Original file line number Diff line number Diff line change
Expand Up @@ -35,3 +35,15 @@ jobs:
environments: default
- run: pixi run --locked test
- run: pixi run --locked package
- run: pixi run --locked typecheck

visualization:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7.0.1
- uses: prefix-dev/setup-pixi@v0.10.2
with:
pixi-version: v0.79.0
cache: true
environments: visualization
- run: pixi run --locked -e visualization typecheck-gui
8 changes: 8 additions & 0 deletions CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ contains only code quality tools.
```sh
pixi run --locked test
pixi run --locked package
pixi run --locked typecheck
pixi run --locked -e visualization typecheck-gui
pixi run --locked -e format format
pixi run --locked -e format lint
```
Expand Down Expand Up @@ -35,3 +37,9 @@ checks.
Changes to scheduling, action outcomes, error recovery, parameter syntax, or
registration should describe compatibility and migration. Use issues to compare
options before introducing a new public contract.

The core, visualization, and packaging checks use strict mypy. Public packages
ship `py.typed`; blackboards are generic and parameter values use `object` to
reflect the existing YAML and ROS inputs. Narrow a parameter with `isinstance`
before using it as a particular type. Untyped ROS integrations are confined to
third-party imports, and the small RQT stubs document the interfaces used here.
2 changes: 1 addition & 1 deletion dynamic_stack_decider/dynamic_stack_decider/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
from .abstract_decision_element import AbstractDecisionElement
from .dsd import DSD

__all__ = [DSD, AbstractDecisionElement, AbstractActionElement]
__all__ = ["DSD", "AbstractDecisionElement", "AbstractActionElement"]
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
from abc import ABCMeta
from __future__ import annotations

from abc import ABCMeta, abstractmethod
from typing import TYPE_CHECKING

from dynamic_stack_decider.abstract_stack_element import AbstractStackElement
from dynamic_stack_decider.types import BlackboardT, ElementData, Parameters

if TYPE_CHECKING:
from dynamic_stack_decider.dsd import DSD


class AbstractActionElement(AbstractStackElement, metaclass=ABCMeta):
class AbstractActionElement(AbstractStackElement[BlackboardT], metaclass=ABCMeta):
"""
One action is similar to a state of an FSM.
As in this case, the system stays in this state in contrast to the decision elements which are called only for determining the active action.
Expand All @@ -18,7 +21,7 @@ class AbstractActionElement(AbstractStackElement, metaclass=ABCMeta):
If the action is complete, it can remove itself from the stack by performing a pop command.
"""

def __init__(self, blackboard, dsd: "DSD", parameters: dict[str, bool | int | float | str]):
def __init__(self, blackboard: BlackboardT, dsd: DSD[BlackboardT], parameters: Parameters) -> None:
"""
Constructor of the action element
:param blackboard: Shared blackboard for data exchange between elements
Expand All @@ -29,14 +32,14 @@ def __init__(self, blackboard, dsd: "DSD", parameters: dict[str, bool | int | fl
# Reevaluation can be disabled by setting 'r' or 'reevaluate' to False
self.never_reevaluate = not parameters.get("r", True) or not parameters.get("reevaluate", True)

def do_not_reevaluate(self):
def do_not_reevaluate(self) -> None:
"""
Prohibits the next reevaluate.
This is useful if you have an action which has to be completed without interruption, e.g. a kick animation.
"""
self._dsd.set_do_not_reevaluate()

def repr_dict(self) -> dict:
def repr_dict(self) -> ElementData:
"""
Represent this stack element as dictionary which is JSON encodable
"""
Expand All @@ -45,3 +48,8 @@ def repr_dict(self) -> dict:
"name": self.name,
"debug_data": self._debug_data,
}

@abstractmethod
def perform(self, reevaluate: bool = False) -> None:
"""Execute the element; decisions return an outcome and actions return nothing."""
raise NotImplementedError(f"Implement perform() in {self.name}")
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
from abc import ABCMeta
from __future__ import annotations

from abc import ABCMeta, abstractmethod

from dynamic_stack_decider.abstract_stack_element import AbstractStackElement
from dynamic_stack_decider.types import BlackboardT, ElementData


class AbstractDecisionElement(AbstractStackElement, metaclass=ABCMeta):
class AbstractDecisionElement(AbstractStackElement[BlackboardT], metaclass=ABCMeta):
"""
The logic is encapsulated in two types of elements.
The decision elements define the logical path similar to a behavior tree.
Expand All @@ -20,7 +23,7 @@ class AbstractDecisionElement(AbstractStackElement, metaclass=ABCMeta):

_reevaluate = False

def repr_dict(self) -> dict:
def repr_dict(self) -> ElementData:
"""
Represent this stack element as dictionary which is JSON encodable
"""
Expand All @@ -30,7 +33,7 @@ def repr_dict(self) -> dict:
"debug_data": self._debug_data,
}

def get_reevaluate(self):
def get_reevaluate(self) -> bool:
"""
Each decision element may define a \textit{reevaluate} criteria.
If the corresponding method returns true, the element will be executed even if it is in the middle of the stack.
Expand All @@ -45,3 +48,8 @@ def get_reevaluate(self):
This can be used to regularly check preconditions, e.g. if the robot knows where the ball is.
"""
return self._reevaluate

@abstractmethod
def perform(self, reevaluate: bool = False) -> str:
"""Execute the element; decisions return an outcome and actions return nothing."""
raise NotImplementedError(f"Implement perform() in {self.name}")
Original file line number Diff line number Diff line change
@@ -1,30 +1,33 @@
from __future__ import annotations

from abc import ABCMeta, abstractmethod
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Generic

from dynamic_stack_decider.logger import get_logger
from dynamic_stack_decider.types import BlackboardT, ElementData, Parameters

if TYPE_CHECKING:
from dynamic_stack_decider.dsd import DSD


class AbstractStackElement(metaclass=ABCMeta):
class AbstractStackElement(Generic[BlackboardT], metaclass=ABCMeta):
"""
The AbstractStackElement is the basis of all elements on the stack.
It provides some help functions which should not be overloaded.
The work of an element is done in the :func:`perform`.
Each element which inherits from the AbstractStackElement can be used as a root element on the stack.
"""

_dsd: "DSD"
parameters: dict[str, bool | int | float | str]
_dsd: DSD[BlackboardT]
parameters: Parameters

def __init__(self, blackboard, dsd: "DSD", parameters: dict[str, bool | int | float | str]):
def __init__(self, blackboard: BlackboardT, dsd: DSD[BlackboardT], parameters: Parameters) -> None:
"""
:param blackboard: Shared blackboard for data exchange between elements
:param dsd: The stack decider which has this element on its stack.
:param parameters: Parameters which serve as arguments to this element
"""
self._debug_data = {}
self._debug_data: dict[str, object] = {}
"""
This is a dict in which data can be saved that should get represented on a __repr__ call.
It can be set using publish_debug_data()
Expand All @@ -41,7 +44,7 @@ def name(self) -> str:
"""
return self.__class__.__name__

def pop(self):
def pop(self) -> None:
"""
Help method which pops the element of the stack.

Expand All @@ -52,7 +55,7 @@ def pop(self):
"""
self._dsd.pop()

def on_pop(self): # noqa
def on_pop(self) -> None:
"""
This method is called when the element is popped from the stack.
It can be used to clean up resources, cancel actions or similar tasks.
Expand All @@ -61,7 +64,7 @@ def on_pop(self): # noqa
pass

@abstractmethod
def perform(self, reevaluate=False):
def perform(self, reevaluate: bool = False) -> str | None:
"""
This method is called when the element is on top of the stack.
This method has to be overloaded by the implementation!
Expand All @@ -71,13 +74,13 @@ def perform(self, reevaluate=False):
msg = f"You should override perform() in {self.__class__.__name__}"
raise NotImplementedError(msg)

def interrupt(self):
def interrupt(self) -> None:
"""
An interrupt leads to a complete clearing of the stack.
"""
self._dsd.interrupt()

def publish_debug_data(self, label: str, data: dict | list | int | float | str | bool):
def publish_debug_data(self, label: str, data: object) -> None:
"""
Publish debug data. Can be viewed using the DSD visualization

Expand All @@ -97,15 +100,15 @@ def publish_debug_data(self, label: str, data: dict | list | int | float | str |
get_logger().debug(f"{label}: {data}")
self._debug_data[label] = data

def clear_debug_data(self):
def clear_debug_data(self) -> None:
"""
Clear existing debug data

This is needed when old values are no longer supposed to be visible
"""
self._debug_data = {}

def repr_dict(self) -> dict:
def repr_dict(self) -> ElementData:
"""Represent this stack element as dictionary which is JSON encodable"""
return {
"type": "abstract",
Expand Down
Loading
Loading