完整调用流程与客户端示例

22.1 从建图到导航

步骤	调用	必须观察的结果
1	systems/device → auth/me	设备 SN 正确，按实际启动模式确认认证行为
2	services/status；按部署启动扫描/运动依赖	相关服务为 running，导航环境已停止
3	mapping/container/start，提供新 scene_name	记录 task_id，不因网络超时直接重发
4	mapping/status	等到 mapping，观察 perceptions_available/map_available，按现场流程完成采集
5	mapping/container/stop	仅表示结束请求受理
6	tasks/{task_id}	等到 Status=2；失败读取 ErrorMsg / ResultJSON
7	maps、maps/{name}/files	地图及核心文件已存在；启用版本时核对保存/安装内容
8	nav/startup，填写地图及实际初始位姿	检查完成步骤；若失败按 failed_step 恢复
9	nav/params/apply_status、nav/navigation_status	参数应用结果满足需求，定位与场景正确
10	nav/missions	记录 mission_id
11	nav/missions/{id} 或 MQTT task/uplink	completed 才是完成；failed/cancelled 分别处理
12	按需取消任务，再停止导航容器	软件取消与容器停止分别确认

初始位姿必须来自实际机器人位置或可靠重定位结果，示例原点不是适用于所有设备的安全默认值。应用的“停止当前任务”操作与现场硬件急停用途不同。

22.2 Python REST 客户端

示例使用 Python requests。通过环境变量提供 IP，启用 Bearer 时另提供 ORSUS_TOKEN；免 token 模式可不设置该变量。不要把生产凭据写入源码。函数区分 HTTP/JSON/业务错误，轮询有本地截止时间；超时不会偷偷重复提交或自动取消设备任务。

import os
import time
from urllib.parse import quote
import requests


class OrsusError(RuntimeError):
    def __init__(self, status, code, message, data=None):
        super().__init__(f"HTTP {status}, code={code}: {message}")
        self.status, self.code, self.data = status, code, data


class OrsusClient:
    def __init__(self, origin, token=None):
        self.base = origin.rstrip('/') + '/v1/api'
        self.session = requests.Session()
        if token:
            self.session.headers['Authorization'] = 'Bearer ' + token

    def call(self, method, path, *, body=None, timeout=(5, 30)):
        response = self.session.request(
            method, self.base + path, json=body, timeout=timeout
        )
        try:
            result = response.json()
        except ValueError as exc:
            raise OrsusError(response.status_code, None, '响应不是 JSON') from exc
        if not isinstance(result, dict) or 'code' not in result:
            raise OrsusError(response.status_code, None, '响应不是 Orsus 信封')
        if not response.ok or result['code'] != 0:
            raise OrsusError(response.status_code, result['code'],
                             result.get('msg', ''), result.get('data'))
        return result.get('data')

    def start_environment(self, scene_name, initial_pose):
        return self.call('POST', '/nav/startup', timeout=(5, 360), body={
            'pose': initial_pose,
            'nav_container': {
                'scene_name': scene_name, 'bringup_mode': 'navigation',
                'use_relocalization': True,
            },
        })

    def submit_goal(self, target):
        return self.call('POST', '/nav/missions', body={
            'mode': 'standard', 'frame_id': 'map', 'target': target,
        })

    def wait_mission(self, mission_id, max_wait_seconds=300):
        path = '/nav/missions/' + quote(mission_id, safe='')
        deadline = time.monotonic() + max_wait_seconds
        while time.monotonic() < deadline:
            state = self.call('GET', path)
            if state['status'] in {'completed', 'failed', 'cancelled'}:
                return state
            time.sleep(1)
        raise TimeoutError(f'本地等待超时，任务 {mission_id} 可能仍在执行，请查询现状')

    def cancel_mission(self, mission_id):
        return self.call('DELETE', '/nav/missions/' + quote(mission_id, safe=''))

    def close(self):
        self.session.close()


if __name__ == '__main__':
    client = OrsusClient(os.environ['ORSUS_ORIGIN'], os.environ.get('ORSUS_TOKEN'))
    try:
        print(client.call('GET', '/auth/me'))
        print(client.call('GET', '/maps'))
        # 完成现场环境和位姿确认后，由业务显式调用以下函数：
        # client.start_environment('office_1f', {'x': 0, 'y': 0, 'theta': 0})
        # mission = client.submit_goal({'x': 3.2, 'y': 1.5, 'theta': 0})
        # state = client.wait_mission(mission['mission_id'])
        # 若 state['status'] != 'completed'，按失败/取消流程处理。
    finally:
        client.close()

遇到 requests 超时或连接异常时保留已知任务 ID，用查询恢复；如果创建请求没有拿到 ID，先检查 current/list，再决定是否创建新任务。不要用简单的“所有 POST 自动重试”HTTP 中间件。

22.3 浏览器 fetch 读取系统 SSE

此示例处理分段、UTF-8 和多行 data，并通过 AbortController 结束连接。断开后由业务层安排退避重连，不在代码中隐藏无限重连。

async function consumeSystemEvents(origin, token, onEvent, signal) {
  const response = await fetch(new URL('/v1/api/systems/stream', origin), {
    headers: { Accept: 'text/event-stream', ...(token ? { Authorization: `Bearer ${token}` } : {}) },
    signal,
  });
  if (!response.ok || !response.headers.get('content-type')?.includes('text/event-stream')) {
    throw new Error(`SSE 连接失败: HTTP ${response.status}`);
  }
  if (!response.body) throw new Error('浏览器不支持流式响应');
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let eventName = 'message';
  let dataLines = [];
  const processLine = line => {
    if (line.endsWith('\r')) line = line.slice(0, -1);
    if (line === '') {
      if (dataLines.length) onEvent(eventName, JSON.parse(dataLines.join('\n')));
      eventName = 'message';
      dataLines = [];
      return;
    }
    if (line.startsWith(':')) return;
    const colon = line.indexOf(':');
    const field = colon < 0 ? line : line.slice(0, colon);
    let value = colon < 0 ? '' : line.slice(colon + 1);
    if (value.startsWith(' ')) value = value.slice(1);
    if (field === 'event') eventName = value;
    if (field === 'data') dataLines.push(value);
  };
  try {
    while (true) {
      const { done, value } = await reader.read();
      buffer += done ? decoder.decode() : decoder.decode(value, { stream: true });
      let newline;
      while ((newline = buffer.indexOf('\n')) >= 0) {
        processLine(buffer.slice(0, newline));
        buffer = buffer.slice(newline + 1);
      }
      if (done) break;
    }
  } finally {
    await reader.cancel().catch(() => {});
    reader.releaseLock();
  }
}

// const controller = new AbortController();
// consumeSystemEvents(origin, token, (event, data) => updatePanel(event, data), controller.signal)
//   .catch(error => showConnectionError(error));
// controller.abort(); // 页面关闭时调用

系统 SSE data 一律是 JSON；ASR 可能含 [DONE]，不能直接复用此示例而不增加上游格式判断。

22.4 MQTT 订阅与恢复

使用部署方指定的 MQTT 客户端、TLS、认证和 ACL。推荐先建立订阅再创建任务，避免漏掉最初事件。

CONNECT broker（部署指定地址、凭据及 TLS）
SUBSCRIBE sys/{region}/{tenant_id}/{device_id}/task/uplink，QoS 1
SUBSCRIBE sys/{region}/{tenant_id}/{device_id}/uplink，QoS 1
接收：按 topic 和 msg_type 区分载荷；关联 task_id / command_id
断连：标记数据已过期，保留任务 ID
重连：重新订阅，再 GET /v1/api/tasks/{id} 获取持久化终态

云端下行命令需要确保 command_id 唯一，expires_at 为实际未来时间。应用不能只等待 command_result=succeeded 就显示“已到达”。

22.5 常见问题定位

现象	优先检查
浏览器能打开设备页，API 401	是否附带唯一 Bearer 头；终端 token 与 API token 是否混用
SSE 一直 401	是否错误使用原生 EventSource；查询参数 token 不适用于系统 SSE
startup 超时	容器、导航 API、参数 API 分段等待；查询是否已有部分服务启动，不直接重试运动任务
code=0 但参数无效	success 分项、offline_save、启动回放结果；是否只保存离线值
code=0 但容器没启动	后台容器日志、互斥策略、Docker 状态；HTTP 受理不等于启动完成
导航暂停但 mission 仍 running	任务没有 paused 状态，结合 NavStatus
建图 stop 后地图不可用	继续查看建图任务 ResultJSON/ErrorMsg 和文件校验，不提前导航
导入失败	ZIP 名称与四个核心文件同名；布局、PGM/PCD/YAML/path 格式与配额
地图已安装但显示旧环境	installed_release_id 只说明磁盘，确认运行中的 navigation_session 与重启/重定位流程
RTK running 但无差分效果	三条连接状态、lastValidGgaAt/lastRtcmAt/lastPublishedAt 和帧计数
日志列表 500	先查 logs/source，使用有效 module；不要依赖空 module 的默认值