[Physical AI W1D1] 5/5 — ROS 2 /clock을 밖으로: WebSocket 서버 + Cloudflare Tunnel

[Physical AI W1D1 · 5/5]

ROS 2 안에서만 돌던 /clock 토픽을, rclpy로 구독하고 WebSocket으로 송출하는 파이썬 스크립트 하나로 바깥에 내보낸다. websockets 버전 함정부터 Cloudflare Tunnel로 외부 주소를 뽑는 법까지 — 스크립트 한 줄씩 "무엇을·왜·어떤 결과"로 해설하는 Day1 5편(실습).

이 글에서 직접 해보는 것

• WebSocket용 파이썬 라이브러리 설치 (그리고 왜 버전을 고정하는가)
• ros_clock_ws_server.py — ROS 2 /clock을 구독해 WebSocket으로 내보내는 스크립트 작성·해설
• rclpy(로봇)와 asyncio(웹)를 한 프로세스에서 같이 돌리는 구조 이해
• cloudflared로 Colab 내부 포트를 외부에서 접속 가능한 주소로 노출
• 시뮬레이터 시간이 외부 브라우저까지 실시간으로 흐르는 전체 경로 검증

(이 글은 앞의 4편에서 이어집니다. 4편에서 ROS 2 Humble + Gazebo headless를 깔고 /clock 브리지까지 만들었다는 전제로 진행합니다.)

---

들어가며 — "ROS 2 안에서만 도는 데이터"의 한계

4편 끝에서 우리는 Gazebo의 시간을 ROS 2 /clock 토픽으로 끌어왔습니다. 그런데 이 데이터는 Colab 리눅스 안에서만 돕니다. 사람이 눈으로 보려면 결국 화면(브라우저) 까지 흘려보내야 하죠.

문제는 ROS 2와 브라우저가 말이 안 통한다는 겁니다. 브라우저는 ROS 2 토픽을 직접 못 읽습니다. 그래서 둘 사이에 번역기 겸 중계소가 필요합니다. 그 역할을 하는 것이 이번 편의 주인공, WebSocket 서버 스크립트입니다.

ROS 2 /clock 토픽
      │  ① rclpy로 구독 (로봇 쪽 언어)
      ▼
ros_clock_ws_server.py   ← 이번 편에서 만드는 스크립트
      │  ② JSON으로 변환해 WebSocket으로 송출 (웹 쪽 언어)
      ▼
Cloudflare Tunnel        ← ③ 외부에서 접속 가능한 wss:// 주소로 노출
      ▼
( 브라우저 )             ← 종착지 (HTML 화면은 이 글 범위 밖)

💡 WebSocket이 뭔가요? — 보통 웹 요청은 "물어보면 한 번 답하고 끝"입니다. WebSocket은 한 번 연결해두면 서버가 계속 데이터를 밀어 넣을 수 있는 통로입니다. 시계처럼 쉴 새 없이 갱신되는 값을 실시간으로 보낼 때 딱 맞습니다.

이번 편은 리눅스 명령어 + 파이썬 스크립트 중심입니다. 브라우저에 그릴 HTML 화면은 이 글 범위에서 빼고, "데이터를 어떻게 ROS 2 밖으로 꺼내 외부 주소까지 뽑아내는가" 에 집중합니다.

---

1. WebSocket 라이브러리 설치 — 그리고 버전을 고정하는 이유

먼저 파이썬용 WebSocket 라이브러리를 설치합니다. 그냥 최신을 받지 않고 버전을 콕 집어 설치하는 게 포인트입니다.

pip install websockets==10.4

• 무엇을: 파이썬에서 WebSocket 서버를 띄울 수 있는 websockets 라이브러리를 10.4 버전으로 고정 설치합니다.
• 왜 하필 10.4인가: 실습 중 더 낮은 버전(9.1)에서 asyncio.Lock 관련 호환 문제가 발생했습니다. 10.4에서 안정적으로 동작하는 것을 확인했기 때문에 버전을 못 박습니다.
• 기대 결과: 설치 완료 후 python3 -c "import websockets; print(websockets.__version__)" → 10.4.

⚠️ 흔한 함정 ① — "버전 고정"의 가치 — 라이브러리는 메이저 버전이 바뀌면 함수 시그니처(예: 핸들러가 받는 인자)가 달라지기도 합니다. 실습/튜토리얼 코드는 검증된 버전에 고정해두는 게 "내 컴퓨터에선 됐는데?"를 막는 가장 싼 보험입니다.

---

2. ros_clock_ws_server.py — 핵심 스크립트 작성

이제 이번 실습의 심장인 스크립트를 만듭니다. 하는 일은 한 문장으로:

ROS 2 /clock 토픽을 구독해서, 들어오는 시간 값을 JSON으로 바꿔 WebSocket으로 연결된 모두에게 0.2초마다 뿌린다.

파일을 엽니다.

nano ros_clock_ws_server.py

• 무엇을: nano는 터미널 안에서 쓰는 가벼운 텍스트 편집기입니다. 아래 코드를 붙여넣고 저장(Ctrl+O → Enter)한 뒤 종료(Ctrl+X)합니다.
• 왜: Colab 터미널엔 GUI 편집기가 없으니, 터미널 안에서 바로 파일을 만드는 nano가 가장 간단합니다.

전체 코드입니다.

import asyncio
import json
import threading
import time

import websockets

import rclpy
from rclpy.node import Node
from rosgraph_msgs.msg import Clock


clients = set()

latest_clock = {
    "sec": 0,
    "nanosec": 0,
    "received_at": time.time()
}


class ClockSubscriber(Node):
    def __init__(self):
        super().__init__("clock_ws_bridge_node")

        self.create_subscription(
            Clock,
            "/clock",
            self.clock_callback,
            10
        )

    def clock_callback(self, msg):
        latest_clock["sec"] = int(msg.clock.sec)
        latest_clock["nanosec"] = int(msg.clock.nanosec)
        latest_clock["received_at"] = time.time()


async def websocket_handler(websocket, path):
    clients.add(websocket)
    print("browser connected path=" + str(path), flush=True)

    try:
        await websocket.send(json.dumps({
            "type": "system",
            "message": "ROS 2 /clock WebSocket connected"
        }))

        async for message in websocket:
            print("browser message: " + str(message), flush=True)

    except Exception as e:
        print("websocket error: " + repr(e), flush=True)

    finally:
        clients.discard(websocket)
        print("browser disconnected", flush=True)


async def broadcast_clock():
    while True:
        if clients:
            payload = json.dumps({
                "type": "clock",
                "sec": latest_clock["sec"],
                "nanosec": latest_clock["nanosec"],
                "received_at": latest_clock["received_at"]
            })

            for ws in list(clients):
                try:
                    await ws.send(payload)
                except Exception:
                    clients.discard(ws)

        await asyncio.sleep(0.2)


def run_ros():
    rclpy.init()

    node = ClockSubscriber()
    print("ROS 2 /clock subscriber started", flush=True)

    try:
        rclpy.spin(node)
    finally:
        node.destroy_node()
        rclpy.shutdown()


async def main():
    ros_thread = threading.Thread(target=run_ros, daemon=True)
    ros_thread.start()

    async with websockets.serve(
        websocket_handler,
        "0.0.0.0",
        9002,
        ping_interval=None,
        compression=None
    ):
        print("WebSocket server running on 0.0.0.0:9002", flush=True)
        await broadcast_clock()


asyncio.run(main())

길어 보이지만, 다섯 덩어리로 끊어 보면 단순합니다.

① 공유 저장소 — latest_clock과 clients

clients = set()
latest_clock = {"sec": 0, "nanosec": 0, "received_at": time.time()}

• 무엇을: clients는 현재 접속한 WebSocket 연결들의 집합, latest_clock은 가장 최근에 받은 시간 값을 담아두는 공용 변수입니다.
• 왜: ROS 2에서 받은 값을 어딘가에 적어두고, WebSocket 쪽이 그걸 읽어 뿌리는 구조입니다. 이 두 변수가 로봇 쪽과 웹 쪽을 잇는 칠판 역할을 합니다.

② ROS 2 구독자 — ClockSubscriber

class ClockSubscriber(Node):
    def __init__(self):
        super().__init__("clock_ws_bridge_node")
        self.create_subscription(Clock, "/clock", self.clock_callback, 10)

    def clock_callback(self, msg):
        latest_clock["sec"] = int(msg.clock.sec)
        latest_clock["nanosec"] = int(msg.clock.nanosec)
        latest_clock["received_at"] = time.time()

• 무엇을: ROS 2 노드를 하나 만들어 /clock 토픽을 구독합니다. 메시지가 올 때마다 clock_callback이 호출되어 latest_clock에 최신 값을 적습니다.
• 왜: 이게 바로 rclpy(ROS 2의 파이썬 라이브러리)로 토픽을 받아오는 표준 방법입니다. 마지막 인자 10은 큐 크기(처리 못 한 메시지를 몇 개까지 쌓아둘지)입니다.
• 결과: 4편에서 만든 /clock이 흐르고 있으면, latest_clock의 sec/nanosec이 실시간으로 갱신됩니다.

③ 연결 처리 — websocket_handler

async def websocket_handler(websocket, path):
    clients.add(websocket)
    await websocket.send(json.dumps({"type": "system", "message": "ROS 2 /clock WebSocket connected"}))
    async for message in websocket:
        ...   # 클라이언트가 보낸 메시지 출력
    # 끝나면 clients에서 제거

• 무엇을: 새 클라이언트(브라우저)가 접속하면 clients에 등록하고, "연결됐다"는 시스템 메시지를 한 번 보냅니다. 연결이 끊기면 finally에서 깔끔히 제거합니다.
• 왜: 누가 접속해 있는지 알아야 그들에게만 데이터를 뿌릴 수 있습니다. 접속/해제를 add/discard로 관리하는 게 그 명단 관리입니다.

④ 송출 루프 — broadcast_clock

async def broadcast_clock():
    while True:
        if clients:
            payload = json.dumps({"type": "clock", "sec": ..., "nanosec": ...})
            for ws in list(clients):
                await ws.send(payload)
        await asyncio.sleep(0.2)

• 무엇을: 0.2초마다(초당 5번) latest_clock을 JSON으로 포장해 접속자 전원에게 보냅니다.
• 왜 JSON인가: 브라우저(자바스크립트)가 가장 다루기 쉬운 형식이 JSON입니다. ROS 2의 메시지 객체를 그대로는 못 보내니, {"sec":..., "nanosec":...} 같은 순수 텍스트 데이터로 번역해 내보냅니다. type 필드로 "시스템 알림"인지 "시계 값"인지 구분합니다.
• 왜 0.2초인가: 너무 자주 보내면 네트워크/CPU 낭비, 너무 뜸하면 끊겨 보입니다. 0.2초(5Hz)는 사람 눈에 부드러우면서 부담 없는 타협점입니다.

⑤ 두 세계를 한 프로세스에서 — run_ros + main

def run_ros():
    rclpy.init()
    node = ClockSubscriber()
    rclpy.spin(node)        # ROS 2 메시지를 계속 받는 무한 루프(블로킹)

async def main():
    ros_thread = threading.Thread(target=run_ros, daemon=True)
    ros_thread.start()      # ROS 2는 별도 스레드에서
    async with websockets.serve(websocket_handler, "0.0.0.0", 9002, ...):
        await broadcast_clock()   # WebSocket은 메인 asyncio에서

asyncio.run(main())

• 여기가 이 스크립트의 가장 영리한 부분입니다. ROS 2의 rclpy.spin()은 "메시지를 계속 받느라 멈춰 있는(blocking)" 무한 루프입니다. 반면 WebSocket 서버는 asyncio(비동기) 위에서 돕니다. 블로킹 루프와 비동기 루프는 한 줄에서 같이 못 돕니다.
• 해결: ROS 2 쪽(run_ros)을 별도 스레드(threading.Thread)로 떼어 돌리고, 메인 자리는 asyncio(WebSocket)가 차지합니다. 둘은 앞서 만든 공용 변수 latest_clock을 통해 칠판에 적고/읽는 방식으로 데이터를 주고받습니다.
• daemon=True는 "메인이 끝나면 이 스레드도 같이 종료"라는 뜻이라 깔끔하게 닫힙니다.
• 0.0.0.0은 "이 머신의 모든 네트워크 인터페이스에서 접속을 받겠다"는 의미로, 다음 단계에서 터널이 붙으려면 필요합니다. 포트는 9002를 씁니다.

💡 핵심 통찰 — 피지컬 AI 시스템은 거의 항상 성격이 다른 두 세계(로봇 제어 루프 + 외부 통신/UI) 를 한 시스템에서 굴립니다. 이 스크립트의 "ROS는 스레드, 웹은 asyncio, 공유 변수로 연결" 패턴은 그 축소판입니다.

---

3. 서버 실행 — 스크립트 띄우기

작성한 스크립트를 실행합니다. ROS 2 환경을 다시 잡아주는 게 포인트입니다.

pkill -f ros_clock_ws_server.py          # 혹시 이전에 띄운 게 있으면 정리
source /opt/ros/humble/setup.bash        # ROS 2 환경 다시 로드
python3 ros_clock_ws_server.py > ros_clock_ws.log 2>&1 &
sleep 3
ss -tunlp | grep 9002                    # 9002 포트가 열렸는지 확인
tail -n 20 ros_clock_ws.log              # 로그로 정상 기동 확인

• 무엇을 / 왜:
  • pkill -f ... — 같은 스크립트가 이미 돌고 있으면 포트 충돌이 나므로 먼저 종료합니다.
  • source ...setup.bash — 4편 흔한 함정 ②와 같은 이유. 백그라운드로 띄우기 전에 이 셸에서 rclpy가 보이도록 ROS 2 환경을 다시 잡습니다. 이걸 빼면 import rclpy에서 죽습니다.
  • > ros_clock_ws.log 2>&1 & — 출력을 로그 파일로 보내고 &로 백그라운드 실행합니다. 터미널을 계속 쓰기 위해서입니다.
  • ss -tunlp | grep 9002 — 9002 포트가 LISTEN 상태인지 확인합니다(서버가 실제로 떴다는 증거).
  • tail ... log — 기동 로그를 봅니다.
• 기대 결과(정상 기준): 로그에 다음 두 줄이 보여야 합니다.

ROS 2 /clock subscriber started
WebSocket server running on 0.0.0.0:9002

⚠️ 흔한 함정 ② — 포트 충돌 — ss에 9002가 안 보이거나 로그에 "address already in use"가 뜨면, 이전 프로세스가 포트를 쥐고 있는 겁니다. pkill -f ros_clock_ws_server.py로 정리하고 다시 실행하세요. (실습 원본에서도 HTTP 서버가 8080↔8081 포트 충돌이 나서 포트를 바꿔 해결한 사례가 있습니다.)

---

4. Cloudflare Tunnel — Colab 내부 포트를 외부로 노출

서버는 떴지만 아직 Colab 안에서만 접근됩니다. 외부 브라우저가 붙으려면 공개 주소가 필요한데, Colab은 외부에서 직접 접속할 수 있는 공인 IP/포트를 열어주지 않습니다. 여기서 Cloudflare Tunnel이 등장합니다.

pkill -f "cloudflared tunnel --url http://localhost:9002"
nohup cloudflared tunnel --url http://localhost:9002 > ros_clock_tunnel.log 2>&1 &
sleep 8
tail -n 20 ros_clock_tunnel.log

• 무엇을: cloudflared가 Colab 내부의 localhost:9002(우리 WebSocket 서버)를 외부에서 접속 가능한 임시 공개 주소와 연결해줍니다.
• 왜 터널인가: 방화벽/NAT 뒤에 숨은 내부 서버를, 포트 개방이나 공인 IP 없이 바깥으로 안전하게 꺼내주는 방식입니다. Cloudflare의 빠른 터널(--url)은 별도 계정 설정 없이 일회용 주소를 즉석에서 만들어줍니다.
• 기대 결과: 로그에 다음과 비슷한 주소가 생성됩니다.

Your quick Tunnel has been created!
https://judicial-nobody-defendant-spaces.trycloudflare.com

이 주소가 HTTPS(https://) 로 나오지만, WebSocket으로 접속할 땐 앞부분을 wss:// 로 바꿔 씁니다(보안 WebSocket).

wss://judicial-nobody-defendant-spaces.trycloudflare.com

💡 여기서부터가 "브라우저로 나가는 단계" — 이 wss:// 주소가 바로 데이터가 ROS 2 밖으로 나가는 출구입니다. 외부 브라우저에서 이 주소로 WebSocket을 열면, 우리 서버가 0.2초마다 보내는 /clock JSON을 실시간으로 받게 됩니다. (브라우저에 띄울 HTML 대시보드 화면은 이 글 범위 밖이라 여기까지만 다룹니다.)

---

5. 전체 흐름 점검 — 시간이 끝까지 흐르는가

지금까지 만든 조각을 이으면, 이번 실습 전체 구조는 다음과 같습니다.

Gazebo headless                (4편)
  └─ /clock 생성
ros_gz_bridge                  (4편)
  └─ Gazebo /clock → ROS 2 /clock
ros_clock_ws_server.py         (5편)
  ├─ rclpy로 /clock 구독
  └─ JSON으로 변환해 WebSocket(9002)으로 송출
Cloudflare Tunnel              (5편)
  └─ 9002를 wss:// 공개 주소로 노출
( 브라우저 )                    ← 종착지 (HTML 범위 밖)

명령줄에서 확인 가능한 성공 기준만 추리면 이렇습니다(브라우저 화면 단계는 제외).

단계	확인 명령	성공 기준
WebSocket 라이브러리	python3 -c "import websockets, websockets.__version__"	10.4
서버 기동	tail -n 20 ros_clock_ws.log	subscriber started + running on 0.0.0.0:9002
포트 열림	ss -tunlp | grep 9002	9002 LISTEN
터널 생성	tail -n 20 ros_clock_tunnel.log	trycloudflare.com 주소 출력
/clock 유입	ros2 topic echo /clock	sec, nanosec 값 증가

이 모든 칸이 통과하면, Gazebo가 만든 시뮬레이션 시간이 → ROS 2 토픽이 되고 → 파이썬 스크립트가 JSON으로 번역해 → 외부에서 접속 가능한 wss:// 주소로 실시간 송출되는 파이프라인이 완성된 것입니다.

---

마무리 — 우리가 검증한 것, 그리고 다음 확장

이번 5편에서 명령어·스크립트로 검증한 것:

• websockets는 검증된 버전(10.4)으로 고정 — 버전 함정 회피
• ros_clock_ws_server.py에서 rclpy(로봇) + asyncio(웹) 를 한 프로세스에 공존시키는 패턴(스레드 분리 + 공유 변수)
• 0.0.0.0:9002로 띄우고 ss/tail로 기동을 객관적으로 확인
• cloudflared로 내부 포트를 외부 wss:// 주소로 노출
• 시뮬레이터 시간이 ROS 2 밖, 외부 주소까지 실시간으로 흐름

이 구조의 진짜 가치는 /clock을 다른 토픽으로 바꾸기만 하면 그대로 재사용된다는 점입니다. 다음 차시의 확장 방향:

• /clock 대신 /robot_status(로봇 위치·방향·센서 상태) 송출
• 브라우저 → 서버 방향으로 /cmd_vel 같은 제어 명령 전송(양방향)
• 웹 버튼으로 ROS 2 제어 명령을 쏘는 원격 조종 대시보드
• 물류 선별 로봇 등 실제 작업 로봇 대시보드로 확대

즉, 이번에 만든 "ROS 2 ↔ WebSocket ↔ 외부" 파이프라인은 앞으로 만들 모든 웹 기반 로봇 모니터링·원격제어의 뼈대가 됩니다.

📚 Week1 Day1 전체 목차 (총 5편)

• 1/5 화면 밖으로 나온 AI — 피지컬 AI란 무엇인가 (개념)
• 2/5 휴머노이드 로봇 전쟁 — 테슬라·보스턴 다이내믹스·중국 비교 (개념)
• 3/5 무엇을 배워야 하나 — ROS 2부터 Sim-to-Real까지 로드맵 (개념)
• 4/5 Colab에서 ROS 2 Humble + Gazebo headless 환경 만들기 (실습)
• 5/5 ROS 2 /clock을 밖으로 — WebSocket 서버 + Cloudflare Tunnel — 이번 글 (실습)