Add per-aircraft state object and automatic parameter sync

Group each locked flight controller's identity, latest observations, battery packs and downloaded parameter table into an Aircraft object, exposed via Telemetry compatibility properties.

After locking a HEARTBEAT, request PARAM_REQUEST_LIST once to download the full parameter table into memory, retrying within param_request_timeout up to param_request_max_attempts. Add request_params_on_start, param_request_timeout and param_request_max_attempts config options, and reuse request_params for DataType 281.

AGENTS.md

@@ -0,0 +1,88 @@
+# AGENTS.md
+
+本项目是**摩天云 MQTT–MAVLink 串口桥**(`motian-mavlink-bridge`)。它连接飞控串口与摩天云 MQTT Broker:被动解析串口上的 MAVLink 帧,按平台 Base JSON 协议上报遥测,并把平台下行指令翻译成 VKFly MAVLink 帧写回串口。详细协议说明见 `README.md`,实施方案见 `doc/实施方案.md`。
+
+## 技术栈与运行环境
+
+- Python **>= 3.10**(代码使用 `X | None` 等语法),Linux 部署。
+- 运行依赖:`paho-mqtt>=2,<3`、`pyserial>=3.5,<4`、`pymavlink>=2.4.50,<3`(见 `pyproject.toml`)。
+- MAVLink 方言 `VKFly` 由 `msg_definitions` Git 子模块的 XML 生成。
+- 测试用标准库 `unittest`,不使用 pytest。
+
+## 目录结构
+
+```
+motian_bridge/
+  cli.py          生产入口,创建 paho-mqtt / pyserial 实例并注入 Bridge
+  bridge.py       Bridge 核心:MQTT 收发、串口读写线程、Base JSON 编解码、handler 注册、参数表自动同步
+  config.py       Config 冻结 dataclass,含校验;密钥从环境变量读取
+  aircraft.py     Aircraft:按飞控聚合身份、最新观测、电池与参数表(ParamTable)
+  telemetry.py    MAVLink 帧解析与上报节流,内部持有一个 Aircraft(仅串口线程使用)
+  commands.py     平台业务指令 -> VKFly MAVLink 帧的 handler
+  dialects/v20/VKFly.py   ⚠️ 生成文件,禁止手工修改
+tests/            unittest 用例(依赖 FakeMqtt/FakeSerial,test_commands 复用 test_bridge 的辅助函数)
+tools/            local_mqtt_broker.py,本地 aMQTT 测试 Broker(非生产依赖)
+doc/              平台接口规范、实施方案、qa.md(待平台澄清问题)
+config.example.json   配置模板
+```
+
+## 环境搭建与常用命令
+
+```bash
+python3 -m venv .venv
+.venv/bin/python -m pip install -e .          # 安装(可编辑模式)
+cp config.example.json config.local.json      # 再按需修改真实参数
+
+# 跑全部测试(必须是 -s tests,用例间有相互导入)
+.venv/bin/python -m unittest discover -s tests -v
+
+# 启动桥接服务
+export MOTIAN_MQTT_PASSWORD='由部署环境提供'
+.venv/bin/motian-mavlink-bridge --config config.local.json --debug
+
+# 本地联调:先起本地 Broker(需先 pip install "amqtt>=0.12,<0.13")
+.venv/bin/python tools/local_mqtt_broker.py
+```
+
+## 架构与代码约定
+
+- **依赖注入**:`Bridge` 只接收抽象的 MQTT client 与 `serial_factory`,具体实现由 `cli.py` 提供,便于测试用 fake 替换。保持这一模式。
+- **线程边界**:串口读取线程(`_serial_loop`)独占 `Telemetry` 解析器与其内部的 `Aircraft` 状态;MQTT 回调线程负责下发。跨线程共享用 `threading.Lock` / `threading.Event`,不要引入全局可变状态。`Aircraft` 的 `latest`/`batteries`/`params` 都只允许串口线程写入,`Telemetry` 保留 `source`/`latest`/`params`/`batteries` 兼容属性。
+- **配置**:所有配置项加到 `Config` 冻结 dataclass,并在 `__post_init__` 里做校验(非法值直接抛 `ValueError`)。**密钥、密码只从环境变量读取**(`mqtt_password_env`),绝不写进 JSON 或代码。
+- **风格**:4 空格缩进、双引号、类型标注、模块级 `LOG = logging.getLogger(__name__)`。判断类型用 `type(x) is int` 这类精确判断(避免 `bool` 被当作 `int` 接受)。
+- **扩展新平台指令**:在 `FlightCommandHandlers.install()` 里 `register_handler(data_type, handler)`,handler 返回时通过 `bridge.publish_report(data_type, data)` 上报。**未实现/未确认的指令只记日志并丢弃,绝不伪造成功 ACK**。
+- 文档、注释用中文;代码标识符和 **git 提交信息用英文祈使句**(如 `Add platform command forwarding callbacks`)。
+
+## 协议约定与红线
+
+- 上行原始透传固定为 `DataType=526` + 大写 HEX `Data`,发到 `/PUBLISH/<SystemID>`;结构化上报发到 `/RECEIVE/<SystemID>`。字段名是 `DataType`(**没有尾随空格**)。
+- MQTT 使用 **QoS 0、不 retain**;收到的 retained 下行一律丢弃;断线不缓存、不重放。
+- 下行报文必须有 `DataType`/`UAVType`/`Data`/`SystemID`/`TimeStamp`,且设备标识须与本机配置一致;`Data` 必须是非空、偶数长度 HEX,不超过 `max_command_bytes`。
+- 业务指令在**尚未收到飞控 HEARTBEAT、未锁定目标 system/component 前禁止下发**(`_target()` 会抛错保护)。
+- **不要修改** `motian_bridge/dialects/v20/VKFly.py`——它是生成文件。XML 变更后重新生成:
+
+  ```bash
+  .venv/bin/python -m pymavlink.tools.mavgen --lang Python --wire-protocol 2.0 \
+      --output motian_bridge/dialects/v20/VKFly.py \
+      msg_definitions/msg_definitions/VKFly.xml
+  ```
+
+- `Telemetry` 中 `VKFly.MAVLINK_IGNORE_CRC = 0` 是安全底线,**不要**为调试方便关闭 CRC 校验。
+- 不要提交 `config.local.json`、`.venv/`、`*.pdf`(见 `.gitignore`);
+- 不要降低 TLS 强度(`mqtt_ca_file` 校验、证书配置等保持开启)。
+
+## 测试要求
+
+- 新增行为必须补 `tests/` 下的 `unittest` 用例;提交前必须通过:
+  `.venv/bin/python -m unittest discover -s tests -v`
+- 涉及串口/MQTT 的用例用 `FakeMqtt` / `FakeSerial`(见 `tests/test_bridge.py`);涉及真实帧编解码时用 `VKFly.MAVLink` 构造帧,参照 `tests/test_telemetry.py`。
+- 涉及配置校验的改动,同步更新 `ValidationTests`。
+
+## 常见陷阱
+
+- 串口**重连会重置 `Telemetry`**(可能换了一架飞控),旧状态必须丢弃,不要跨重连复用遥测。
+- `Telemetry` 逐字节喂给 `parse_char`,每次只返回一条消息;分块读取不影响帧边界,不要改成整块喂入。
+- `STATUSTEXT` 的文本需绕过生成的 Python 解码器、手动按 `char[50]` 读 UTF-8(生成代码按 ASCII 处理会丢字符),改动这里前先看 `telemetry.py` 注释。
+- 除**锁定 HEARTBEAT 后主动同步一次参数表**外,桥接程序只被动接收,不主动向飞控请求消息流;缺字段时先确认飞控是否在广播对应消息(`--debug` 可看到 missing/stale 日志)。参数同步由 `request_params_on_start`(默认开)控制,超时/重试见 `param_request_timeout`、`param_request_max_attempts`;`#281` 与自动同步共用 `commands.py` 的 `request_params()`,改动需同步更新 `test_commands.ParameterSyncTests`。
+- `#274` 的 `true` 表示**解锁**,不是物理离地;`DataType 258` 的 `IsFlying` 也按解锁解释。
+- 部分平台参数映射尚未由飞控协议确认(`commands.py` 的 `TODO_PARAMS`)以及 `#2050 Data=true` 的“接管”语义,均保留 TODO,**不要擅自映射或发送**;需先澄清可记入 `doc/qa.md`。

README.md

@@ -13,7 +13,11 @@
 | 260 提示 | STATUSTEXT | 收到就上报，保留原始严重级别和 UTF-8 文本 |
 | 274 飞行状态 | HEARTBEAT 解锁标志 | 首次获知及改变时上报布尔值；**true=解锁、false=上锁，并非实际离地/着陆** |
 
-只解析首个有效飞控 HEARTBEAT 对应的 system/component ID。串口重连重选来源并清空缓存；心跳超过 5 秒暂停 #258，其他来源字段超过 5 秒自动剔除。只被动接收，不请求飞控遥测；如缺少字段，请先确认飞控串口是否广播对应消息。缺失消息类型可用 `--debug` 查看。`#257` 在双电池时不假定并联/串联或容量一致，不填写推算的总电压/电量。
+只解析首个有效飞控 HEARTBEAT 对应的 system/component ID。串口重连重选来源并清空缓存；心跳超过 5 秒暂停 #258，其他来源字段超过 5 秒自动剔除。除下面所述的参数表自动同步外，只被动接收，不请求飞控遥测；如缺少字段，请先确认飞控串口是否广播对应消息。缺失消息类型可用 `--debug` 查看。`#257` 在双电池时不假定并联/串联或容量一致，不填写推算的总电压/电量。
+
+## 参数表自动同步
+
+锁定飞控 HEARTBEAT 后，默认会主动发送一次 `PARAM_REQUEST_LIST`，把飞控回传的 `PARAM_VALUE` 累积到本机该飞机的参数表（仅存内存，不上报、不落盘）。`param_request_timeout` 秒内未收齐时按 `param_request_max_attempts` 重试（默认 15 秒、最多 3 次），收齐或放弃后停止。串口重连视为换了飞控，会重新同步一次。将 `request_params_on_start` 设为 `false` 可关闭该行为，回到完全被动接收。`#282` 中需要读取现值的参数（如 `FS_CONF_A`）会用到这张表；平台下发的 `#281` 复用同一套请求逻辑，不受该开关影响。
 
 ## 安装和运行
 
@@ -60,11 +64,11 @@ Bridge 在收到飞控 HEARTBEAT 并锁定目标 system/component 后，支持
 | `obstacle_avoidance_distance` | `OBAVOID_DIST` |
 | `return_type` | `RTL_EXEC_MODE`，直线/正序/逆序分别为 0/1/2 |
 | `precision_land_mode` | `RTL_PLD_EN`，关闭/二维码引导分别为 0/1 |
-| `low_battery_action` | 修改 `FS_CONF_A` 的低两级电量保护位；必须先执行 #281 并收到当前 `FS_CONF_A`，以保留其他保护位 |
+| `low_battery_action` | 修改 `FS_CONF_A` 的低两级电量保护位；需先获得当前 `FS_CONF_A`（参数表自动同步或执行 #281 后均可），以保留其他保护位 |
 
 以下参数映射尚未由飞控协议明确，代码仅记录 TODO 并拒绝写串口：`battery_emergency_threshold`、`geofence_action`、`visual_sensor_enable`、`april_tag_id`、`data_link_loss_action`、`rc_loss_action`、`parachute_safety_enable`、`four_g_lost_action`、`rtk_switch`。未知参数同样不会直接作为 param_id 发送。
 
-这些业务指令不会在尚未收到飞控 HEARTBEAT 时发送，以避免使用错误的目标 system/component。当前只负责下发帧；COMMAND_ACK、PARAM_VALUE 到平台业务应答的关联与超时重试尚未实现。
+这些业务指令不会在尚未收到飞控 HEARTBEAT 时发送，以避免使用错误的目标 system/component。当前只负责下发帧；COMMAND_ACK 到平台业务应答的关联与超时重试尚未实现。`PARAM_VALUE` 会缓存在本机飞机对象的参数表中，但不向平台转发。
 
 ## 没有远程平台时的串口实测
 
@@ -119,7 +123,8 @@ cp config.example.json config.local.json
   "mqtt_username": null,
   "mqtt_password_env": null,
   "mqtt_tls": false,
-  "publish_raw_mavlink": true
+  "publish_raw_mavlink": true,
+  "request_params_on_start": true
 }
 ```
 

config.example.json

@@ -16,5 +16,8 @@
   "max_chunk_bytes": 1024,
   "max_command_bytes": 4096,
   "reconnect_seconds": 2.0,
-  "publish_raw_mavlink": true
+  "publish_raw_mavlink": true,
+  "request_params_on_start": true,
+  "param_request_timeout": 15.0,
+  "param_request_max_attempts": 3
 }

motian_bridge/aircraft.py

@@ -0,0 +1,229 @@
+"""Per-flight-controller state grouped into a single aircraft object.
+
+An :class:`Aircraft` is created only after the first valid flight-controller
+HEARTBEAT is locked by :class:`~motian_bridge.telemetry.Telemetry`. It owns the
+identity (system/component), the latest raw observations, the vendor battery
+packs and the downloaded parameter table. A serial reconnect discards the whole
+object, exactly like :class:`Telemetry` itself.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import math
+from typing import Any
+
+from .config import Config
+from .dialects.v20 import VKFly
+
+
+# Observations older than this are never used as current flight state.
+STALE_SECONDS = 5.0
+# VKFly does not define a parameter index for unknown entries.
+UNKNOWN_PARAM_INDEX = 0xFFFF
+
+
+def enum_label(enum: str, value: int, prefix: str) -> str:
+    entry = VKFly.enums.get(enum, {}).get(value)
+    return entry.name.removeprefix(prefix) if entry else f"UNKNOWN_{value}"
+
+
+def distance_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    r1, r2 = math.radians(lat1), math.radians(lat2)
+    dr = r2 - r1
+    dl = math.radians(lon2 - lon1)
+    a = math.sin(dr / 2) ** 2 + math.cos(r1) * math.cos(r2) * math.sin(dl / 2) ** 2
+    return 6371000.0 * 2 * math.asin(min(1.0, math.sqrt(a)))
+
+
+def _put_finite(data: dict[str, Any], name: str, value: float) -> None:
+    if math.isfinite(value):
+        data[name] = value
+
+
+@dataclass
+class ParamEntry:
+    """One downloaded parameter, with the metadata VKFly reports."""
+
+    value: float
+    param_type: int | None = None
+    index: int | None = None
+
+
+class ParamTable:
+    """The full parameter table of one aircraft, filled from PARAM_VALUE frames."""
+
+    def __init__(self) -> None:
+        self.entries: dict[str, ParamEntry] = {}
+        # param_count from the flight controller, once it has been seen.
+        self.count: int | None = None
+        # param_index values already received, used to detect a complete download.
+        self.received: set[int] = set()
+
+    def update(self, message: Any) -> None:
+        """Record one PARAM_VALUE message."""
+        param_id = message.param_id
+        if isinstance(param_id, bytes):
+            param_id = param_id.decode("ascii", "replace")
+        param_id = param_id.rstrip("\x00")
+        index = int(message.param_index)
+        self.entries[param_id] = ParamEntry(
+            value=float(message.param_value),
+            param_type=int(message.param_type),
+            index=index,
+        )
+        count = int(message.param_count)
+        if count > 0:
+            self.count = count
+        if index != UNKNOWN_PARAM_INDEX:
+            self.received.add(index)
+
+    def __setitem__(self, param_id: str, value: float | ParamEntry) -> None:
+        entry = value if isinstance(value, ParamEntry) else ParamEntry(float(value))
+        self.entries[param_id] = entry
+
+    def __contains__(self, param_id: object) -> bool:
+        return param_id in self.entries
+
+    def __len__(self) -> int:
+        return len(self.entries)
+
+    def get(self, param_id: str, default: float | None = None) -> float | None:
+        entry = self.entries.get(param_id)
+        return entry.value if entry is not None else default
+
+    def complete(self) -> bool:
+        """True once every indexed parameter has been received."""
+        return self.count is not None and self.count > 0 and len(self.received) >= self.count
+
+
+class Aircraft:
+    """State that belongs to a single flight controller."""
+
+    def __init__(self, config: Config, system: int, component: int,
+                 autopilot: int, uav_type: int) -> None:
+        self.config = config
+        self.system = system
+        self.component = component
+        self.autopilot = autopilot
+        self.uav_type = uav_type
+        self.latest: dict[str, tuple[Any, float]] = {}
+        self.batteries: dict[int, tuple[Any, float]] = {}
+        self.params = ParamTable()
+        # Automatic parameter download bookkeeping.
+        self.param_sync_requested = False
+        self.param_sync_at: float | None = None
+        self.param_request_attempts = 0
+
+    @property
+    def source(self) -> tuple[int, int]:
+        return (self.system, self.component)
+
+    def fresh(self, kind: str, now: float) -> Any | None:
+        entry = self.latest.get(kind)
+        return entry[0] if entry and 0 <= now - entry[1] <= STALE_SECONDS else None
+
+    def param_value(self, param_id: str) -> float | None:
+        return self.params.get(param_id)
+
+    def flight_data(self, now: float, heartbeat: Any) -> dict[str, Any]:
+        data: dict[str, Any] = {
+            "SystemID": self.config.system_id,
+            "Armed": bool(heartbeat.base_mode & VKFly.MAV_MODE_FLAG_SAFETY_ARMED),
+            "IsFlying": bool(heartbeat.base_mode & VKFly.MAV_MODE_FLAG_SAFETY_ARMED),
+            "DroneAutopilot": int(heartbeat.autopilot),
+            "UavType": int(heartbeat.type),
+            "FlightMode": enum_label(
+                "VKFLY_CUSTOM_MODE", heartbeat.custom_mode, "VKFLY_CUSTOM_MODE_"
+            ),
+        }
+        attitude = self.fresh("ATTITUDE", now)
+        if attitude:
+            _put_finite(data, "Roll", math.degrees(attitude.roll))
+            _put_finite(data, "Pitch", math.degrees(attitude.pitch))
+            _put_finite(data, "Yaw", math.degrees(attitude.yaw) % 360)
+        position = self.fresh("GLOBAL_POSITION_INT", now)
+        if position:
+            if abs(position.lat) <= 900000000 and abs(position.lon) <= 1800000000:
+                data.update(Latitude=position.lat / 1e7, Longitude=position.lon / 1e7)
+            data.update(
+                Altitude=position.relative_alt / 1000,
+                VelocityX=position.vx / 100,
+                VelocityY=position.vy / 100,
+                VelocityZ=position.vz / 100,
+            )
+        hud = self.fresh("VFR_HUD", now)
+        if hud:
+            for key, value in (
+                ("AirSpeed", hud.airspeed),
+                ("GroundSpeed", hud.groundspeed),
+                ("ClimbRate", hud.climb),
+            ):
+                _put_finite(data, key, value)
+        home = self.fresh("HOME_POSITION", now)
+        if (
+            home
+            and abs(home.latitude) <= 900000000
+            and abs(home.longitude) <= 1800000000
+        ):
+            data.update(
+                HomeLatitude=home.latitude / 1e7,
+                HomeLongitude=home.longitude / 1e7,
+                HomeAltitude=home.altitude / 1000,
+                IsHomeLocationSet=True,
+            )
+            if position and "Latitude" in data:
+                distance = distance_m(
+                    data["Latitude"],
+                    data["Longitude"],
+                    data["HomeLatitude"],
+                    data["HomeLongitude"],
+                )
+                data["DistanceToHome"] = distance
+                data["SpaceDistToHome"] = math.hypot(
+                    distance, position.alt / 1000 - data["HomeAltitude"]
+                )
+        gps = self.fresh("GPS_RAW_INT", now)
+        if gps:
+            info: dict[str, Any] = {
+                "Type": 15,
+                "FixType": enum_label("GPS_FIX_TYPE", gps.fix_type, "GPS_FIX_TYPE_"),
+            }
+            if gps.fix_type >= 2:
+                info["AbsoluteAlt"] = gps.alt / 1000
+            if gps.eph != 65535:
+                info["LevelDivisor"] = gps.eph / 100
+            if gps.epv != 65535:
+                info["VerticalDivisor"] = gps.epv / 100
+            if gps.satellites_visible != 255:
+                info["SatelliteCount"] = gps.satellites_visible
+            data["Gps"] = info
+        return data
+
+    def battery_data(self, now: float) -> dict[str, Any] | None:
+        packs = {
+            id_: item[0]
+            for id_, item in self.batteries.items()
+            if 0 <= now - item[1] <= STALE_SECONDS
+        }
+        if not packs:
+            return None
+        data: dict[str, Any] = {"SystemID": self.config.system_id}
+        for id_, pack in packs.items():
+            side = "Left" if id_ == 0 else "Right"
+            if pack.voltage:
+                data["Battery_Voltage" + side] = pack.voltage / 1000
+            data["Battery_Current" + side] = pack.current / 100
+            if 0 <= pack.cap_percent <= 100:
+                data["Battery_Remaining" + side] = pack.cap_percent
+        if len(packs) == 1:
+            pack = next(iter(packs.values()))
+            if pack.voltage:
+                data["Battery_Voltage"] = pack.voltage / 1000
+            data["Battery_Current"] = pack.current / 100
+            if 0 <= pack.cap_percent <= 100:
+                data["Battery_Remaining"] = pack.cap_percent
+            data["Battery_Temperature"] = pack.temperature
+            data["cycleIndex"] = pack.cyc_cnt
+            data["flags"] = pack.err_code
+        return data

motian_bridge/bridge.py

@@ -6,6 +6,7 @@ import json
 import logging
 import re
 import threading
+import time
 from typing import Any, Callable
 
 from .config import Config
@@ -249,5 +250,38 @@ class Bridge:
                         self.publish_report(data_type, report)
                 for data_type, report in self._telemetry.tick():
                     self.publish_report(data_type, report)
+                self._sync_parameters(time.monotonic())
             except Exception:
                 LOG.exception("telemetry processing failed; serial link preserved")
+
+    def _sync_parameters(self, now: float) -> None:
+        """Download the full parameter table once after locking a flight controller."""
+        if not self.config.request_params_on_start:
+            return
+        aircraft = self._telemetry.aircraft
+        if aircraft is None:
+            return
+        if not aircraft.param_sync_requested:
+            aircraft.param_sync_requested = True
+            aircraft.param_sync_at = now
+            aircraft.param_request_attempts = 1
+            LOG.info("requesting full parameter table from sys=%s comp=%s", *aircraft.source)
+            self._flight_commands.request_params(*aircraft.source)
+            return
+        if aircraft.params.complete():
+            return
+        if aircraft.param_request_attempts >= self.config.param_request_max_attempts:
+            return
+        last_request = aircraft.param_sync_at if aircraft.param_sync_at is not None else now
+        if now - last_request < self.config.param_request_timeout:
+            return
+        aircraft.param_sync_at = now
+        aircraft.param_request_attempts += 1
+        LOG.warning(
+            "parameter table incomplete (%d/%s); re-requesting (%d/%d)",
+            len(aircraft.params.received),
+            aircraft.params.count if aircraft.params.count is not None else "?",
+            aircraft.param_request_attempts,
+            self.config.param_request_max_attempts,
+        )
+        self._flight_commands.request_params(*aircraft.source)

motian_bridge/commands.py

@@ -107,6 +107,10 @@ class FlightCommandHandlers:
         if command.data is not None:
             raise ValueError("DataType 281 Data must be null")
         system, component = self._target()
+        self.request_params(system, component)
+
+    def request_params(self, system: int, component: int) -> None:
+        """Send PARAM_REQUEST_LIST to download the flight controller's table."""
         self._send(self.encoder.param_request_list_encode(system, component))
 
     def set_param(self, command: BaseMessage, bridge: Bridge) -> None:

motian_bridge/config.py

@@ -26,6 +26,9 @@ class Config:
     max_command_bytes: int = 4096
     reconnect_seconds: float = 2.0
     publish_raw_mavlink: bool = True
+    request_params_on_start: bool = True
+    param_request_timeout: float = 15.0
+    param_request_max_attempts: int = 3
 
     def __post_init__(self) -> None:
         if not self.system_id or any(c in self.system_id for c in "/+#\x00"):
@@ -42,6 +45,12 @@ class Config:
             raise ValueError("invalid keepalive or reconnect interval")
         if type(self.publish_raw_mavlink) is not bool:
             raise ValueError("publish_raw_mavlink must be a boolean")
+        if type(self.request_params_on_start) is not bool:
+            raise ValueError("request_params_on_start must be a boolean")
+        if self.param_request_timeout <= 0:
+            raise ValueError("param_request_timeout must be positive")
+        if type(self.param_request_max_attempts) is not int or self.param_request_max_attempts < 1:
+            raise ValueError("param_request_max_attempts must be a positive integer")
         if self.mqtt_password_env and not self.mqtt_username:
             raise ValueError("mqtt_username is required when mqtt_password_env is set")
         if (self.mqtt_cert_file or self.mqtt_key_file or self.mqtt_ca_file) and not self.mqtt_tls:

motian_bridge/telemetry.py

@@ -1,14 +1,16 @@
 """State derived from checksummed VKFly MAVLink messages.
 
 The wire parser and cached message sources live on the serial reader thread only.
-Times below are monotonic receive times, never the flight controller's boot clock.
+Observations are grouped per flight controller in an
+:class:`~motian_bridge.aircraft.Aircraft`. Times below are monotonic receive
+times, never the flight controller's boot clock.
 """
 
 import logging
-import math
 import time
 from typing import Any
 
+from .aircraft import Aircraft, ParamTable
 from .config import Config
 from .dialects.v20 import VKFly
 
@@ -17,51 +19,52 @@ LOG = logging.getLogger(__name__)
 # The generated module honors MAV_IGNORE_CRC from the environment by default.
 # Never permit that override for telemetry used as flight state.
 VKFly.MAVLINK_IGNORE_CRC = 0
-STALE_SECONDS = 5.0
 REPORT_SECONDS = 1.0
 _INVALID_AUTOPILOT = VKFly.MAV_AUTOPILOT_INVALID
 _AUTOPILOT_COMPONENT = VKFly.MAV_COMP_ID_AUTOPILOT1
 _ARMED_FLAG = VKFly.MAV_MODE_FLAG_SAFETY_ARMED
 
 
-def _enum_label(enum: str, value: int, prefix: str) -> str:
-    entry = VKFly.enums.get(enum, {}).get(value)
-    return entry.name.removeprefix(prefix) if entry else f"UNKNOWN_{value}"
-
-
-def _distance_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
-    r1, r2 = math.radians(lat1), math.radians(lat2)
-    dr = r2 - r1
-    dl = math.radians(lon2 - lon1)
-    a = math.sin(dr / 2) ** 2 + math.cos(r1) * math.cos(r2) * math.sin(dl / 2) ** 2
-    return 6371000.0 * 2 * math.asin(min(1.0, math.sqrt(a)))
-
-
-def _put_finite(data: dict[str, Any], name: str, value: float) -> None:
-    if math.isfinite(value):
-        data[name] = value
-
-
 class Telemetry:
     """Parse MAVLink v1/v2 bytes and emit (DataType, Data) reports.
 
-    Neither the parser nor state is shared across serial reconnects. Non-flight
-    components and other systems never contribute to the selected UAV's state.
+    Neither the parser nor the current :class:`Aircraft` is shared across serial
+    reconnects. Non-flight components and other systems never contribute to the
+    selected UAV's state.
     """
 
     def __init__(self, config: Config):
         self.config = config
         self.parser = VKFly.MAVLink(None)
-        self.source: tuple[int, int] | None = None
-        self.latest: dict[str, tuple[Any, float]] = {}
-        self.batteries: dict[int, tuple[Any, float]] = {}
-        self.params: dict[str, float] = {}
+        self.aircraft: Aircraft | None = None
+        self._empty_params = ParamTable()
         self.battery_dirty = False
         self.last_flight_report: float | None = None
         self.last_battery_report: float | None = None
         self.last_armed: bool | None = None
         self.last_missing_log: float | None = None
 
+    # Backwards-compatible views onto the currently locked aircraft. These are
+    # empty until the first valid flight-controller HEARTBEAT arrives.
+    @property
+    def source(self) -> tuple[int, int] | None:
+        return self.aircraft.source if self.aircraft is not None else None
+
+    @property
+    def latest(self) -> dict[str, tuple[Any, float]]:
+        return self.aircraft.latest if self.aircraft is not None else {}
+
+    @property
+    def batteries(self) -> dict[int, tuple[Any, float]]:
+        return self.aircraft.batteries if self.aircraft is not None else {}
+
+    @property
+    def params(self) -> ParamTable:
+        return self.aircraft.params if self.aircraft is not None else self._empty_params
+
+    def param_value(self, param_id: str) -> float | None:
+        return self.params.get(param_id)
+
     def feed(self, chunk: bytes, now: float | None = None) -> list[tuple[int, Any]]:
         now = time.monotonic() if now is None else now
         reports: list[tuple[int, Any]] = []
@@ -77,25 +80,30 @@ class Telemetry:
                 reports.extend(self._handle(message, now))
         return reports
 
+    def _lock(self, message: Any, src: tuple[int, int]) -> None:
+        self.aircraft = Aircraft(
+            self.config, src[0], src[1], int(message.autopilot), int(message.type)
+        )
+        LOG.info("locked MAVLink flight controller sys=%s comp=%s", *src)
+
     def _handle(self, message: Any, now: float) -> list[tuple[int, Any]]:
         kind = message.get_type()
         src = (message.get_srcSystem(), message.get_srcComponent())
-        if self.source is None:
+        if self.aircraft is None:
             if (
                 kind != "HEARTBEAT"
                 or src[1] != _AUTOPILOT_COMPONENT
                 or message.autopilot == _INVALID_AUTOPILOT
             ):
                 return []
-            self.source = src
-            LOG.info("locked MAVLink flight controller sys=%s comp=%s", *src)
-        if src != self.source:
+            self._lock(message, src)
+        if src != self.aircraft.source:
             return []
 
         # A BMS frame with an unexpected ID must not appear as a usable pack.
         if kind == "VK_BMS_STATUS" and message.bat_id not in (0, 1):
             return []
-        self.latest[kind] = (message, now)
+        self.aircraft.latest[kind] = (message, now)
         if kind == "HEARTBEAT":
             armed = bool(message.base_mode & _ARMED_FLAG)
             if armed != self.last_armed:
@@ -103,7 +111,7 @@ class Telemetry:
                 # Platform convention: "flying" means armed, not airborne.
                 return [(274, armed)]
         elif kind == "VK_BMS_STATUS" and message.bat_id in (0, 1):
-            self.batteries[message.bat_id] = (message, now)
+            self.aircraft.batteries[message.bat_id] = (message, now)
             self.battery_dirty = True
         elif kind == "STATUSTEXT":
             # Generated Python text decoding assumes ASCII and irreversibly
@@ -117,39 +125,35 @@ class Telemetry:
             if text:
                 return [(260, {"severity": int(message.severity), "msg": text})]
         elif kind == "PARAM_VALUE":
-            param_id = message.param_id
-            if isinstance(param_id, bytes):
-                param_id = param_id.decode("ascii", "replace")
-            self.params[param_id.rstrip("\x00")] = float(message.param_value)
+            self.aircraft.params.update(message)
         return []
 
-    def _fresh(self, kind: str, now: float) -> Any | None:
-        entry = self.latest.get(kind)
-        return entry[0] if entry and 0 <= now - entry[1] <= STALE_SECONDS else None
-
     def tick(self, now: float | None = None) -> list[tuple[int, Any]]:
         now = time.monotonic() if now is None else now
         reports: list[tuple[int, Any]] = []
-        heartbeat = self._fresh("HEARTBEAT", now)
+        aircraft = self.aircraft
+        if aircraft is None:
+            return reports
+        heartbeat = aircraft.fresh("HEARTBEAT", now)
         if heartbeat is not None and (
             self.last_flight_report is None
             or now - self.last_flight_report >= REPORT_SECONDS
         ):
-            reports.append((258, self.flight_data(now, heartbeat)))
+            reports.append((258, aircraft.flight_data(now, heartbeat)))
             self.last_flight_report = now
-            self._log_missing(now)
+            self._log_missing(aircraft, now)
         if self.battery_dirty and (
             self.last_battery_report is None
             or now - self.last_battery_report >= REPORT_SECONDS
         ):
-            battery = self.battery_data(now)
+            battery = aircraft.battery_data(now)
             self.battery_dirty = False
             if battery is not None:
                 reports.append((257, battery))
                 self.last_battery_report = now
         return reports
 
-    def _log_missing(self, now: float) -> None:
+    def _log_missing(self, aircraft: Aircraft, now: float) -> None:
         if self.last_missing_log is not None and now - self.last_missing_log < 60:
             return
         missing = [
@@ -162,110 +166,8 @@ class Telemetry:
                 "HOME_POSITION",
                 "VK_BMS_STATUS",
             )
-            if self._fresh(name, now) is None
+            if aircraft.fresh(name, now) is None
         ]
         if missing:
             LOG.debug("missing/stale MAVLink telemetry: %s", ", ".join(missing))
         self.last_missing_log = now
-
-    def flight_data(self, now: float, heartbeat: Any) -> dict[str, Any]:
-        data: dict[str, Any] = {
-            "SystemID": self.config.system_id,
-            "Armed": bool(heartbeat.base_mode & _ARMED_FLAG),
-            "IsFlying": bool(heartbeat.base_mode & _ARMED_FLAG),
-            "DroneAutopilot": int(heartbeat.autopilot),
-            "UavType": int(heartbeat.type),
-            "FlightMode": _enum_label(
-                "VKFLY_CUSTOM_MODE", heartbeat.custom_mode, "VKFLY_CUSTOM_MODE_"
-            ),
-        }
-        attitude = self._fresh("ATTITUDE", now)
-        if attitude:
-            _put_finite(data, "Roll", math.degrees(attitude.roll))
-            _put_finite(data, "Pitch", math.degrees(attitude.pitch))
-            _put_finite(data, "Yaw", math.degrees(attitude.yaw) % 360)
-        position = self._fresh("GLOBAL_POSITION_INT", now)
-        if position:
-            if abs(position.lat) <= 900000000 and abs(position.lon) <= 1800000000:
-                data.update(Latitude=position.lat / 1e7, Longitude=position.lon / 1e7)
-            data.update(
-                Altitude=position.relative_alt / 1000,
-                VelocityX=position.vx / 100,
-                VelocityY=position.vy / 100,
-                VelocityZ=position.vz / 100,
-            )
-        hud = self._fresh("VFR_HUD", now)
-        if hud:
-            for key, value in (
-                ("AirSpeed", hud.airspeed),
-                ("GroundSpeed", hud.groundspeed),
-                ("ClimbRate", hud.climb),
-            ):
-                _put_finite(data, key, value)
-        home = self._fresh("HOME_POSITION", now)
-        if (
-            home
-            and abs(home.latitude) <= 900000000
-            and abs(home.longitude) <= 1800000000
-        ):
-            data.update(
-                HomeLatitude=home.latitude / 1e7,
-                HomeLongitude=home.longitude / 1e7,
-                HomeAltitude=home.altitude / 1000,
-                IsHomeLocationSet=True,
-            )
-            if position and "Latitude" in data:
-                distance = _distance_m(
-                    data["Latitude"],
-                    data["Longitude"],
-                    data["HomeLatitude"],
-                    data["HomeLongitude"],
-                )
-                data["DistanceToHome"] = distance
-                data["SpaceDistToHome"] = math.hypot(
-                    distance, position.alt / 1000 - data["HomeAltitude"]
-                )
-        gps = self._fresh("GPS_RAW_INT", now)
-        if gps:
-            info: dict[str, Any] = {
-                "Type": 15,
-                "FixType": _enum_label("GPS_FIX_TYPE", gps.fix_type, "GPS_FIX_TYPE_"),
-            }
-            if gps.fix_type >= 2:
-                info["AbsoluteAlt"] = gps.alt / 1000
-            if gps.eph != 65535:
-                info["LevelDivisor"] = gps.eph / 100
-            if gps.epv != 65535:
-                info["VerticalDivisor"] = gps.epv / 100
-            if gps.satellites_visible != 255:
-                info["SatelliteCount"] = gps.satellites_visible
-            data["Gps"] = info
-        return data
-
-    def battery_data(self, now: float) -> dict[str, Any] | None:
-        packs = {
-            id_: item[0]
-            for id_, item in self.batteries.items()
-            if 0 <= now - item[1] <= STALE_SECONDS
-        }
-        if not packs:
-            return None
-        data: dict[str, Any] = {"SystemID": self.config.system_id}
-        for id_, pack in packs.items():
-            side = "Left" if id_ == 0 else "Right"
-            if pack.voltage:
-                data["Battery_Voltage" + side] = pack.voltage / 1000
-            data["Battery_Current" + side] = pack.current / 100
-            if 0 <= pack.cap_percent <= 100:
-                data["Battery_Remaining" + side] = pack.cap_percent
-        if len(packs) == 1:
-            pack = next(iter(packs.values()))
-            if pack.voltage:
-                data["Battery_Voltage"] = pack.voltage / 1000
-            data["Battery_Current"] = pack.current / 100
-            if 0 <= pack.cap_percent <= 100:
-                data["Battery_Remaining"] = pack.cap_percent
-            data["Battery_Temperature"] = pack.temperature
-            data["cycleIndex"] = pack.cyc_cnt
-            data["flags"] = pack.err_code
-        return data

tests/test_aircraft.py

@@ -0,0 +1,85 @@
+from types import SimpleNamespace
+import unittest
+
+from motian_bridge import Config
+from motian_bridge.aircraft import ParamEntry, ParamTable
+from motian_bridge.dialects.v20 import VKFly
+from motian_bridge.telemetry import Telemetry
+
+
+def cfg(**kwargs):
+    fields = dict(system_id="UAV000001", uav_type=7, serial_port="loop://",
+                  serial_baudrate=57600, mqtt_host="localhost")
+    fields.update(kwargs)
+    return Config(**fields)
+
+
+def sender(system=1, component=1):
+    return VKFly.MAVLink(None, srcSystem=system, srcComponent=component)
+
+
+def heartbeat(mav, armed=False, mode=4):
+    return mav.heartbeat_encode(
+        VKFly.MAV_TYPE_QUADROTOR, VKFly.MAV_AUTOPILOT_GENERIC,
+        VKFly.MAV_MODE_FLAG_SAFETY_ARMED if armed else 0,
+        mode, VKFly.MAV_STATE_ACTIVE).pack(mav)
+
+
+def param_value_frame(mav, param_id, value, index=0, count=1,
+                      type_=VKFly.MAV_PARAM_TYPE_REAL32):
+    return mav.param_value_encode(param_id, value, type_, count, index).pack(mav)
+
+
+class AircraftTests(unittest.TestCase):
+    def test_aircraft_created_from_heartbeat(self):
+        telemetry = Telemetry(cfg())
+        self.assertIsNone(telemetry.aircraft)
+        self.assertIsNone(telemetry.source)
+        telemetry.feed(heartbeat(sender(4, 1)), now=0)
+        aircraft = telemetry.aircraft
+        self.assertIsNotNone(aircraft)
+        self.assertEqual(aircraft.source, (4, 1))
+        self.assertEqual(telemetry.source, (4, 1))
+        self.assertEqual(aircraft.uav_type, VKFly.MAV_TYPE_QUADROTOR)
+
+    def test_param_table_accumulates_and_completes(self):
+        telemetry = Telemetry(cfg())
+        mav = sender()
+        telemetry.feed(heartbeat(mav), now=0)
+        table = telemetry.aircraft.params
+        self.assertFalse(table.complete())
+        telemetry.feed(param_value_frame(mav, b"FS_CONF_A", 4660, index=0, count=2), now=0)
+        self.assertFalse(table.complete())
+        telemetry.feed(param_value_frame(mav, b"ALT_LIM_UP1", 120, index=1, count=2), now=0)
+        self.assertTrue(table.complete())
+        self.assertEqual(table.get("FS_CONF_A"), 4660)
+        self.assertEqual(table.get("ALT_LIM_UP1"), 120)
+        self.assertEqual(telemetry.param_value("FS_CONF_A"), 4660)
+        self.assertIsNone(telemetry.param_value("MISSING"))
+        self.assertIn("FS_CONF_A", table)
+
+    def test_duplicate_index_and_unknown_index_do_not_complete(self):
+        table = ParamTable()
+        known = SimpleNamespace(param_id=b"P", param_value=1.0, param_type=9,
+                                param_count=2, param_index=0)
+        table.update(known)
+        table.update(known)
+        self.assertEqual(len(table.received), 1)
+        self.assertEqual(len(table), 1)
+        unknown = SimpleNamespace(param_id=b"Q", param_value=2.0, param_type=9,
+                                  param_count=2, param_index=0xFFFF)
+        table.update(unknown)
+        self.assertEqual(table.received, {0})
+        self.assertFalse(table.complete())
+
+    def test_param_table_setitem_accepts_values_and_entries(self):
+        table = ParamTable()
+        table["X"] = 5.0
+        table["Y"] = ParamEntry(3.0, VKFly.MAV_PARAM_TYPE_UINT8, 1)
+        self.assertEqual(table.get("X"), 5.0)
+        self.assertEqual(table.get("Y"), 3.0)
+        self.assertEqual(table.entries["Y"].param_type, VKFly.MAV_PARAM_TYPE_UINT8)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_bridge.py

@@ -156,7 +156,9 @@ class ValidationTests(unittest.TestCase):
 
     def test_config_validation(self):
         for changes in ({"system_id": "x/#"}, {"uav_type": -1},
-                        {"max_chunk_bytes": 0}, {"mqtt_password_env": "PASS"}):
+                        {"max_chunk_bytes": 0}, {"mqtt_password_env": "PASS"},
+                        {"request_params_on_start": 1}, {"param_request_timeout": 0},
+                        {"param_request_max_attempts": 0}):
             with self.subTest(changes=changes), self.assertRaises(ValueError):
                 config(**changes)
 

tests/test_commands.py

@@ -3,6 +3,7 @@ import unittest
 
 from motian_bridge import Bridge
 from motian_bridge.dialects.v20 import VKFly
+from motian_bridge.telemetry import Telemetry
 from test_bridge import FakeMqtt, FakeSerial, command, config
 
 
@@ -167,5 +168,66 @@ class CommandHandlerTests(unittest.TestCase):
         self.assertEqual(len(self.serial.written), before)
 
 
+class ParameterSyncTests(unittest.TestCase):
+    def setUp(self):
+        self.cfg = config(param_request_timeout=1.0, param_request_max_attempts=2)
+        self.mqtt = FakeMqtt()
+        self.serial = FakeSerial()
+        self.bridge = Bridge(self.cfg, self.mqtt, lambda *args, **kwargs: self.serial)
+        self.bridge._serial = self.serial
+        self.bridge._on_connect(self.mqtt, None, None, 0, None)
+        self.bridge._telemetry.feed(heartbeat(42, 1), now=0)
+
+    def request_lists(self):
+        messages = [decode(frame) for frame in self.serial.written]
+        return [message for message in messages if message.get_type() == "PARAM_REQUEST_LIST"]
+
+    def test_auto_request_once_targets_locked_controller(self):
+        self.bridge._sync_parameters(now=0)
+        frames = self.request_lists()
+        self.assertEqual(len(frames), 1)
+        self.assertEqual((frames[0].target_system, frames[0].target_component), (42, 1))
+        self.bridge._sync_parameters(now=0.1)
+        self.assertEqual(len(self.serial.written), 1)
+
+    def test_disabled_config_never_requests(self):
+        bridge = Bridge(config(request_params_on_start=False), self.mqtt,
+                        lambda *args, **kwargs: self.serial)
+        bridge._serial = self.serial
+        bridge._telemetry.feed(heartbeat(42, 1), now=0)
+        bridge._sync_parameters(now=0)
+        self.assertEqual(self.serial.written, [])
+
+    def test_retry_until_max_attempts_then_give_up(self):
+        self.bridge._sync_parameters(now=0)
+        mav = VKFly.MAVLink(None, srcSystem=42, srcComponent=1)
+        self.bridge._telemetry.feed(mav.param_value_encode(
+            b"P", 1.0, VKFly.MAV_PARAM_TYPE_REAL32, 3, 0).pack(mav), now=0.1)
+        self.bridge._sync_parameters(now=0.5)   # timeout not reached yet
+        self.assertEqual(len(self.serial.written), 1)
+        self.bridge._sync_parameters(now=1.0)   # second attempt
+        self.assertEqual(len(self.serial.written), 2)
+        self.bridge._sync_parameters(now=2.0)   # attempts exhausted
+        self.assertEqual(len(self.serial.written), 2)
+
+    def test_complete_table_stops_retry(self):
+        self.bridge._sync_parameters(now=0)
+        mav = VKFly.MAVLink(None, srcSystem=42, srcComponent=1)
+        self.bridge._telemetry.feed(mav.param_value_encode(
+            b"P", 1.0, VKFly.MAV_PARAM_TYPE_REAL32, 1, 0).pack(mav), now=0.1)
+        self.assertTrue(self.bridge._telemetry.aircraft.params.complete())
+        self.bridge._sync_parameters(now=99.0)
+        self.assertEqual(len(self.serial.written), 1)
+
+    def test_new_aircraft_requests_again(self):
+        self.bridge._sync_parameters(now=0)
+        self.assertEqual(len(self.serial.written), 1)
+        self.bridge._telemetry = Telemetry(self.cfg)
+        self.bridge._telemetry.feed(heartbeat(7, 1), now=0)
+        self.bridge._sync_parameters(now=0)
+        self.assertEqual(len(self.serial.written), 2)
+        self.assertEqual(decode(self.serial.written[-1]).target_system, 7)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_telemetry.py

@@ -38,6 +38,16 @@ def position(mav):
         520, 110, -30, 8730).pack(mav)
 
 
+def decode_frame(frame):
+    parser = VKFly.MAVLink(None)
+    result = None
+    for byte in frame:
+        message = parser.parse_char(bytes((byte,)))
+        if message is not None:
+            result = message
+    return result
+
+
 class ParserTests(unittest.TestCase):
     def test_split_frames_crc_and_v1_heartbeat(self):
         telemetry = Telemetry(cfg())
@@ -254,5 +264,29 @@ class BridgeTelemetryTests(unittest.TestCase):
             bridge.stop()
 
 
+    def test_serial_loop_auto_downloads_parameters(self):
+        from test_bridge import FakeMqtt, FakeSerial
+
+        mav = sender()
+        fake_serial = FakeSerial()
+        fake_serial.to_read.append(heartbeat(mav, True))
+        conf = cfg(publish_raw_mavlink=False)
+        mqtt = FakeMqtt()
+        bridge = Bridge(conf, mqtt, lambda *args, **kwargs: fake_serial)
+        bridge._on_connect(mqtt, None, None, 0, None)
+        bridge.start()
+        try:
+            deadline = time.monotonic() + 2
+            while not fake_serial.written and time.monotonic() < deadline:
+                time.sleep(0.01)
+            types = [decode_frame(frame).get_type() for frame in fake_serial.written]
+            self.assertIn("PARAM_REQUEST_LIST", types)
+            request = next(decode_frame(frame) for frame in fake_serial.written
+                           if decode_frame(frame).get_type() == "PARAM_REQUEST_LIST")
+            self.assertEqual((request.target_system, request.target_component), (1, 1))
+        finally:
+            bridge.stop()
+
+
 if __name__ == "__main__":
     unittest.main()