Fix voltage display and request flight controller SN

README.md

@@ -1,13 +1,14 @@
 # VK 飞控遥测监控
 
-基于 PyQt6 的 Windows 串口上位机。**只被动接收** MAVLink，不发送心跳、参数、解锁或飞行控制命令。支持 MAVLink 2 VKFly 自定义方言及其包含的标准消息；校验失败的帧丢弃。连接后从串口收到的数据源并不自动区分多架飞机，多机共用链路时请勿把仪表作为单机状态依据。
+基于 PyQt6 的 Windows 串口上位机。除请求飞控 SN 外只接收 MAVLink，不发送心跳、参数、解锁或飞行控制命令。支持 MAVLink 2 VKFly 自定义方言及其包含的标准消息；校验失败的帧丢弃。连接后从串口收到的数据源并不自动区分多架飞机，多机共用链路时请勿把仪表作为单机状态依据。
 
 ## 功能
 
 - 选择串口和波特率，显示连接状态。
 - 图形化姿态仪、机体朝向避障雷达；显示位置、地速、高度、GPS、电池、惯导状态和可用的 SN。
 - 消息观察器列出收到的所有已识别消息、ID、数量和最近状态；点击可查看原始字段。3 秒未更新的消息标记为陈旧。
-- SN 只在收到 `AUTOPILOT_VERSION` 或飞控 `VK_COMP_VERSION` 后显示；本程序不会主动请求，未收到显示等待数据。
+- 电源卡片使用最后收到的 `SYS_STATUS` 或 `VK_BMS_STATUS`：前者按 VKFly 协议的 cV 换算（不同于标准 MAVLink 的 mV），后者按 mV 换算；两者分别表示飞控供电与 BMS 电压。
+- 连接后首次收到飞控心跳时，向该飞控发送一次 `MAV_CMD_REQUEST_MESSAGE`，请求飞控的 `VK_COMP_VERSION`；收到回复后显示其中的 SN。未回复时可点击“重新请求 SN”，不自动请求其他消息；多机链路会选择第一个识别到的飞控。
 - 避障图只按 `BODY_FRD`（12）解释数据，过滤协议无效距离；其他坐标系不画图。10m 是显示量程，不代表传感器上限。
 
 **安全说明：**此软件是监视工具，不能替代地面站及飞行安全检查。页面卡片暂显示最后收到的值；判断实时性请看消息观察器中的更新时间。不要在旋翼安装螺旋桨时进行桌面调试。

app.py

@@ -1,8 +1,9 @@
-"""VKFly telemetry monitor. Read-only: never sends arm or flight commands."""
+"""VKFly telemetry monitor. Sends only flight-controller SN requests."""
 import math
 import sys
 import time
 from collections import Counter
+from queue import Empty, Queue
 from PyQt6.QtCore import Qt, QThread, pyqtSignal, QTimer, QRectF
 from PyQt6.QtGui import QColor, QPainter, QPen, QFont, QPolygonF
 from PyQt6.QtCore import QPointF
@@ -20,19 +21,31 @@ class Link(QThread):
     status = pyqtSignal(str)
     def __init__(self, port, baud):
         super().__init__(); self.port=port; self.baud=baud; self.running=True
+        self.sn_requests=Queue()
+    def request_sn(self, target_system):
+        self.sn_requests.put(target_system)
     def run(self):
         try:
-            with serial.Serial(self.port, self.baud, timeout=0.15) as conn:
+            with serial.Serial(self.port, self.baud, timeout=0.15, write_timeout=1) as conn:
                 parser=vkfly.MAVLink(None); parser.robust_parsing=True
-                self.status.emit(f'已连接 {self.port} · {self.baud} · 只读')
+                sender=vkfly.MAVLink(conn, srcSystem=255, srcComponent=vkfly.MAV_COMP_ID_MISSIONPLANNER)
+                self.status.emit(f'已连接 {self.port} · {self.baud} · 仅请求 SN')
                 while self.running:
+                    try:
+                        target_system=self.sn_requests.get_nowait()
+                    except Empty:
+                        pass
+                    else:
+                        sender.command_long_send(target_system, vkfly.MAV_COMP_ID_AUTOPILOT1,
+                            vkfly.MAV_CMD_REQUEST_MESSAGE, 0, vkfly.MAVLINK_MSG_ID_VK_COMP_VERSION,
+                            1, 0, 0, 0, 0, 0)
                     data=conn.read(min(conn.in_waiting or 1, 4096))
                     for byte in data:
                         msg=parser.parse_char(bytes((byte,)))
                         if msg is not None and msg.get_type() != 'BAD_DATA': self.message.emit(msg)
         except Exception as exc:
             self.status.emit(f'串口错误：{exc}')
-        finally:
+        else:
             self.status.emit('连接已关闭')
     def stop(self):
         self.running=False
@@ -82,10 +95,11 @@ class Radar(QWidget):
 
 class Monitor(QMainWindow):
     def __init__(self):
-        super().__init__();self.setWindowTitle('VK 飞控遥测监控');self.resize(1400,900);self.link=None;self.counts=Counter();self.rows={};self.seen={};self.values={};self.selected=None
+        super().__init__();self.setWindowTitle('VK 飞控遥测监控');self.resize(1400,900);self.link=None;self.counts=Counter();self.rows={};self.seen={};self.values={};self.selected=None;self.flight_system=None
         root=QWidget();self.setCentralWidget(root); outer=QVBoxLayout(root)
         bar=QHBoxLayout();self.ports=QComboBox();self.refresh_ports();refresh=QPushButton('刷新串口');refresh.clicked.connect(self.refresh_ports);self.baud=QComboBox();self.baud.addItems(['115200','230400','460800','57600','921600']);self.button=QPushButton('连接');self.button.clicked.connect(self.toggle)
-        self.status=QLabel('未连接 · 被动接收，不发送控制指令');bar.addWidget(QLabel('串口'));bar.addWidget(self.ports);bar.addWidget(refresh);bar.addWidget(QLabel('波特率'));bar.addWidget(self.baud);bar.addWidget(self.button);bar.addWidget(self.status,1);outer.addLayout(bar)
+        self.sn_button=QPushButton('重新请求 SN');self.sn_button.setEnabled(False);self.sn_button.clicked.connect(self.request_sn)
+        self.status=QLabel('未连接 · 仅请求 SN，不发送控制指令');bar.addWidget(QLabel('串口'));bar.addWidget(self.ports);bar.addWidget(refresh);bar.addWidget(QLabel('波特率'));bar.addWidget(self.baud);bar.addWidget(self.button);bar.addWidget(self.sn_button);bar.addWidget(self.status,1);outer.addLayout(bar)
         splitter=QSplitter(Qt.Orientation.Horizontal);outer.addWidget(splitter,1)
         left=QWidget();ll=QVBoxLayout(left); title=QLabel('飞行态势');title.setObjectName('title');ll.addWidget(title)
         visuals=QHBoxLayout();self.att=Attitude();self.radar=Radar();visuals.addWidget(self.att,1);visuals.addWidget(self.radar,1);ll.addLayout(visuals)
@@ -102,16 +116,21 @@ class Monitor(QMainWindow):
         if idx>=0:self.ports.setCurrentIndex(idx)
     def toggle(self):
         if self.link:
-            self.link.stop();self.link=None;self.button.setText('连接');return
+            self.link.stop();self.link=None;self.flight_system=None;self.sn_button.setEnabled(False);self.button.setText('连接');return
         port=self.ports.currentText()
         if not port:self.status.setText('没有可用串口，请刷新');return
-        self.counts.clear();self.rows.clear();self.seen.clear();self.values.clear();self.table.setRowCount(0);self.detail.clear();self.selected=None
+        self.counts.clear();self.rows.clear();self.seen.clear();self.values.clear();self.table.setRowCount(0);self.detail.clear();self.selected=None;self.flight_system=None;self.sn_button.setEnabled(False)
         for label in self.labels.values(): label.setText('等待数据')
         self.att.roll=self.att.pitch=self.att.yaw=None;self.att.update();self.radar.points=[];self.radar.update()
         self.link=Link(port,int(self.baud.currentText()));self.link.message.connect(self.receive);self.link.status.connect(self.set_status);self.link.finished.connect(self.finished);self.button.setText('断开');self.link.start()
     def set_status(self,s):self.status.setText(s)
     def finished(self):
-        if self.link and not self.link.isRunning():self.link=None;self.button.setText('连接')
+        if self.link and not self.link.isRunning():
+            self.link=None;self.flight_system=None;self.sn_button.setEnabled(False);self.button.setText('连接')
+    def request_sn(self):
+        if self.link and self.flight_system is not None:
+            self.link.request_sn(self.flight_system)
+            self.labels['sn'].setText('已请求，等待飞控回复')
     def receive(self,m):
         name=m.get_type(); self.counts[name]+=1;self.seen[name]=time.monotonic();self.values[name]=m.to_dict()
         if name not in self.rows:
@@ -123,16 +142,18 @@ class Monitor(QMainWindow):
             self.att.roll=m.roll;self.att.pitch=m.pitch;self.att.yaw=m.yaw;self.att.update()
         elif name=='HEARTBEAT':
             armed=bool(m.base_mode & 128);self.labels['mode'].setText(f'{"已解锁" if armed else "已上锁"}  ·  模式 {m.custom_mode}\n系统 {m.get_srcSystem()} / 组件 {m.get_srcComponent()}')
+            if self.link and self.flight_system is None and m.get_srcSystem()>0 and m.get_srcComponent()==vkfly.MAV_COMP_ID_AUTOPILOT1 and m.autopilot != vkfly.MAV_AUTOPILOT_INVALID:
+                self.flight_system=m.get_srcSystem();self.sn_button.setEnabled(True);self.request_sn()
         elif name=='GLOBAL_POSITION_INT':
             self.labels['location'].setText(f'{m.lat/1e7:.7f}°, {m.lon/1e7:.7f}°');self.labels['altitude'].setText(f'海拔 {m.alt/1000:.1f} m · 相对 {m.relative_alt/1000:.1f} m');self.labels['velocity'].setText(f'水平 {math.hypot(m.vx,m.vy)/100:.1f} m/s · 上升 {-m.vz/100:.1f} m/s')
         elif name=='GPS2_RAW':self.labels['gps'].setText(f'{m.satellites_visible} 颗 · fix_type {m.fix_type}')
         elif name=='SYS_STATUS':
-            v='未知' if m.voltage_battery==65535 else f'{m.voltage_battery/1000:.1f} V'; pct='未知' if m.battery_remaining<0 else f'{m.battery_remaining}%';self.labels['power'].setText(f'{v} · 剩余 {pct}')
+            # VKFly uses centivolts here, unlike standard MAVLink's millivolts.
+            v='未知' if m.voltage_battery==65535 else f'{m.voltage_battery/100:.1f} V'; pct='未知' if m.battery_remaining<0 else f'{m.battery_remaining}%';self.labels['power'].setText(f'{v} · 剩余 {pct}')
         elif name=='VK_BMS_STATUS':self.labels['power'].setText(f'{m.voltage/1000:.1f} V · 剩余 {m.cap_percent}% · 电池 {m.bat_id}')
         elif name=='VKINS_STATUS':self.labels['ins'].setText(f'状态位 0x{m.nav_status:02X} · 定位星 {m.solv_psat_num} · 定向星 {m.solv_hsat_num}')
-        elif name=='AUTOPILOT_VERSION':self.labels['sn'].setText(str(m.uid))
-        elif name=='VK_COMP_VERSION' and m.comp_id==1:
-            val=m.SN; self.labels['sn'].setText(bytes(val).split(b'\0')[0].decode('ascii','replace'))
+        elif name=='VK_COMP_VERSION' and m.comp_id==1 and (self.flight_system is None or m.get_srcSystem()==self.flight_system):
+            val=bytes(m.SN).split(b'\0')[0].decode('ascii','replace');self.labels['sn'].setText(val or '飞控未提供 SN')
         elif name=='OBSTACLE_DISTANCE':
             # This plot assumes MAV_FRAME_BODY_FRD and the MAVLink angular convention.
             if m.frame != 12:self.radar.points=[]

tests/test_power.py

@@ -0,0 +1,48 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+
+os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from PyQt6.QtWidgets import QApplication
+from app import Monitor
+from generated import vkfly
+
+app = QApplication.instance() or QApplication([])
+
+
+class PowerDisplayTest(unittest.TestCase):
+    def setUp(self):
+        self.monitor = Monitor()
+        self.writer = vkfly.MAVLink(None, srcSystem=1, srcComponent=1)
+
+    def tearDown(self):
+        self.monitor.close()
+
+    def deliver(self, message):
+        reader = vkfly.MAVLink(None)
+        decoded = None
+        for byte in message.pack(self.writer):
+            result = reader.parse_char(bytes([byte]))
+            if result is not None:
+                decoded = result
+        self.assertIsNotNone(decoded)
+        self.monitor.receive(decoded)
+        return self.monitor.labels['power'].text()
+
+    def test_sys_status_uses_centivolts(self):
+        msg = self.writer.sys_status_encode(0, 0, 0, 0, 2400, 0, 75, 0, 0, 0, 0, 0, 0)
+        self.assertEqual(self.deliver(msg), '24.0 V · 剩余 75%')
+
+    def test_sys_status_unknown_voltage_and_remaining(self):
+        msg = self.writer.sys_status_encode(0, 0, 0, 0, 65535, 0, -1, 0, 0, 0, 0, 0, 0)
+        self.assertEqual(self.deliver(msg), '未知 · 剩余 未知')
+
+    def test_bms_uses_millivolts(self):
+        msg = self.writer.vk_bms_status_encode(0, 24000, 0, 0, 80, 1, 0, 0, [0] * 30, 0, 0)
+        self.assertEqual(self.deliver(msg), '24.0 V · 剩余 80% · 电池 1')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_sn_request.py

@@ -0,0 +1,117 @@
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from PyQt6.QtWidgets import QApplication
+from app import Link, Monitor
+from generated import vkfly
+
+app = QApplication.instance() or QApplication([])
+
+
+class FakeLink:
+    def __init__(self):
+        self.requests = []
+
+    def request_sn(self, target_system):
+        self.requests.append(target_system)
+
+    def stop(self):
+        pass
+
+
+class SnRequestTest(unittest.TestCase):
+    def setUp(self):
+        self.monitor = Monitor()
+        self.link = FakeLink()
+        self.monitor.link = self.link
+        self.writer = vkfly.MAVLink(None, srcSystem=42, srcComponent=1)
+
+    def tearDown(self):
+        self.monitor.close()
+
+    def decode(self, message, writer):
+        reader = vkfly.MAVLink(None)
+        decoded = None
+        for byte in message.pack(writer):
+            result = reader.parse_char(bytes([byte]))
+            if result is not None:
+                decoded = result
+        self.assertIsNotNone(decoded)
+        return decoded
+
+    def test_auto_request_once_and_manual_retry(self):
+        heartbeat = self.decode(self.writer.heartbeat_encode(2, 3, 0, 0, 4, 3), self.writer)
+        self.monitor.receive(heartbeat)
+        self.monitor.receive(heartbeat)
+        self.assertEqual(self.link.requests, [42])
+        self.assertTrue(self.monitor.sn_button.isEnabled())
+        self.monitor.sn_button.click()
+        self.assertEqual(self.link.requests, [42, 42])
+        self.assertEqual(self.monitor.labels['sn'].text(), '已请求，等待飞控回复')
+
+    def test_ignores_non_autopilot_heartbeat(self):
+        other = vkfly.MAVLink(None, srcSystem=99, srcComponent=1)
+        self.monitor.receive(self.decode(other.heartbeat_encode(6, vkfly.MAV_AUTOPILOT_INVALID, 0, 0, 0, 3), other))
+        self.assertEqual(self.link.requests, [])
+        self.assertFalse(self.monitor.sn_button.isEnabled())
+
+    def test_displays_flight_controller_sn_not_other_component(self):
+        self.monitor.receive(self.decode(self.writer.heartbeat_encode(2, 3, 0, 0, 4, 3), self.writer))
+        zeros = [0] * 16
+        sn = list(b'VK123') + [0] * 11
+        msg = self.writer.vk_comp_version_encode(1, zeros, zeros, sn, zeros, zeros, zeros)
+        reader = vkfly.MAVLink(None)
+        decoded = None
+        for byte in msg.pack(self.writer):
+            result = reader.parse_char(bytes([byte]))
+            if result is not None:
+                decoded = result
+        self.monitor.receive(decoded)
+        self.assertEqual(self.monitor.labels['sn'].text(), 'VK123')
+        other = self.writer.vk_comp_version_encode(3, zeros, zeros, zeros, zeros, zeros, zeros)
+        self.monitor.receive(other)
+        self.assertEqual(self.monitor.labels['sn'].text(), 'VK123')
+
+    def test_serial_writes_only_sn_request(self):
+        link = Link('fake', 115200)
+        link.request_sn(42)
+        writes = []
+
+        class FakeSerial:
+            in_waiting = 0
+            def __init__(self, *args, **kwargs):
+                pass
+            def __enter__(self):
+                return self
+            def __exit__(self, *args):
+                pass
+            def write(self, data):
+                writes.append(bytes(data))
+                return len(data)
+            def read(self, size):
+                link.running = False
+                return b''
+
+        with patch('app.serial.Serial', FakeSerial):
+            link.run()
+        self.assertEqual(len(writes), 1)
+        reader = vkfly.MAVLink(None)
+        msg = None
+        for byte in writes[0]:
+            result = reader.parse_char(bytes([byte]))
+            if result is not None:
+                msg = result
+        self.assertEqual(msg.get_type(), 'COMMAND_LONG')
+        self.assertEqual((msg.target_system, msg.target_component), (42, 1))
+        self.assertEqual(msg.command, vkfly.MAV_CMD_REQUEST_MESSAGE)
+        self.assertEqual((msg.param1, msg.param2), (vkfly.MAVLINK_MSG_ID_VK_COMP_VERSION, 1))
+        self.assertEqual((msg.param3, msg.param4, msg.param5, msg.param6, msg.param7), (0, 0, 0, 0, 0))
+
+
+if __name__ == '__main__':
+    unittest.main()