Skip to content

Commit a9a280a

Browse files
committed
[ports] Convert constructor kwargs to types defined in parent classes
1 parent 7fe4562 commit a9a280a

2 files changed

Lines changed: 97 additions & 21 deletions

File tree

py_trees/ports_utils.py

Lines changed: 47 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,52 @@ def convert_str_to_type(value: str, target_type: type | UnionType, logger: Ports
182182
return value
183183

184184

185+
def collect_type_hints(constructor: Callable) -> dict[str, Any]:
186+
"""Collect parameter type hints from a constructor.
187+
188+
When ``constructor`` is a class, the method resolution order (MRO) is walked so
189+
that a type hint declared on a parent's ``__init__`` is still found when the
190+
subclass forwards ``**kwargs`` to ``super().__init__()``, and therefore does
191+
not declare the parameter itself, or declares it without an annotation.
192+
193+
The most-derived annotation for a given parameter name wins: walking the MRO
194+
from subclass to base and only recording the first hint seen per parameter.
195+
196+
Args:
197+
constructor: A class (whose ``__init__`` chain is inspected) or a callable.
198+
199+
Returns:
200+
dict[str, Any]: Mapping of parameter name to its type annotation. Parameters
201+
without an annotation anywhere in the hierarchy are omitted, as are
202+
``self`` and any ``*args``/``**kwargs`` catch-alls.
203+
"""
204+
classes = constructor.__mro__ if inspect.isclass(constructor) else (constructor,)
205+
206+
hints: dict[str, Any] = {}
207+
for klass in classes:
208+
func = klass.__init__ if inspect.isclass(klass) else klass
209+
try:
210+
sig = inspect.signature(func)
211+
except (TypeError, ValueError):
212+
# Built-ins (e.g. object.__init__ on some interpreters) may not be
213+
# introspectable; just skip them and keep walking the hierarchy.
214+
continue
215+
216+
for pname, param in sig.parameters.items():
217+
if pname == "self":
218+
continue
219+
if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
220+
continue
221+
if param.annotation is inspect._empty:
222+
continue
223+
# The first hint found wins => the most-derived class in the MRO.
224+
if pname in hints:
225+
continue
226+
hints[pname] = param.annotation
227+
228+
return hints
229+
230+
185231
def apply_type_hints(
186232
constructor: Callable,
187233
kwargs: dict[str, Any],
@@ -207,13 +253,7 @@ def apply_type_hints(
207253
if logger is None:
208254
logger = NOOP_LOGGER
209255

210-
sig = inspect.signature(constructor.__init__ if inspect.isclass(constructor) else constructor)
211-
hints: dict[str, Any] = {}
212-
for pname, param in sig.parameters.items():
213-
if pname == "self":
214-
continue
215-
if param.annotation is not inspect._empty:
216-
hints[pname] = param.annotation
256+
hints = collect_type_hints(constructor)
217257

218258
converted: dict[str, Any] = {}
219259
success = True

tests/test_ports_xml_parser.py

Lines changed: 50 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,30 @@ def duration_value_ms(self) -> Any:
8383
return self.get_input(self.INPUT_DURATION_MS_PORT)
8484

8585

86+
class EchoCtorArgs(BehaviourWithPorts):
87+
"""Behaviour that tests interpreting constructor type hints for type coercion from XML ports."""
88+
@classmethod
89+
def input_ports(cls) -> dict:
90+
return {"in": PortInformation(data_type=str, required=False)} # not used here
91+
92+
@classmethod
93+
def output_ports(cls) -> dict:
94+
return {"out": PortInformation(data_type=str, required=False)} # not used here
95+
96+
def __init__(self, name: str, greeting: str, times: float | None, flag: bool, **kwargs: Any) -> None:
97+
super().__init__(name, **kwargs)
98+
self.greeting = greeting
99+
self.times = times
100+
self.flag = flag
101+
102+
class EchoCtorArgsChild(EchoCtorArgs):
103+
"""
104+
Behaviour that tests interpreting constructor type hints for type coercion from XML ports,
105+
but from its parent class.
106+
"""
107+
def __init__(self, **kwargs):
108+
super().__init__(**kwargs)
109+
86110
@dataclass
87111
class RobotData:
88112
type: str
@@ -471,25 +495,37 @@ def test_ctor_args_passed_as_kwargs(self) -> None:
471495
Values are automatically converted based on the type hints in the constructor.
472496
"""
473497

474-
class EchoCtorArgs(BehaviourWithPorts):
475-
@classmethod
476-
def input_ports(cls) -> dict:
477-
return {"in": PortInformation(data_type=str, required=False)} # not used here
498+
xml = """<root main_tree_to_execute="Main">
499+
<BehaviorTree ID="Main">
500+
<Sequence>
501+
<EchoCtorArgs name="E1" greeting="hello" times="3" flag="true"/>
502+
</Sequence>
503+
</BehaviorTree>
504+
</root>"""
478505

479-
@classmethod
480-
def output_ports(cls) -> dict:
481-
return {"out": PortInformation(data_type=str, required=False)} # not used here
506+
with tempfile.NamedTemporaryFile(delete=False, mode="w", suffix=".xml") as tf:
507+
tf.write(xml)
508+
path = tf.name
482509

483-
def __init__(self, name: str, greeting: str, times: float | None, flag: bool, **kwargs: Any) -> None:
484-
super().__init__(name, **kwargs)
485-
self.greeting = greeting
486-
self.times = times
487-
self.flag = flag
510+
root = parse_behaviour_tree_xml(path, logger=StdoutLogger())
511+
node = find_node_by_class(root, EchoCtorArgs)
512+
self.assertIsNotNone(node)
513+
self.assertEqual(node.greeting, "hello")
514+
self.assertEqual(node.times, 3.0)
515+
self.assertEqual(node.flag, True)
516+
517+
os.unlink(path)
518+
519+
def test_ctor_args_passed_as_kwargs_in_parent_class(self) -> None:
520+
"""
521+
Non-port XML attributes must be passed as constructor kwargs.
522+
Values are automatically converted based on the type hints in the parent class constructor.
523+
"""
488524

489525
xml = """<root main_tree_to_execute="Main">
490526
<BehaviorTree ID="Main">
491527
<Sequence>
492-
<EchoCtorArgs name="E1" greeting="hello" times="3" flag="true"/>
528+
<EchoCtorArgsChild name="E1" greeting="hello" times="3" flag="true"/>
493529
</Sequence>
494530
</BehaviorTree>
495531
</root>"""
@@ -499,7 +535,7 @@ def __init__(self, name: str, greeting: str, times: float | None, flag: bool, **
499535
path = tf.name
500536

501537
root = parse_behaviour_tree_xml(path, logger=StdoutLogger())
502-
node = find_node_by_class(root, EchoCtorArgs)
538+
node = find_node_by_class(root, EchoCtorArgsChild)
503539
self.assertIsNotNone(node)
504540
self.assertEqual(node.greeting, "hello")
505541
self.assertEqual(node.times, 3.0)

0 commit comments

Comments
 (0)