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。
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""UltralyticsDetector —— ultralytics YOLO .pt 直载的检测后端。
|
||
|
||
⚠ 移植注记（C++ 侧需自行复现，详见 README）：letterbox（缩放 + 填充到
|
||
imgsz）、输出张量解码、NMS、框映射回输入图原生坐标系，本例由 ultralytics
|
||
库内完成；移植时必须逐项复现同一几何（步长对齐、填充方式、IoU 门限、
|
||
原生坐标回映），否则视差取点会引入系统性偏差。
|
||
"""
|
||
|
||
import queue
|
||
from pathlib import Path
|
||
from typing import List
|
||
|
||
from stereo_ranger.types import Detection
|
||
|
||
from .config import StereoYoloConfig
|
||
|
||
|
||
class UltralyticsDetector:
|
||
"""双实例池：ultralytics 单实例非线程安全，左右并发各持一份（§3/§4）。
|
||
|
||
接口约定（可注入替换）：``detect(image) -> list[Detection]``，
|
||
image 为 BGR ndarray（cv2 约定），返回框位于该图原生坐标系。
|
||
"""
|
||
|
||
def __init__(self, config: StereoYoloConfig) -> None:
|
||
self._cfg = config
|
||
self._model_path = _resolve_model_path(config.model_path)
|
||
self._pool: "queue.SimpleQueue" = queue.SimpleQueue()
|
||
from ultralytics import YOLO # 延迟导入：仅真实推理付加载成本，单测注入假检测器（§6）
|
||
|
||
for _ in range(2):
|
||
self._pool.put(YOLO(str(self._model_path)))
|
||
|
||
def detect(self, image) -> List[Detection]:
|
||
model = self._pool.get()
|
||
try:
|
||
results = model.predict(
|
||
source=image,
|
||
imgsz=self._cfg.imgsz,
|
||
conf=self._cfg.conf_threshold,
|
||
iou=self._cfg.iou_threshold,
|
||
device=self._cfg.device,
|
||
verbose=False,
|
||
)
|
||
finally:
|
||
self._pool.put(model)
|
||
if not results:
|
||
return []
|
||
return _result_to_detections(results[0])
|
||
|
||
|
||
def _result_to_detections(result) -> List[Detection]:
|
||
boxes = getattr(result, "boxes", None)
|
||
if boxes is None or len(boxes) == 0:
|
||
return []
|
||
xyxy = boxes.xyxy.cpu().numpy()
|
||
conf = boxes.conf.cpu().numpy()
|
||
cls = boxes.cls.cpu().numpy()
|
||
return [
|
||
Detection(
|
||
float(box[0]),
|
||
float(box[1]),
|
||
float(box[2]),
|
||
float(box[3]),
|
||
confidence=float(score),
|
||
class_id=int(class_id),
|
||
)
|
||
for box, score, class_id in zip(xyxy, conf, cls)
|
||
]
|
||
|
||
|
||
def _resolve_model_path(raw: str) -> Path:
|
||
path = Path(raw).expanduser()
|
||
if not path.is_absolute():
|
||
path = Path.cwd() / path
|
||
if not path.is_file():
|
||
raise FileNotFoundError(f"模型权重不存在: {path}（model_path 相对路径按启动工作目录解析）")
|
||
return path.resolve()
|