模型调用
This commit is contained in:
@@ -12,7 +12,6 @@ files:
|
||||
- at_client.py
|
||||
- camera_manager.py
|
||||
- cameraParameters.xml
|
||||
- charging_exit.sh
|
||||
- config.py
|
||||
- hardware.py
|
||||
- laser_detector.py
|
||||
|
||||
@@ -262,6 +262,16 @@ TRIANGLE_SAMPLE_PATCH_HALF_PX = 2
|
||||
# 开机阶段预加载 YOLO detector;detect 使用 dual_buff=False,避免返回上一帧结果。
|
||||
TRIANGLE_YOLO_PRELOAD_ON_BOOT = False
|
||||
|
||||
# YOLO target size classification: class 0=20cm, class 1=40cm.
|
||||
TARGET_CLASS_YOLO_ENABLE = True
|
||||
TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_285484.mud"
|
||||
TARGET_CLASS_YOLO_LABELS = (20, 40)
|
||||
TARGET_CLASS_YOLO_CONF_TH = 0.50
|
||||
TARGET_CLASS_YOLO_IOU_TH = 0.45
|
||||
TARGET_CLASS_YOLO_RETRY_ON_EMPTY = False
|
||||
TARGET_CLASS_YOLO_RETRY_CONF_TH = 0.25
|
||||
TARGET_CLASS_YOLO_PRELOAD_ON_BOOT = True
|
||||
|
||||
# ── 第二段 YOLO:仅在 Stage1 裁切出的靶环图上推理(与合成 stage2 训练数据一致)→ 子框内传统算法取直角点 ──
|
||||
# Stage1 靶环裁切内如何找黑三角标记(对比耗时时可切换):
|
||||
# "yolo" — 调 Stage2 黑三角模型得子框,再子框内传统提取(需 TRIANGLE_BLACK_YOLO_ENABLE=True)。
|
||||
|
||||
@@ -162,7 +162,11 @@ def cmd_str():
|
||||
and _loc_black == "yolo"
|
||||
and bool(getattr(config, "TRIANGLE_BLACK_YOLO_PRELOAD_ON_BOOT", True))
|
||||
)
|
||||
_preload_yolo = _preload_yolo or _need_black_preload
|
||||
_need_target_preload = (
|
||||
bool(getattr(config, "TARGET_CLASS_YOLO_ENABLE", False))
|
||||
and bool(getattr(config, "TARGET_CLASS_YOLO_PRELOAD_ON_BOOT", True))
|
||||
)
|
||||
_preload_yolo = _preload_yolo or _need_black_preload or _need_target_preload
|
||||
if _preload_yolo:
|
||||
preload_yolo_detector(logger)
|
||||
except Exception as e:
|
||||
|
||||
@@ -325,6 +325,18 @@ def process_shot(adc_val):
|
||||
# 网络事件移到拍照之后,避免阻塞拍照
|
||||
network_manager.safe_enqueue({"shoot_event": "start"}, msg_type=2, high=True)
|
||||
|
||||
# Classify only the current shot frame; never reuse a previous result.
|
||||
target_class_result = None
|
||||
try:
|
||||
from target_roi_yolo import try_get_target_class_from_yolo
|
||||
|
||||
target_class_result = try_get_target_class_from_yolo(frame, logger=logger)
|
||||
if logger:
|
||||
logger.info(f"[YOLO-TARGET] 当前箭业务结果: {target_class_result}")
|
||||
except Exception as exc:
|
||||
if logger:
|
||||
logger.warning(f"[YOLO-TARGET] 当前箭分类失败,按未知处理: {exc}")
|
||||
|
||||
# 调用算法分析
|
||||
analysis_result = analyze_shot(frame)
|
||||
|
||||
@@ -384,11 +396,25 @@ def process_shot(adc_val):
|
||||
srv_y = round(float(dy), 4) if dy is not None else 200.0
|
||||
|
||||
# 构造上报数据
|
||||
target_label = (
|
||||
target_class_result.get("label")
|
||||
if isinstance(target_class_result, dict)
|
||||
else None
|
||||
)
|
||||
target_confidence = (
|
||||
target_class_result.get("confidence")
|
||||
if isinstance(target_class_result, dict)
|
||||
else None
|
||||
)
|
||||
inner_data = {
|
||||
"shot_id": shot_id,
|
||||
"x": srv_x,
|
||||
"y": srv_y,
|
||||
"r": 20.0, # 保留字段(服务端当前忽略,物理外环半径 cm)
|
||||
"target_class": target_label,
|
||||
"target_class_confidence": (
|
||||
float(target_confidence) if target_confidence is not None else None
|
||||
),
|
||||
"d": round((distance_m or 0.0) * 100),
|
||||
"d_laser": round((laser_distance_m or 0.0) * 100),
|
||||
"d_laser_quality": laser_signal_quality,
|
||||
@@ -416,6 +442,11 @@ def process_shot(adc_val):
|
||||
inner_data["ellipse_center_y"] = None
|
||||
|
||||
report_data = {"cmd": 1, "data": inner_data}
|
||||
if logger:
|
||||
logger.info(
|
||||
f"[REPORT-TARGET] enqueue shot_id={shot_id}, "
|
||||
f"target_class={target_label}, confidence={target_confidence}"
|
||||
)
|
||||
network_manager.safe_enqueue(report_data, msg_type=2, high=True)
|
||||
|
||||
# 数据上报后再画标注,不干扰检测阶段的原始画面
|
||||
|
||||
+143
-1
@@ -89,6 +89,29 @@ def _stage2_roi_crop_save_worker(
|
||||
_detector_by_path = {}
|
||||
|
||||
|
||||
def _resolve_model_path(model_path: str):
|
||||
"""Resolve a model in either the installed app or MaixVision run directory."""
|
||||
model_path = (model_path or "").strip()
|
||||
if model_path and os.path.isfile(model_path):
|
||||
return model_path
|
||||
if not model_path:
|
||||
return ""
|
||||
name = os.path.basename(model_path)
|
||||
module_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
candidates = (
|
||||
os.path.join(module_dir, name),
|
||||
os.path.join(module_dir, "test", name),
|
||||
os.path.join("/tmp/maixpy_run", name),
|
||||
os.path.join("/tmp/maixpy_run", "test", name),
|
||||
os.path.join(os.getcwd(), name),
|
||||
os.path.join(os.getcwd(), "test", name),
|
||||
)
|
||||
for candidate in candidates:
|
||||
if os.path.isfile(candidate):
|
||||
return candidate
|
||||
return model_path
|
||||
|
||||
|
||||
def reset_yolo_detector_cache():
|
||||
"""切换模型路径时可调用(通常不必)。"""
|
||||
global _detector_by_path
|
||||
@@ -175,6 +198,23 @@ def preload_yolo_detector(logger=None):
|
||||
% (_loc_black,)
|
||||
)
|
||||
|
||||
if bool(getattr(cfg, "TARGET_CLASS_YOLO_ENABLE", False)) and bool(
|
||||
getattr(cfg, "TARGET_CLASS_YOLO_PRELOAD_ON_BOOT", True)
|
||||
):
|
||||
class_model_path = _resolve_model_path(
|
||||
getattr(cfg, "TARGET_CLASS_YOLO_MODEL_PATH", "") or ""
|
||||
)
|
||||
class_detector = _get_detector(class_model_path)
|
||||
if class_detector is None:
|
||||
if logger:
|
||||
logger.warning(
|
||||
f"[YOLO-TARGET] 预加载失败:无法加载模型 {class_model_path}"
|
||||
)
|
||||
else:
|
||||
ok = True
|
||||
if logger:
|
||||
logger.info(f"[YOLO-TARGET] 靶规格模型已预加载: {class_model_path}")
|
||||
|
||||
return ok
|
||||
|
||||
|
||||
@@ -206,8 +246,10 @@ def _det_obj_class_id(o):
|
||||
if v is None:
|
||||
continue
|
||||
try:
|
||||
if callable(v):
|
||||
v = v()
|
||||
return int(float(v))
|
||||
except (TypeError, ValueError):
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
continue
|
||||
return None
|
||||
|
||||
@@ -242,6 +284,106 @@ def _normalize_objs(objs):
|
||||
return out
|
||||
|
||||
|
||||
def _det_obj_score(o):
|
||||
"""Return confidence across supported Maix YOLO result formats."""
|
||||
for key in ("score", "confidence", "conf", "prob"):
|
||||
if hasattr(o, key):
|
||||
try:
|
||||
value = getattr(o, key)
|
||||
if callable(value):
|
||||
value = value()
|
||||
value = float(value)
|
||||
if value == value:
|
||||
return value
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
pass
|
||||
return 0.0
|
||||
|
||||
|
||||
def try_get_target_class_from_yolo(maix_frame, logger=None):
|
||||
"""Classify the current target as 20cm or 40cm; return None if unknown."""
|
||||
try:
|
||||
import config as cfg
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
if not bool(getattr(cfg, "TARGET_CLASS_YOLO_ENABLE", False)):
|
||||
return None
|
||||
model_path = _resolve_model_path(
|
||||
getattr(cfg, "TARGET_CLASS_YOLO_MODEL_PATH", "") or ""
|
||||
)
|
||||
if not os.path.isfile(model_path):
|
||||
if logger:
|
||||
logger.warning(f"[YOLO-TARGET] 模型文件不存在: {model_path}")
|
||||
return None
|
||||
detector = _get_detector(model_path)
|
||||
if detector is None:
|
||||
if logger:
|
||||
logger.warning("[YOLO-TARGET] 无法加载 nn.YOLOv5")
|
||||
return None
|
||||
|
||||
conf_th = float(getattr(cfg, "TARGET_CLASS_YOLO_CONF_TH", 0.5))
|
||||
iou_th = float(getattr(cfg, "TARGET_CLASS_YOLO_IOU_TH", 0.45))
|
||||
labels = getattr(cfg, "TARGET_CLASS_YOLO_LABELS", (20, 40))
|
||||
if isinstance(labels, str):
|
||||
labels = tuple(x.strip() for x in labels.split(",") if x.strip())
|
||||
labels = tuple(labels)
|
||||
|
||||
def _detect(threshold):
|
||||
try:
|
||||
raw = detector.detect(maix_frame, conf_th=threshold, iou_th=iou_th)
|
||||
except Exception as exc:
|
||||
if logger:
|
||||
logger.warning(f"[YOLO-TARGET] detect 异常: {exc}")
|
||||
return []
|
||||
return _normalize_objs(raw if raw is not None else [])
|
||||
|
||||
def _candidates(objs):
|
||||
found = []
|
||||
for obj in objs:
|
||||
class_id = _det_obj_class_id(obj)
|
||||
if class_id is None or class_id < 0 or class_id >= len(labels):
|
||||
continue
|
||||
try:
|
||||
label = int(float(labels[class_id]))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if label in (20, 40):
|
||||
found.append((label, class_id, _det_obj_score(obj)))
|
||||
return found
|
||||
|
||||
objects = _detect(conf_th)
|
||||
candidates = _candidates(objects)
|
||||
if logger and objects:
|
||||
logger.info(
|
||||
"[YOLO-TARGET] 原始框=%d, 解析类别=%s"
|
||||
% (
|
||||
len(objects),
|
||||
[(_det_obj_class_id(o), _det_obj_score(o)) for o in objects[:8]],
|
||||
)
|
||||
)
|
||||
if not candidates and bool(
|
||||
getattr(cfg, "TARGET_CLASS_YOLO_RETRY_ON_EMPTY", False)
|
||||
):
|
||||
retry_th = float(getattr(cfg, "TARGET_CLASS_YOLO_RETRY_CONF_TH", conf_th))
|
||||
if 0 < retry_th < conf_th:
|
||||
candidates = _candidates(_detect(retry_th))
|
||||
|
||||
if not candidates:
|
||||
if logger:
|
||||
logger.warning("[YOLO-TARGET] 当前帧未识别到 20/40,按未知处理")
|
||||
return None
|
||||
|
||||
label, class_id, confidence = max(candidates, key=lambda item: item[2])
|
||||
result = {"label": label, "class_id": class_id, "confidence": confidence}
|
||||
if logger:
|
||||
logger.info(
|
||||
f"[YOLO-TARGET] 当前帧分类={label}, class_id={class_id}, "
|
||||
f"conf={confidence:.3f}"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _det_to_src_xyxy(o, coord_mode: str, src_w: int, src_h: int, net_w: int, net_h: int):
|
||||
"""把单个检测框转为全图坐标系下的 xyxy(半开区间语义与后续 clip 一致)。"""
|
||||
x, y, w, h = float(o.x), float(o.y), float(o.w), float(o.h)
|
||||
|
||||
Reference in New Issue
Block a user