Skip to content

[ports] Change input/output port declarations to class attributes #511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sea-bass merged 1 commit into from
Aug 26, 2026
Merged

[ports] Change input/output port declarations to class attributes #511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
51 changes: 17 additions & 34 deletions docs/ports.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ Why use ports?
Using ports instead of ad-hoc blackboard reads and writes pays off in several concrete ways:

* **Explicit data contracts.**
A node's ``input_ports()`` and ``output_ports()`` declarations *are* its data-flow API.
A node's ``INPUT_PORTS`` and ``OUTPUT_PORTS`` declarations *are* its data-flow API.
A reader can see at a glance what a node consumes and produces without reading through its :meth:`~py_trees.behaviour.Behaviour.update` method.

* **Structured, early error detection.**
Expand Down Expand Up @@ -60,18 +60,13 @@ Concrete nodes typically inherit from the convenience base
from py_trees.ports import BehaviourWithPorts, PortInformation

class Multiply(BehaviourWithPorts):
@classmethod
def input_ports(cls):
return {
"a": PortInformation(data_type=float, required=True),
"b": PortInformation(data_type=float, required=True),
}

@classmethod
def output_ports(cls):
return {
"product": PortInformation(data_type=float, required=True),
}
INPUT_PORTS = {
"a": PortInformation(data_type=float, required=True),
"b": PortInformation(data_type=float, required=True),
}
OUTPUT_PORTS = {
"product": PortInformation(data_type=float, required=True),
}

def update(self):
self._set_output("product", self.get_input("a") * self.get_input("b"))
Expand Down Expand Up @@ -108,11 +103,9 @@ A port can declare a ``default_value`` next to its type, so the fallback lives w

.. code-block:: python

@classmethod
def input_ports(cls):
return {
"timeout": PortInformation(data_type=float, default_value=5.0),
}
INPUT_PORTS = {
"timeout": PortInformation(data_type=float, default_value=5.0),
}

def update(self):
timeout = self.get_input("timeout") # 5.0 unless something wrote to the port
Expand Down Expand Up @@ -232,7 +225,7 @@ then resolves under the alias in the ``"auto"`` path:

Pass ``register=False`` on a class definition to keep a particular subclass out of the
registry. Still-abstract classes (e.g. :class:`~py_trees.ports.BehaviourWithPorts` itself,
which does not implement ``input_ports`` / ``output_ports``) are never registered.
which does not declare ``INPUT_PORTS`` / ``OUTPUT_PORTS``) are never registered.

.. _ports-xml-attributes-label:

Expand All @@ -241,7 +234,7 @@ XML attributes: ports *and* constructor arguments

Attributes on a node's XML tag serve **two** distinct purposes:

1. Attribute names that match a declared port (``input_ports()`` or ``output_ports()``) are treated as **port remappings**.
1. Attribute names that match a declared port (``INPUT_PORTS`` or ``OUTPUT_PORTS``) are treated as **port remappings**.
Values may be ``{curly_key}`` references (wired to the remapping table) or literal constants (type-converted according to the port's declared type).

2. Attribute names that do **not** match any declared port are treated as **constructor keyword arguments** and forwarded to the class constructor.
Expand All @@ -252,13 +245,8 @@ Example::

class Greeting(BehaviourWithPorts):

@classmethod
def input_ports(cls):
return {"name_key": PortInformation(data_type=str, required=True)}

@classmethod
def output_ports(cls):
return {"greeting": PortInformation(data_type=str, required=True)}
INPUT_PORTS = {"name_key": PortInformation(data_type=str, required=True)}
OUTPUT_PORTS = {"greeting": PortInformation(data_type=str, required=True)}

def __init__(self, name: str, prefix: str = "Hello", **kwargs):
super().__init__(name=name, **kwargs)
Expand Down Expand Up @@ -456,13 +444,8 @@ A few things you should be aware of, and suggestions on how to fill the gaps you
class Retry(PortsMixin, py_trees.decorators.Retry):
"""Retry that reads its failure budget from an input port."""

@classmethod
def input_ports(cls):
return {"num_failures": PortInformation(data_type=int, required=True)}

@classmethod
def output_ports(cls):
return {}
INPUT_PORTS = {"num_failures": PortInformation(data_type=int, required=True)}
OUTPUT_PORTS = {}

def __init__(
self,
Expand Down
21 changes: 7 additions & 14 deletions py_trees/demos/ports/basic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,13 @@
class Multiply(BehaviourWithPorts):
"""Read two floats and write their product."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {
"a": PortInformation(data_type=float, required=True),
"b": PortInformation(data_type=float, required=True),
}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {
"product": PortInformation(data_type=float, required=True),
}
INPUT_PORTS = {
"a": PortInformation(data_type=float, required=True),
"b": PortInformation(data_type=float, required=True),
}
OUTPUT_PORTS = {
"product": PortInformation(data_type=float, required=True),
}

def update(self) -> py_trees.common.Status:
"""Multiply the two inputs and write the result to the output port."""
Expand Down
45 changes: 12 additions & 33 deletions py_trees/demos/ports/nested_subtrees.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,18 +28,11 @@
class StartMissionReport(BehaviourWithPorts):
"""Create the initial mission report text."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {
"robot_name": PortInformation(data_type=str, required=True),
"mission_name": PortInformation(data_type=str, required=True),
}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {"report": PortInformation(data_type=str, required=True)}
INPUT_PORTS = {
"robot_name": PortInformation(data_type=str, required=True),
"mission_name": PortInformation(data_type=str, required=True),
}
OUTPUT_PORTS = {"report": PortInformation(data_type=str, required=True)}

def update(self) -> py_trees.common.Status:
"""Combine ``robot_name`` and ``mission_name`` into the initial report."""
Expand All @@ -52,18 +45,11 @@ def update(self) -> py_trees.common.Status:
class AddMissionStep(BehaviourWithPorts):
"""Append one mission step to the report text."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {
"report_in": PortInformation(data_type=str, required=True),
"step": PortInformation(data_type=str, required=True),
}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {"report_out": PortInformation(data_type=str, required=True)}
INPUT_PORTS = {
"report_in": PortInformation(data_type=str, required=True),
"step": PortInformation(data_type=str, required=True),
}
OUTPUT_PORTS = {"report_out": PortInformation(data_type=str, required=True)}

def update(self) -> py_trees.common.Status:
"""Append ``step`` to ``report_in`` and write to ``report_out``."""
Expand All @@ -76,15 +62,8 @@ def update(self) -> py_trees.common.Status:
class ReadMissionReport(BehaviourWithPorts):
"""Final node that exposes the finished report for printing."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {"report": PortInformation(data_type=str, required=True)}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {}
INPUT_PORTS = {"report": PortInformation(data_type=str, required=True)}
OUTPUT_PORTS = {}

def update(self) -> py_trees.common.Status:
"""Return SUCCESS; the report is read via :attr:`final_report`."""
Expand Down
43 changes: 11 additions & 32 deletions py_trees/demos/ports/remapping.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,21 +28,14 @@
class GenerateValue(BehaviourWithPorts):
"""Step 1: generate a random value and write it to an output port."""

INPUT_PORTS = {}
OUTPUT_PORTS = {"value": PortInformation(data_type=str, required=True)}

def __init__(self, name: str, prefix: str = "value", **kwargs: Any) -> None:
"""Initialise the behaviour with *prefix* used in the generated value."""
super().__init__(name=name, **kwargs)
self._prefix = prefix

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {"value": PortInformation(data_type=str, required=True)}

def update(self) -> py_trees.common.Status:
"""Generate a random value and write it to the output port."""
token = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
Expand All @@ -53,24 +46,17 @@ def update(self) -> py_trees.common.Status:
class AppendSuffix(BehaviourWithPorts):
"""Step 2: read one string, append a suffix, and write the updated string."""

INPUT_PORTS = {"text_in": PortInformation(data_type=str, required=True)}
OUTPUT_PORTS = {
"text_out": PortInformation(data_type=str, required=True),
"value": PortInformation(data_type=str, required=True),
}

def __init__(self, name: str, suffix: str, **kwargs: Any) -> None:
"""Initialise the behaviour with a *suffix* to append."""
super().__init__(name=name, **kwargs)
self._suffix = suffix

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {"text_in": PortInformation(data_type=str, required=True)}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {
"text_out": PortInformation(data_type=str, required=True),
"value": PortInformation(data_type=str, required=True),
}

def update(self) -> py_trees.common.Status:
"""Append the suffix and write both the processed text and a status string."""
processed_text = f"{self.get_input('text_in')}{self._suffix}"
Expand All @@ -82,15 +68,8 @@ def update(self) -> py_trees.common.Status:
class ReadResult(BehaviourWithPorts):
"""Step 3: read the final pipeline value so it can be inspected."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {"value": PortInformation(data_type=str, required=True)}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {}
INPUT_PORTS = {"value": PortInformation(data_type=str, required=True)}
OUTPUT_PORTS = {}

def update(self) -> py_trees.common.Status:
"""Return SUCCESS; the value is read via :attr:`result`."""
Expand Down
39 changes: 9 additions & 30 deletions py_trees/demos/ports/xml_tree.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,8 @@
class GreetingProducer(BehaviourWithPorts):
"""Produce a greeting string."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {"output": PortInformation(data_type=str, required=True)}
INPUT_PORTS = {}
OUTPUT_PORTS = {"output": PortInformation(data_type=str, required=True)}

def update(self) -> py_trees.common.Status:
"""Write a fixed greeting to the output port."""
Expand All @@ -47,18 +40,11 @@ def update(self) -> py_trees.common.Status:
class AddSuffix(BehaviourWithPorts):
"""Append a configurable suffix to an input string."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {
"input": PortInformation(data_type=str, required=True),
"suffix": PortInformation(data_type=str, required=True),
}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {"output": PortInformation(data_type=str, required=True)}
INPUT_PORTS = {
"input": PortInformation(data_type=str, required=True),
"suffix": PortInformation(data_type=str, required=True),
}
OUTPUT_PORTS = {"output": PortInformation(data_type=str, required=True)}

def update(self) -> py_trees.common.Status:
"""Append ``suffix`` to ``input`` and write the result to ``output``."""
Expand All @@ -69,15 +55,8 @@ def update(self) -> py_trees.common.Status:
class PrintConsumer(BehaviourWithPorts):
"""Capture the final value for display."""

@classmethod
def input_ports(cls) -> dict:
"""Return the input port declarations."""
return {"input": PortInformation(data_type=str, required=True)}

@classmethod
def output_ports(cls) -> dict:
"""Return the output port declarations."""
return {}
INPUT_PORTS = {"input": PortInformation(data_type=str, required=True)}
OUTPUT_PORTS = {}

def update(self) -> py_trees.common.Status:
"""Return SUCCESS; the value is read via :attr:`consumed_value`."""
Expand Down
Loading
Loading