[CARLA] CARLA Simulator camera

[김위제]

ubuntu 24.04 desktop 
ros2 jazzy

carla에서 카메라 토픽을 받아서 출력과 저장

ros2 topic info /carla/ego_vehicle/rgb/camera_info  -v

Topic type: sensor_msgs/msg/CameraInfo

패키지 생성

cd ros2_ws/src

ros2 pkg create camera --build-type ament_cmake --dependencies rclcpp sensor_msgs cv_bridge OpenCV

cd ros2_ws

colcon build --packages-select camera

source install/setup.bash

ros2 run camera camera 

camera/src/sub.cpp

/carla/ego_vehicle/rgb/camera_info 토픽을 받지 않고
/carla/ego_vehicle/rgb/image로 토픽을 받음 

frame_id가 hero로 되어있는 경우

#include "rclcpp/rclcpp.hpp"
#include "sensor_msgs/msg/image.hpp"
#include "cv_bridge/cv_bridge.hpp"
#include "opencv2/opencv.hpp"
#include <memory>

class CameraVisualizer : public rclcpp::Node {
public:
CameraVisualizer() : Node("camera") {

auto qos_profile = rclcpp::QoS(10);
qos_profile.reliability(RMW_QOS_POLICY_RELIABILITY_RELIABLE);
qos_profile.durability(RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL);

// 토픽명 설정
subscription_ = this->create_subscription<sensor_msgs::msg::Image>(
"/carla/ego_vehicle/rgb/image", qos_profile,
std::bind(&CameraVisualizer::callback, this, std::placeholders::_1));

// OpenCV 창 생성
cv::namedWindow("CARLA Camera Output", cv::WINDOW_NORMAL);
cv::resizeWindow("CARLA Camera Output", 640, 360);

// 비디오 저장 설정
video_writer_.open("output_video.mp4", cv::VideoWriter::fourcc('m', 'p', '4', 'v'), 20.0, cv::Size(800, 600));

RCLCPP_INFO(this->get_logger(), "Camera Visualizer Node Started.");
}

~CameraVisualizer() {
if (video_writer_.isOpened()) video_writer_.release();
cv::destroyAllWindows();
}

private:
void callback(const sensor_msgs::msg::Image::SharedPtr msg) {
try {
// ROS Image 메시지를 OpenCV Mat으로 변환
cv::Mat frame = cv_bridge::toCvShare(msg, "bgr8")->image;

if (frame.empty()) return;

// 비디오 초기 설정 (첫 프레임 기준)
if (!is_writer_init_) {
video_writer_.open("output_video.mp4", cv::VideoWriter::fourcc('m','p','4','v'), 20.0, frame.size());
is_writer_init_ = true;
}

// 영상 파일 기록
if (video_writer_.isOpened()) {
video_writer_.write(frame);
}

// 실시간 화면 출력
cv::imshow("CARLA Camera Output", frame);
cv::waitKey(1);

} catch (cv_bridge::Exception& e) {
RCLCPP_ERROR(this->get_logger(), "cv_bridge exception: %s", e.what());
}
}

rclcpp::Subscription<sensor_msgs::msg::Image>::SharedPtr subscription_;
cv::VideoWriter video_writer_;
bool is_writer_init_ = false;
};

int main(int argc, char** argv) {
rclcpp::init(argc, argv);
rclcpp::spin(std::make_shared<CameraVisualizer>());
rclcpp::shutdown();
return 0;
}

/camera/CMakeLists.txt

cmake_minimum_required(VERSION 3.8)
project(camera)

if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
add_compile_options(-Wall -Wextra -Wpedantic)
endif()

find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)
find_package(sensor_msgs REQUIRED)
find_package(cv_bridge REQUIRED)
find_package(OpenCV REQUIRED)

add_executable(camera src/sub.cpp)
target_include_directories(camera PRIVATE ${OpenCV_INCLUDE_DIRS})
ament_target_dependencies(camera rclcpp sensor_msgs cv_bridge)
target_link_libraries(camera ${OpenCV_LIBRARIES})

install(TARGETS
camera
DESTINATION lib/${PROJECT_NAME})

ament_package()

카메라 topic을 받기 위해

ros2_native 예제를 사용

ros2_native.py

#!/usr/bin/env python

# Copyright (c) 2024 Computer Vision Center (CVC) at the Universitat Autonoma de
# Barcelona (UAB).
#
# This work is licensed under the terms of the MIT license.
# For a copy, see <https://opensource.org/licenses/MIT>.

# Allows controlling a vehicle with a keyboard. For a simpler and more
# documented example, please take a look at tutorial.py.

import argparse
import json
import logging

import carla


def _setup_vehicle(world, config):
logging.debug("Spawning vehicle: {}".format(config.get("type")))

bp_library = world.get_blueprint_library()
map_ = world.get_map()

bp = bp_library.filter(config.get("type"))[0]
bp.set_attribute("role_name", config.get("id"))
#bp.set_attribute('role_name', 'ego_vehicle')
bp.set_attribute("ros_name", config.get("id"))
return world.spawn_actor(
bp,
map_.get_spawn_points()[0],
attach_to=None)


def _setup_sensors(world, vehicle, sensors_config):
bp_library = world.get_blueprint_library()

sensors = []
for sensor in sensors_config:
logging.debug("Spawning sensor: {}".format(sensor))

bp = bp_library.filter(sensor.get("type"))[0]
bp.set_attribute("ros_name", sensor.get("id"))
bp.set_attribute("role_name", sensor.get("id"))
for key, value in sensor.get("attributes", {}).items():
bp.set_attribute(str(key), str(value))

wp = carla.Transform(
location=carla.Location(x=sensor["spawn_point"]["x"], y=-sensor["spawn_point"]["y"], z=sensor["spawn_point"]["z"]),
rotation=carla.Rotation(roll=sensor["spawn_point"]["roll"], pitch=-sensor["spawn_point"]["pitch"], yaw=-sensor["spawn_point"]["yaw"])
)

sensors.append(
world.spawn_actor(
bp,
wp,
attach_to=vehicle
)
)

sensors[-1].enable_for_ros()

return sensors


def main(args):

world = None
vehicle = None
sensors = []
original_settings = None

try:
client = carla.Client(args.host, args.port)
client.set_timeout(60.0)

world = client.get_world()

original_settings = world.get_settings()
settings = world.get_settings()
settings.synchronous_mode = True
settings.fixed_delta_seconds = 0.05
world.apply_settings(settings)

traffic_manager = client.get_trafficmanager()
traffic_manager.set_synchronous_mode(True)

with open(args.file) as f:
config = json.load(f)

vehicle = _setup_vehicle(world, config)
sensors = _setup_sensors(world, vehicle, config.get("sensors", []))

_ = world.tick()

vehicle.set_autopilot(True)

# # ######## add
# vehicle.set_autopilot(True)

# # Traffic Manager 설정 추가
traffic_manager = client.get_trafficmanager()


# # 특정 차량(hero)만 아주 천천히 가게 하고 싶다면:
traffic_manager.distance_to_leading_vehicle(vehicle, 5.0) # 앞차와의 간격
traffic_manager.vehicle_percentage_speed_difference(vehicle, 95.0) # 90% 감속 (매우 느림)
# ##########
# logging.info("Running...")

while True:
_ = world.tick()

except KeyboardInterrupt:
print('\nCancelled by user. Bye!')

finally:
if original_settings:
world.apply_settings(original_settings)

for sensor in sensors:
sensor.destroy()

if vehicle:
vehicle.destroy()


if __name__ == '__main__':
argparser = argparse.ArgumentParser(description='CARLA ROS2 native')
argparser.add_argument('--host', metavar='H', default='localhost', help='IP of the host CARLA Simulator (default: localhost)')
argparser.add_argument('--port', metavar='P', default=2000, type=int, help='TCP port of CARLA Simulator (default: 2000)')
argparser.add_argument('-f', '--file', default='', required=True, help='File to be executed')
argparser.add_argument('-v', '--verbose', action='store_true', dest='debug', help='print debug information')

args = argparser.parse_args()

log_level = logging.DEBUG if args.debug else logging.INFO
logging.basicConfig(format='%(levelname)s: %(message)s', level=log_level)

logging.info('Listening to server %s:%s', args.host, args.port)

main(args)

이 소스를 실행하기 위해서는 설정파일이 필요함

설정 파일

stack.json

{
"type": "vehicle.lincoln.mkz",
"id": "ego_vehicle",
"sensors": [
{
"type": "sensor.camera.rgb",
"id": "rgb",
"spawn_point": {"x": 2.0, "y": 0.0, "z": 1.5, "roll": 0.0, "pitch": 20.0, "yaw": 0.0},
"attributes": {
"image_size_x": 400,
"image_size_y": 200,
"fov": 90.0
}
},
{
"type": "sensor.lidar.ray_cast",
"id": "lidar",
"spawn_point": {"x": 0.0, "y": 0.0, "z": 2.4, "roll": 0.0, "pitch": 0.0, "yaw": 0.0},
"attributes": {
"range": 85,
"channels": 64,
"points_per_second": 600000,
"rotation_frequency": 20,
"upper_fov": 10,
"lower_fov": -30,
"atmosphere_attenuation_rate": 0.004,
"dropoff_general_rate": 0.45,
"dropoff_intensity_limit": 0.8,
"dropoff_zero_intensity": 0.4
}
}
]
}

센서 종류

실행 영상

저장된 영상

[Sungryul Lee]

제목 CARLA Simulator camer -> CARLA Simulator camera
/carla/ego_vehicle/rgb/camera_info 토픽을 받는건가요?
carla 서버에 카메라 설정하는 소스도 추가하고 서버실행및 초기화 명령어도 추가할것

[김위제]

수정했습니다