#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Standalone live camera + single YOLO runner. 不复用项目内的 `camera_manager` / `target_roi_yolo` / `config` / `logger_manager`。 功能： - 独立初始化摄像头 - 实时读取帧 - 独立加载单个 YOLO 模型并推理 - 画出检测框、ROI、FPS 适用场景： - 单独验证一个模型是否能跑 - 验证实时帧率 - 验证 ROI 是否裁对 - 不进入主业务射箭流程 """ from __future__ import annotations import os import time from dataclasses import dataclass @dataclass class RunnerConfig: camera_width: int = 640 camera_height: int = 480 model_path: str = "/root/model_278702.mud" conf_th: float = 0.7 retry_conf_th: float = 0.5 class_ids: tuple = (0,) merge_mode: str = "union" coord_mode: str = "native" roi_margin_frac: float = 0.11 min_box_side_px: int = 8 def log(msg: str): print(msg) class DummyLogger: def info(self, msg): log(msg) def warning(self, msg): log(msg) def error(self, msg): log(msg) class StandaloneYOLORunner: def __init__(self, cfg: RunnerConfig): self.cfg = cfg self.logger = DummyLogger() self._last_fps_t = time.perf_counter() self._frames = 0 self._fps = 0.0 self._camera = None self._det = None def _import_maix(self): try: from maix import camera, image, nn return camera, image, nn except Exception as e: raise RuntimeError(f"maix import failed: {e}") def _init_camera(self): camera, _, _ = self._import_maix() if self._camera is not None: return self._camera try: self._camera = camera.Camera( width=self.cfg.camera_width, height=self.cfg.camera_height, format=camera.RGB888, ) except Exception: self._camera = camera.Camera(width=self.cfg.camera_width, height=self.cfg.camera_height) return self._camera def _load_detector(self, model_path: str): _, _, nn = self._import_maix() if not model_path or not os.path.isfile(model_path): return None return nn.YOLOv5(model=model_path, dual_buff=False) @staticmethod def _get_class_id(obj): for key in ("class_id", "cls", "label", "category", "cat_id", "id"): if hasattr(obj, key): v = getattr(obj, key) if v is None: continue try: return int(float(v)) except Exception: pass return None @staticmethod def _normalize_boxes(raw): out = [] for o in raw or []: if isinstance(o, (list, tuple)) and len(o) >= 6: class Box: pass b = Box() b.x, b.y, b.w, b.h, b.score, b.class_id = map(float, o[:6]) out.append(b) else: out.append(o) return out def _det_to_xyxy(self, det, obj): x = float(getattr(obj, "x", 0.0)) y = float(getattr(obj, "y", 0.0)) w = float(getattr(obj, "w", 0.0)) h = float(getattr(obj, "h", 0.0)) return x, y, x + w, y + h def _run_detector(self, det, img, conf_th, class_ids): if det is None: return [] raw = det.detect(img, conf_th=conf_th) objs = self._normalize_boxes(raw if raw is not None else []) out = [] for o in objs: cid = self._get_class_id(o) if cid is not None and cid not in class_ids: continue out.append(o) return out def _calc_fps(self): self._frames += 1 now = time.perf_counter() dt = now - self._last_fps_t if dt >= 1.0: self._fps = self._frames / dt self._frames = 0 self._last_fps_t = now return self._fps def _draw_text(self, img, lines): try: import cv2 y = 24 for line in lines: cv2.putText(img, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 255, 0), 1, cv2.LINE_AA) y += 20 except Exception: pass def _clip_roi(self, x0, y0, x1, y1, w, h): x0 = max(0, min(int(x0), w - 1)) y0 = max(0, min(int(y0), h - 1)) x1 = max(x0 + 1, min(int(x1), w)) y1 = max(y0 + 1, min(int(y1), h)) return x0, y0, x1, y1 def _merge_boxes(self, boxes): if not boxes: return None x0 = min(b[0] for b in boxes) y0 = min(b[1] for b in boxes) x1 = max(b[2] for b in boxes) y1 = max(b[3] for b in boxes) return x0, y0, x1, y1 def _run_single_yolo(self, frame, img_cv): h, w = int(img_cv.shape[0]), int(img_cv.shape[1]) if self._det is None: self._det = self._load_detector(self.cfg.model_path) det = self._det if det is None: return [] boxes = self._run_detector(det, frame, self.cfg.conf_th, self.cfg.class_ids) if not boxes and self.cfg.retry_conf_th < self.cfg.conf_th: boxes = self._run_detector(det, frame, self.cfg.retry_conf_th, self.cfg.class_ids) xyxy = [] for obj in boxes: x0, y0, x1, y1 = self._det_to_xyxy(det, obj) if (x1 - x0) < self.cfg.min_box_side_px or (y1 - y0) < self.cfg.min_box_side_px: continue if self.cfg.coord_mode == "native": x0, y0, x1, y1 = self._clip_roi(x0, y0, x1, y1, w, h) xyxy.append((x0, y0, x1, y1)) return xyxy def run(self): _, image, _ = self._import_maix() cam = self._init_camera() log("[YOLOTE] standalone runner started") while True: try: frame = cam.read() except Exception as e: log(f"[YOLOTE] camera read failed: {e}") time.sleep(0.02) continue if frame is None: time.sleep(0.01) continue try: img_cv = image.image2cv(frame, False, False) except Exception as e: log(f"[YOLOTE] image2cv failed: {e}") time.sleep(0.01) continue import cv2 t0 = time.perf_counter() boxes = self._run_single_yolo(frame, img_cv) t1 = time.perf_counter() for i, (bx0, by0, bx1, by1) in enumerate(boxes): cv2.rectangle(img_cv, (int(bx0), int(by0)), (int(bx1) - 1, int(by1) - 1), (0, 255, 0), 2) cv2.putText(img_cv, f"B{i}", (int(bx0), max(0, int(by0) - 4)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1, cv2.LINE_AA) fps = self._calc_fps() self._draw_text( img_cv, [ f"FPS: {fps:.1f}", f"YOLO: {(t1 - t0)*1000.0:.1f} ms", f"Boxes: {len(boxes)}", "Ctrl+C to exit", ], ) try: frame_out = image.cv2image(img_cv, False, False) if hasattr(cam, "show"): cam.show(frame_out) else: try: frame_out.show() except Exception: pass except Exception as e: log(f"[YOLOTE] show failed: {e}") time.sleep(0.001) def main(): cfg = RunnerConfig() runner = StandaloneYOLORunner(cfg) try: runner.run() except KeyboardInterrupt: log("[YOLOTE] interrupted") if __name__ == "__main__": main()