def generate_launch_description()

in cloudwatch_metrics_collector/launch/cloudwatch_metrics_collector.launch.py [0:0]
53 lines of code
8 McCabe index (conditional complexity)

def generate_launch_description():

  # Default to included config file
  default_config = os.path.join(get_package_share_directory('cloudwatch_metrics_collector'),
    'config', 'sample_configuration.yaml')

  with open(default_config, 'r') as f:
      config_text = f.read()
  config_yaml = yaml.safe_load(config_text)
  default_aws_metrics_namespace = config_yaml['cloudwatch_metrics_collector']['ros__parameters']['aws_metrics_namespace']
  default_aws_region = config_yaml['cloudwatch_metrics_collector']['ros__parameters']['aws_client_configuration']['region']
  
  aws_robomaker_metric_namespace = None
  aws_default_metric_dimensions = None
  if RoboMakerEnvVariables.CAPABILITY in os.environ:
    aws_robomaker_metric_namespace = os.environ[RoboMakerEnvVariables.CAPABILITY]
    # If this is the RoboMaker Fleet Management environment then add the robot
    # name as a default dimension
    if os.environ[RoboMakerEnvVariables.CAPABILITY] == "AWSRoboMakerFleetManagement" and RoboMakerEnvVariables.ROBOT_NAME in os.environ:
      aws_default_metric_dimensions = "RobotName:" + os.environ.get(RoboMakerEnvVariables.ROBOT_NAME)
    # If this is the RoboMaker Simulation envrionment then add the simulation job
    # ID as a default dimension
    if os.environ[RoboMakerEnvVariables.CAPABILITY] == "AWSRoboMakerSimulation" and RoboMakerEnvVariables.SIMULATION_JOB_ID in os.environ:
       aws_default_metric_dimensions = "SimulationJobId:" + os.environ.get(RoboMakerEnvVariables.SIMULATION_JOB_ID)
      
  parameters = [launch.substitutions.LaunchConfiguration(CONFIG), ]
  if aws_default_metric_dimensions:
    parameters.append({"aws_default_metric_dimensions": aws_default_metric_dimensions})
  parameters.append({"aws_metrics_namespace": launch.substitutions.LaunchConfiguration("aws_metrics_namespace")})
  parameters.append({"aws_client_configuration": { "region": launch.substitutions.LaunchConfiguration("aws_region") }})

  ld = launch.LaunchDescription([
    launch.actions.DeclareLaunchArgument(
      NODE_NAME,
      default_value="cloudwatch_metrics_collector",
    ),
    launch.actions.DeclareLaunchArgument(
      CONFIG,
      default_value=default_config
    ),
    launch.actions.DeclareLaunchArgument(
      "aws_region",
      default_value=default_aws_region
    ),
    launch.actions.DeclareLaunchArgument(
      "aws_metrics_namespace",
      default_value=default_aws_metrics_namespace
    )
  ])

  if aws_robomaker_metric_namespace:
    ld.add_action(launch.actions.SetLaunchConfiguration(
      name="aws_metrics_namespace",
      value=aws_robomaker_metric_namespace
    ))

  encoder_node = launch_ros.actions.Node(
    package="cloudwatch_metrics_collector",
    node_executable="cloudwatch_metrics_collector",
    node_name=launch.substitutions.LaunchConfiguration(NODE_NAME),
    parameters=parameters,
    output="screen"
  )

  ld.add_action(encoder_node)

  return ld