ROS2(Cyclone DDS)를 네트워크로 연결하여 실험하기

✅ 목표: ROS2(Cyclone DDS)를 네트워크로 연결하여 실험하기

[퍼블리셔 노트북1] ─────→ [Cyclone DDS (P2P)] ─────→ [서브스크라이버 노트북2]

❗ROS2는 MQTT와 다르게 브로커가 없고, Peer-to-Peer(DDS) 구조이다.
그래서 네트워크 연결 조건이 조금 더 까다롭지만, 잘만 설정하면 MQTT보다 낮은 레이턴시를 보여준다

---

Ubuntu에서 ROS2 Humble 설치 (공식 방식, Ubuntu 22.04 기준)

 

📦 1단계: 설치 전 기본 설정

sudo apt update && sudo apt install locales
sudo locale-gen en_US en_US.UTF-8
sudo update-locale LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8
export LANG=en_US.UTF-8

 

🔑 2단계: ROS2 패키지 저장소 설정

sudo apt install software-properties-common
sudo add-apt-repository universe
sudo apt update && sudo apt install curl -y
sudo curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.asc | sudo tee /usr/share/keyrings/ros-archive-keyring.gpg > /dev/null

echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] \
https://packages.ros.org/ros2/ubuntu $(. /etc/os-release && echo $UBUNTU_CODENAME) main" | \
sudo tee /etc/apt/sources.list.d/ros2.list > /dev/null

 

📥 3단계: ROS2 Humble 설치

sudo apt update
sudo apt install ros-humble-desktop -y

📌 (디스크 여유 없을 경우 ros-humble-ros-base로 최소 설치도 가능)

 

4단계: ROS2 환경 변수 설정

echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc
source ~/.bashrc

 

설치 확인

ros2 --version
# 예: ros2 0.9.8 (humble)

ros2 run demo_nodes_cpp talker
# → talker 노드가 실행되면 성공!

 

colcon 및 의존 패키지 설치 (빌드용)

sudo apt install python3-colcon-common-extensions -y

---

✅ 1. Cyclone DDS 네트워크 통신 조건

기본 전제:

• ROS2 노드는 Cyclone DDS를 통해 멀티캐스트 UDP를 사용해서 서로 탐색
• 단, 네트워크 설정이 다르면 서로 못 찾을 수 있다

📌 꼭 지켜야 할 조건:

조건	설명
✅ 같은 서브넷 (예: 192.168.0.x)	Multicast 탐색 가능해야 함
✅ 방화벽 허용	UDP 7400 ~ 7500 포트 범위 열려야 함
✅ 브리지 네트워크 or 유선 연결	Parallels, VirtualBox 사용할 경우 “브리지 모드” 필수
✅ 동일 ROS2 버전 & RMW	둘 다 rmw_cyclonedds_cpp 사용하도록 설정

---

✅ 2. Cyclone DDS 통신 확인 실험

A. 퍼블리셔 노트북에서:

ros2 run demo_nodes_cpp talker

B. 서브스크라이버 노트북에서:

ros2 run demo_nodes_cpp listener

 

 정상 연결되면 listener 쪽에서 메시지 출력됨

[publisher]

[subscriber]

---

✅ 3. Cyclone DDS에서 통신이 안 될 경우 대처법

📁 ~/.ros/ros2/에 cyclonedds.xml 설정 파일 만들기

 

<CycloneDDS>
  <Domain>
    <General>
      <NetworkInterfaceAddress>192.168.0.101</NetworkInterfaceAddress> <!-- 본인의 IP -->
    </General>
    <Discovery>
      <Peers>
        <Peer address="192.168.0.102"/> <!-- 상대방 IP -->
      </Peers>
    </Discovery>
  </Domain>
</CycloneDDS>

그리고 환경변수 설정:

export CYCLONEDDS_URI=file:///home/devjang/.ros/ros2/cyclonedds.xml

또는 .bashrc에 추가!

---

✅ 4. ROS2 퍼포먼스 실험을 위한 코드 구성

 

역할	예시	파일설명
퍼블리셔	dds_pub.py	rclcpp::Publisher, dummy JSON 전송
서브스크라이버	dds_sub.py	수신 후 latency 계산, 손실률 측정 가능
빌드 시스템	colcon build + setup.py
시각화	visualize_result.ipynb로 동일하게 분석 가능

---

setup.py와 package.xml은 ROS2에서 패키지를 정의하고 빌드/배포할 때 사용하는 핵심 구성 파일들이다, 둘 다 중요한 역할을 하지만 용도와 포맷이 다르기 때문에 함께 써야한다.

[setup.py 코드]

Python 기반 ROS2 패키지의 설치 및 실행 스크립트

🔹 역할

• Python setuptools 기반으로 ROS2 노드를 정의
• 실행 가능한 스크립트(ros2 run ...)로 등록
• 설치할 파일, 이름, 버전, 실행 entrypoint 등 지정

from setuptools import setup

package_name = 'dds_test_py'

setup(
    name=package_name,
    version='0.1.0',
    packages=[package_name],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='Your Name',
    maintainer_email='you@example.com',
    description='Python ROS2 pub/sub test',
    license='Apache License 2.0',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'dds_pub = dds_pub.dds_pub:main',
            'dds_sub = dds_sub.dds_sub:main',
        ],
    },
)

🔺 주의: packages=[package_name] ← 이건 __init__.py 포함된 폴더 이름과 같아야 한다

 

---

[package.xml 코드]

ROS2가 패키지를 인식하고, 의존성/메타정보를 정의하는 파일

 

🔹 역할

• ROS2의 빌드 시스템(colcon)이 이 패키지를 인식하게 해줌
• 어떤 패키지인지 설명 (이름, 버전, 유지관리자, 라이선스 등)
• 어떤 ROS2 패키지나 라이브러리에 의존하는지 선언

<package format="3">
  <name>dds_test_py</name>
  <version>0.1.0</version>
  <description>Python ROS2 pub/sub test</description>
  <maintainer email="you@example.com">Your Name</maintainer>
  <license>Apache 2.0</license>
  <!-- 핵심: Python 패키지로 선언 -->
  <build_type>ament_python</build_type>
  <!-- 실행 시 필요한 의존성 -->
  <exec_depend>rclpy</exec_depend>
  <exec_depend>std_msgs</exec_depend>
</package>

---

✅ 둘의 관계 요약

항목	package.xml	setup.py
사용 대상	ROS2 시스템	Python/colcon 설치
기능	패키지 정보, ROS 의존성	Python 설치/실행 설정
필수 여부	✅ ROS2에서 반드시 필요	✅ Python ROS2 노드에 필요
비유	"책 표지와 목차"	"책 내용물의 설치 설명서"

---

함께 쓰일 때의 흐름

1. colcon build 명령 실행 시
   • package.xml 읽고 의존성 확인
   • Python 패키지인 경우 setup.py를 실행해 빌드/설치
2. 빌드 후:
   • ros2 run dds_test_py dds_pub → setup.py에서 등록한 main() 함수 실행

---

실행 예시

빌드:

colcon build --packages-select dds_test_py

 

소스:

source install/setup.bash

 

실행:

ros2 run dds_test_py dds_pub
ros2 run dds_test_py dds_sub

---

✅ 목표

노드	기능
dds_pub.py	id, timestamp, payload 포함 메시지 전송 + pub_log.csv 저장
dds_sub.py	수신 후 latency 계산 + sub_log.csv, loss_result.csv 저장

---

📁 폴더 구조 (기본)

ros2/
├── dds_pub/
│   ├── __init__.py
│   └── dds_pub.py
├── dds_sub/
│   ├── __init__.py
│   └── dds_sub.py
├── log/
│   ├── dds_pub_log.csv
│   ├── dds_sub_log.csv
│   └── loss_result.csv
├── setup.py
├── package.xml

---

[dds_pub.py (퍼블리셔) 코드]

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
import json, time, csv, os

class DDSPublisher(Node):
    def __init__(self):
        super().__init__('dds_publisher')
        self.publisher_ = self.create_publisher(String, 'dds_topic', 10)
        self.timer_period = 0.1  # 10Hz
        self.timer = self.create_timer(self.timer_period, self.timer_callback)
        self.msg_id = 0
        self.log_path = 'log/dds_pub_log.csv'
        os.makedirs('log', exist_ok=True)
        self.log_file = open(self.log_path, 'w', newline='')
        self.writer = csv.writer(self.log_file)
        self.writer.writerow(['id', 'timestamp', 'payload'])
        self.get_logger().info("DDS Publisher Started")

    def timer_callback(self):
        now = time.time()
        payload = {
            "id": self.msg_id,
            "timestamp": now,
            "data": {"sensor": "dds", "value": 123}
        }
        msg_str = json.dumps(payload)
        msg = String()
        msg.data = msg_str
        self.publisher_.publish(msg)

        self.writer.writerow([self.msg_id, now, msg_str])
        self.msg_id += 1

    def destroy_node(self):
        self.log_file.close()
        super().destroy_node()

def main(args=None):
    rclpy.init(args=args)
    node = DDSPublisher()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info('Keyboard interrupt → stopping.')
    finally:
        node.destroy_node()
        rclpy.shutdown()

---

[dds_sub.py (서브스크라이버 + latency + 손실률)]

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
import json, time, csv, os

class DDSSubscriber(Node):
    def __init__(self):
        super().__init__('dds_subscriber')
        self.subscription = self.create_subscription(
            String,
            'dds_topic',
            self.listener_callback,
            10)
        self.subscription  # prevent unused variable warning

        self.received_ids = set()
        self.log_path = 'log/dds_sub_log.csv'
        self.loss_path = 'log/loss_result.csv'
        os.makedirs('log', exist_ok=True)
        self.log_file = open(self.log_path, 'w', newline='')
        self.writer = csv.writer(self.log_file)
        self.writer.writerow(['id', 'recv_time', 'latency_ms'])
        self.get_logger().info("DDS Subscriber Started")

    def listener_callback(self, msg):
        try:
            now = time.time()
            raw = json.loads(msg.data)
            msg_id = int(raw.get("id", -1))
            sent_time = float(raw.get("timestamp", 0))
            latency = (now - sent_time) * 1000

            self.received_ids.add(msg_id)
            self.writer.writerow([msg_id, now, round(latency, 3)])
        except Exception as e:
            self.get_logger().error(f"Error decoding message: {e}")

    def destroy_node(self):
        self.log_file.close()
        self.calculate_loss()
        super().destroy_node()

    def calculate_loss(self):
        try:
            with open('log/dds_pub_log.csv', newline='') as f:
                reader = csv.DictReader(f)
                pub_ids = set(int(row['id']) for row in reader)
        except FileNotFoundError:
            self.get_logger().warn("dds_pub_log.csv not found")
            return

        total_sent = len(pub_ids)
        total_recv = len(self.received_ids)
        loss = total_sent - total_recv
        loss_rate = (loss / total_sent) * 100 if total_sent > 0 else 0

        with open(self.loss_path, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(["total_sent", "total_received", "loss_count", "loss_rate(%)"])
            writer.writerow([total_sent, total_recv, loss, round(loss_rate, 2)])

        self.get_logger().info(f"손실률: {loss_rate:.2f}% ({loss}/{total_sent})")

def main(args=None):
    rclpy.init(args=args)
    node = DDSSubscriber()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info('Keyboard interrupt → stopping.')
    finally:
        node.destroy_node()
        rclpy.shutdown()

---

현재 경로에는 아직 실험 결과 로그 파일들이 존재하지 않아서 시각화를 바로 진행할 수는 없으며 실험을 끝내고 로그 파일들만 생성하면 바로 쓸 수 있도록, visualize_result.ipynb를 구성하였다.

[visualize_result.ipynb 템플릿]

import pandas as pd
import matplotlib.pyplot as plt

# 파일 경로
pub_log_path = "log/dds_pub_log.csv"
sub_log_path = "log/dds_sub_log.csv"
loss_result_path = "log/loss_result.csv"

# 퍼블리셔 로그 (선택)
try:
    pub_df = pd.read_csv(pub_log_path)
    print("총 퍼블리시 수:", len(pub_df))
except:
    print("⚠️ 퍼블리셔 로그 없음")

# 서브스크라이버 로그 (latency)
sub_df = pd.read_csv(sub_log_path)

plt.figure(figsize=(10, 4))
plt.plot(sub_df["recv_time"], sub_df["latency_ms"], label="Latency (ms)")
plt.xlabel("Receive Time (s)")
plt.ylabel("Latency (ms)")
plt.title("ROS2 Latency Over Time")
plt.grid(True)
plt.legend()
plt.show()

# 손실률 출력
loss_df = pd.read_csv(loss_result_path)
print("\n📊 Message Loss Summary:")
display(loss_df)

---

실험 순서 & 명령어

터미널에서 ROS2 퍼블리셔 실행

ros2 run dds_test_py dds_pub

실험 종료하려면: Ctrl + C

 

다른 터미널에서 ROS2 서브스크라이버 실행

ros2 run dds_test_py dds_sub

 

실험 종료하려면: Ctrl + C

종료 시 자동으로 loss_result.csv 생성됨

로그 파일 확인

ls log/
# 예상: dds_pub_log.csv, dds_sub_log.csv, loss_result.csv

시각화

jupyter notebook

→ visualize_result.ipynb 열고 latency plot + 손실률 표 확인