keita
ded17d7995
- utils/task_runner: 6态状态机(WAIT→ARM_TAKEOFF→GOTO(L2)→TRACK(L1)→APPROACH→RTL) + 编排器(快照采集/状态动作) + 双层日志(事件JSONL/遥测CSV) + 吊舱控制指令(0x07/0x0F/0x0E/0x08) - pod_bridge: ground_link 地面坐标UDP注入(超龄判无效,无激光替代表4); set_heading_servo_enabled() 运行时L1/L2切换(切换即reset清积分防输出跳变); pod_locked 提为公共门控; config 新增 ground_inject 段 - virtual_pod 模拟吊舱(无电机语义) + verify_px4 增量④仅飞控验证 + pod_sitl_test L1 SITL全流程(飞行编排不塞进 run_bridge, 保持纯offboard闭环入口) - 入口: task_runner_sitl(VirtualPod+虚拟目标, SITL端到端跑通约1m24s) / task_runner_real(真串口+UDP注入+吊舱控制) - 地面站 ground_station: 坐标注入 + 状态可视化(Flask) - S1标定回填(2026-10-09): beta_sign=+1 / alpha_sign=-1, 加标定回归测试守护 - 新增测试: task_runner x4 + ground_link/virtual_pod*/bridge/px4_gateway 增量; 全量 483 passed - 文档: 预定流程(需求) + 预定流程-操作指南(操作/待办) + L1方案补S1结论与实现状态
184 lines
7.5 KiB
Python
184 lines
7.5 KiB
Python
"""config_test.py —— 桥接配置加载与校验单元测试。"""
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from pod_bridge.config import (BridgeConfig, GuidanceConfig, HeadingServoConfig,
|
||
PodSerialConfig, load_from_json)
|
||
from visual_track_control.config import NoDistanceAction, TrackingConfig
|
||
|
||
_DEFAULT_CONFIG = Path(__file__).resolve().parents[2] / "config" / "pod_bridge.json"
|
||
|
||
|
||
def _write(tmp_path, obj) -> str:
|
||
p = tmp_path / "cfg.json"
|
||
p.write_text(json.dumps(obj), encoding="utf-8")
|
||
return str(p)
|
||
|
||
|
||
def test_shipped_default_config_loads():
|
||
cfg = load_from_json(str(_DEFAULT_CONFIG))
|
||
assert cfg.pod_serial.baudrate == 115200
|
||
assert cfg.px4.connection_string == "udp://:14540"
|
||
assert cfg.link.attitude_downlink_hz == pytest.approx(20.0)
|
||
assert cfg.guidance.max_speed_mps == pytest.approx(5.0)
|
||
assert cfg.guidance.stop_radius_m == pytest.approx(5.0)
|
||
|
||
|
||
def test_empty_json_uses_all_defaults(tmp_path):
|
||
cfg = load_from_json(_write(tmp_path, {}))
|
||
default = BridgeConfig()
|
||
assert cfg.pod_serial == default.pod_serial
|
||
assert cfg.px4 == default.px4
|
||
assert cfg.link == default.link
|
||
assert cfg.guidance == default.guidance
|
||
|
||
|
||
def test_partial_override_keeps_other_defaults(tmp_path):
|
||
cfg = load_from_json(_write(tmp_path, {
|
||
"pod_serial": {"port": "COM3"},
|
||
"guidance": {"max_speed_mps": 8.0},
|
||
}))
|
||
assert cfg.pod_serial.port == "COM3"
|
||
assert cfg.pod_serial.baudrate == 115200 # 默认回退
|
||
assert cfg.guidance.max_speed_mps == 8.0
|
||
assert cfg.guidance.stop_radius_m == GuidanceConfig().stop_radius_m # 默认回退
|
||
|
||
|
||
def test_full_override_roundtrip(tmp_path):
|
||
obj = {
|
||
"pod_serial": {"port": "/dev/ttyUSB0", "baudrate": 115200},
|
||
"px4": {"connection_string": "serial:///dev/ttyS1:57600"},
|
||
"link": {"attitude_downlink_hz": 30.0, "telemetry_timeout_ms": 800},
|
||
"guidance": {"frequency_hz": 20.0, "stop_radius_m": 3.0,
|
||
"max_speed_mps": 7.5, "speed_kp": 0.8, "yaw_kp": 1.5},
|
||
}
|
||
cfg = load_from_json(_write(tmp_path, obj))
|
||
assert cfg.px4.connection_string == "serial:///dev/ttyS1:57600"
|
||
assert cfg.link.attitude_downlink_hz == 30.0
|
||
assert cfg.link.telemetry_timeout_ms == 800
|
||
assert cfg.guidance.frequency_hz == 20.0
|
||
assert cfg.guidance.yaw_kp == 1.5
|
||
|
||
|
||
# --- 校验错误 ---------------------------------------------------------------
|
||
def test_missing_file_raises():
|
||
with pytest.raises(ValueError, match="打开失败"):
|
||
load_from_json(str(_DEFAULT_CONFIG.parent / "no_such_file.json"))
|
||
|
||
|
||
def test_non_object_root_raises(tmp_path):
|
||
p = tmp_path / "bad.json"
|
||
p.write_text("[1, 2, 3]", encoding="utf-8")
|
||
with pytest.raises(ValueError, match="根节点必须为对象"):
|
||
load_from_json(str(p))
|
||
|
||
|
||
def test_malformed_json_raises(tmp_path):
|
||
p = tmp_path / "bad.json"
|
||
p.write_text("{not json", encoding="utf-8")
|
||
with pytest.raises(ValueError, match="解析失败"):
|
||
load_from_json(str(p))
|
||
|
||
|
||
def test_wrong_type_raises_with_field_name(tmp_path):
|
||
with pytest.raises(ValueError, match="guidance.max_speed_mps"):
|
||
load_from_json(_write(tmp_path, {"guidance": {"max_speed_mps": "fast"}}))
|
||
|
||
|
||
def test_unknown_key_rejected(tmp_path):
|
||
with pytest.raises(ValueError, match="未知字段"):
|
||
load_from_json(_write(tmp_path, {"guidance": {"max_speed": 5.0}}))
|
||
|
||
|
||
def test_empty_connection_string_raises(tmp_path):
|
||
with pytest.raises(ValueError, match="px4.connection_string"):
|
||
load_from_json(_write(tmp_path, {"px4": {"connection_string": ""}}))
|
||
|
||
|
||
@pytest.mark.parametrize("obj,field", [
|
||
({"link": {"attitude_downlink_hz": 100.0}}, "link.attitude_downlink_hz"), # >60
|
||
({"link": {"attitude_downlink_hz": 0.5}}, "link.attitude_downlink_hz"), # <1
|
||
({"guidance": {"max_speed_mps": -1.0}}, "guidance.max_speed_mps"), # 非正
|
||
({"guidance": {"stop_radius_m": -2.0}}, "guidance.stop_radius_m"), # 负
|
||
({"guidance": {"speed_derivative_filter_coef": 1.0}}, # 上界开
|
||
"guidance.speed_derivative_filter_coef"),
|
||
({"pod_serial": {"baudrate": 0}}, "pod_serial.baudrate"), # 非正
|
||
])
|
||
def test_out_of_range_raises(tmp_path, obj, field):
|
||
with pytest.raises(ValueError, match=field.replace(".", r"\.")):
|
||
load_from_json(_write(tmp_path, obj))
|
||
|
||
|
||
def test_pod_serial_default_port_is_placeholder():
|
||
assert PodSerialConfig().port == ""
|
||
|
||
|
||
# --- L1 航向伺服配置（方案 §4，决策 1/2/3）----------------------------------
|
||
def test_empty_json_heading_servo_and_tracking_default(tmp_path):
|
||
cfg = load_from_json(_write(tmp_path, {}))
|
||
assert cfg.heading_servo == HeadingServoConfig()
|
||
assert cfg.heading_servo.enabled is False # 默认关，保持 L2 行为
|
||
assert cfg.tracking == TrackingConfig() # 缺 tracking 段 → 默认
|
||
|
||
|
||
def test_heading_servo_loads():
|
||
cfg = load_from_json(str(_DEFAULT_CONFIG))
|
||
hs = cfg.heading_servo
|
||
assert hs.enabled is True # 部署配置走 L1
|
||
assert hs.beta_from_off_target is True # 无电机：脱靶量为主信号
|
||
assert hs.beta_from_frame is False # 框架角当前无用
|
||
assert hs.vertical_enabled is True # 决策3：无人机 L1 开垂直
|
||
|
||
|
||
def test_deployed_config_has_calibrated_signs():
|
||
"""S1 台架标定回归(2026-10-09)：目标在左下时 off_target_x/y 均为负 →
|
||
吊舱 X 轴右正(与控制律 β 同向) beta_sign=+1；Y 轴上正(与 α 下正相反) alpha_sign=-1。
|
||
厂商确认本版电机无用 → 框架角路径必须关。改动这些值前必须重新台架标定。"""
|
||
cfg = load_from_json(str(_DEFAULT_CONFIG))
|
||
assert cfg.heading_servo.beta_sign == 1
|
||
assert cfg.heading_servo.alpha_sign == -1
|
||
assert cfg.heading_servo.beta_from_frame is False
|
||
assert cfg.heading_servo.alpha_from_frame is False
|
||
|
||
|
||
def test_heading_servo_partial_override(tmp_path):
|
||
cfg = load_from_json(_write(tmp_path, {
|
||
"heading_servo": {"beta_sign": -1, "beta_from_frame": True},
|
||
}))
|
||
assert cfg.heading_servo.beta_sign == -1
|
||
assert cfg.heading_servo.beta_from_frame is True
|
||
assert cfg.heading_servo.alpha_sign == 1 # 默认回退
|
||
|
||
|
||
@pytest.mark.parametrize("obj,field", [
|
||
({"heading_servo": {"beta_sign": 0}}, "heading_servo.beta_sign"), # 非 ±1
|
||
({"heading_servo": {"alpha_sign": 2}}, "heading_servo.alpha_sign"),
|
||
({"heading_servo": {"beta_sign": 1.0}}, "heading_servo.beta_sign"), # 非整数
|
||
({"heading_servo": {"unknown": 1}}, "未知字段"),
|
||
])
|
||
def test_heading_servo_bad_sign_rejected(tmp_path, obj, field):
|
||
with pytest.raises(ValueError, match=field.replace(".", r"\.")):
|
||
load_from_json(_write(tmp_path, obj))
|
||
|
||
|
||
def test_tracking_incomplete_section_raises(tmp_path):
|
||
# tracking 段一旦提供即按 vtc 严格解析：缺分组报错
|
||
with pytest.raises(ValueError, match="camera"):
|
||
load_from_json(_write(tmp_path, {"tracking": {
|
||
"control": {"frequency_hz": 20.0, "visual_stale_ms": 200,
|
||
"distance_stale_ms": 500, "attitude_stale_ms": 300},
|
||
}}))
|
||
|
||
|
||
def test_shipped_tracking_l1_values():
|
||
cfg = load_from_json(str(_DEFAULT_CONFIG))
|
||
d = cfg.tracking.distance
|
||
assert d.no_distance_action is NoDistanceAction.SLOW_APPROACH # 决策3：低速接近
|
||
assert d.no_distance_approach_limit_mps == pytest.approx(0.2) # 极低限速
|
||
assert cfg.tracking.vertical.vz_limit_mps == pytest.approx(0.5) # vz 压低
|
||
assert cfg.tracking.heading.mode.value == "rate"
|
||
|