Quick Start
This page walks you through one complete request-response with the shortest path: connect to the device → prepare the message codec → send a request → receive the response. The examples use the project's Python client wrapper and do not depend on the GUI.
Prerequisites
Before you start, make sure:
- The device is powered on and networked with your host (USB or Wi-Fi, see Connection & Authentication).
- You have an OpenAPI Key (how to obtain it: Connection & Authentication · Requesting an OpenAPI Key).
- The host has the example project and a Python environment with the dependencies (see below).
Environment and dependencies
| Item | Requirement |
|---|---|
| OS | Linux, Windows (the GUI uses PySide2 and PySide6 respectively) |
| Python | 3.8 or newer |
| MQTT client | paho-mqtt, 2.0.0 or newer |
| Codec | protobuf, grpcio-tools (to compile the proto) |
| Point cloud decompression (optional) | lz4, zstandard (only needed when subscribing to the real-time point cloud) |
Getting the example project and installing dependencies
The example project is the public repository Lixel_OpenAPI. The commands below use Ubuntu 22.04; for Windows see README.md in the repository:
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 # repository root
python3 -m venv venv
source venv/bin/activate # after activation you can use python and pip directly
unset PYTHONPATH # needed on machines with ROS installed, to avoid ROS packages leaking in
pip install --upgrade pip
pip install -r python/requirements.txt # includes paho-mqtt, protobuf, grpcio-tools
pip install lz4 zstandard # install when you need the real-time point cloud
Note the following:
- Use a virtual environment, not the system Python packages: Ubuntu 22.04 ships
python3-protobuf3.12.4, below the 5.26.0 required bypython/requirements.txt. - Ubuntu 22.04 has no
pip,python3-venv, orpythoncommand by default, so install the packages above first and usepythononly after activating the virtual environment. - All Python snippets below must run inside the
python/directory (cd pythonfirst, e.g. save a snippet aspython/quick.pyand runpython quick.py), otherwisefrom example.core import …cannot find theexamplepackage. Commands with thepython/prefix, such assrc/generate_proto.py, run from the repository root. python/requirements.txtdoes not include the GUI dependencies; to run the GUI, additionally installpython/requirements-linux.txt(PySide2) orpython/requirements-windows.txt(PySide6).
Step 1: Prepare the Protobuf module
All business messages use Protobuf. First compile the proto definition into a Python module:
python python/src/generate_proto.py
Note: compilation uses the current environment's
grpc_tools.protoc, keeping the gencode version consistent with the runtime protobuf version. At runtime it is recommended to setPROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=pythonto avoid C-extension incompatibilities across environments.
Step 2: Connect to the device
Use the client wrapper to connect. For a wired connection the device default address is 192.168.55.1 and the MQTT port is 1883. The example below applies to K1 / L2PRO; for MQTTS models see Connection & Authentication, which needs use_tls=True, port=8883, and the vendor-provided credentials:
from example.core import ClientOptions, OpenApiClient
opts = ClientOptions(
host="192.168.55.1",
port=1883,
openapi_key="<provided by XGRIDS>",
)
with OpenApiClient(opts) as client:
print("connected:", client.is_connected)
The with statement connects on entry and disconnects on exit. A failed connection raises an exception: an unreachable address or closed port raises OSError (including ConnectionRefusedError; a network timeout on Python 3.10+ is TimeoutError); if the TCP connection is up but no successful CONNACK arrives within 10 seconds (including the broker rejecting the connection) it raises TimeoutError, with the timeout set by ClientOptions.connect_timeout.
Step 3: Send your first request
Send the simplest "query device info" request and wait for the response. Request-response is a synchronous call that first subscribes to the response topic, then publishes the request, then blocks for the reply:
from example.core import ClientOptions, OpenApiClient
opts = ClientOptions(host="192.168.55.1", openapi_key="<provided by XGRIDS>")
with OpenApiClient(opts) as client:
rsp = client.request(
req_topic="lixel/application/request/device_info",
req_message="DeviceInfoRequest",
payload={}, # an empty payload is fine
rsp_topic="lixel/application/response/device_info",
rsp_message="DeviceInfoResponse",
)
print(rsp) # already decoded into a dict
request() fills the OpenAPI Key into the header automatically. The return value is a decoded dict with the device model, firmware version, camera list, and so on.
Step 4: Subscribe to reports
Besides request-response, much data is actively reported by the device. Subscribe to the device status, received once per second:
import time
from example.core import ClientOptions, OpenApiClient
def on_status(topic, data): # data is already decoded into a dict
print(topic, data)
opts = ClientOptions(host="192.168.55.1", openapi_key="<provided by XGRIDS>")
with OpenApiClient(opts) as client:
client.subscribe(
"lixel/application/report/device_status",
"DeviceStatusReport",
on_status,
)
time.sleep(5) # keep the connection alive to receive reports
subscribe() returns immediately, so keep the program running for the callback to fire. An exception raised in the callback is logged (Python prints to the terminal by default) and does not interrupt other subscriptions.
Note: among the interfaces exposed by the OpenAPI, only the heartbeat uses JSON text. When subscribing to the heartbeat, pass an empty string for
messageand the callback receives a string, not a dict.Note: receiving the heartbeat only means the device is networked, not that it is ready for business. Right after boot the port is open and the heartbeat is already sent, but requests like
device_infomay have no response for a while. After boot, retrydevice_infofirst and send other requests once it succeeds.
Full example: one scan
A typical end-to-end flow is: connect → confirm standby → start scanning → wait until the device is really scanning → scan for a while → stop scanning → wait for standby.
connect → subscribe device_status → confirm standby → AC_START → wait for scanning → scan → AC_STOP → wait for standby → disconnect
Starting and stopping are both asynchronous. The response to AC_START only means the device received the request; the device then goes standby → starting → scanning; measured to reach scanning 14 to 15 seconds after AC_START. An AC_STOP sent before scanning is deferred, and the device does not enter scanning. After AC_STOP the device goes through stopping and back to standby; measured from entering stopping, 43 to 46 seconds to standby (3 runs); from sending AC_STOP, a single run measured 52.1 seconds (Lixel L2Pro, firmware V3.2.1, status reported once per second, 1-second precision). So the example subscribes to report/device_status, reads modeling_state, and waits for the target state rather than a fixed sleep. For modeling_state values, see Error Codes · Scan status codes.
import threading
import time
from example.core import ClientOptions, OpenApiClient
SCAN_READY = 0x1204012C # EC_ScanReady, standby
SCAN_MODELING = 0x1204012E # EC_ScanModeling, scanning
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="<provided by 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("device not in standby: 0x%X" % (state["modeling_state"] or 0)) # values: see scan status codes on page 07
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("device did not accept the start request: %s" % rsp.get("error"))
if wait_for(SCAN_MODELING, 90): # measured: scanning 14 to 15 s after AC_START
time.sleep(10) # scan for 10 s
else:
print("timed out waiting for scanning")
modeling(client, {"action": "AC_STOP"})
wait_for(SCAN_READY, 180) # measured: up to 52 s back to standby after AC_STOP, 180 s is margin
This example is a simplified version of python/example/usage/modeling.py in the repository: the same flow, but without the Ctrl-C handling, the extra AC_STOP after a wait timeout, or exit codes. A rejected start request usually means a wrong key or a malformed message; the device does not reject it with a return code just because it is not in standby, so the example checks modeling_state first. For real use, run modeling.py directly. From 1.0.1 on, it waits until the device is really scanning, exits when a request is not accepted, and sends a stop when interrupted with Ctrl-C or on a wait timeout: python python/example/usage/modeling.py --host 192.168.55.1 --key <provided by XGRIDS> --project demo. A full run takes a minute or more. The 1.0.0 modeling.py only waits a few fixed seconds before stopping, before the device can enter scanning — use 1.0.1 or newer.
Subscribing to the pose in standby produced no data within 3 seconds (measured); the pose and point cloud are only available while scanning. To receive them during a scan, subscribe after the device enters scanning, e.g. put client.subscribe("lixel/application/report/lio_pose", "LioPoseReport", callback) after wait_for(SCAN_MODELING, 90) succeeds.
Note: for scan parameters (record mode, scan mode, mount type, etc.), see Interface Catalog · Scan mapping. The real-time point cloud needs extra decompression, see Real-time Data.
Next steps
- For the connection and authentication details of each model, see Connection & Authentication.
- To understand the topic rules, header, and codec, see Message Model & Codec.
- For the full interface catalog (scan, points, projects, RTK, etc.), see Interface Catalog.
- For more runnable scripts, see Client Wrapper & Tools.