feat: launch grant experiment by id

This commit is contained in:
Даниил Грабарь
2026-11-08 10:52:22 +10:00
parent 814922f8ca
commit 568696d7b5
6 changed files with 610 additions and 0 deletions
+95
View File
@@ -0,0 +1,95 @@
"""Launch validation and action wiring."""
import importlib.util
from pathlib import Path
import pytest
from launch.actions import DeclareLaunchArgument, IncludeLaunchDescription, OpaqueFunction
from launch_ros.actions import Node
from iiwa_experement.runtime_config import validate_experiment_id
PACKAGE = Path(__file__).resolve().parents[1]
LAUNCH = PACKAGE / 'launch' / 'experiment.launch.py'
def load_launch_module():
spec = importlib.util.spec_from_file_location('experiment_launch', LAUNCH)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
@pytest.mark.parametrize('raw, expected', [('1', 1), ('20', 20), (' 11 ', 11)])
def test_valid_experiment_ids(raw, expected):
assert validate_experiment_id(raw) == expected
@pytest.mark.parametrize('raw', ['', '0', '21', 'abc', '-1', '1.0', '01', ' '])
def test_invalid_experiment_ids_are_rejected(raw):
with pytest.raises(ValueError):
validate_experiment_id(raw)
def test_runtime_actions_include_bringup_and_one_scene_spawner():
module = load_launch_module()
actions = module.runtime_actions(
experiment_id=19,
scene_config='/tmp/custom scenes.yaml',
setting_path='/tmp/run/setting.yaml',
rviz='false',
)
includes = [a for a in actions if isinstance(a, IncludeLaunchDescription)]
nodes = [a for a in actions if isinstance(a, Node)]
assert len(includes) == 1
assert len(nodes) == 1
assert 'iiwa_bringup' in repr(includes[0].launch_description_source.location)
arguments = dict(includes[0].launch_arguments)
assert arguments['simulate'] == 'true'
assert arguments['rviz'] == 'false'
assert arguments['setting'] == '/tmp/run/setting.yaml'
node = nodes[0]
assert node.node_package == 'iiwa_experement'
assert node.node_executable == 'scene_spawner'
assert vars(node)['_Node__node_name'] == 'scene_spawner'
assert module.spawner_parameters(19, '/tmp/custom scenes.yaml') == {
'experiment_id': 19,
'scene_config': '/tmp/custom scenes.yaml',
}
def test_runtime_actions_can_register_cleanup(tmp_path):
from launch.actions import RegisterEventHandler
module = load_launch_module()
actions = module.runtime_actions(
experiment_id=1,
scene_config='/tmp/scenes.yaml',
setting_path='/tmp/setting.yaml',
cleanup_dir=str(tmp_path / 'run'),
)
assert len(actions) == 3
assert isinstance(actions[-1], RegisterEventHandler)
target = tmp_path / 'run'
target.mkdir()
(target / 'world.wbt').write_text('temporary', encoding='utf-8')
module._remove_directory(None, directory=str(target))
assert not target.exists()
def test_launch_description_declares_all_arguments_and_validates_in_opaque_function():
module = load_launch_module()
description = module.generate_launch_description()
arguments = {
entity.name: entity.default_value
for entity in description.entities
if isinstance(entity, DeclareLaunchArgument)
}
assert set(arguments) == {'experiment_id', 'rviz', 'scene_config', 'base_setting'}
assert [substitution.text for substitution in arguments['experiment_id']] == ['']
assert [substitution.text for substitution in arguments['rviz']] == ['false']
assert 'scenes.yaml' in repr(arguments['scene_config'])
assert 'setting.yaml' in repr(arguments['base_setting'])
opaque = [e for e in description.entities if isinstance(e, OpaqueFunction)]
assert len(opaque) == 1
@@ -42,8 +42,10 @@ def test_expected_resources_are_installed():
'package.xml',
'doc/experements.md',
'doc/experiment-scenes.md',
'doc/README.md',
'worlds/grant/iiwa-grant.wbt',
'config/scenes.yaml',
'launch/experiment.launch.py',
):
path = share / relative
assert path.is_file(), path
@@ -0,0 +1,112 @@
"""Temporary runtime files point at installed assets and keep patron."""
from pathlib import Path
import pytest
import yaml
from ament_index_python.packages import get_package_share_directory
from iiwa_experement.runtime_config import (
LOCAL_PROTOS,
prepare_runtime_files,
)
PACKAGE = Path(__file__).resolve().parents[1]
CONFIG_SHARE = Path(get_package_share_directory('iiwa_config'))
DESCRIPTION_SHARE = Path(get_package_share_directory('iiwa_description'))
PACKAGE_SHARE = Path(get_package_share_directory('iiwa_experement'))
BASE_SETTING = CONFIG_SHARE / 'config' / 'setting.yaml'
WORLD_TEMPLATE = PACKAGE_SHARE / 'worlds' / 'grant' / 'iiwa-grant.wbt'
def prepare(tmp_path, **overrides):
arguments = {
'base_setting': BASE_SETTING,
'world_template': WORLD_TEMPLATE,
'description_share': DESCRIPTION_SHARE,
'output_dir': tmp_path / 'run',
}
arguments.update(overrides)
return prepare_runtime_files(**arguments)
def test_world_copy_resolves_only_local_protos(tmp_path):
world_path, _ = prepare(tmp_path)
text = world_path.read_text(encoding='utf-8')
for name in LOCAL_PROTOS:
absolute = (DESCRIPTION_SHARE / 'resource' / 'protos' / name).resolve()
assert absolute.is_file()
assert f'EXTERNPROTO "{absolute}"' in text
assert 'EXTERNPROTO "../../../iiwa_description' not in text
assert text.count('https://raw.githubusercontent.com/cyberbotics/webots/R2025a/') == 3
assert world_path == tmp_path / 'run' / 'iiwa-grant.wbt'
def test_setting_copy_disables_random_objects_and_keeps_patron_and_camera(tmp_path):
world_path, setting_path = prepare(tmp_path)
data = yaml.safe_load(setting_path.read_text(encoding='utf-8'))
webots = data['digital_twin']['webots']
assert webots['world'] == str(world_path.resolve())
assert webots['objects']['enabled'] is False
assert webots['objects']['count'] == 0
assert data['tool']['active'] == 'patron'
assert webots['transform'] == '-0.25 0 0.79'
assert webots['rotation'] == '0 0 1 0'
assert webots['controller_timer'] == '50'
assert any(
str(camera).endswith('config/cameras/d455_top.yaml')
for camera in webots['cameras']
)
assert data['controller']['controller_path'].endswith('moveit/iiwa_controller.yaml')
assert data['digital_twin']['rviz']['config'].endswith('rviz_moveit.rviz')
def test_prepared_setting_is_accepted_by_the_existing_loader(tmp_path):
_, setting_path = prepare(tmp_path)
from iiwa_utils.setting_loader import build_settings
settings = build_settings(str(setting_path), check_files=True)
assert settings.tool.active == 'patron'
assert settings.digital_twin.webots.objects.enabled is False
assert settings.digital_twin.webots.objects.count == 0
assert settings.digital_twin.webots.cameras
assert settings.digital_twin.webots.cameras[0].endswith('d455_top.yaml')
def test_tool_active_xacro_and_generated_urdf_include_patron(tmp_path):
from iiwa_utils import converter
from iiwa_utils.setting_loader import build_settings
tool_active = DESCRIPTION_SHARE / 'urdf' / 'tools' / 'tool_active.xacro'
assert 'patron.xacro' in tool_active.read_text(encoding='utf-8')
_, setting_path = prepare(tmp_path)
settings = build_settings(str(setting_path), check_files=True)
urdf = converter.load_robot_description(
settings.robot.description,
settings.robot.name,
{
'simulate': 'true',
'initial_positions_file': settings.controller.moveit.initial_positions,
'joint_limits_file': settings.controller.moveit.joint_limits,
},
)
assert 'name="patron"' in urdf
assert 'link name="patron"' in urdf
assert 'patron.dae' in urdf
def test_missing_world_template_is_reported(tmp_path):
with pytest.raises(FileNotFoundError, match='world template'):
prepare(tmp_path, world_template=tmp_path / 'missing.wbt')
def test_missing_proto_is_reported(tmp_path):
empty_description = tmp_path / 'description'
(empty_description / 'resource' / 'protos').mkdir(parents=True)
with pytest.raises(FileNotFoundError, match='CollaborativeRobotTable.proto'):
prepare(tmp_path, description_share=empty_description)
def test_missing_base_setting_is_reported(tmp_path):
with pytest.raises(FileNotFoundError, match='base settings'):
prepare(tmp_path, base_setting=tmp_path / 'missing.yaml')