Global Extended Remote Robot Interface
한 로봇을 제어하는 온-로봇 제어 소프트웨어.
synapse 표준을 import하고 자기 일(단일 로봇)만 한다 — 플릿(Fleet)은 모른다.
Base(게이트웨이) · Sub(Capability 구현) · Mission(시퀀스)의 3계층으로 구성된다.
물리 HW(TOM)·외부 통신 데몬(AND)·제어 SW(GERRI) 셋이 모여 한 로봇 분(分)의 묶음을 이룬다. 이 레포 tom_and_gerri는 그 중 AND 구현 + GERRI를 담는다.
tom_and_gerri는 버전 0.12.0이다(스택 세대 정렬 — MINOR는 전 레포가 함께 올린다). 의존성은 synapse>=0.0.1·pypubsub>=4.0이며, 무거운 백엔드는 extras로 분리한다 — [and](rtc2kng WebRTC: aiortc·av·opencv·websockets·requests)·[and-nexus](nexus 직결 WS)·[and-rubberneck](ketirtc 미디어), [rmi](fastapi·uvicorn)·[cockpit](fastapi·uvicorn), [vision](anthropic)·[obb](numpy — OBB·파지계획)·[detect](numpy·ultralytics — YOLO 디텍션 풀스택)·[voice](sounddevice·edge-tts·faster-whisper — 음성 인터페이스)·[mcap](raasdata — 상시 녹화)·[camera]·[ik](pinocchio)·[dev](pytest). 로봇별 코드는 tag_{robot} 별도 레포로 분리된다.Figure 1. TAG 묶음. 로봇 하나는 TOM·AND·GERRI 셋의 합으로 정의된다. GERRI는 synapse만 의존하고, 전송 방식(AND)·로봇 SDK는 바깥의 교체 가능한 세부사항이다.
각 계층은 명확한 단일 책임을 가진다. Sub는 외부 통신을 전혀 모르고, Base가 외부와의 유일한 창구(게이트웨이)다.
Topic.Robot.STATUS 구독 → robot_id를 채워 외부 Topic.Network.SEND로 내보낸다.Topic.Network.RECV 수신 → receiver(target) 기준으로 자기/해당 sub에 라우팅한다.@property capability — 자기 + 모든 sub의 {"common":[…], "custom":[…]}를 종합한다.stop() — 진행 미션을 abort하고 각 sub의 stop()을 호출한다(라우팅 분기 최우선 매칭).pub.sendMessage(Topic.Robot.STATUS, data=self.get_status()). 외부 토픽 사용 금지.self.capability(공용 enum) + self.custom_capability(메서드 이름 list)를 선언한다.stop()은 모든 로봇이 반드시 구현 — 베이스는 NotImplementedError를 던진다.dispatch와 sender를 주입한다.wait_input / wait_teleop(NEED_HELP → TELEOP) 처리. "실패는 정상, 인간 개입 우선."Figure 2. 3계층 데이터 흐름. 외부 통신은 오직 Base를 통한다. Sub는 내부 버스(pypubsub)로만 status를 올리고, Base가 게이트웨이에서 robot_id를 채워 외부로 중계한다 — 의존 방향은 안쪽(synapse)을 향한다.
통신 단위 하나당 하나의 Base가 존재한다. __init__(info)는 info["id"]를 요구하며, 이것이 robot_id(외부 식별자)가 된다.
Base는 생성 시 세 토픽을 구독해 게이트웨이를 연다 — 내부 Robot.STATUS(sub→base), Mission.EVENT(runner→base), 외부 Network.RECV(AND→base). 게이트웨이로서 sub의 내부 status에는 robot_id를 채워 외부 Network.SEND로 중계하고, 외부 메시지는 receiver 값으로 자기/해당 sub에 라우팅한다.
| 멤버 | 책임 |
|---|---|
| robot_id | info["id"]에서 받은 외부 식별자. 게이트웨이에서 sub status·외부 메시지에 채워진다. |
| sub_controllers | 항상 dict. 등록된 Sub들의 모음으로, receiver(sub_id) 기준 라우팅 대상이 된다. |
| missions | name → RobotMission 카탈로그. NEXUS가 이름으로 트리거한다. |
| mission_runner | MissionRunner(dispatch=self.dispatch_capability, sender=self.robot_id) — base가 디스패치와 sender를 주입한다. |
| teleop_handler | 선택 — command.teleop를 해석하는 핸들러. |
| @property capability | self + 모든 sub의 {"common":[…], "custom":[…]}를 종합해 status에 노출한다. |
| dispatch_capability | (capability, params=None, target=None) — capability 이름으로 메서드를 getattr 조회·호출해 Result 반환. target(sub_id) 지정 시 그 sub로(예: 양팔 left/right), 미지정이면 base 자신 → 첫 매칭 sub 순. |
| software_update | 전 로봇 공통 유지보수 동작(base가 광고). 워크스페이스 형제 레포를 git pull(synapse.update) 후 재시작한다 — pull은 워커 스레드에서 돌려 status/ping 동결을 막고, 재진입 락으로 이중 실행(동시 pull)을 차단한다. BUSY/TELEOP·미션 중엔 거부, 변경 없으면 재시작 안 함. 매 status의 robot_info.version에 배포 semver가 자동 주입되고, 재시작은 systemd(SYNAPSE_SYSTEMD_UNIT)나 os.execv 자기재실행. |
| command_device | 로봇이 다른 노드(엘리베이터·자동문 등 IoT, 타 로봇)에 표준 capability 명령을 보낸다 — 이미 연결된 Network.SEND(→AND→NEXUS)로 표준 봉투를 실으면 서버가 대상 WS로 중계한다(별도 HTTP 불필요). 피드백은 응답 토픽이 아니라 대상의 다음 robot.status 관측이 SSOT다(BUSY→IDLE=스텝 완료). |
로봇 SDK를 직접 호출하는 곳으로, __init__(info)는 info["id"](=sub_id)를 요구한다. status는 get_status()로 만들어 내부 토픽으로만 발행하고, robot_id는 Base가 게이트웨이에서 채운다.
from gerri import SubControllerCommon
from synapse import Capability, Result
class TemplateSub(SubControllerCommon):
def __init__(self, info, sdk=None):
super().__init__(info)
self._sdk = sdk or _FakeSDK()
self.capability = [
Capability.NAVIGATE_TO_POSE,
Capability.VELOCITY_CTRL,
Capability.DOCK, Capability.UNDOCK,
Capability.STOP, # ← 필수
]
def navigate_to_pose(self, x, y, theta=0.0, timeout_s=30.0) -> Result:
"""좌표로 이동. x,y(미터, map frame) · theta(도, deg) · timeout_s(초)."""
try:
if self._sdk.move_to(x, y, theta):
return Result.ok(message="도착", reached=[x, y, theta])
return Result.fail("not_reached", message="도달 실패")
except TimeoutError:
return Result.fail("timeout", message=f"{timeout_s}초 초과")
def velocity_ctrl(self, linear, angular) -> Result:
"""직접 속도 제어. linear(m/s) · angular(deg/s)."""
self._sdk.set_velocity(linear, angular)
return Result.ok()
def dock(self, charger_id=None) -> Result: return Result.ok(message="도킹")
def undock(self) -> Result: return Result.ok()
def stop(self) -> Result:
"""필수 — 로봇을 IDLE 로."""
self._sdk.halt()
return Result.ok(message="정지")
Figure 3. 실제 TemplateSub 구현(scaffold). 한국어 docstring에 단위·범위·좌표계를 명시하는 것이 컨벤션이다. 모든 메서드는 synapse Result를 반환하고, stop()은 예외 없이 구현된다.
capability 값이 곧 메서드 이름이다. Base는 getattr(sub, name)(**params)로 자동 매칭해 호출한다 — 분기(if robot == …) 없이 라우팅이 성립한다.
synapse.Capability enum으로 정의된 표준 동작. 시그니처가 고정되어 어떤 로봇이든 같은 인터페이스로 부른다.self.custom_capability(메서드 이름 문자열 list)로 선언. 시그니처가 자유로워 로봇 특수 기능을 노출한다. 일부는 Sub 공통으로 자동 제공된다(아래 파란 칩).relocate는 표준 pose 보정 — 로봇 pose 추정치를 자기 map 프레임에서 (x,y,theta)로 보정하되, 맵 정렬 T(서버 소유)는 건드리지 않는다. pose를 리포트하는 sub만 자동 광고하고, 정적(고정) sub는 그 보정을 robot.config.json에 영속한다. software_update는 base가 전 로봇 공통으로 광고하는 유지보수 동작이다. custom 쪽 set_status_rate(적응형 status 주기)·set_footprint(실측 크기)·set_pose(표시 pose 배치)·speak(발화)·ai_chat(대화)은 SubControllerCommon이 공통 제공한다. ⚠️ 뒤의 둘은 조건부 광고다 — speak는 로봇이 publish_speech를 오버라이드해 발화 슬롯을 가졌을 때만, ai_chat은 러너가 attach_chat을 붙였을 때만 광고된다. 광고해놓고 못 하면 콘솔에 누를 수 있는 죽은 버튼이 생긴다. 그래서 정본은 sub.advertised_capability() 하나다 — 선언한 것과 광고되는 것이 갈라지면 두뇌 화이트리스트가 화이트리스트 ∩ 광고라 넣은 항목이 조용히 탈락한다(오류도 경고도 없이 로봇이 그냥 말을 안 한다).
and-rx-<tag>) — 오퍼레이터가 2명이면 같은 sub의 capability가 동시에 들어온다. 직렬화하는 장치가 아무 데도 없었고, 실측으로 6스레드가 같은 capability를 동시에 돌렸다. _capability_lock(RLock)이 sub 디스패치와 base 자기 capability를 같은 락으로 묶는다 — 재진입 가능이어야 하는 이유는 capability 안에서 다른 capability를 부르는 경로(미션·두뇌)가 같은 스레드로 들어오기 때문이다(비재진입 Lock이면 교착). ⛔ STOP은 이 락을 타지 않는다 — _handle_for_self가 dispatch 전에 분기한다. 느린 capability 뒤에서 기다리는 STOP은 STOP이 아니다.
Result.ok(message="", **data), 실패는 Result.fail(error, message="")이다. @property capability는 self + 모든 sub의 것을 {"common":[…], "custom":[…]}로 종합해 status에 노출한다 — 관제 UI는 이 목록만 보고 무엇을 시킬 수 있는지 안다.
자료구조(Task·RobotMission)는 synapse.missions, 상태/이벤트는 synapse.enums에 있다. Runner는 실행만 하고, mission_state는 runner가 소유한다(edge-authoritative).
dispatch(task.capability, task.params) → Resulttask.func() 실행 — 테스트·빠른 개발Figure 4. Task 5타입. task.policy.on_failure는 stop(미션 FAILED) / continue(다음 task) / retry(retry_count만큼 재시도) / help(사람 개입 요청 → NEED_HELP·TELEOP, 무응답 timeout 시 FAILED 폴백)를 결정한다. 실패는 정상으로 취급된다.
개입 진입 두 경로(A4): ① task 실패(on_failure=help)나 wait_teleop 호출 → RUNNING → NEED_HELP → TELEOP, ② 오퍼레이터 자발 takeover(HelpAction.TAKEOVER) → 다음 task 경계에서 TELEOP 인수. 두 경로 모두 개입 세션으로 합류한다.
세션에서는 태스크마다 TeleopExitAction 5종으로 해소한다 — resume_done·skip=다음 task, resume_retry=로봇에 재위임, mission_done=남은 전부 완료→SUCCEEDED, abort=미션 ABORTED. 무응답 timeout은 정책 폴백(FAILED).
Figure 5. 상태기계와 인간 개입. 모니터/UI는 status(TELEOP)와 event(TELEOP_STARTED/ENDED)로 operator 개입 구간을 감지한다. STOP은 abort() → 다음 경계에서 ABORTED.
from synapse.missions import RobotMission, cap, notify, wait_input
def pickup_with_confirm() -> RobotMission:
"""이동 → 사람 확인(계획된 입력 대기) → 복귀."""
return RobotMission(name="pickup_with_confirm", description="확인 후 픽업", tasks=[
cap("go", "navigate_to_pose", x=2.0, y=0.0),
wait_input("confirm", prompt="집을까요?",
buttons={"yes": "예", "no": "아니오"}),
cap("home", "navigate_to_pose", x=0.0, y=0.0),
])
Figure 6. 실제 RobotMission 예시. cap·wait_input·notify 헬퍼로 Task 시퀀스를 간결히 정의한다. 미션은 Base의 self.missions 카탈로그에 등록되어 NEXUS가 트리거할 수 있다.
available_missions가 종전에 dict 키(name)만 실어, 한국어 이름이 이미 있는데도 발행 경계에서 버려졌다 — 콕핏이 fetch — 물건을 집어서…를 띄웠고 설명은 한국어인데 제목만 코드 식별자였다(사용자가 fetch를 “패치”로 읽었다). 지금은 {name, label, description}이다. ⚠️ name은 바꾸지 않는다 — mission.trigger가 그 값으로 되쏘기 때문이고, 표시용은 따로 싣는다(label이 식별자와 같으면 빈 문자열 — 같은 말을 두 번 쓰지 않는다). 두뇌는 이미 mission_specs로 같은 규약을 쓰고 있었다 — 콘솔만 못 받고 있었다.synapse는 표준 입력 인덱스만 정하고, 의미 해석은 핸들러가 한다. 모두 통일 시그니처 handle(axes, buttons, poses, target=None)를 따른다. 핸들러는 dispatch(capability, params)만 알면 된다.
velocity_ctrl, B=STOP.home_position, B=STOP, 트리거(L2/R2)=gripper_ctrl(0~100%).synapse.teleop_mapping의 JoyAxis·JoyButton 인덱스를 읽는다. 입력은 sparse해도 되며 없는 인덱스는 0으로 친다.
poses[PoseChannel.LEFT/RIGHT_HAND](6DOF) → cartesian_ctrl, 아날로그 그립(axes) → gripper_ctrl(0~100%).target(receiver)으로 결정 — 별도 dual-arm 핸들러가 필요 없다.
mobile_manipulator.py — 합성/단독 팔 텔레옵. MobileManipulatorTeleopHandler는 한 command.teleop 번들의 축을 두 sub로 분배한다 — WASD/좌스틱은 base velocity_ctrl로, 마우스(팬/틸트/휠)는 arm 관절 증분으로 흘려 모바일 매니퓰레이터를 동시에 몬다(B=STOP 양쪽). ManipulatorTeleopHandler는 이를 상속해 velocity_ctrl 분배만 빼고 단독 팔(고정 매니퓰레이터)만 제어한다.
표준이 있으니 scaffold를 복사해 SDK만 연결하면 된다. Base는 sub 등록·핸들러·미션 카탈로그만 구성하면, 게이트웨이·STOP·디스패치·미션 실행을 BaseControllerCommon이 제공한다(올인원).
gerri --init — 템플릿에서 레포 생성tag_{robot}/tag_{robot}/ 패키지 + 루트에 {robot}_robot.py·tests/·pyproject.toml·setup.sh. tag_template 토큰은 전부 자동 치환된다(실측 10개 파일 — 종전엔 사람이 pyproject 3곳 + setup.sh 3곳을 손으로 고쳐야 했다).self.capability list를 설정하고, capability 메서드에서 로봇 SDK를 호출한다. 반드시 stop() 구현, Result 반환, 네트워크 토픽은 절대 건드리지 않는다.self.sub_controllers = {"main": MySub(...)}로 등록. 선택적으로 self.teleop_handler = MobileJoyHandler(self.dispatch_capability), 그리고 self.missions 카탈로그 등록.RobotRunner로 배선 — 여기가 조용히 빠진다cleanup_errors로 남긴다. 로봇은 콜백·인자로만 받으므로 러너는 tag_*를 import하지 않는다. 문서만 주고 신규 로봇 3대를 만들게 한 실험에서 3/3이 남의 러너를 베꼈고 그 종료 경로의 except Exception: pass 4개까지 복제했다 — 선택 배선도 제각각 빠졌다(WebRTC 1/3 · RMI 1/3 · sync_description 0/3 = 콕핏에 영상·3D 트윈 없음).tag_template/base_controller.py — TemplateBase
from gerri import BaseControllerCommon
from gerri.handlers import MobileJoyHandler
from .missions import patrol_demo, pickup_with_confirm
from .sub_controller import TemplateSub
class TemplateBase(BaseControllerCommon):
def __init__(self, info):
super().__init__(info)
# 1:1 — sub_controllers 는 항상 dict
self.sub_controllers = {
"main": TemplateSub({
"id": info["id"],
"category": info.get("category", "mobile"),
"model": info.get("model"),
}),
}
# 텔레옵 핸들러(선택) — 게임패드로 주행
self.teleop_handler = MobileJoyHandler(self.dispatch_capability)
# 미션 카탈로그
self.missions = {
"patrol_demo": patrol_demo(),
"pickup_with_confirm": pickup_with_confirm(),
}
로봇별 구현은 SEPARATE 레포다. 현재는 gerri/ 밑에 단순히 git clone되어 있다(git submodule 아님). 각 레포는 tom_and_gerri와 synapse를 import한다.
| 경로 | 역할 |
|---|---|
| __init__.py | exports: BaseControllerCommon, SubControllerCommon |
| base_controller.py | Base — 게이트웨이 |
| sub_controller.py | Sub — capability 구현 |
| mission/runner.py | MissionRunner |
| handlers/ | joy.py · vr.py · mobile_manipulator.py (합성/단독 팔 텔레옵) |
| vision/ | 비전 공통 라이브러리 — ① VisionPicker + 교체형 VLMBackend(프레임→VLM→픽셀) · ② 디텍션→OBB→파지계획(detect.py·obb.py·grasp.py·pipeline.py) |
| interface/ | cockpit(1:1 WebRTC) + rmi(로컬 관리 웹 UI) |
| utils/camera/ | CameraSource 레지스트리 → WebRTC 비디오 트랙 |
| config_loader.py | robot.config.json override 로더(설치 마법사·영속 pose/footprint) |
| setup_wizard.py / .html | 설치 마법사(config 생성 UI) |
| tag_template/ | scaffold (복사 출발점) |
| tag_piper/ | 로봇별 별도 레포 (clone, submodule 아님) |
| tag_rby1/ | 로봇별 별도 레포 (clone) |
Figure 7. 레포 구조. 코어(gerri/)는 안정적으로 관리하고, 로봇별 tag_*는 독립 레포로 자유롭게 진화한다. utils/camera/는 webcam·realsense·dummy 소스를 레지스트리로 추상화해 WebRTC 비디오 트랙에 공급한다.
단일 로봇과 WebRTC P2P로 직결되는 1:1 정밀 조작 인터페이스다. 카메라·오디오·DataChannel·조이스틱을 한 화면에서 다룬다. 코드는 이 레포의 gerri/interface/cockpit/에 있다 — 별도 컴포넌트가 아니라 GERRI가 제공하는 운전석이다.
TOM의 standalone_server가 직접 정적 파일을 서빙하고, 로컬 /ws/{room} 릴레이로 시그널링한다. 서버 없이 현장에서 바로 로봇에 붙어 조작.
NEXUS 게이트웨이가 /{site}/{building}/{floor}/tag_{robot}/에서 콕핏을 서빙하고, NEXUS /ws/{room}가 시그널링을 중계한다. 대시보드에서 클릭 한 번으로 진입.
Figure 8. 두 가지 호스팅 모드. 어느 쪽이든 제어·미디어는 P2P로 흐르고 서버는 시그널링만 중계한다.
서버는 정적 index.html에 런타임 설정만 문자열 치환으로 주입한다. 빌드된 정적 파일 안의 마커를 window.__COCKPIT__ = {robotId, signalingUrl, mode} 한 줄로 바꿔치기하는 방식이라, 호스팅 모드에 따라 robot_id·시그널링 URL·mode만 달라진다.
index.html · cockpit.js · cockpit.css)는 순수 정적 파일이라 로직을 전혀 갖지 않는다. 주입된 window.__COCKPIT__ = {robotId, signalingUrl, mode} 객체만 읽고 동작하므로, 동일한 정적 파일이 Standalone·NEXUS 양쪽에서 그대로 재사용된다.
targets(room, peer, message)는 전송 계층과 무관한 순수 함수다. 메시지에 receiver가 있으면 그 peer에게만 1:1 전달하고, 없으면(예: "ready") 같은 room의 나머지 전부에게 broadcast한다. 내용은 해석하지 않고 JSON을 그대로 중계한다.
tom_and_gerri/tag 코드를 import하지 않는다. 콕핏 프론트는 데이터(정적 파일)일 뿐 — tag_{robot}/ 폴더가 있으면 그걸 서빙하고, 없으면 기본 콕핏으로 폴백한다. 로봇별로 자기 콕핏을 자유롭게 커스터마이즈할 수 있다.로봇 자체에서 도는 로컬 관리 웹 UI다 — 코드는 gerri/interface/rmi/에 있고, 외부 의존 0으로 오프라인에서 그대로 동작한다. 핵심은 이것이 NetworkBackend 구현이라는 점이다 — AdaptiveNetworkDaemon으로 감싸면 nexus와 동급의 "관제망 하나"로 합류한다(base·sub·synapse 코드 수정 0).
/ 대시보드/ux 현장 운영자/models/<name>Figure 9. RMI 라우트. connect()가 uvicorn을 데몬 스레드로 띄우고, WebSocket으로 status·mission_event를 화면에 push하며 teleop·capability·mission·help를 수신해 base 게이트웨이로 주입한다.
로봇 → 화면은 Network.SEND의 robot.status·mission.event를 그대로 push하고, 화면 → 로봇은 표준 봉투(command.teleop·command.capability·mission.trigger·help.action)를 _deliver로 주입해 — nexus 명령과 동일 경로로 처리된다. 즉 멀티 데몬 팬아웃/병합에 그대로 편입되어, 로컬 웹 UI가 nexus·rubberneck과 나란히 또 하나의 관제망이 된다. 설치는 extras [rmi](fastapi·uvicorn).
대시보드의 조작 콘솔은 현장에서 쓰던 기존 tag_rmi 콘솔의 듀얼 조이스틱 UI를 그대로 이식했다 — 좌/우 스틱이 각각 axes[0..1]·axes[2..3]에 실리고, 버튼 16개·키보드(WASD/IJKL)·전체화면↔PiP(드래그로 옮기는 떠 있는 창) 토글을 함께 제공한다. 다만 Tailwind CDN은 인라인 CSS로 바꿔 오프라인 100%를 지켰고, 전송은 표준 command.teleop 봉투(axes 8 · buttons 16)로 맞췄으며, 극성은 위/W = 전진(+) 규약을 따른다. /ux(현장 운영자)에는 미션 컨트롤 4버튼이 있다 — 일시중지(capability.pause)·이어하기(resume)·스킵(help.skip)·미션 중지(help.abort + stop)이며, 하드 긴급정지는 별도 버튼으로 남는다.
robot.config.json override(설치 마법사) 체계와 충돌하므로 제외한다. 템플릿은 자체 완결 단일 파일이라 오프라인 100%.
카메라 프레임에서 피킹 값을 뽑아내는 공통 라이브러리다 — 어떤 로봇이든 가져다 쓴다. 갈래는 둘이다 — ① VLM 픽커는 제로샷으로 픽셀 좌표를 내고([vision]), ② 디텍션→OBB→파지계획은 온디바이스 디텍터로 미터 단위 파지 자세까지 낸다([obb]·[detect]). 코드는 gerri/vision/이고, 무거운 의존은 전부 지연 임포트라 코어 로봇은 numpy·torch 없이도 돈다.
학습·가중치 없이 '무엇을 어디서 집을지'를 바로 얻는 경로다. 반환 좌표는 전부 픽셀 공간이고, 3D 변환·hand-eye는 로봇 몫으로 남긴다(프로바이더·카메라 중립).
analyze_frame(bgr) — 다운스케일 후 원본 좌표 복원. PickObject(center·grasp_angle_deg·bbox·thin_width_px·shape·approach)와 추천 best를 반환한다. center는 잡을 지점이지 기하 중심이 아니다. 교체형 VLMBackend를 주입받는 '모델 교체 시임(seam)'.pixel_to_point는 depth·intrinsics로 3D 점을 내는 선택 유틸, draw_result는 피킹 결과를 프레임 위에 bbox·중심·닫힘선으로 오버레이한다(데모·디버그).claude-sonnet-5, 집기 직전 확정은 claude-opus-5 1회(오집기 비용이 호출 1회 비용보다 훨씬 크다). 구조화 출력으로 스키마 준수 JSON을 보장한다. API 키는 ANTHROPIC_API_KEY.claude CLI(헤드리스)를 호출하는 구독 인증 백엔드 — anthropic SDK도, API 키도 불필요(로봇 PC에 CLI 설치·로그인 전제). 데모·개발용.Figure 10. 비전 피킹 구조. VisionPicker(계약)는 백엔드를 모르고, VLMBackend 인터페이스만 본다 — Claude(SDK)·Claude Code(CLI 구독)를 같은 계약으로 갈아끼우고, Gemini/GPT도 같은 방식으로 추가한다(로봇 코드 무변경).
VLM 왕복 없이 로봇 위에서 돌리는 실시간 경로다. 디텍터가 인스턴스 마스크를 주면 → OBB가 그 마스크를 베이스 3D로 역투영해 지향 경계상자를 뽑고 → 파지계획이 눕힘/세움을 판정해 파지점·자세·개폐폭을 계산으로 낸다. 셋 다 로봇 비의존이라 RBY1·Piper 등 어떤 매니퓰레이터든 그대로 재사용한다.
GraspPipeline.scan()Figure 11. 파지 파이프라인. scan(frame, deproject, plane_z, labels) 한 번이 {class, score, center_px, box2d, bbox, plan, feasible} 목록을 파지 가능·고신뢰 순으로 돌려준다. 카메라 역투영은 주입된 콜백(deproject(u,v) → [x,y,z])이라 extrinsic·광학 리맵은 로봇 소유로 남는다 — 마스크 없는 디텍션(박스만)은 OBB 산출 불가라 스킵.
.pt(개발)/.engine(Orin TensorRT) 둘 다. Jetson Orin 실측 49 FPS.labels가 곧 어휘다(["cup","snack box","tape"]). COCO 밖 물체를 이름으로 찾는다(실측: 과자박스·테이프가 YOLO26 미검출이던 원인). Orin 26.5 FPS.FakeDetector/FakeSegmenter는 torch 없는 테스트용.detect.backend(yolo26·yoloe·vlm_sam) 한 줄로 갈아끼운다 — 플래그를 늘리지 않는 config-driven 원칙. model은 백엔드 소유라, backend만 바꾼 config에 남아 있는 타 백엔드 기본 가중치는 자동으로 무시된다.Figure 12. 디텍터 브릿지. 셋 다 detect(frame_bgr, labels) → [Detection] 한 프로토콜만 지키므로 파이프라인 아래는 무엇이 꽂혀도 동일하다. 어휘를 미리 알면 YOLOE로 빠르게, 말로 이해해야 하면 VLM+SAM으로 한 번만.
OBB는 마스크 픽셀을 베이스 3D로 역투영한 뒤, 수평 footprint에 PCA를 걸어 장축·단축과 yaw_deg를, z 범위로 높이를 뽑는다 — {center, size, yaw_deg}가 전부이고 open3d 없이 순수 numpy로 끝난다(plane_z를 주면 책상면 아래 점을 잘라 마스크 가장자리의 테이블 혼입을 막는다). 이어서 plan_grasp가 그 상자만 보고 결정한다 — 납작하게 눕힌 물체는 위에서(top), 세워 길쭉하면 옆에서(side) 문다. 반환은 mode·target·orient{rx,ry,rz}·grip_width·feasible·reason이고, 판정 기준(standing_ratio)·그리퍼 한계(gripper_max_width)·손가락 개폐축 캘리브(finger_offset_deg)는 전부 config다.
결과는 extension.vision.objects로 발행되어 콕핏 오버레이의 클릭 대상이 되거나, 미션이 pick_at(bbox=…)으로 그대로 실행한다 — 즉 비전과 실행 사이의 계약도 표준 status 필드 하나로 끝난다.
center를 "물체의 기하 중심"으로 못박고 있어서 머그는 손잡이가 아니라 몸통을, 도넛·그릇은 구멍을 찍었다. shape(cylinder·box·ring·handle·flat·irregular)와 approach(top·side)를 함께 내게 해 의미는 VLM, metric 보정은 로봇으로 역할을 갈랐다. 둘 다 선택 필드라 기존 소비자는 무영향이다.
obb_from_points가 극단값(min/max)으로 박스를 만들어 이상치 하나가 중심을 그 거리의 절반만큼 끌었다. trim_pct(축별 백분위 트림)와 method="caliper"(최소폭 회전 캘리퍼)를 더했다. ⚠️ 기본값은 종전 그대로다(trim_pct=0.0·method="pca") — 소비자가 전부 실기 파지 경로라 기본을 바꾸면 되던 파지가 바뀐다.
orient은 EE 항등(rx=ry=rz=0) = 그리퍼가 아래(-z)를 향하는 top-down 기준이다(RBY1 플랜지). 플랜지 항등이 그리퍼-다운이 아닌 로봇(Piper 류)은 소비자 쪽에서 자기 플랜지→그리퍼 오프셋을 합성해 쓴다 — 공통 레이어에 로봇별 분기를 넣지 않는다.
_pick_sequence)가 수직(+z) 접근 고정이라, 임의 접근·측면 파지라는 grasp net의 핵심 이득이 사장된다 — 즉 6-DOF의 게이트는 비전이 아니라 실행기다. pre/lift를 접근축 기준 오프셋으로 일반화하면 그때 다시 연다(금지가 아니라 재평가 트리거가 걸린 판단이다).
입력은 두 갈래다 — 마이크(STT)와 콕핏 텍스트(ai_chat). 그 뒤는 완전히 같다: 같은 LLM 두뇌, 같은 출력 경로, 같은 안전 계약. 코어는 이 폴더를 import하지 않는다 — base도 sub도 두뇌를 모르고, 러너가 배선한다(MissionRunner와 동형).
| 단계 | 무엇이 일어나나 |
|---|---|
| 입력 ① 마이크 | MicListener(발화 분할) → FasterWhisperSTT → handle_transcript. 로봇이 말하는 동안 들어온 소리는 캡처 시점에 폐기한다(에코) |
| 입력 ② 텍스트 | 콕핏 💬 → ai_chat {text} capability → handle_text. 마이크 없는 로봇·시끄러운 현장·원격 콘솔용 |
| 두뇌 | VoiceBrain.decide — 백엔드는 gerri.vision.backends를 텍스트 전용으로 재사용한다(auto = 키 있으면 SDK, 없으면 claude CLI 구독) |
| 출력 하나 | 말하기 → publish_speech() → status.extension.speech. status 워처가 그걸 읽어 TTS한다 — 발화의 단일 소스(콕핏 말풍선도 같은 자리). ⚠️ 슬롯이 하나라 마지막에 쓴 것이 이긴다 → 발화는 모았다 보내지 않고 on_say 콜백으로 나오는 즉시 흘려보낸다 |
| 실행 | capability 디스패치 또는 미션 트리거. 둘 다 아래 안전 계약을 코드가 강제한다 |
Figure 13. AI 인터페이스 파이프라인. handle_text는 스스로 TTS를 부르지 않고 on_say 콜백으로 발화를 흘려보내며(콜백을 안 주면 종전대로 마지막 발화를 반환 — 구 배선 호환), 실제 발행은 ai_chat이 한다 — 직접 say와 워처 say가 겹치면 이중 발화가 된다. 배선은 러너 한 곳이다(VoiceService(...) + sub.attach_chat(...)).
ai다 — 폴더 이름은 voice로 남는다: 텍스트도 같은 두뇌를 쓰게 되면서 voice라는 config 이름이 거짓말이 됐다. capabilities/missions는 "로봇이 무엇을 해도 되는가"라는 권한이라 마이크와 무관하다. 두뇌·권한(backend·model·robot_name·capabilities·missions)은 ai 직속, 마이크 전용(listen·wake_*·stt·tts·mic)은 ai.voice 하위다. listen:false면 두뇌만 켠다 — 수백MB STT 모델도 마이크 스레드도 만들지 않는다. 구 스키마는 승격 + 워닝으로 받는다(robot.config.json은 gitignore라 코드 pull로 안 바뀐다).ai.capabilities ∩ 로봇 광고이고 정보성만 넣는다. LLM이 목록 밖을 내놔도 decide가 코드에서 무효화한다(프롬프트 불신). ⚠️ 그런데 두뇌가 params를 자유 생성하므로 기본값이 안전해도 인자로 뒤집히는 것은 넣지 않는다 — visual_pick(execute=True)·detect_mode(on=True)·pick_at이 그 부류다.ai_chat은 화이트리스트에 넣을 수 없다: attach_chat이 붙는 순간 자동 광고되므로 교집합을 통과할 수 있고, 통과하면 두뇌→ai_chat→두뇌 무한 재귀다(handle_text는 침묵 게이트도 우회한다). _FORBIDDEN_CAPABILITIES가 config에 뭐가 오든 코드에서 걷어낸다.fetch(대상=프링글스). 위험 동작은 미션 안 wait_input이 되묻고 나서야 지나간다(fetch = 인식 → 확인 → 집기 → 건네기 → 복귀). 인자를 못 채운 채 미션을 걸면 decide가 막고 되묻는다 — 안 막으면 "$대상"이 문자열 그대로 capability까지 흘러 팔이 엉뚱한 곳으로 간다. 필수 인자 목록은 미션 정의에서 자동 추출된다.sounddevice가 기본이고, PipeWire에서 무음이 잡히는 환경은 parec로 우회한다.describe_status 같은 동기 capability가 dispatch 중에 진짜 답을 extension.speech에 쓰는데, 호출자가 다 끝나고 추임새를 발행해 답을 덮었다 — 대화가 "확인해볼게요"에서 끝났다. ⚠️ 그래서 어떤 건 되고 어떤 건 안 되는 것처럼 보였다 — describe_scene은 워커라 답이 늦게 도착해 우연히 살아남았다. 지금은 추임새 → (dispatch) → 답 순서가 자연히 지켜진다.AttributeError 트레이스백이 로그를 덮었다 — 트레이스백이 진짜 결함을 덮는다. 지금은 TTS가 없으면 조용히 건너뛴다(발화 텍스트는 이미 extension.speech로 나갔다). "TTS 실패"는 고장을 뜻하는 말이라 정상 구성(텍스트 전용)에 쓰면 다음 사람이 원인을 거기서 찾는다 — 폴백은 상태를 구분해 말한다._start_mission과 같은 규약으로 데몬 스레드에 뺐고, 중복 진입은 chat_busy로 거부한다. 이 문제의 일반해는 AND 데몬화다(AND 참조) — 거기서 송수신 자체를 워커 스레드로 끊었다.로봇이 10Hz로 status를 밀면 안 바뀐 정적 메타가 대역의 대부분을 차지한다. 그래서 생략하는데, 생략과 소멸을 구분하지 않으면 소비자 캐시에 유령이 남는다.
_omit_unchanged) — 안 바뀐 블록은 안 보낸다: robot_info가 와이어의 61%인데 거의 안 변한다. 소비자(NEXUS registry·콕핏)는 "온 블록은 그대로, 안 온 블록은 직전 것 유지"로 병합한다. ⚠️ digest가 블록 통째로 걸리므로 무거운 정적 메타와 가벼운 런타임 값이 같은 블록에 있으면 후자가 바뀔 때마다 전자까지 끌려 나온다 — 실측: 토글 하나(~30B 변경)가 robot_info 3,869B 전체를 재전송, 약 130배다.robot_info.extension(self.robot_info_extension), 런타임에 바뀌는 설정값(옆잡기 토글·status 주기·티칭 파일 목록)은 robot_state.extension.settings(self.robot_state_settings)다 — robot_state는 어차피 매 틱 나가므로 거기 두면 위의 130배 증폭이 사라진다. 레거시 robot_info_extra는 최상위에 얹는데 그것이 세 가지를 깬다 — ① 소멸 불가(키 단위 merge + 블록 단위 툼스톤이라 한 번 실린 최상위 키는 소비자 캐시에 영구 잔존) ② 각도 정규화 우회(to_standard_angles는 robot_state만 접는다 → 조용한 57.3배) ③ static/dynamic 혼동. 그래도 robot_info_extra를 살려 뒀다 — 모든 tag_*가 의존하고 있어 지금 끊으면 전 로봇이 깨진다(3박자 ADD→SWITCH→REMOVE의 ADD). 콘솔은 이미 extension 우선 → 최상위 폴백으로 읽으므로 로봇은 지금 바로 옮겨도 된다.capability_info)처럼 끝나면 키가 사라지는 블록은, 와이어에서 "안 바뀜"과 "없어짐"이 구분되지 않아 소비자 병합 캐시에 영구 고착됐다. 직전에 보냈던 블록이 없어지면 명시적 None을 한 번 실어 삭제를 표현한다 — 전체 프레임·부분 프레임 공통이다(소멸 시점이 전체 전송 틱이면 툼스톤을 영영 못 보내는 창이 있었다). ⚠️ 계약의 절반은 소비자에게 있다. 콘솔의 병합이 {...prev, ...value}라 블록이 사라지지 않아 끝난 capability_info(진행 표시)가 콕핏에 영구 고착됐고, 생략이 걸린 틱의 빈 껍데기 robot_info를 병합 전에 읽어 version·robot_shape가 통째로 날아갔다. 규칙 본체를 한 벌로 합쳐(mergeStatusBlocks) 대시보드와 콕핏이 같은 함수를 쓴다 — 같은 규칙의 독립 구현이 여러 벌이면 그중 하나는 반드시 툼스톤을 빼먹는다.robot_info였다. 병합은 화면이 하고, 캐시는 완전본으로 남겨 신규·재접속 클라이언트의 기준선으로 쓴다.gerri/recording/mcap_bridge.py가 새 채널 없이 Topic.Robot.STATUS를 (생략 전) 구독한다. 콜백은 직렬화 + 큐 + 즉시 반환만 하고 디스크는 전용 워커가 만진다 — 발행 스레드 추가비용 약 0.08 ms/건(10Hz 예산의 0.08%), 그래서 켜둔다. 드롭은 실시간성 우선(reject-new)이고 누적 건수를 로그와 파일 안 recording.stats에 남긴다 — 조용한 드롭은 없다. 영상은 비용 자릿수가 200배라 안 넣었다. 패키지가 없어도 로봇은 그대로 뜬다(이유를 로그로 말하고 녹화만 끈다).계약(TraceHop·TraceBatch·토픽별 정책)은 synapse가 정본이고, gerri/trace.py는 로봇측 발행기뿐이다 — 어디서 hop을 줍고, 얼마나 접어서, 얼마나 자주 보내는가. 집계와 화면은 NEXUS가 맡는다.
| 계기를 얹은 자리(단일 초크포인트) | 무엇이 실리나 |
|---|---|
| _and_/daemon.py :: _log_raw | AND의 TX/RX — 노드 robot:{id}/and. 이미 전 송수신이 지나는 자리다. |
| base_controller :: _standardize_outgoing | normalized · dropped_fields — “내 필드가 왜 콘솔에 안 보이지?”에 답하는 데이터. |
| base_controller :: _on_external_message | 명령이 어느 sub의 어느 capability로 갔나. |
| base_controller :: __init__ / disconnect | enable(info) / disable() — 러너가 넘긴 robot config가 여기로 온다. |
Figure 14. 새 훅 포인트를 만들지 않았다(死코드 금지). 경로 테스트도 모듈 함수를 직접 부르는 것이 아니라 실제 발행 경로로 메시지를 흘려보내 와이어에 hop이 실리는지 본다 — 위 지점 중 하나를 지우면 그 테스트가 빨개진다.
"trace": {"enabled": false, "batch_seconds": 1.0, "tick_seconds": 1.0}
record()는 전역 bool 하나를 읽고 즉시 돌아온다. enabled는 엄격 판정이다(true/1/yes/on만) — 문자열 "false"가 truthy라서 켜지는 사고를 막는다. 기본 꺼짐이 이 기능의 1번 제약이라 오탐 방향(켜짐)이 위험하다. config로만 켠다 — CLI 플래그를 늘리지 않는다.robot.status·command.teleop)은 (topic, node, direction) 조합마다 창당 요약 hop 1건이고 개별 메시지는 포기한다(uuid="", 남는 것은 count·창 평균 latency_ms·dropped_fields 합집합). 실측 근거 — teleop을 건별로 추적했더니 hop 트래픽이 원본 업링크의 148%였다. 창은 두 가지로 닫힌다 — ① 다음 메시지가 tick_seconds를 넘겨 도착할 때 ② 배치가 나갈 때(②가 없으면 흐름이 멎은 창이 다음 메시지까지 붙들려 “마지막 구간”이 영영 안 보인다).