综述

Lichtblick是一个开源的机器人监控调试工具，支持3d空间场景显示，地图显示，波形显示等功能。将它直接运用到飞行器的遥测数据实时显示，可以极大节约开发工作量。

Lichtblick原名foxglove studio。在原版转为闭源开发之后，宝马公司继续维护开源分支并改名。界面、使用方法和api接口与foxglove studio基本相同。

本文展示了一个最简单的遥测数据实时显示、录制、回放的python程序。这是真正的到手即用，迅速实现功能。

该程序代码大部分由ai生成，功能经过人工测试验证。

数据流向

遥测数据(json) -----------------> foxglove_json_server.py -----------------> Lichtblick （实时显示）

UDP数据包 | websocket

------------------------------> mcap文件（录制）

服务端在 udp://0.0.0.0:9999 接收json数据输入

服务端在 ws://localhost:8765 等待litchblick连接传入，随后转发数据到litchblick

服务端将收到的数据保存到mcap文件

mcap文件随后可以被Lichtblick读取

数据格式

支持标准的json数据输入，可以自行任意编辑json字段，可以选择任意字段绘制曲线图。

使用litchblick支持的数据格式，可以获得额外功能（3d对象显示，2d图像显示等）

{ "voltage": 12.5, "temperature": 33.7, "status": "OK" }

如果要使用真实的遥测接收机，只需要将收到的二进制数据包转换成json格式（如上），作为udp数据包发到本机的9999端口即可

安装方法

安装Lichtblick (foxglove-studio的开源替代品) XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX

安装依赖项 pip install foxglove-websocket mcap

启动python服务端 python foxglove_json_server.py

运行Lichtblick，点击左上角 Add connection -> WebSocket -> ws://localhost:8765

运行演示程序 python foxglove_json_demo.py

运行效果

源代码

foxglove_json_XXXXXXXXX

import asyncio, json, time

from datetime import datetime

from mcap.writer import Writer

from foxglove_websocket import run_cancellable

from foxglove_websocket.server import FoxgloveServer, FoxgloveServerListener

from foxglove_websocket.types import ChannelId, ClientChannelId, ServiceId

# ---------- Config parameters ---------

fname = datetime.now().strftime("%Y%m%d_%H%M%S") + '.mcap'

udp_listen = ("0.0.0.0", 9999)

foxglove_listen = ("0.0.0.0", 8765)

channel_name = "tlm" # Channel name

# ---------- Create schema ---------

tlm_schema = (

"Telemetry", # Channel type (schema name)

json.dumps({

"type": "object",

"additionalProperties": True,

})

)

# ---------- UDP protocol ----------

class UdpJsonProtocol(asyncio.DatagramProtocol):

def __init__(self, loop, foxglove_server=None, foxglove_chan=None, mcap_writer=None, mcap_chan=None):

self.loop = loop

self.foxglove_server = foxglove_server

self.foxglove_chan = foxglove_chan

self.mcap_writer = mcap_writer

self.mcap_chan = mcap_chan

def datagram_received(self, data: bytes, addr):

ts = time.time_ns()

# send to foxglove

if self.foxglove_server:

self.loop.create_task(self.foxglove_server.send_message(

self.foxglove_chan, ts, data

))

# write to mcap file

if self.mcap_writer:

self.mcap_writer.add_message(

channel_id=self.mcap_chan,

log_time=ts,

publish_time=ts,

data=data,

)

def error_received(self, exc):

print(f"[UDP] Error: {exc}")

# ---------- Server setup ----------

async def main():

server = FoxgloveServer(

*foxglove_listen,

"Telemetry server",

supported_encodings=["json"],

)

server.start()

mcap_file = open(fname, "wb")

mcap_writer = Writer(mcap_file)

mcap_writer.start()

# Channel: UDP JSON payloads

foxglove_chan: ChannelId = await server.add_channel({

"topic": channel_name,

"encoding": "json",

"schemaName": tlm_schema[0],

"schema": tlm_schema[1],

"schemaEncoding": "jsonschema"

})

mcap_chan = mcap_writer.register_channel(

topic=channel_name,

message_encoding="json",

schema_id=mcap_writer.register_schema(

name=tlm_schema[0],

encoding="jsonschema",

data=tlm_schema[1].encode("utf-8"),

),

)

loop = asyncio.get_running_loop()

transport, _protocol = await loop.create_datagram_endpoint(

lambda: UdpJsonProtocol(loop, server, foxglove_chan, mcap_writer, mcap_chan),

local_addr=udp_listen,

)

print("UDP JSON listener: %s:%d" % udp_listen)

print("Foxglove WebSocket: ws://%s:%d" % foxglove_listen)

print("MCAP recoder: %s" % mcap_file.name)

print("")

try:

await asyncio.Event().wait()

except KeyboardInterrupt:

print("Exiting...")

finally:

transport.close()

server.close()

mcap_writer.finish()

mcap_file.close()

if __name__ == "__main__":

run_cancellable(main())

foxglove_json_XXXXXXX

import socket, json, time

UDP_IP = "127.0.0.1" # or your receiver address

UDP_PORT = 9999 # match your server

HZ = 50 # 50Hz

DT = 1.0 / HZ # 20ms

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

velocity = 0.0

position = 0.0

def get_accel(t):

if t < 1: return 100 * t # 0 to 100 in first 1s

elif t < 4: return 100 # hold 100 until 4s

elif t < 5: return 100 - 100*(t-4) # ramp down to 0 by 5s

else: return -position/1000

start = time.time()

while True:

t = time.time() - start

accel = get_accel(t)

velocity += accel * DT

position += velocity * DT

msg = {

"time": round(t, 3),

"acceleration": accel,

"velocity": velocity,

"position": position,

}

encoded_msg = json.dumps(msg).encode("utf-8")

sock.sendto(encoded_msg, (UDP_IP, UDP_PORT))

print(msg)

time.sleep(DT)

开源授权协议

标准的MIT协议