快速开始

本篇带你用最短路径跑通一次完整的请求-响应：连接设备 → 准备消息编解码 → 发出请求 → 收到响应。示例基于项目提供的 Python 客户端封装，不依赖图形界面。

准备工作

开始前，确认以下几项：

设备已上电，并与你的主机建立网络连接（USB 或 Wi-Fi，见连接与鉴权）。
已拿到 OpenAPI Key（获取方式见连接与鉴权 · 申请 OpenAPI Key）。
主机已获取示例工程，并装好 Python 环境与依赖（见下）。

环境与依赖

项目	要求
操作系统	Linux、Windows（图形界面分别使用 PySide2、PySide6）
Python	3.8 及以上
MQTT 客户端	`paho-mqtt`，2.0.0 及以上
编解码	`protobuf`、`grpcio-tools`（用于编译 proto）
点云解压（可选）	`lz4`、`zstandard`（只在订阅实时点云时需要）

获取示例工程并安装依赖

示例工程是公开仓库 Lixel_OpenAPI。以下命令以 Ubuntu 22.04 为例，Windows 的步骤见仓库里的 README.zh-CN.md：

sudo apt-get update
sudo apt-get install -y python3-pip python3-venv git
git clone --branch 1.1.0 https://github.com/xgrids/Lixel_OpenAPI.git
cd Lixel_OpenAPI                    # 仓库根目录
python3 -m venv venv
source venv/bin/activate            # 激活后可以直接使用 python 和 pip
unset PYTHONPATH                    # 机器上装过 ROS 时需要，避免 ROS 的包混入
pip install --upgrade pip
pip install -r python/requirements.txt     # 含 paho-mqtt、protobuf、grpcio-tools
pip install lz4 zstandard           # 需要实时点云时再装

注意以下几点：

要用虚拟环境，不要用系统自带的 Python 包：Ubuntu 22.04 自带的 python3-protobuf 是 3.12.4，低于 python/requirements.txt 要求的 5.26.0。
Ubuntu 22.04 默认没有 pip 和 python3-venv，也没有 python 命令，所以先装上面的软件包，并在虚拟环境激活后再使用 python。
下面所有 Python 片段都要在 python/ 目录下运行（先 cd python，例如把片段保存成 python/quick.py 后执行 python quick.py），否则 from example.core import … 会找不到 example 包。src/generate_proto.py 等带 python/ 前缀的命令则在仓库根目录运行。
python/requirements.txt 不含图形界面依赖；运行图形界面需要另外安装 python/requirements-linux.txt（PySide2）或 python/requirements-windows.txt（PySide6）。

第一步：准备 Protobuf 模块

所有业务消息都用 Protobuf 编解码。先把 proto 定义编译为 Python 模块：

python python/src/generate_proto.py

注：编译使用当前环境的 grpc_tools.protoc，确保生成代码与运行时的 protobuf 版本一致。运行时建议设置 PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python，避免不同环境下的 C 扩展不兼容。

第二步：连接设备

用客户端封装建立连接。有线连接时设备默认地址为 192.168.55.1，MQTT 端口为 1883。下面的示例适用于 K1 / L2PRO；使用 MQTTS 的型号见连接设备，需要 use_tls=True、port=8883 和厂商提供的账号：

from example.core import ClientOptions, OpenApiClient

opts = ClientOptions(
    host="192.168.55.1",
    port=1883,
    openapi_key="<由 XGRIDS 提供>",
)

with OpenApiClient(opts) as client:
    print("connected:", client.is_connected)

with 语句会在进入时连接、退出时断开。连接失败会抛出异常：地址不可达或端口未开放时抛出 OSError（含 ConnectionRefusedError；Python 3.10 及以上的网络超时是 TimeoutError）；已连上 TCP 但 10 秒内没有收到成功的 CONNACK（包括 broker 拒绝连接）时抛出 TimeoutError，超时时间由 ClientOptions.connect_timeout 设置。

第三步：发出第一个请求

发一个最简单的"查询设备信息"请求，并等待响应。请求-响应是同步调用，内部会先订阅响应 Topic、再发布请求、阻塞等待应答：

from example.core import ClientOptions, OpenApiClient

opts = ClientOptions(host="192.168.55.1", openapi_key="<由 XGRIDS 提供>")

with OpenApiClient(opts) as client:
    rsp = client.request(
        req_topic="lixel/application/request/device_info",
        req_message="DeviceInfoRequest",
        payload={},                                   # 空负载即可
        rsp_topic="lixel/application/response/device_info",
        rsp_message="DeviceInfoResponse",
    )
    print(rsp)                                        # 已解码为 dict

request() 会自动把 OpenAPI Key 填入消息头。返回值是解码后的字典，包含设备型号、固件版本、相机列表等信息。

第四步：订阅实时上报

除了请求-响应，很多数据是设备主动上报的。订阅设备状态，每秒会收到一次：

import time
from example.core import ClientOptions, OpenApiClient


def on_status(topic, data):      # data 已解码为 dict
    print(topic, data)


opts = ClientOptions(host="192.168.55.1", openapi_key="<由 XGRIDS 提供>")
with OpenApiClient(opts) as client:
    client.subscribe(
        "lixel/application/report/device_status",
        "DeviceStatusReport",
        on_status,
    )
    time.sleep(5)                # 保持连接，等待上报

subscribe() 会立即返回，所以要让程序保持运行，回调才会被触发。回调里抛出的异常会被记录到日志（Python 默认输出到终端），不会中断其他订阅。

注：OpenAPI 公开的接口中，只有心跳采用 JSON 文本。订阅心跳时 message 传空字符串，回调会收到字符串而非字典。
注：收到心跳只说明设备已联网，不表示业务已就绪。设备刚开机时，端口已开放、心跳已经在发，但 device_info 等请求可能在接下来一段时间内没有响应。开机后先重试 device_info，成功后再发其他请求。

完整示例：扫描一次

一次典型的端到端业务流程是：连接 → 确认设备待机 → 启动扫描 → 等设备真正进入扫描中 → 扫描一段时间 → 停止扫描 → 等设备回到待机。

连接设备 → 订阅 device_status → 确认待机 → AC_START → 等待扫描中 → 扫描 → AC_STOP → 等待待机 → 断开

扫描的开始和停止都是异步的。AC_START 的响应只表示设备收到了请求，设备随后依次经过待机、开始扫描中、扫描中；实测 AC_START 后 14 到 15 秒才进入扫描中。进入扫描中之前发出的 AC_STOP 实测会被推迟，设备不会进入扫描中。AC_STOP 之后设备经停止扫描中回到待机，实测从进入停止扫描中算起，43 到 46 秒回到待机（3 次）；从发出 AC_STOP 算起，单次实测 52.1 秒（Lixel L2Pro，固件 V3.2.1，状态每秒上报一次，精度 1 秒）。所以示例订阅 report/device_status，读取 modeling_state，等到目标状态再继续，不使用固定的 sleep 等待。modeling_state 的取值见错误码 · 扫描状态码。

import threading
import time
from example.core import ClientOptions, OpenApiClient

SCAN_READY = 0x1204012C      # EC_ScanReady，待机
SCAN_MODELING = 0x1204012E   # EC_ScanModeling，扫描中

state = {"modeling_state": None}
cond = threading.Condition()


def on_status(topic, data):
    with cond:
        state["modeling_state"] = data["modeling_state"]
        cond.notify_all()


def wait_for(target, timeout):
    deadline = time.monotonic() + timeout
    with cond:
        while state["modeling_state"] != target:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return False
            cond.wait(remaining)
    return True


def modeling(client, payload):
    return client.request(
        req_topic="lixel/application/request/modeling",
        req_message="ModelingRequest",
        payload=payload,
        rsp_topic="lixel/application/response/modeling",
        rsp_message="ModelingResponse",
    )


opts = ClientOptions(host="192.168.55.1", openapi_key="<由 XGRIDS 提供>")
with OpenApiClient(opts) as client:
    client.subscribe("lixel/application/report/device_status",
                     "DeviceStatusReport", on_status)
    if not wait_for(SCAN_READY, 10):
        raise SystemExit("设备不在待机状态: 0x%X" % (state["modeling_state"] or 0))   # 取值见 08 页的扫描状态码
    rsp = modeling(client, {"action": "AC_START", "project_name": "demo",
                            "record_mode": "RM_RECORD_CALC", "scan_mode": "SM_PCL",
                            "mount_type": "MT_HAND"})
    code = rsp.get("error", {}).get("code")
    if code not in (0x12040001, 0x12040002):   # EC_Success / EC_SuccessAndReboot
        raise SystemExit("设备未接受启动请求: %s" % rsp.get("error"))
    if wait_for(SCAN_MODELING, 90):      # 实测 AC_START 后 14 到 15 秒进入扫描中
        time.sleep(10)                   # 扫描 10 秒
    else:
        print("等待进入扫描中超时")
    modeling(client, {"action": "AC_STOP"})
    wait_for(SCAN_READY, 180)            # 实测 AC_STOP 后最长 52 秒回到待机，180 秒是余量

这段示例是仓库里 python/example/usage/modeling.py 的简化版：流程相同，但省略了 Ctrl-C 和等待超时之后补发 AC_STOP 的善后处理，也没有退出码。启动请求未被接受通常是密钥错误或消息格式错误，设备不会因为当前不在待机而用返回码拒绝，所以示例先检查 modeling_state。正式使用请直接运行 modeling.py。从 1.0.1 起，它会等设备真正进入扫描中，请求未被接受时退出，被 Ctrl-C 中断或等待超时时也会发出停止：python python/example/usage/modeling.py --host 192.168.55.1 --key <由 XGRIDS 提供> --project demo。完整流程需要一分钟以上。1.0.0 的 modeling.py 只固定等待几秒就发出停止，设备来不及进入扫描中，请使用 1.0.1 及以上版本。

待机时订阅位姿，实测 3 秒内没有数据；位姿和点云都是在扫描中取到的。要在扫描过程中接收，请在设备进入扫描中之后订阅，例如把 client.subscribe("lixel/application/report/lio_pose", "LioPoseReport", 回调) 放在 wait_for(SCAN_MODELING, 90) 成功之后。

注：扫描参数（录制模式、扫描模式、挂载方式等）的取值见接口清单 · 扫描建图。实时点云需要额外解压，见实时数据。

下一步

不同型号的连接方式与鉴权细节，见连接与鉴权。
想了解 Topic 规则、消息头与编解码原理，见消息模型与编解码。
完整接口清单（扫描、点位、工程、RTK 等），见接口清单。
更多可直接运行的脚本，见客户端封装与工具。