From 53873cfdcf7fcf0699c78fc177f439ffaa8dd159 Mon Sep 17 00:00:00 2001 From: Sebastian Castro Date: Mon, 24 Aug 2026 17:36:46 -0400 Subject: [PATCH] [ports] Change input/output port declarations to class attributes --- docs/ports.rst | 51 +++------ py_trees/demos/ports/basic.py | 21 ++-- py_trees/demos/ports/nested_subtrees.py | 45 +++----- py_trees/demos/ports/remapping.py | 43 ++------ py_trees/demos/ports/xml_tree.py | 39 ++----- py_trees/ports.py | 138 ++++++++++++------------ tests/test_ports.py | 9 +- tests/test_ports_helpers.py | 93 +++++----------- tests/test_ports_xml_parser.py | 78 ++++---------- 9 files changed, 169 insertions(+), 348 deletions(-) diff --git a/docs/ports.rst b/docs/ports.rst index 73fb22cd..b5b6c7b0 100644 --- a/docs/ports.rst +++ b/docs/ports.rst @@ -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.** @@ -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")) @@ -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 @@ -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: @@ -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. @@ -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) @@ -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, diff --git a/py_trees/demos/ports/basic.py b/py_trees/demos/ports/basic.py index 9a1cfbb2..9ab08b15 100644 --- a/py_trees/demos/ports/basic.py +++ b/py_trees/demos/ports/basic.py @@ -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.""" diff --git a/py_trees/demos/ports/nested_subtrees.py b/py_trees/demos/ports/nested_subtrees.py index ce484975..9fa87b31 100644 --- a/py_trees/demos/ports/nested_subtrees.py +++ b/py_trees/demos/ports/nested_subtrees.py @@ -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.""" @@ -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``.""" @@ -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`.""" diff --git a/py_trees/demos/ports/remapping.py b/py_trees/demos/ports/remapping.py index e1cc623a..ddd8a238 100644 --- a/py_trees/demos/ports/remapping.py +++ b/py_trees/demos/ports/remapping.py @@ -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)) @@ -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}" @@ -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`.""" diff --git a/py_trees/demos/ports/xml_tree.py b/py_trees/demos/ports/xml_tree.py index 9bfaf835..d91d3ba4 100644 --- a/py_trees/demos/ports/xml_tree.py +++ b/py_trees/demos/ports/xml_tree.py @@ -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.""" @@ -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``.""" @@ -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`.""" diff --git a/py_trees/ports.py b/py_trees/ports.py index 13f7d673..816c644f 100644 --- a/py_trees/ports.py +++ b/py_trees/ports.py @@ -15,7 +15,7 @@ import copy import warnings -from abc import ABC, abstractmethod +from abc import ABC from dataclasses import dataclass from typing import TYPE_CHECKING, Any @@ -140,13 +140,16 @@ def get_ports_registry() -> dict[str, type["PortsMixin"]]: def _ports_class_is_abstract(cls: type) -> bool: """ - Return whether *cls* still has unimplemented abstract methods. + Return whether *cls* is still abstract. - This intentionally scans the resolved attributes rather than reading - ``cls.__abstractmethods__``: ``__init_subclass__`` runs *before* ``ABCMeta`` - populates ``__abstractmethods__`` on the new class, so that attribute is not - yet reliable at registration time. + A class is abstract if has not declared its ports (``INPUT_PORTS`` and ``OUTPUT_PORTS``), + or while it still has unimplemented abstract methods of its own. + The latter check scans the resolved attributes rather than reading ``cls.__abstractmethods__``: + ``__init_subclass__`` runs *before* ``ABCMeta`` populates ``__abstractmethods__`` + on the new class, so that attribute is not yet reliable at registration time. """ + if not (hasattr(cls, "INPUT_PORTS") and hasattr(cls, "OUTPUT_PORTS")): + return True return any(getattr(getattr(cls, name, None), "__isabstractmethod__", False) for name in dir(cls)) @@ -162,21 +165,22 @@ class PortsMixin(_MixinBase): 1. Inherit from ``PortsMixin`` first, followed by a concrete py_trees class (e.g. ``py_trees.behaviour.Behaviour``). - 2. Define its input and output ports as class-level information by implementing the - ``@classmethod`` s ``input_ports`` and ``output_ports``. + 2. Declare its input and output ports as class-level information by assigning the + class attributes ``INPUT_PORTS`` and ``OUTPUT_PORTS``. A ``PortsMixin`` represents a modular unit that interacts with input and output data through well-defined ports. These ports are typed and validated at runtime to ensure consistency and facilitate composability between different nodes. - Subclasses must define their input and output ports as class-level information by implementing - the ``@classmethod`` s ``input_ports`` and ``output_ports``. + Subclasses must declare their input and output ports as class-level information by assigning + the class attributes ``INPUT_PORTS`` and ``OUTPUT_PORTS``. - * ``input_ports(cls)``: returns a dictionary mapping input port names to port information. - * ``output_ports(cls)``: returns a dictionary mapping output port names to port information. + * ``INPUT_PORTS``: a mapping of input port names to port information. + * ``OUTPUT_PORTS``: a mapping of output port names to port information. - These methods return the expected port definitions for the class and do not change at runtime. - These port definitions are used to: + The declarations are evaluated once at class definition and read through the + ``input_ports()`` / ``output_ports()`` classmethods. They are shared class-level state + and must not be mutated at runtime. These port definitions are used to: 1. Register blackboard keys for communication. 2. Enforce type, default values, and presence validation at runtime. @@ -185,13 +189,8 @@ class PortsMixin(_MixinBase): Example usage:: class MyBehaviour(PortsMixin, py_trees.behaviour.Behaviour): - @classmethod - def input_ports(cls): - return {"input": PortInformation(data_type=str, default_value="foo", required=True)} - - @classmethod - def output_ports(cls): - return {"output": PortInformation(data_type=str, required=True)} + INPUT_PORTS = {"input": PortInformation(data_type=str, default_value="foo", required=True)} + OUTPUT_PORTS = {"output": PortInformation(data_type=str, required=True)} def __init__(self, name: str): super().__init__(name=name) @@ -201,14 +200,14 @@ def update(self): self._set_output("output", f"Processed({input_val})") return py_trees.common.Status.SUCCESS - Port specification format in ``input_ports()`` and ``output_ports()``:: + Port specification format in ``INPUT_PORTS`` and ``OUTPUT_PORTS``:: { "": PortInformation(data_type=, required=), } Input and output port names must be unique across both sets; overlapping names are not allowed and - will raise a ``ValueError`` at instantiation. + will raise a ``ValueError`` at class definition. **Subtrees** @@ -265,13 +264,8 @@ def update(self): **Example**:: class ConsumerProducer(PortsMixin, py_trees.behaviour.Behaviour): - @classmethod - def input_ports(cls): - return {"input": PortInformation(data_type=str, required=True)} - - @classmethod - def output_ports(cls): - return {"output": PortInformation(data_type=str, required=True)} + INPUT_PORTS = {"input": PortInformation(data_type=str, required=True)} + OUTPUT_PORTS = {"output": PortInformation(data_type=str, required=True)} def update(self): input_val = self.get_input("input") @@ -279,37 +273,48 @@ def update(self): return py_trees.common.Status.SUCCESS """ + # Port declarations, assigned by concrete subclasses. + # Deliberately annotation-only here, as their absence marks a class as still abstract + # and therefore prevents automatic registration. + INPUT_PORTS: dict[str, PortInformation] + OUTPUT_PORTS: dict[str, PortInformation] + def __init_subclass__(cls, *, tag: str | None = None, register: bool = True, **kwargs: Any) -> None: """ - Auto-register concrete subclasses so parsers can resolve them by tag. + Validate port declarations and auto-register concrete subclasses. - Defining a concrete ``PortsMixin`` subclass registers it under its class - name (or *tag*, if given) in the global ports registry, so a parser can - resolve it by tag. + Defining a concrete ``PortsMixin`` subclass (one that assigns the ``INPUT_PORTS`` + and ``OUTPUT_PORTS`` class attributes) registers it under its class name (or *tag*, + if given) in the global ports registry, so a parser can resolve it by tag. Args: tag: Optional explicit tag (lookup name). Defaults to ``cls.__name__``. register: Set to ``False`` to skip auto-registration for this class. + Raises: + ValueError: If any port name appears in both ``INPUT_PORTS`` and ``OUTPUT_PORTS``. + Still-abstract subclasses (e.g. :class:`BehaviourWithPorts`, which does - not implement ``input_ports`` / ``output_ports``) are never registered. + not declare ``INPUT_PORTS`` / ``OUTPUT_PORTS``) are never registered. """ super().__init_subclass__(**kwargs) - if not register or _ports_class_is_abstract(cls): + if _ports_class_is_abstract(cls): + return + for port in cls.INPUT_PORTS.keys() & cls.OUTPUT_PORTS.keys(): + raise ValueError(f"Port '{port}' appears in both input and output ports") + if not register: return register_ports_class(tag if tag is not None else cls.__name__, cls) @classmethod - @abstractmethod def input_ports(cls) -> dict[str, PortInformation]: - """Return a mapping of input port names to port information.""" - raise NotImplementedError("Subclasses must implement input_ports()") + """Return the mapping of input port names to port information (treat as read-only).""" + return cls.INPUT_PORTS @classmethod - @abstractmethod def output_ports(cls) -> dict[str, PortInformation]: - """Return a mapping of output port names to port information.""" - raise NotImplementedError("Subclasses must implement output_ports()") + """Return the mapping of output port names to port information (treat as read-only).""" + return cls.OUTPUT_PORTS @classmethod def get_port_type(cls, port_name: str) -> type: @@ -324,12 +329,10 @@ def get_port_type(cls, port_name: str) -> type: Raises: KeyError: If the port name is not defined in either input or output ports. """ - if port_name in cls.input_ports(): - return cls.input_ports()[port_name].data_type # type: ignore[no-any-return] - elif port_name in cls.output_ports(): - return cls.output_ports()[port_name].data_type # type: ignore[no-any-return] - else: - raise KeyError(f"Port '{port_name}' not defined.") + for ports in (cls.input_ports(), cls.output_ports()): + if port_name in ports: + return ports[port_name].data_type # type: ignore[no-any-return] + raise KeyError(f"Port '{port_name}' not defined.") @classmethod def is_port_required(cls, port_name: str) -> bool: @@ -347,12 +350,10 @@ def is_port_required(cls, port_name: str) -> bool: Raises: KeyError: If the port name is not defined in either input or output ports. """ - if port_name in cls.input_ports(): - return cls.input_ports()[port_name].required - elif port_name in cls.output_ports(): - return cls.output_ports()[port_name].required - else: - raise KeyError(f"Port '{port_name}' not defined.") + for ports in (cls.input_ports(), cls.output_ports()): + if port_name in ports: + return ports[port_name].required + raise KeyError(f"Port '{port_name}' not defined.") def __init__(self, *args: Any, behaviour_class_name: str | None = None, **kwargs: Any) -> None: """ @@ -367,8 +368,17 @@ def __init__(self, *args: Any, behaviour_class_name: str | None = None, **kwargs **kwargs: Additional keyword arguments passed to the parent class. Raises: - ValueError: If any port name appears in both input_ports() and output_ports(). + TypeError: If the class has not declared its ``INPUT_PORTS`` / ``OUTPUT_PORTS`` + class attributes (i.e. it is still abstract). """ + # ABC cannot guard instantiation here since the port declarations are attributes, + # so mirror the abstract-class TypeError explicitly. + if _ports_class_is_abstract(type(self)): + raise TypeError( + f"Can't instantiate abstract class {type(self).__name__} without the " + "INPUT_PORTS and OUTPUT_PORTS class attribute declarations." + ) + super().__init__(*args, **kwargs) # The following fields will be added in the setup_ports function and are non-functional intentionally till it # gets added to the setup_ports function. @@ -382,11 +392,6 @@ def __init__(self, *args: Any, behaviour_class_name: str | None = None, **kwargs behaviour_class_name if behaviour_class_name is not None else self.__class__.__name__ ) - # Consistency check: no value can appear in both input and output ports - for port in self.input_ports(): - if port in self.output_ports(): - raise ValueError(f"Port '{port}' appears in both input and output ports") - def setup_ports( self, port_remappings: dict | None = None, @@ -810,21 +815,16 @@ class BehaviourWithPorts(PortsMixin, py_trees.behaviour.Behaviour): Subclassing requirements: - - Each subclass must implement the ``input_ports`` and ``output_ports`` class methods to specify - its input and output ports. + - Each subclass must declare its ports by assigning the ``INPUT_PORTS`` and ``OUTPUT_PORTS`` + class attributes. - Each subclass must implement the ``update()`` method to define its behaviour. - Other methods from :class:`py_trees.behaviour.Behaviour` may be overridden as needed. Example usage:: class ExampleBehaviour(BehaviourWithPorts): - @classmethod - def input_ports(cls): - return {"input_data": PortInformation(data_type=str, required=True)} - - @classmethod - def output_ports(cls): - return {"output_data": PortInformation(data_type=str, required=True)} + INPUT_PORTS = {"input_data": PortInformation(data_type=str, required=True)} + OUTPUT_PORTS = {"output_data": PortInformation(data_type=str, required=True)} def update(self): # Implementation of the behaviour diff --git a/tests/test_ports.py b/tests/test_ports.py index 302cf360..6893bddd 100644 --- a/tests/test_ports.py +++ b/tests/test_ports.py @@ -397,13 +397,8 @@ def test_seeded_output_default_is_not_shared(self) -> None: class _RegistryLeaf(BehaviourWithPorts, register=False): """Concrete leaf used to exercise the registry; itself kept out of it.""" - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} + OUTPUT_PORTS = {} def update(self) -> py_trees.common.Status: return py_trees.common.Status.SUCCESS diff --git a/tests/test_ports_helpers.py b/tests/test_ports_helpers.py index 0e8ab3ed..23e151ce 100644 --- a/tests/test_ports_helpers.py +++ b/tests/test_ports_helpers.py @@ -17,13 +17,8 @@ class Producer(BehaviourWithPorts): OUTPUT_PORT = "output" - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {cls.OUTPUT_PORT: PortInformation(data_type=str, required=True)} + INPUT_PORTS = {} + OUTPUT_PORTS = {OUTPUT_PORT: PortInformation(data_type=str, required=True)} def update(self) -> py_trees.common.Status: self._set_output(self.OUTPUT_PORT, f"Producer[{self.subtree_namespace}:{self.name}]") @@ -34,13 +29,8 @@ class ConsumerProducer(BehaviourWithPorts): OUTPUT_PORT = "output" INPUT_PORT = "input" - @classmethod - def input_ports(cls) -> dict: - return {cls.INPUT_PORT: PortInformation(data_type=str, required=True)} - - @classmethod - def output_ports(cls) -> dict: - return {cls.OUTPUT_PORT: PortInformation(data_type=str, required=True)} + INPUT_PORTS = {INPUT_PORT: PortInformation(data_type=str, required=True)} + OUTPUT_PORTS = {OUTPUT_PORT: PortInformation(data_type=str, required=True)} def update(self) -> py_trees.common.Status: input_value = self.get_input(self.INPUT_PORT) @@ -51,13 +41,8 @@ def update(self) -> py_trees.common.Status: class Consumer(BehaviourWithPorts): INPUT_PORT = "input" - @classmethod - def input_ports(cls) -> dict: - return {cls.INPUT_PORT: PortInformation(data_type=str, required=True)} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {INPUT_PORT: PortInformation(data_type=str, required=True)} + OUTPUT_PORTS = {} def update(self) -> py_trees.common.Status: return py_trees.common.Status.SUCCESS @@ -70,13 +55,8 @@ def consumed_value(self) -> Any: class FloatConsumer(BehaviourWithPorts): INPUT_PORT = "input" - @classmethod - def input_ports(cls) -> dict: - return {cls.INPUT_PORT: PortInformation(data_type=float, required=True)} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {INPUT_PORT: PortInformation(data_type=float, required=True)} + OUTPUT_PORTS = {} def update(self) -> py_trees.common.Status: return py_trees.common.Status.SUCCESS @@ -89,16 +69,11 @@ def consumed_value(self) -> Any: class DefaultingConsumer(BehaviourWithPorts): """Consumer whose input ports declare default values.""" - @classmethod - def input_ports(cls) -> dict: - return { - "input": PortInformation(data_type=str, required=True, default_value="fallback"), - "items": PortInformation(data_type=list[int], required=False, default_value=[1, 2]), - } - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = { + "input": PortInformation(data_type=str, required=True, default_value="fallback"), + "items": PortInformation(data_type=list[int], required=False, default_value=[1, 2]), + } + OUTPUT_PORTS = {} def update(self) -> py_trees.common.Status: return py_trees.common.Status.SUCCESS @@ -111,16 +86,11 @@ def consumed_value(self) -> Any: class DefaultingProducer(BehaviourWithPorts): """Producer whose output ports declare default values.""" - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return { - "output": PortInformation(data_type=str, default_value="initial"), - "items": PortInformation(data_type=list[int], default_value=[1, 2]), - } + INPUT_PORTS = {} + OUTPUT_PORTS = { + "output": PortInformation(data_type=str, default_value="initial"), + "items": PortInformation(data_type=list[int], default_value=[1, 2]), + } def update(self) -> py_trees.common.Status: self._set_output("output", "produced") @@ -176,13 +146,8 @@ def update(self) -> py_trees.common.Status: class AlwaysSuccessBP(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} + OUTPUT_PORTS = {} def __init__(self, name: str, **kwargs: Any) -> None: super().__init__(name=name, **kwargs) @@ -192,13 +157,8 @@ def update(self) -> py_trees.common.Status: class AlwaysFailureBP(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} + OUTPUT_PORTS = {} def __init__(self, name: str, **kwargs: Any) -> None: super().__init__(name=name, **kwargs) @@ -208,13 +168,8 @@ def update(self) -> py_trees.common.Status: class AlwaysRunningBP(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} + OUTPUT_PORTS = {} def __init__(self, name: str, **kwargs: Any) -> None: super().__init__(name=name, **kwargs) diff --git a/tests/test_ports_xml_parser.py b/tests/test_ports_xml_parser.py index 36bbe840..c0e0b7a4 100644 --- a/tests/test_ports_xml_parser.py +++ b/tests/test_ports_xml_parser.py @@ -57,21 +57,14 @@ class DummyFactory: class Wait(BehaviourWithPorts): - INPUT_DURATION_MS_PORT = "input_duration_ms" + INPUT_PORTS = {"input_duration_ms": PortInformation(data_type=int, required=True)} + OUTPUT_PORTS = {} def __init__(self, name: str, factory: DummyFactory, **kwargs: Any) -> None: super().__init__(name=name, **kwargs) self._factory = factory self.start_time = 0.0 - @classmethod - def input_ports(cls) -> dict: - return {cls.INPUT_DURATION_MS_PORT: PortInformation(data_type=int, required=True)} - - @classmethod - def output_ports(cls) -> dict: - return {} - def initialise(self) -> None: self.start_time = time.time() @@ -86,19 +79,14 @@ def update(self) -> py_trees.common.Status: @property def duration_value_ms(self) -> Any: - return self.get_input(self.INPUT_DURATION_MS_PORT) + return self.get_input("input_duration_ms") class EchoCtorArgs(BehaviourWithPorts): """Behaviour that tests interpreting constructor type hints for type coercion from XML ports.""" - @classmethod - def input_ports(cls) -> dict: - return {"in": PortInformation(data_type=str, required=False)} # not used here - - @classmethod - def output_ports(cls) -> dict: - return {"out": PortInformation(data_type=str, required=False)} # not used here + INPUT_PORTS = {"in": PortInformation(data_type=str, required=False)} # not used here + OUTPUT_PORTS = {"out": PortInformation(data_type=str, required=False)} # not used here def __init__(self, name: str, greeting: str, times: float | None, flag: bool, **kwargs: Any) -> None: super().__init__(name, **kwargs) @@ -211,13 +199,8 @@ def test_custom_behavior_with_extra_arg(self) -> None: """Test custom behavior with an additional argument.""" class CustomBehaviourWithPorts(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {"in": PortInformation(data_type=str, required=False)} - - @classmethod - def output_ports(cls) -> dict: - return {"out": PortInformation(data_type=str, required=False)} + INPUT_PORTS = {"in": PortInformation(data_type=str, required=False)} + OUTPUT_PORTS = {"out": PortInformation(data_type=str, required=False)} def __init__(self, name: str, extra_arg: str, **kwargs: Any) -> None: super().__init__(name, **kwargs) @@ -683,13 +666,8 @@ def test_mixed_ports_and_ctor_kwargs(self) -> None: """ class PortAndCtor(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {"in": PortInformation(data_type=str, required=True)} # only this is a port - - @classmethod - def output_ports(cls) -> dict: - return {"out": PortInformation(data_type=str, required=False)} + INPUT_PORTS = {"in": PortInformation(data_type=str, required=True)} # only this is a port + OUTPUT_PORTS = {"out": PortInformation(data_type=str, required=False)} def __init__(self, name: str, label: str, **kwargs: Any) -> None: super().__init__(name, **kwargs) @@ -734,13 +712,8 @@ def test_ctor_arg_with_curly_value_is_not_resolved(self) -> None: """ class TakesKeyString(BehaviourWithPorts): - @classmethod - def input_ports(cls) -> dict: - return {} # no ports at all - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} # no ports at all + OUTPUT_PORTS = {} def __init__(self, name: str, token: str) -> None: super().__init__(name) @@ -799,13 +772,8 @@ def test_decorator_ports_node(self) -> None: class RepeatWithPorts(PortsMixin, py_trees.decorators.Repeat): """A `py_trees.decorators.Repeat` decorator that also exposes ports.""" - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {"count": PortInformation(data_type=int, required=False)} + INPUT_PORTS = {} + OUTPUT_PORTS = {"count": PortInformation(data_type=int, required=False)} xml = """ @@ -842,13 +810,8 @@ def test_composite_ports_node(self) -> None: class SequenceWithPorts(PortsMixin, py_trees.composites.Sequence): """A `py_trees.composites.Sequence` composite that also exposes ports.""" - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {} + INPUT_PORTS = {} + OUTPUT_PORTS = {} xml = """ @@ -892,17 +855,12 @@ def test_direct_portsmixin_leaf_accepted(self) -> None: class DirectPortsLeaf(PortsMixin, py_trees.behaviour.Behaviour): """PortsMixin leaf that does NOT go through BehaviourWithPorts.""" + INPUT_PORTS = {} + OUTPUT_PORTS = {"out": PortInformation(data_type=str, required=True)} + def __init__(self, name: str, **kwargs: Any) -> None: super().__init__(name=name, **kwargs) - @classmethod - def input_ports(cls) -> dict: - return {} - - @classmethod - def output_ports(cls) -> dict: - return {"out": PortInformation(data_type=str, required=True)} - def update(self) -> py_trees.common.Status: self._set_output("out", "direct-ports-leaf-ran") return py_trees.common.Status.SUCCESS