keita
df91b1110e
归档 legacy/：stereo_yolo/stereo_ranger/target_tracker 三包 + 175 条测试 + demo/配置/权重/三份规格；运行链路不再依赖，默认 pytest 446 条，归档测试按需运行。 目标有效性第二判据改为 LOS 运动学一致性（pod_bridge/los_consistency.py），删除 require_vision；pod_locked() 叠加该判据，状态机随之后退重搜。 自定义返航（task_runner/return_home.py）：保持转场高度回起飞点 → 恢复初始航向 → 垂直降落，不再爬 RTL_RETURN_ALT；未起飞/无起飞点/超时/异常仍退回 PX4 原生 RTL。 任务链安全兜底：主循环异常→紧急返航、stop 幂等、landing_stop 保留遥测、in_air 流复用+超时、解锁/起飞与吊舱识别仅执行一次、rtl_timeout 上限。 入口共享件拆到 utils/sitl_support.py；pod_sitl_test.py→l1_sitl_test.py；删除旧架构 test.py；相机脚本移入 tools/。 新增 .gitignore 并剔除历史误跟踪的 59 个 .pyc；requirements 按实际依赖重写；文档全面对齐，新增 legacy/README.md 与 tools/README.md。
156 lines
5.4 KiB
Python
156 lines
5.4 KiB
Python
"""StereoYoloFrontend 出口契约测试（需求说明 §2.2/§2.3/§2.5/§5）。
|
||
|
||
全部用注入的假检测器驱动：不需要模型权重、不需要图像素材、不需要硬件；
|
||
真并行用双人屏障断言（串行实现会屏障超时）。
|
||
"""
|
||
|
||
import logging
|
||
import threading
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from stereo_ranger.types import Detection
|
||
from stereo_yolo import StereoYoloConfig, StereoYoloFrontend
|
||
|
||
|
||
class ScriptedDetector:
|
||
"""按调用顺序返回脚本项：list[Detection] 正常返回；Exception 抛出。"""
|
||
|
||
def __init__(self, script):
|
||
self._script = list(script)
|
||
self.calls = 0
|
||
|
||
def detect(self, image):
|
||
self.calls += 1
|
||
if not self._script:
|
||
raise AssertionError("假检测器脚本已耗尽（调用次数多于预期）")
|
||
item = self._script.pop(0)
|
||
if isinstance(item, Exception):
|
||
raise item
|
||
return list(item)
|
||
|
||
|
||
class BarrierDetector:
|
||
"""两次调用必须同时在途（屏障超时即证明串行或单线程执行）。"""
|
||
|
||
def __init__(self, barrier: threading.Barrier, detections):
|
||
self._barrier = barrier
|
||
self._detections = list(detections)
|
||
|
||
def detect(self, image):
|
||
self._barrier.wait(timeout=2.0)
|
||
return list(self._detections)
|
||
|
||
|
||
def _img(fill=0, shape=(8, 8, 3)) -> np.ndarray:
|
||
return np.full(shape, fill, dtype=np.uint8)
|
||
|
||
|
||
def _det(x1=1.0, y1=2.0, x2=11.0, y2=12.0, class_id=0, confidence=0.9, track_id=None) -> Detection:
|
||
return Detection(x1, y1, x2, y2, confidence=confidence, class_id=class_id, track_id=track_id)
|
||
|
||
|
||
def _frontend(script, **cfg_kwargs) -> StereoYoloFrontend:
|
||
kwargs = {"concurrency": "sequential"}
|
||
kwargs.update(cfg_kwargs)
|
||
return StereoYoloFrontend(StereoYoloConfig(**kwargs), detector=ScriptedDetector(script))
|
||
|
||
|
||
def test_frames_share_frame_id_and_keep_boxes():
|
||
left_box, right_box = _det(1.0, 2.0, 11.0, 12.0, confidence=0.77), _det(3.0, 4.0, 13.0, 14.0)
|
||
frontend = _frontend([[left_box], [right_box]])
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=7)
|
||
|
||
assert left_frame.frame_id == right_frame.frame_id == 7
|
||
assert left_frame.detections == (left_box,)
|
||
assert right_frame.detections == (right_box,)
|
||
assert left_frame.detections[0].confidence == pytest.approx(0.77)
|
||
assert left_frame.detections[0].track_id is None # 无追踪器时留空（§2.1）
|
||
assert frontend._detector.calls == 2
|
||
|
||
|
||
def test_whitelist_filters_by_class():
|
||
frontend = _frontend([[_det(class_id=3)], [_det(class_id=0)]], class_whitelist=(0,))
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=1)
|
||
|
||
assert left_frame.detections == ()
|
||
assert len(right_frame.detections) == 1 and right_frame.detections[0].class_id == 0
|
||
|
||
|
||
def test_no_whitelist_keeps_all_classes():
|
||
frontend = _frontend([[_det(class_id=3)], [_det(class_id=5)]])
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=1)
|
||
|
||
assert len(left_frame.detections) == 1 and len(right_frame.detections) == 1
|
||
|
||
|
||
def test_degenerate_boxes_dropped(caplog):
|
||
nan_box = _det(x1=float("nan"))
|
||
zero_width = _det(x1=5.0, x2=5.0)
|
||
inverted = _det(x1=20.0, x2=10.0)
|
||
good = _det()
|
||
frontend = _frontend([[nan_box, zero_width, inverted, good], []])
|
||
|
||
with caplog.at_level(logging.WARNING, logger="stereo_yolo"):
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=2)
|
||
|
||
assert left_frame.detections == (good,)
|
||
assert right_frame.detections == ()
|
||
assert "丢弃退化检测框" in caplog.text
|
||
assert "累计 1 次" in caplog.text
|
||
|
||
|
||
def test_boundary_touching_box_kept():
|
||
hairline = _det(x1=10.0, x2=10.0 + 1e-9)
|
||
frontend = _frontend([[hairline], []])
|
||
left_frame, _ = frontend.detect(_img(), _img(), frame_id=3)
|
||
assert left_frame.detections == (hairline,)
|
||
|
||
|
||
def test_one_eye_failure_isolated(caplog):
|
||
good = _det()
|
||
frontend = _frontend([[good], RuntimeError("boom")])
|
||
|
||
with caplog.at_level(logging.WARNING, logger="stereo_yolo"):
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=4)
|
||
|
||
assert left_frame.detections == (good,)
|
||
assert right_frame.detections == ()
|
||
assert left_frame.frame_id == right_frame.frame_id == 4
|
||
assert "右目推理失败" in caplog.text
|
||
|
||
|
||
def test_shape_mismatch_rejected(caplog):
|
||
frontend = _frontend([]) # 脚本为空：若检测器被调用会抛断言
|
||
|
||
with caplog.at_level(logging.WARNING, logger="stereo_yolo"):
|
||
left_frame, right_frame = frontend.detect(_img(shape=(8, 8, 3)), _img(shape=(8, 6, 3)), frame_id=5)
|
||
|
||
assert left_frame.detections == () and right_frame.detections == ()
|
||
assert left_frame.frame_id == right_frame.frame_id == 5
|
||
assert frontend._detector.calls == 0
|
||
assert "左右图尺寸不一致" in caplog.text
|
||
|
||
|
||
def test_parallel_eyes_run_concurrently():
|
||
barrier = threading.Barrier(2)
|
||
box = _det()
|
||
frontend = StereoYoloFrontend(
|
||
StereoYoloConfig(concurrency="parallel"),
|
||
detector=BarrierDetector(barrier, [box]),
|
||
)
|
||
try:
|
||
left_frame, right_frame = frontend.detect(_img(), _img(), frame_id=6)
|
||
finally:
|
||
frontend.close()
|
||
|
||
assert left_frame.detections == (box,)
|
||
assert right_frame.detections == (box,)
|
||
|
||
|
||
def test_close_is_idempotent():
|
||
frontend = StereoYoloFrontend(StereoYoloConfig(concurrency="parallel"), detector=ScriptedDetector([]))
|
||
frontend.close()
|
||
frontend.close()
|