yolo最新选择

This commit is contained in:
yrx
2026-08-28 14:57:56 +08:00
parent 70aa072164
commit 231937afba
27 changed files with 418 additions and 29 deletions
+3
View File
@@ -0,0 +1,3 @@
{
"cmake.sourceDirectory": "E:/code/code/code/new/new/new/new/new/nw/2.17.0/archery/cpp_ext"
}
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+2 -2
View File
@@ -18,8 +18,8 @@ files:
- laser_manager.py
- logger_manager.py
- main.py
- model_285484.cvimodel
- model_285484.mud
- model_317211.cvimodel
- model_317211.mud
- network.py
- ota_curl.sh
- ota_manager.py
+10 -7
View File
@@ -234,10 +234,10 @@ TRIANGLE_BLACKHAT_KERNEL_FRAC = 0.018 # 核大小 ≈ min(h,w)*frac,取奇数
# ── YOLO(NPU) 靶环 ROI → 裁剪后再跑三角形(减小 CPU 处理面积)──────────────────
# 日志里 net_in=W×H 来自 .mud 模型(det.input_width/height),不是这里配置的。
TRIANGLE_YOLO_ROI_ENABLE = True
TRIANGLE_YOLO_MODEL_PATH = APP_DIR + "/model_270139.mud"
TRIANGLE_YOLO_MODEL_PATH = APP_DIR + "/model_317211.mud"
# 参与 ROI 的类别:多类时只填「整靶/靶环」的 id;不要填角标类,否则 union 仍可对,但 largest 会偏小。
TRIANGLE_YOLO_RING_CLASS_IDS = (0,)
TRIANGLE_YOLO_CONF_TH = 0.7
TRIANGLE_YOLO_CONF_TH = 0.9
TRIANGLE_YOLO_IOU_TH = 0.45
# YOLO 首次/临界帧可能在高阈值下 0 框;启用后仅在 0 候选时用较低阈值重试一次。
# 后续仍会经过 min_box_side、ROI aspect、三角形几何校验,避免直接放大假阳性。
@@ -264,9 +264,9 @@ 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_MODEL_PATH = APP_DIR + "/model_317704.mud"
TARGET_CLASS_YOLO_LABELS = (20, 40)
TARGET_CLASS_YOLO_CONF_TH = 0.50
TARGET_CLASS_YOLO_CONF_TH = 0.66
TARGET_CLASS_YOLO_IOU_TH = 0.45
TARGET_CLASS_YOLO_RETRY_ON_EMPTY = False
TARGET_CLASS_YOLO_RETRY_CONF_TH = 0.25
@@ -326,14 +326,17 @@ LOG_QUEUE_MAXSIZE = 10000 # 日志队列上限
MAX_CMD_THREADS = 10 # 并发命令线程上限(防止服务器下发命令时无限创建线程)
# ==================== 图像保存配置 ====================
SAVE_IMAGE_ENABLED = False # 是否保存图像(True=保存,False=不保存)
SAVE_IMAGE_ON_FAILURE = True # 检测失败时是否强制保存图像(供调试测试用)
SAVE_IMAGE_ENABLED = True # 是否保存图像(True=保存,False=不保存)
SAVE_IMAGE_ON_FAILURE = False # 检测失败时是否强制保存图像(供调试测试用)
PHOTO_DIR = "/root/phot" # 照片存储目录
MAX_IMAGES = 1000
SAVE_RAW_IMAGE_ENABLED = True # 额外保存完整原始帧(不画框、不画点、不裁剪)
RAW_IMAGE_DIR = PHOTO_DIR + "/raw"
RAW_IMAGE_MAX_IMAGES = MAX_IMAGES
# Stage2 调试目录(默认 PHOTO_DIR/stage2_roi)内 JPEG 最多保留张数;None 表示与 MAX_IMAGES 相同
TRIANGLE_BLACK_YOLO_STAGE2_ROI_MAX_IMAGES = None
SHOW_CAMERA_PHOTO_WHILE_SHOOTING = False # 是否在拍摄时显示摄像头图像(True=显示,False=不显示),建议在连着USB测试过程中打开
SHOW_CAMERA_PHOTO_WHILE_SHOOTING = True # 是否在拍摄时显示摄像头图像(True=显示,False=不显示),建议在连着USB测试过程中打开
# ==================== OTA配置 ====================
MAX_BACKUPS = 5
+10 -3
View File
@@ -132,6 +132,7 @@ def cmd_str():
sync_system_time_from_4g()
# 2.1 WiFi 热点配网兜底:仅当 STA 与 4G 均不可用时起 AP + HTTP;提交后删 /boot/wifi.ap、建 wifi.sta 并 reboot
_ota_pending_path = f"{config.APP_DIR}/ota_pending.json"
try:
from wifi_config_httpd import maybe_start_wifi_ap_fallback
@@ -167,14 +168,16 @@ def cmd_str():
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:
if _preload_yolo and not os.path.exists(_ota_pending_path):
preload_yolo_detector(logger)
elif _preload_yolo and logger:
logger.warning("[YOLO] ota_pending.json found; skip model preload until rollback check")
except Exception as e:
if logger:
logger.warning(f"[YOLO-ROI] 启动预加载异常(不影响后续射箭): {e}")
# 3. 启动时检查:是否需要恢复备份
pending_path = f"{config.APP_DIR}/ota_pending.json"
pending_path = _ota_pending_path
if os.path.exists(pending_path):
try:
with open(pending_path, 'r', encoding='utf-8') as f:
@@ -250,7 +253,11 @@ def cmd_str():
network_manager.read_device_id()
# 5. 创建照片存储目录(如果启用图像保存或检测失败时强制保存)
if config.SAVE_IMAGE_ENABLED or getattr(config, "SAVE_IMAGE_ON_FAILURE", False):
if (
config.SAVE_IMAGE_ENABLED
or getattr(config, "SAVE_IMAGE_ON_FAILURE", False)
or getattr(config, "SAVE_RAW_IMAGE_ENABLED", False)
):
photo_dir = config.PHOTO_DIR
if photo_dir not in os.listdir("/root"):
try:
Binary file not shown.
+2 -2
View File
@@ -1,7 +1,7 @@
[basic]
type = cvimodel
model = model_270820.cvimodel
model = model_317189.cvimodel
[extra]
model_type = yolov5
@@ -9,5 +9,5 @@ input_type = rgb
mean = 0, 0, 0
scale = 0.00392156862745098, 0.00392156862745098, 0.00392156862745098
anchors = 10, 13, 16, 30, 33, 23, 30, 61, 62, 45, 59, 119, 116, 90, 156, 198, 373, 326
labels = triangle
labels = circle, triangle
Binary file not shown.
+2 -2
View File
@@ -1,7 +1,7 @@
[basic]
type = cvimodel
model = model_270139.cvimodel
model = model_317211.cvimodel
[extra]
model_type = yolov5
@@ -9,5 +9,5 @@ input_type = rgb
mean = 0, 0, 0
scale = 0.00392156862745098, 0.00392156862745098, 0.00392156862745098
anchors = 10, 13, 16, 30, 33, 23, 30, 61, 62, 45, 59, 119, 116, 90, 156, 198, 373, 326
labels = 黑三角和圆环
labels = circle, triangle
Binary file not shown.
+2 -2
View File
@@ -1,7 +1,7 @@
[basic]
type = cvimodel
model = model_285484.cvimodel
model = model_317423.cvimodel
[extra]
model_type = yolov5
@@ -9,5 +9,5 @@ input_type = rgb
mean = 0, 0, 0
scale = 0.00392156862745098, 0.00392156862745098, 0.00392156862745098
anchors = 10, 13, 16, 30, 33, 23, 30, 61, 62, 45, 59, 119, 116, 90, 156, 198, 373, 326
labels = 20, 40
labels = 20, 10, 40
+18 -5
View File
@@ -8,7 +8,12 @@ from laser_manager import laser_manager
from logger_manager import logger_manager
from network import network_manager
from triangle_target import load_camera_from_xml, load_triangle_positions, try_triangle_scoring
from vision import estimate_distance, detect_circle_v3, enqueue_save_shot
from vision import (
estimate_distance,
detect_circle_v3,
enqueue_save_shot,
enqueue_save_raw_shot,
)
from maix import image, time
# 缓存相机标定与三角形位置,避免每次射箭重复读磁盘
@@ -322,6 +327,11 @@ def process_shot(adc_val):
try:
frame = camera_manager.read_frame()
# Copy the untouched frame before any detection or drawing.
from shot_id_generator import shot_id_generator
shot_id = shot_id_generator.generate_id()
enqueue_save_raw_shot(frame, shot_id)
# 网络事件移到拍照之后,避免阻塞拍照
network_manager.safe_enqueue({"shoot_event": "start"}, msg_type=2, high=True)
@@ -380,10 +390,6 @@ def process_shot(adc_val):
if dx is None and dy is None and logger:
logger.warning("[MAIN] 未检测到偏移量(三角形与圆形均失败),但会保存图像")
# 生成射箭ID
from shot_id_generator import shot_id_generator
shot_id = shot_id_generator.generate_id()
if logger:
logger.info(f"[MAIN] 射箭ID: {shot_id}")
@@ -441,6 +447,13 @@ def process_shot(adc_val):
inner_data["ellipse_center_x"] = None
inner_data["ellipse_center_y"] = None
# 记录这组 inner_data 即将进入上报队列的本地时间,精确到毫秒。
upload_time_ms = int(time_std.time() * 1000)
upload_time_sec, upload_time_millis = divmod(upload_time_ms, 1000)
inner_data["upload_time"] = (
time_std.strftime("%Y-%m-%d %H:%M:%S", time_std.localtime(upload_time_sec))
+ f".{upload_time_millis:03d}"
)
report_data = {"cmd": 1, "data": inner_data}
if logger:
logger.info(
+12 -3
View File
@@ -126,10 +126,19 @@ def _get_detector(model_path: str):
return _detector_by_path[model_path]
try:
from maix import nn
except ImportError:
except Exception:
return None
_detector_by_path[model_path] = nn.YOLOv5(model=model_path, dual_buff=False)
return _detector_by_path[model_path]
# YOLO is an optional capability. A broken/incompatible model must not
# abort boot (especially before the OTA rollback check).
try:
detector = nn.YOLOv5(model=model_path, dual_buff=False)
except Exception:
# Cache the failure to avoid retrying a broken native load every frame.
# reset_yolo_detector_cache() clears this after a model replacement.
_detector_by_path[model_path] = None
return None
_detector_by_path[model_path] = detector
return detector
def preload_yolo_detector(logger=None):
+108
View File
@@ -0,0 +1,108 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Run from MaixVision on PC to inspect the box's live 20/40 YOLO output."""
import os
from maix import app, camera, display, image, nn, time
# This file is sent to /tmp/maixpy_run by MaixVision. Keep the model path
# absolute so the script uses the model already installed on the box.
MODEL_PATH = "/maixapp/apps/t11/model_317181.mud"
CAMERA_WIDTH = 640
CAMERA_HEIGHT = 480
CONF_TH = 0.65
IOU_TH = 0.45
def _flatten_objects(raw):
if raw is None:
return []
if isinstance(raw, (list, tuple)):
result = []
for item in raw:
if isinstance(item, (list, tuple)):
result.extend(_flatten_objects(item))
else:
result.append(item)
return result
return [raw]
def main():
if not os.path.isfile(MODEL_PATH):
raise FileNotFoundError("model not found on box: " + MODEL_PATH)
detector = nn.YOLOv5(model=MODEL_PATH, dual_buff=False)
cam = camera.Camera(CAMERA_WIDTH, CAMERA_HEIGHT)
disp = display.Display()
labels = tuple(str(label) for label in detector.labels)
print("[YOLO] model:", MODEL_PATH)
print("[YOLO] labels:", labels)
print("[YOLO] conf=%.2f iou=%.2f" % (CONF_TH, IOU_TH))
fps = 0.0
frame_count = 0
last_log_ms = time.ticks_ms()
while not app.need_exit():
loop_start_ms = time.ticks_ms()
img = cam.read()
detect_start_ms = time.ticks_ms()
raw = detector.detect(img, conf_th=CONF_TH, iou_th=IOU_TH)
detect_ms = max(0, time.ticks_diff(time.ticks_ms(), detect_start_ms))
objects = _flatten_objects(raw)
candidates = []
for obj in objects:
class_id = int(obj.class_id)
score = float(obj.score)
label = labels[class_id] if 0 <= class_id < len(labels) else "unknown"
color = image.COLOR_GREEN if label in ("20", "40") else image.COLOR_RED
img.draw_rect(obj.x, obj.y, obj.w, obj.h, color=color)
img.draw_string(
obj.x,
max(0, obj.y - 16),
"%scm %.2f" % (label, score),
color=color,
)
if label in ("20", "40"):
candidates.append((score, label))
loop_ms = max(1, time.ticks_diff(time.ticks_ms(), loop_start_ms))
instant_fps = 1000.0 / float(loop_ms)
fps = instant_fps if frame_count == 0 else fps * 0.9 + instant_fps * 0.1
if candidates:
best_score, best_label = max(candidates, key=lambda item: item[0])
status = "TARGET %scm %.2f" % (best_label, best_score)
status_color = image.COLOR_GREEN
else:
status = "TARGET UNKNOWN"
status_color = image.COLOR_RED
img.draw_string(5, 5, status, color=status_color)
img.draw_string(
5,
25,
"infer=%dms fps=%.1f boxes=%d" % (detect_ms, fps, len(objects)),
color=image.COLOR_YELLOW,
)
disp.show(img)
frame_count += 1
now_ms = time.ticks_ms()
if time.ticks_diff(now_ms, last_log_ms) >= 1000:
print(
"[YOLO] %s infer=%dms fps=%.1f boxes=%d"
% (status, detect_ms, fps, len(objects))
)
last_log_ms = now_ms
if __name__ == "__main__":
main()
+184
View File
@@ -0,0 +1,184 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Offline baseline for traditional target-paper detection.
Dataset format: sibling .txt files use YOLO boxes and classes.txt maps ids
(the supplied dataset uses 0=40, 1=20, 2=10). This intentionally simple
baseline uses grayscale segmentation and contour geometry; it is useful as a
reference before adding more specialized black-triangle grouping.
"""
from __future__ import annotations
import argparse
import csv
import glob
import itertools
import os
import cv2
import numpy as np
def detect_white_papers(image: np.ndarray) -> list[tuple[int, int, int, int]]:
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
h, w = gray.shape[:2]
mask = cv2.inRange(gray, 120, 255)
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((9, 9), np.uint8))
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((5, 5), np.uint8))
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
out = []
for contour in contours:
x, y, bw, bh = cv2.boundingRect(contour)
area = float(bw * bh)
if area < 0.05 * w * h or min(bw, bh) < 80:
continue
fill = cv2.contourArea(contour) / max(area, 1.0)
aspect = bw / max(float(bh), 1.0)
if fill >= 0.45 and 0.4 <= aspect <= 2.5:
out.append((x, y, x + bw, y + bh))
return out
def detect_black_triangle_papers(image: np.ndarray):
"""Infer paper boxes from the four small black corner marks."""
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
mask = cv2.inRange(gray, 0, 100)
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((2, 2), np.uint8))
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
points = []
for contour in contours:
x, y, bw, bh = cv2.boundingRect(contour)
area = cv2.contourArea(contour)
vertices = cv2.approxPolyDP(contour, 0.08 * cv2.arcLength(contour, True), True)
if 60 <= area <= 400 and 8 <= bw <= 24 and 8 <= bh <= 24:
if 3 <= len(vertices) <= 5 and 0.5 <= bw / max(bh, 1) <= 2.0:
points.append((x + bw / 2.0, y + bh / 2.0))
candidates = []
for group in itertools.combinations(points, 4):
xs = sorted(p[0] for p in group)
ys = sorted(p[1] for p in group)
span_x, span_y = xs[-1] - xs[0], ys[-1] - ys[0]
if span_x < 50 or span_y < 50 or not 0.45 < span_x / span_y < 1.5:
continue
corners = ((xs[0], ys[0]), (xs[-1], ys[0]),
(xs[0], ys[-1]), (xs[-1], ys[-1]))
error = max(min(np.hypot(p[0] - c[0], p[1] - c[1]) for c in corners)
for p in group) / max(span_x, span_y)
if error > 0.22:
continue
ex, ey = 0.12 * span_x, 0.12 * span_y
candidates.append((xs[0] - ex, ys[0] - ey,
xs[-1] + ex, ys[-1] + ey, error))
# A colored target ring supplies an independent center check. Hough is
# deliberately low-cost here because it runs only on the already small
# candidate list's source frame.
hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
color = cv2.inRange(hsv, (0, 70, 45), (179, 255, 255))
color = cv2.morphologyEx(color, cv2.MORPH_OPEN, np.ones((5, 5), np.uint8))
ring_centers = []
for contour in cv2.findContours(color, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)[0]:
area = cv2.contourArea(contour)
if area < 150:
continue
moments = cv2.moments(contour)
if moments["m00"]:
ring_centers.append((moments["m10"] / moments["m00"], moments["m01"] / moments["m00"]))
checked = []
for box in candidates:
if not ring_centers:
checked.append(box)
continue
x0, y0, x1, y1, err = box
inside = any(x0 - .15 * (x1 - x0) <= cx <= x1 + .15 * (x1 - x0)
and y0 - .15 * (y1 - y0) <= cy <= y1 + .15 * (y1 - y0)
for cx, cy in ring_centers)
if inside:
checked.append(box)
return sorted(checked, key=lambda x: x[-1])
def iou(a, b):
x0, y0 = max(a[0], b[0]), max(a[1], b[1])
x1, y1 = min(a[2], b[2]), min(a[3], b[3])
inter = max(0, x1 - x0) * max(0, y1 - y0)
aa = max(0, a[2] - a[0]) * max(0, a[3] - a[1])
bb = max(0, b[2] - b[0]) * max(0, b[3] - b[1])
return inter / max(aa + bb - inter, 1)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("dataset", help="directory containing jpg and YOLO txt files")
ap.add_argument("--iou", type=float, default=0.5)
ap.add_argument("--out", default="traditional_eval_results.csv",
help="CSV output path; relative paths are next to the dataset")
ap.add_argument("--vis-dir", default="traditional_eval_images",
help="directory for annotated result images; empty disables")
args = ap.parse_args()
stats = {0: [0, 0], 1: [0, 0]}
rows = []
# OpenCV on some Windows builds cannot decode non-ASCII filenames. Work
# relative to the dataset directory so the supplied Chinese path is safe.
dataset = os.path.abspath(args.dataset)
os.chdir(dataset)
# cwd is now the dataset, so a relative output avoids Windows console
# encoding issues with the Chinese parent path.
vis_dir = args.vis_dir if args.vis_dir else ""
if vis_dir:
os.makedirs(vis_dir, exist_ok=True)
files = glob.glob(os.path.join("**", "*.jpg"), recursive=True)
for image_path in files:
label_path = os.path.splitext(image_path)[0] + ".txt"
if not os.path.isfile(label_path):
continue
image = cv2.imread(image_path)
if image is None:
continue
h, w = image.shape[:2]
predictions = detect_black_triangle_papers(image)
vis = image.copy()
for p in predictions:
cv2.rectangle(vis, (int(p[0]), int(p[1])), (int(p[2]), int(p[3])), (0, 255, 255), 2)
for line in open(label_path, encoding="utf-8", errors="ignore"):
z = line.split()
if len(z) < 5 or int(float(z[0])) not in stats:
continue
cls, cx, cy, bw, bh = int(float(z[0])), *map(float, z[1:5])
truth = (int((cx - bw / 2) * w), int((cy - bh / 2) * h),
int((cx + bw / 2) * w), int((cy + bh / 2) * h))
best = max((iou(truth, p) for p in predictions), default=0.0)
best_box = max(predictions, key=lambda p: iou(truth, p), default=())
stats[cls][0] += 1
stats[cls][1] += int(best >= args.iou)
rows.append({
"image": image_path,
"class_id": cls,
"truth_xyxy": ",".join(map(str, truth[:4])),
"pred_xyxy": ",".join(map(str, best_box[:4])) if best_box else "",
"iou": f"{best:.4f}",
"pass": int(best >= args.iou),
})
color = (0, 255, 0) if best >= args.iou else (0, 0, 255)
cv2.rectangle(vis, truth[:2], truth[2:4], color, 2)
cv2.putText(vis, f"GT {cls} IoU {best:.2f}",
(truth[0], max(16, truth[1] - 4)),
cv2.FONT_HERSHEY_SIMPLEX, 0.45, color, 1, cv2.LINE_AA)
if vis_dir:
name = os.path.splitext(os.path.basename(image_path))[0] + "_result.jpg"
cv2.imwrite(os.path.join(vis_dir, name), vis)
total = sum(v[0] for v in stats.values())
good = sum(v[1] for v in stats.values())
print(f"paper objects: {good}/{total} = {good / max(total, 1):.2%} (IoU >= {args.iou})")
for cls, (n, ok) in stats.items():
print(f"class {cls}: {ok}/{n} = {ok / max(n, 1):.2%}")
out_path = args.out if os.path.isabs(args.out) else os.path.join(dataset, args.out)
with open(out_path, "w", newline="", encoding="utf-8-sig") as fp:
writer = csv.DictWriter(fp, fieldnames=("image", "class_id", "truth_xyxy",
"pred_xyxy", "iou", "pass"))
writer.writeheader()
writer.writerows(rows)
print(f"details csv: {out_path}")
if __name__ == "__main__":
main()
+7
View File
@@ -31,3 +31,10 @@
# 2.15.18 wifi连接成功重新登录
# 2.16.4 优化射箭延迟
# 2.17.0 yolo标靶类别识别
# 2.17.1 26-08-19 1739 压力传感修改 增量方式
# 2.17.2 26-08-24 1756 靶纸识别模型更替
# 2.17.3 26-08-25 957 原图拍摄开关
# 2.17.4 26-08-25 1457 模型修改
+1 -1
View File
@@ -4,6 +4,6 @@
应用版本号
每次 OTA 更新时,只需要更新这个文件中的版本号
"""
VERSION = '2.17.0'
VERSION = '2.17.4'
+56 -1
View File
@@ -908,7 +908,12 @@ def _save_worker_loop():
item = _save_queue.get()
if item is None:
break
_save_shot_image_impl(*item)
if isinstance(item, dict) and item.get("kind") == "raw":
_save_raw_image_impl(
item["img_cv"], item["shot_id"], item["photo_dir"]
)
else:
_save_shot_image_impl(*item)
except Exception as e:
logger = logger_manager.logger
if logger:
@@ -936,6 +941,56 @@ def start_save_shot_worker():
logger.info("[VISION] 存图 worker 线程已启动")
def _save_raw_image_impl(img_cv, shot_id, photo_dir):
"""保存相机完整原始帧,不添加任何检测标注。"""
logger = logger_manager.logger
try:
os.makedirs(photo_dir, exist_ok=True)
filename = os.path.join(photo_dir, f"shot_{shot_id}_raw.jpg")
image.cv2image(img_cv, False, False).save(filename)
prune_old_images_in_dir(
photo_dir,
getattr(config, "RAW_IMAGE_MAX_IMAGES", config.MAX_IMAGES),
logger,
"[VISION-RAW]",
)
if logger:
logger.info(f"[VISION-RAW] 已保存纯原图: {filename}")
return filename
except Exception as e:
if logger:
logger.error(f"[VISION-RAW] 保存纯原图失败: {e}")
return None
def enqueue_save_raw_shot(frame, shot_id, photo_dir=None):
"""立即复制相机帧并异步保存,避免后续识别和绘图修改原图。"""
if not getattr(config, "SAVE_RAW_IMAGE_ENABLED", False):
return
if photo_dir is None:
photo_dir = getattr(
config, "RAW_IMAGE_DIR", os.path.join(config.PHOTO_DIR, "raw")
)
try:
img_copy = np.copy(image.image2cv(frame, False, False))
_save_queue.put_nowait(
{
"kind": "raw",
"img_cv": img_copy,
"shot_id": shot_id,
"photo_dir": photo_dir,
}
)
except queue.Full:
logger = logger_manager.logger
if logger:
logger.warning("[VISION-RAW] 存图队列已满,跳过本次纯原图保存")
except Exception as e:
logger = logger_manager.logger
if logger:
logger.error(f"[VISION-RAW] 复制纯原图失败: {e}")
def enqueue_save_shot(result_img, center, radius, method, ellipse_params,
laser_point, distance_m, shot_id=None, photo_dir=None,
yolo_roi_xyxy=None, force_save=False):