Skip to content

[Jazzy] Improve the compatibility of processing YAML parameter files #1549

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
ahcorde merged 6 commits into from
Dec 23, 2025
Merged

[Jazzy] Improve the compatibility of processing YAML parameter files #1549

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
950cc59
Improve the compatibility of processing YAML parameter files
Barry-Xu-2018 Dec 2, 2025
9477152
Enhance parameter_dict_from_yaml_file to handle empty node names and …
Barry-Xu-2018 Dec 4, 2025
2a5b27c
Address review comments from Coiplit
Barry-Xu-2018 Dec 4, 2025
8ada33e
Simplify namespace handling and improve error messaging
Barry-Xu-2018 Dec 4, 2025
5b1634f
Improve checking parameter file structure for wildcard "/*"
Barry-Xu-2018 Dec 4, 2025
468c615
Adjust error messages
Barry-Xu-2018 Dec 4, 2025
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
151 changes: 143 additions & 8 deletions rclpy/rclpy/parameter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,30 +298,165 @@ def parameter_dict_from_yaml_file(
param_keys = []
param_dict = {}

if use_wildcard and '/**' in param_file:
param_keys.append('/**')
if use_wildcard:
if '/**' in param_file:
param_keys.append('/**')

if not target_nodes:
# /*
# node_name:
# ros__parameters:
# ...
if '/*' in param_file:
param_keys.append('/*')

# /**/node_name
# ros__parameters:
# ...
for k, v in param_file.items():
if k.startswith('/**/'):
param_keys.append(k)

if target_nodes:
for n in target_nodes:
if n not in param_file.keys():
if n is None or n == '':
continue

# Get absolute node name
if n[0] != '/':
n = '/' + n

# target node doesn't include namespace
# Match definition in param file
# /node_name:
# ros__parameters:
# ...
# or
# node_name:
# ros__parameters:
# ...
if '/' not in n[1:]:
is_found = False
# In the param file, there may be two ways to write a node_name.
# So need to handle all of them.
# /node_name:
# ros__parameters:
# ...
# or
# node_name:
# ros__parameters:
#
if n in param_file:
param_keys.append(n)
is_found = True
if n[1:] in param_file:
param_keys.append(n[1:])
is_found = True
if is_found:
continue
raise RuntimeError(f'Param file does not contain parameters for {n},'
f'only for nodes: {list(param_file.keys())} ')
param_keys.append(n)

# target node include namespaces, e.g. /namespace/node_name
# Matched definition in param file
# /namespace
# node_name:
# ros__parameters:
# ...
# or
# /namespace/node_name:
# ros__parameters:
# ...
ns, node_name = n.rsplit('/', 1)

# If ns is single level namespace and wildcard is true
# Match definition in param file
# /*
# node_name:
# ros__parameters:
# ...
if use_wildcard and '/' not in ns[1:] and '/*' in param_file:
if not isinstance(param_file['/*'], dict):
raise RuntimeError(
'YAML file is not a valid ROS parameter file for namespace "/*"')
if node_name in param_file['/*']:
param_keys.append('/*#' + node_name)

# If 'ns' in target node is single level or multi level namespace and
# wildcard is true,
# Match definition in param file
# /**/node_name:
# ros__parameters:
# ...
if use_wildcard and '/**/' + node_name in param_file:
param_keys.append('/**/' + node_name)

# If 'ns' in target node is single level or multi level namespace,
# Match definition in param file.
# /namespace
# node_name:
# ros__parameters:
# ...
if ns in param_file and isinstance(param_file[ns], dict):
if node_name in param_file[ns]:
param_keys.append(ns + '#' + node_name)

# if 'ns' in target node is single level or multi level namespace,
# Match definition in param file.
# /namespace/node_name:
# ros__parameters:
# ...
if n in param_file:
param_keys.append(n)
else:
# wildcard key must go to the front of param_keys so that
# node-namespaced parameters will override the wildcard parameters
keys = set(param_file.keys())
keys.discard('/**')
keys.discard('/*')
keys_to_remove = [item for item in keys if item.startswith('/**/')]
for key in keys_to_remove:
keys.discard(key)
param_keys.extend(keys)

if len(param_keys) == 0:
raise RuntimeError('Param file does not contain selected parameters')

for n in param_keys:
value = param_file[n]
if not isinstance(value, dict) or 'ros__parameters' not in value:
raise RuntimeError(f'YAML file is not a valid ROS parameter file for node {n}')
param_dict.update(value['ros__parameters'])
# "namespace#node_name" means the following parameters need to be collected in
# the param file.
# namespace
# node_name:
# ros__parameters:
# ...
if '#' in n:
ns, node_name = n.split('#', 1)
if (isinstance(param_file[ns][node_name], dict) and
'ros__parameters' in param_file[ns][node_name]):
param_dict.update(param_file[ns][node_name]['ros__parameters'])
continue
else:
value = param_file[n]
if isinstance(value, dict):
if 'ros__parameters' in value:
param_dict.update(value['ros__parameters'])
continue
else:
if n in param_file:
# n is namespace (e.g. '/*'). Add all node parameters under
# this namespace
if isinstance(param_file[n], dict):
for node_name, param_value in param_file[n].items():
if (isinstance(param_value, dict) and
'ros__parameters' in param_value):
param_dict.update(param_value['ros__parameters'])
else:
raise RuntimeError(
f'YAML file is not a valid ROS parameter file for node'
f' {n}/{node_name}')
continue
raise RuntimeError(f'YAML file is not a valid ROS parameter file for node {n}')

return _unpack_parameter_dict(namespace, param_dict)


Expand Down
175 changes: 175 additions & 0 deletions rclpy/test/test_parameter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,181 @@ def test_parameter_dict_from_yaml_file(self):
os.unlink(f.name)
self.assertRaises(FileNotFoundError, parameter_dict_from_yaml_file, 'unknown_file')

def test_parameter_dict_from_yaml_file2(self):
yaml_string = """
/**:
ros__parameters:
wildcard: true
/param_test_target1:
ros__parameters:
abs-nodename: true
param_test_target1:
ros__parameters:
base-nodename: false
/foo/param_test_target2:
ros__parameters:
abs-ns-nodename: true
/foo:
param_test_target2:
ros__parameters:
abs-ns-base-nodename: false
/*:
param_test_target2:
ros__parameters:
abs-wildcard-ns-base-nodename: true
/**/param_test_target2:
ros__parameters:
abs-wildcard-ns-nodename: false
/ns1/ns2/param_test_target3:
ros__parameters:
deep-ns-nodename: true
/ns1/ns2:
param_test_target3:
ros__parameters:
deep-ns-base-nodename: false
/**/param_test_target3:
ros__parameters:
abs-wildcard-deep-ns-nodename: true
"""
# Not set target nodes and wildcard is false
expected_no_target_nodes_and_no_wildcard = {
'abs-nodename': Parameter(
'abs-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-ns-base-nodename': Parameter(
'abs-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'abs-ns-nodename': Parameter(
'abs-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'base-nodename': Parameter(
'base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-base-nodename': Parameter(
'deep-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-nodename': Parameter(
'deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
}
# Not set target nodes and wildcard is true
expected_no_target_nodes_and_wildcard = {
'abs-nodename': Parameter(
'abs-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-ns-base-nodename': Parameter(
'abs-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'abs-ns-nodename': Parameter(
'abs-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-wildcard-deep-ns-nodename': Parameter(
'abs-wildcard-deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-wildcard-ns-base-nodename': Parameter(
'abs-wildcard-ns-base-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-wildcard-ns-nodename': Parameter(
'abs-wildcard-ns-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'base-nodename': Parameter(
'base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-base-nodename': Parameter(
'deep-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-nodename': Parameter(
'deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'wildcard': Parameter(
'wildcard', Parameter.Type.BOOL, True).to_parameter_msg(),
}
# Set one target node without ns and wildcard is false
expected_one_target_node_and_no_wildcard = {
'abs-nodename': Parameter(
'abs-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'base-nodename': Parameter(
'base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
}
# Set one target node without ns and wildcard is true
expected_one_target_node_and_wildcard = {
'abs-nodename': Parameter(
'abs-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'base-nodename': Parameter(
'base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'wildcard': Parameter(
'wildcard', Parameter.Type.BOOL, True).to_parameter_msg(),
}
# Set one target node with single level namespace and wildcard is false
expected_one_target_node_with_single_ns_and_no_wildcard = {
'abs-ns-base-nodename': Parameter(
'abs-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'abs-ns-nodename': Parameter(
'abs-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
}
# Set one target node with single level namespace and wildcard is true
expected_one_target_node_with_single_ns_and_wildcard = {
'abs-ns-base-nodename': Parameter(
'abs-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'abs-ns-nodename': Parameter(
'abs-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-wildcard-ns-base-nodename': Parameter(
'abs-wildcard-ns-base-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'abs-wildcard-ns-nodename': Parameter(
'abs-wildcard-ns-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'wildcard': Parameter(
'wildcard', Parameter.Type.BOOL, True).to_parameter_msg(),
}

# Set one target node with multi level namespace and wildcard is false
expected_one_target_node_with_multi_ns_and_no_wildcard = {
'deep-ns-base-nodename': Parameter(
'deep-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-nodename': Parameter(
'deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
}
# Set one target node with multi level namespace and wildcard is true
expected_one_target_node_with_multi_ns_and_wildcard = {
'abs-wildcard-deep-ns-nodename': Parameter(
'abs-wildcard-deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'deep-ns-base-nodename': Parameter(
'deep-ns-base-nodename', Parameter.Type.BOOL, False).to_parameter_msg(),
'deep-ns-nodename': Parameter(
'deep-ns-nodename', Parameter.Type.BOOL, True).to_parameter_msg(),
'wildcard': Parameter(
'wildcard', Parameter.Type.BOOL, True).to_parameter_msg(),
}

try:
with NamedTemporaryFile(mode='w', delete=False) as f:
f.write(yaml_string)
f.flush()
f.close()
parameter_dict = parameter_dict_from_yaml_file(f.name, use_wildcard=False)
assert parameter_dict == expected_no_target_nodes_and_no_wildcard
parameter_dict = parameter_dict_from_yaml_file(f.name, use_wildcard=True)
assert parameter_dict == expected_no_target_nodes_and_wildcard

parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=False, target_nodes=['param_test_target1'])
assert parameter_dict == expected_one_target_node_and_no_wildcard
parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=True, target_nodes=['param_test_target1'])
assert parameter_dict == expected_one_target_node_and_wildcard

parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=False, target_nodes=['/foo/param_test_target2'])
assert parameter_dict == expected_one_target_node_with_single_ns_and_no_wildcard
parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=True, target_nodes=['/foo/param_test_target2'])
assert parameter_dict == expected_one_target_node_with_single_ns_and_wildcard

parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=False, target_nodes=['/ns1/ns2/param_test_target3'])
assert parameter_dict == expected_one_target_node_with_multi_ns_and_no_wildcard
parameter_dict = parameter_dict_from_yaml_file(
f.name, use_wildcard=True, target_nodes=['/ns1/ns2/param_test_target3'])
assert parameter_dict == expected_one_target_node_with_multi_ns_and_wildcard

with pytest.raises(RuntimeError,
match='Param file does not contain selected parameters'):
parameter_dict = parameter_dict_from_yaml_file(
f.name, False, target_nodes=['/abc/cde/param_test_target3'])
with pytest.raises(RuntimeError,
match='Param file does not contain selected parameters'):
parameter_dict = parameter_dict_from_yaml_file(
f.name, False, target_nodes=['/abc/param_test_target2'])

finally:
if os.path.exists(f.name):
os.unlink(f.name)
self.assertRaises(FileNotFoundError, parameter_dict_from_yaml_file, 'unknown_file')


if __name__ == '__main__':
unittest.main()