feat: 2.18.10 靶规格复检与置信度门槛

- 检出靶框裁剪放大后二次 detect 复检(conf 0.25)，以复检结果为准
- 全图结果上报门槛 MIN_REPORT_CONF=0.5，复检结果 >0.25 即采纳
- 0 结果时按 RETRY_CONF_TH 自动重检；class_id=1 解析阶段过滤
- 换用 model_326162，去除调试 print 改走 logger

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "cmake.sourceDirectory": "E:/code/code/code/new/new/new/new/new/nw/2.17.0/archery/cpp_ext"
+}

app.yaml

@@ -1,6 +1,6 @@
 id: t11
 name: t11
-version: 2.15.35
+version: 2.18.10
 author: t11
 icon: ''
 desc: t11
@@ -18,8 +18,8 @@ files:
   - laser_manager.py
   - logger_manager.py
   - main.py
-  - model_285484.cvimodel
+  - model_326162.cvimodel
-  - model_285484.mud
+  - model_326162.mud
   - network.py
   - ota_curl.sh
   - ota_manager.py
@@ -28,6 +28,7 @@ files:
   - shoot_manager.py
   - shot_id_generator.py
   - target_roi_yolo.py
+  - tcp_messages_pb2.py
   - time_sync.py
   - triangle_positions.json
   - triangle_target.py

config.py

@@ -48,7 +48,7 @@ WIFI_CONFIG_AP_IP = "192.168.66.1"         # 与 MaixPy Wifi.start_ap 默认一
 # ===== TCP over SSL(TLS) 配置 =====
 USE_TCP_SSL = True          # True=按手册走 MSSLCFG/MIPCFG 绑定 SSL
 TCP_LINK_ID = 2             # 
-TCP_SSL_PORT = 50006          # TLS 端口（不一定必须 443，以服务器为准）
+TCP_SSL_PORT = 50007          # TLS 端口（不一定必须 443，以服务器为准）
 
 # SSL profile
 SSL_ID = 1                  # ssl_id=1
@@ -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、三角形几何校验，避免直接放大假阳性。
@@ -262,14 +262,25 @@ 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.
+# YOLO target size classification: class_id → label，下标即 class_id：0=40cm, 1=0(靶心), 2=20cm.
 TARGET_CLASS_YOLO_ENABLE = True
-TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_285484.mud"
+TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_326162.mud"
-TARGET_CLASS_YOLO_LABELS = (20, 40)
+TARGET_CLASS_YOLO_LABELS = (40, 0, 20)
-TARGET_CLASS_YOLO_CONF_TH = 0.50
+# 只有映射结果落在该集合内才用于上报（如 class 1→label 0 直接忽略）
-TARGET_CLASS_YOLO_IOU_TH = 0.45
+TARGET_CLASS_YOLO_VALID_LABELS = (20, 40)
-TARGET_CLASS_YOLO_RETRY_ON_EMPTY = False
+# 解析阶段直接丢弃的 class_id（1=靶心），不参与任何选择/上报
+TARGET_CLASS_YOLO_IGNORE_CLASS_IDS = (1,)
+TARGET_CLASS_YOLO_CONF_TH = 0.5
+# 首次检测 0 结果时用该阈值重新检测一次（≥CONF_TH 时按 CONF_TH 同参重跑）
 TARGET_CLASS_YOLO_RETRY_CONF_TH = 0.25
+# 检出靶框后裁剪放大再 detect 一次，与全图结果互相印证（不一致以复检为准）
+TARGET_CLASS_YOLO_RECHECK_CROP = True
+TARGET_CLASS_YOLO_RECHECK_MARGIN_FRAC = 0.1
+# 复检的 detect 阈值与接受门槛：置信度大于该值即采纳复检结果
+TARGET_CLASS_YOLO_RECHECK_CONF_TH = 0.25
+# 全图结果（无复检时）的上报置信度门槛，None/0 表示不启用
+TARGET_CLASS_YOLO_MIN_REPORT_CONF = 0.5
+TARGET_CLASS_YOLO_IOU_TH = 0.45
 TARGET_CLASS_YOLO_PRELOAD_ON_BOOT = True
 
 # ── 第二段 YOLO：仅在 Stage1 裁切出的靶环图上推理（与合成 stage2 训练数据一致）→ 子框内传统算法取直角点 ──
@@ -326,10 +337,13 @@ LOG_QUEUE_MAXSIZE = 10000       # 日志队列上限
 MAX_CMD_THREADS = 10            # 并发命令线程上限（防止服务器下发命令时无限创建线程）
 
 # ==================== 图像保存配置 ====================
-SAVE_IMAGE_ENABLED = False  # 是否保存图像（True=保存，False=不保存）
+SAVE_IMAGE_ENABLED = True  # 是否保存图像（True=保存，False=不保存）
 SAVE_IMAGE_ON_FAILURE = True  # 检测失败时是否强制保存图像（供调试测试用）
 PHOTO_DIR = "/root/phot"    # 照片存储目录
 MAX_IMAGES = 1000
+SAVE_RAW_IMAGE_ENABLED = False  # 额外保存完整原始帧（不画框、不画点、不裁剪）
+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
 
@@ -352,7 +366,22 @@ PIN_MAPPINGS = {
 }
 
 # ==================== 电源配置 ====================
-AUTO_POWER_OFF_IN_SECONDS = 10 * 60  # 自动关机时间（秒），0表示不自动关机
+AUTO_POWER_OFF_IN_SECONDS = 100 * 60  # 自动关机时间（秒），0表示不自动关机
+# 充电时自动关机暂时禁用；需要恢复时改为 True。
+CHARGING_AUTO_POWER_OFF_ENABLED = False
+
+# 一代电源控制：A24 由电源板负责按键/关机信号，软件关机时输出高电平。
+
+# 电源状态指示灯
+STATUS_LED_GREEN_GPIO = "GPIOA25"
+STATUS_LED_RED_GPIO = "GPIOA23"
+STATUS_LED_GREEN_ENABLED = True
+STATUS_LED_RED_ENABLED = True
+STATUS_LED_ACTIVE_LEVEL = 1
+STATUS_LED_LOW_BATTERY_PERCENT = 10
+STATUS_LED_FULL_BATTERY_PERCENT = 90
+STATUS_LED_CHARGING_BLINK_MS = 500
+STATUS_LED_POLL_MS = 1000
 
 BATTERY_SOC_LPF_ALPHA = 0.5
 BATTERY_SOC_AVG_WINDOW = 5

main.py

@@ -13,7 +13,9 @@ from maix import camera, display, image, app, time, uart, pinmap, i2c
 from maix.peripheral import adc
 import _thread
 import os
+import sys
 import json
+import shutil
 import time as wall_time
 
 # 导入新模块
@@ -132,6 +134,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 +170,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:
@@ -249,8 +254,61 @@ def cmd_str():
     # 4. 初始化设备ID（network_manager 内部会自动设置 device_id 和 password）
     network_manager.read_device_id()
 
+    # 4.1 检查是否有 OTA 待更新文件（从临时目录移动到实际目录）
+    staging_dir = f"{config.APP_DIR}/ota_staging"
+    if os.path.exists(staging_dir):
+        try:
+            moved_count = 0
+            for root, dirs, files in os.walk(staging_dir):
+                for f in files:
+                    src = os.path.join(root, f)
+                    rel = os.path.relpath(src, staging_dir)
+                    dest = os.path.join(config.APP_DIR, rel)
+                    dest_dir = os.path.dirname(dest)
+                    if dest_dir:
+                        try:
+                            os.makedirs(dest_dir, exist_ok=True)
+                        except:
+                            pass
+                    try:
+                        shutil.copy2(src, dest)
+                        moved_count += 1
+                    except Exception as e:
+                        if logger:
+                            logger.error(f"[OTA] 移动文件失败 {rel}: {e}")
+            # 删除临时目录
+            try:
+                shutil.rmtree(staging_dir, ignore_errors=True)
+            except:
+                pass
+            if logger:
+                logger.info(f"[OTA] 已从临时目录更新 {moved_count} 个文件，重启应用...")
+            # 清理硬件资源，然后重启应用
+            try:
+                laser_manager.turn_off_laser()
+            except:
+                pass
+            try:
+                camera_manager.release()
+            except:
+                pass
+            try:
+                os.sync()
+            except:
+                pass
+            import sys
+            os.execv(sys.executable, [sys.executable, os.path.join(config.APP_DIR, "main.py")])
+            return
+        except Exception as e:
+            if logger:
+                logger.error(f"[OTA] 处理临时目录失败: {e}")
+
     # 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:
@@ -282,12 +340,14 @@ def cmd_str():
         logger.info("系统准备完成...")
 
     last_adc_trigger = 0
+    trigger_adc_val = 0  # 触发时的气压值，气压需降回此值以下才能再次触发
     # 读取一次ADC初始值，防止开机时传感器已有压力导致误触发
+    enable_check = True
+    _should_reboot = False
     try:
         last_adc_val = hardware_manager.adc_obj.read()
     except Exception:
         last_adc_val = 0
-    peak_adc_val = 0  # 当前周期内的压力峰值
     # 气压采样：减少日志频率（每 N 个点输出一条），避免 logger.debug 拖慢采样
     PRESSURE_BATCH_SIZE = 100
 
@@ -339,6 +399,14 @@ def cmd_str():
                 time.sleep_ms(250)
                 continue
 
+            # OTA 完成后需要重启，从主循环退出（由启动时 staging 检测处理重启）
+            if network_manager.ota_restart_pending:
+                network_manager.ota_restart_pending = False
+                _should_reboot = True
+                if logger:
+                    logger.info("[MAIN] OTA重启标志已设置，退出主循环...")
+                break
+
             # 不在 OTA 状态下，检测是否空闲足够长，自动关机
             # print(f"[MAIN] 空闲时间: {hardware_manager.get_idle_time_in_sec() }秒")
             # print(f"配置关机时间：{config.AUTO_POWER_OFF_IN_SECONDS} 秒")
@@ -377,22 +445,16 @@ def cmd_str():
                 pressure_max = adc_val
             if len(pressure_buf) >= PRESSURE_BATCH_SIZE:
                 _flush_pressure_buf("batch")
-            # 峰值检测：压力从峰值下降时触发，确保捕获到最大冲击时刻
+            # 突变增量检测：压力增量大于400时触发
-            if adc_val > peak_adc_val:
+            # 触发后需等气压降到触发值以下才重新检测增量
-                peak_adc_val = adc_val  # 更新峰值
+            if adc_val < trigger_adc_val :
-            if (peak_adc_val >= config.ADC_TRIGGER_THRESHOLD
+                enable_check = True
-                    and adc_val < peak_adc_val
+            if (adc_val - last_adc_val) > 200 and enable_check:
-                    and last_adc_val >= peak_adc_val):
-                # 封顶后下降沿触发：peak是最大值，当前值开始下降，且上次值还在peak位置
                 hardware_manager.start_idle_timer()  # 重新计时
-                diff_ms = current_time - last_adc_trigger
-                if diff_ms < 3000:
-                    peak_adc_val = 0  # 去抖期间重置峰值
-                    time.sleep_ms(5)
-                    continue
                 last_adc_trigger = current_time
-                peak_adc_val = 0  # 触发后重置峰值
+                trigger_adc_val = adc_val  # 记录触发时的气压值
-                # 触发前先把缓存刷出来，避免波形被长耗时处理截断
+                last_adc_val = adc_val  # 更新基准值，防止连续增量误触发
+                enable_check = False
                 _flush_pressure_buf("before_trigger")
 
                 try:
@@ -411,7 +473,7 @@ def cmd_str():
                         camera_manager.show(camera_manager.read_frame())
                     except Exception as e:
                         pass
-                time.sleep_ms(5)
+                time.sleep_ms(1)
             last_adc_val = adc_val
 
         except Exception as e:
@@ -430,10 +492,62 @@ def cmd_str():
                 _flush_pressure_buf("exception")
             except:
                 pass
-            time.sleep_ms(1000)  # 等待1秒后继续
+            time.sleep_ms(1000)  # 等待1秒后 continue
-
-
 
+    # 主循环退出后，如果是由 OTA 触发的，移动 staging 文件并重启应用
+    if _should_reboot:
+        staging_dir = f"{config.APP_DIR}/ota_staging"
+        if os.path.exists(staging_dir):
+            try:
+                moved_count = 0
+                for root, dirs, files in os.walk(staging_dir):
+                    for f in files:
+                        src = os.path.join(root, f)
+                        rel = os.path.relpath(src, staging_dir)
+                        dest = os.path.join(config.APP_DIR, rel)
+                        dest_dir = os.path.dirname(dest)
+                        if dest_dir:
+                            try:
+                                os.makedirs(dest_dir, exist_ok=True)
+                            except:
+                                pass
+                        try:
+                            shutil.copy2(src, dest)
+                            moved_count += 1
+                        except Exception as e:
+                            if logger:
+                                logger.error(f"[OTA] 移动文件失败 {rel}: {e}")
+                try:
+                    shutil.rmtree(staging_dir, ignore_errors=True)
+                except:
+                    pass
+                if logger:
+                    logger.info(f"[MAIN] OTA 更新完成，已应用 {moved_count} 个文件，重启应用...")
+            except Exception as e:
+                if logger:
+                    logger.error(f"[OTA] 处理 staging 目录失败: {e}")
+        else:
+            if logger:
+                logger.info("[MAIN] OTA 更新完成，重启应用...")
+        # 清理硬件资源，然后重启应用（不重启设备）
+        try:
+            laser_manager.turn_off_laser()
+        except:
+            pass
+        try:
+            camera_manager.release()
+        except:
+            pass
+        try:
+            hardware_manager.stop_idle_timer()
+        except:
+            pass
+        try:
+            os.sync()
+        except:
+            pass
+        import sys
+        os.execv(sys.executable, [sys.executable, os.path.join(config.APP_DIR, "main.py")])
 
 
 # 主程序入口

model_270139.mud

@@ -1,13 +0,0 @@
-
-[basic]
-type = cvimodel
-model = model_270139.cvimodel
-
-[extra]
-model_type = yolov5
-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 = 黑三角和圆环
-

model_270820.mud

@@ -1,13 +0,0 @@
-
-[basic]
-type = cvimodel
-model = model_270820.cvimodel
-
-[extra]
-model_type = yolov5
-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
-

model_326162.mud

@@ -1,7 +1,7 @@
 
 [basic]
 type = cvimodel
-model = model_285484.cvimodel
+model = model_326162.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 = 40, 00, 20
 

network.py

@@ -13,6 +13,7 @@ import hmac
 import hashlib
 import ujson
 import os
+import sys
 import threading
 import socket
 import config
@@ -21,7 +22,13 @@ from hardware import hardware_manager
 from power import get_bus_voltage, voltage_to_percent, is_charging
 from logger_manager import logger_manager
 from wifi import wifi_manager
-import subprocess
+
+# protobuf 支持（纯 proto 协议，必须可用）
+try:
+    import tcp_messages_pb2 as pb
+except ImportError:
+    print("[NET] tcp_messages_pb2 not found, protobuf disabled")
+    raise
 
 
 def _wifi_tls_would_block(exc):
@@ -69,9 +76,18 @@ class NetworkManager:
         self._uart4g_lock = threading.Lock()
         self._device_id = None
         self._password = None
-        self._raw_line_data = []
+
         self._manual_trigger_flag = False
 
+        # OTA 防重复：上次 OTA 完成时间戳，30秒内不重复 OTA
+        self._last_ota_time = 0
+        self._ota_cooldown_sec = 30
+
+        # OTA 重启标志：OTA线程设置，主循环检测到后从主循环退出再重启
+        self.ota_restart_pending = False
+
+        # protobuf 协议（纯 proto，无 JSON 兼容）
+
         # 限制并发命令线程数
         self._cmd_thread_lock = threading.Lock()
         self._cmd_thread_count = 0
@@ -190,13 +206,7 @@ class NetworkManager:
                 return self._normal_send_queue.pop(0)
             return None
 
-    def _set_raw_line_data(self, data):
-        """设置原始行数据（内部方法）"""
-        self._raw_line_data = data
 
-    def _get_raw_line_data(self):
-        """获取原始行数据（内部方法）"""
-        return self._raw_line_data
 
     def get_uart_lock(self):
         """获取UART锁（用于with语句）"""
@@ -615,53 +625,113 @@ class NetworkManager:
         except Exception as e:
             self.logger.error(f"[LASER] cmd200 检测异常: {e}")
 
+    def _cmd5_ota(self, ota_url):
+        """后台线程执行 cmd5 OTA"""
+        hardware_manager.start_idle_timer()
+        self.logger.info(f"[Ota] cmd5 开始OTA: {ota_url}")
+        self.safe_enqueue({"result": "ota start..."}, 2)
+
+        try:
+            from ota_manager import ota_manager
+            ok, msg = ota_manager.perform_ota(ota_url)
+            if ok:
+                self.safe_enqueue({"result": "success"}, 2)
+                time.sleep_ms(500)
+                os.execv(sys.executable, [sys.executable, os.path.join(config.APP_DIR, "main.py")])
+            else:
+                self.logger.error(f"[ota] cmd5 失败: {msg}")
+                self.safe_enqueue({"result": "ota fail", "reason": msg}, 2)
+        except Exception as e:
+            self.logger.error(f"[ota] cmd5 异常: {e}")
+            self.safe_enqueue({"result": "ota fail", "reason": str(e)}, 2)
+
     def _cmd300_ota(self, data_obj):
-        """后台线程执行 cmd300 OTA，避免阻塞主循环"""
+        """后台线程执行 cmd300 OTA，避免阻塞主循环
+        流程：检查WiFi → 下载ZIP → 解压覆盖项目 → 重启程序
+        """
         hardware_manager.start_idle_timer()
         inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
         self.logger.info(f"[New Ota] cmd300 , data: {inner_data}")
-        ssid = inner_data.get("ssid")
-        password = inner_data.get("password")
         ota_res_url = inner_data.get("url")
-        try:
+
-            for _f in ("/etc/wpa_supplicant.conf", "/boot/wpa_supplicant.conf", "/boot/wifi.ssid", "/boot/wifi.pass"):
+        if not ota_res_url:
+            self.logger.error("[ota] cmd300 缺少 url 参数")
+            self.safe_enqueue({"cmd": 300, "result": "ota fail", "reason": "missing url"}, 2)
+            return
+
+        # OTA 冷却期检查：防止服务器重复下发导致无限 OTA 循环
+        now = time.time()
+        if self._last_ota_time > 0:
+            elapsed = int(now - self._last_ota_time)
+            if elapsed < self._ota_cooldown_sec:
+                remaining = self._ota_cooldown_sec - elapsed
+                self.logger.warning(f"[ota] cmd300 冷却期内，跳过 (剩余 {remaining}s)")
                 try:
-                    os.remove(_f)
+                    pkt = self._make_send_packet(2, {"cmd": 300, "result": "ota skip", "reason": f"cooldown {remaining}s"})
-                except OSError:
+                    self.tcp_send_raw(pkt)
+                except Exception:
                     pass
-            w = network.wifi.Wifi()
+                return
-            e = w.connect(ssid, password, wait=True, timeout=15)
+
-            err.check_raise(e, "connect wifi failed")
+        if not wifi_manager.is_wifi_connected():
-            if self.logger:
+            self.logger.warning("[ota] cmd300 当前未连接WiFi，拒绝OTA")
-                self.logger.info(f"[ota] Connect success, got ip{w.get_ip()}")
+            self.safe_enqueue({"cmd": 300, "result": "ota fail", "reason": "wifi not connected"}, 2)
-            self.safe_enqueue(
+            return
-                {
+
+        self.logger.info(f"[ota] WiFi已连接，开始OTA: {ota_res_url}")
+        self.safe_enqueue({"cmd": 300, "result": "ota start..."}, 2)
+
+        def _ota_progress(phase, progress):
+            """OTA进度回调，通过tcp_send_raw直接发送（绕过被暂停的发送队列）"""
+            try:
+                if not self._tcp_connected:
+                    self.logger.warning(f"[ota] 进度发送跳过: tcp未连接 phase={phase} progress={progress}")
+                    return
+                pkt = self._make_send_packet(2, {
                     "cmd": 300,
-                    "result": "ota start...",
+                    "result": f"ota {phase}",
-                    "wifi": w.get_ip(),
+                    "progress": progress,
-                },
+                    "phase": phase,
-                2,
+                })
-            )
+                ok = self.tcp_send_raw(pkt)
-            subprocess.run(
+                if not ok:
-                ["sh", "/maixapp/apps/t11/ota_curl.sh", ota_res_url])
+                    self.logger.warning(f"[ota] 进度发送失败: phase={phase} progress={progress}")
-            self.safe_enqueue(
+            except Exception as e:
-                {
+                self.logger.error(f"[ota] 发送进度异常: {e}")
-                    "cmd": 300,
+
-                    "result": "success",
+        try:
-                    "wifi": w.get_ip(),
+            from ota_manager import ota_manager
-                },
+            ok, msg = ota_manager.perform_ota(ota_res_url, progress_callback=_ota_progress)
-                2,
+            self._last_ota_time = time.time()
-            )
+            if ok:
+                self.logger.info("[ota] OTA成功，准备重启程序...")
+                # 直接通过tcp发送success，不走发送队列（主循环可能未drain）
+                try:
+                    pkt = self._make_send_packet(2, {"cmd": 300, "result": "success", "progress": 51, "phase": "rebooting"})
+                    ok_send = self.tcp_send_raw(pkt)
+                    self.logger.info(f"[ota] success包发送结果: {ok_send}, tcp_connected={self._tcp_connected}")
+                except Exception as e:
+                    self.logger.error(f"[ota] 发送success失败: {e}")
+                # 设置重启标志，由主循环检测到后从主循环退出再重启
+                # （后台线程调 os.execv 会导致 ISP 线程残留，新进程摄像头初始化失败）
+                self.logger.info("[ota] 设置重启标志，等待主循环退出...")
+                self.ota_restart_pending = True
+            else:
+                self.logger.error(f"[ota] cmd300 失败: {msg}")
+                self._last_ota_time = time.time()
+                try:
+                    pkt = self._make_send_packet(2, {"cmd": 300, "result": "ota fail", "reason": msg})
+                    self.tcp_send_raw(pkt)
+                except Exception as e:
+                    self.logger.error(f"[ota] 发送失败结果异常: {e}")
         except Exception as e:
-            self.logger.error(f"[ota] cmd300 失败: {e}")
+            self.logger.error(f"[ota] cmd300 异常: {e}")
-            self.safe_enqueue(
+            self._last_ota_time = time.time()
-                {
+            try:
-                    "cmd": 300,
+                pkt = self._make_send_packet(2, {"cmd": 300, "result": "ota fail", "reason": str(e)})
-                    "result": "ota fail",
+                self.tcp_send_raw(pkt)
-                    "reason": str(e),
+            except Exception as ex:
-                },
+                self.logger.error(f"[ota] 发送失败结果异常: {ex}")
-                2,
-            )
 
     def _cmd600_conn_wifi(self, data_obj):
         hardware_manager.start_idle_timer()
@@ -669,6 +739,7 @@ class NetworkManager:
         self.logger.info(f"[conn wifi] cmd600 , data: {inner_data}")
         ssid = inner_data.get("ssid")
         password = inner_data.get("password")
+        prev_network_type = self._network_type
         # 停止旧的WiFi质量监测（无论当前是WiFi还是4G连接）
         self._stop_wifi_quality_monitor()
         try:
@@ -678,18 +749,10 @@ class NetworkManager:
                 except OSError:
                     pass
             w = network.wifi.Wifi()
-            e = w.connect(ssid, password, wait=True, timeout=15)
+            e = w.connect(ssid, password, wait=True, timeout=5)
             err.check_raise(e, "connect wifi failed")
             if self.logger:
                 self.logger.info(f"[ota] Connect success, got ip{w.get_ip()}")
-            self.safe_enqueue(
-                {
-                    "cmd": 600,
-                    "result": "success",
-                    "wifi": w.get_ip(),
-                },
-                2,
-            )
             self._session_force_4g = False
             self.disconnect_server()
             self._tcp_connected = False
@@ -697,20 +760,186 @@ class NetworkManager:
             self.logger.info("[conn wifi] WiFi已连接，等待主循环重新登录")
         except Exception as e:
             self.logger.error(f"cmd600 失败: {e}")
-            self.safe_enqueue(
+            # 同步发送失败结果（旧连接仍存活时直接发送）
-                {
+            if prev_network_type == "4g":
-                    "cmd": 600,
+                pkt = self._make_send_packet(2, {"cmd": 600, "result": "conn fail", "reason": str(e)})
-                    "result": "conn fail",
+                self.tcp_send_raw(pkt)
-                    "reason": str(e),
+            else:
-                },
+                self.safe_enqueue(
-                2,
+                    {
-            )
+                        "cmd": 600,
-            self._switch_to_4g_due_to_poor_wifi()
+                        "result": "conn fail",
+                        "reason": str(e),
+                    },
+                    2,
+                )
+            # 当前是4G在线，旧连接未断，无需切换；当前是WiFi，旧WiFi已被w.connect()断开，需回退4G
+            if prev_network_type == "wifi":
+                self._switch_to_4g_due_to_poor_wifi()
 
     def safe_enqueue(self, data_dict, msg_type=2, high=False):
         """线程安全地将消息加入队列（公共方法）"""
         self._enqueue((msg_type, data_dict), high)
 
+    def _make_send_packet(self, msg_type, data_dict):
+        """使用 protobuf 构造发送数据包"""
+        return self._make_proto_packet(msg_type, data_dict)
+
+    def _build_logic_body(self, cmd, data_dict):
+        """根据 cmd 构造对应的 LogicBody oneof payload"""
+        d = data_dict.get("data", data_dict) if isinstance(data_dict.get("data"), dict) else data_dict
+
+        if cmd == 1:
+            return pb.LogicBody(
+                cmd=cmd,
+                shoot_data=pb.ShootData(
+                    shot_id=d.get("shot_id", ""),
+                    x=d.get("x", 0.0),
+                    y=d.get("y", 0.0),
+                    r=d.get("r", 0.0),
+                    d=d.get("d", 0.0),
+                    adc=d.get("adc", 0.0),
+                    target_class=str(d.get("target_class", "")),
+                    target_class_confidence=d.get("target_class_confidence", 0.0),
+                    d_laser=d.get("d_laser", 0.0),
+                    d_laser_quality=d.get("d_laser_quality", 0.0),
+                    m=d.get("m", ""),
+                    laser_method=d.get("laser_method", ""),
+                    target_x=d.get("target_x", 0.0),
+                    target_y=d.get("target_y", 0.0),
+                    offset_method=d.get("offset_method", ""),
+                    distance_method=d.get("distance_method", ""),
+                )
+            )
+        elif cmd == 4:
+            return pb.LogicBody(
+                cmd=cmd,
+                battery_report=pb.BatteryReport(
+                    battery=d.get("battery", 0.0),
+                    voltage=d.get("voltage", 0.0),
+                    net_type=d.get("netType", ""),
+                    charging=d.get("charging", False),
+                )
+            )
+        elif cmd == 200:
+            return pb.LogicBody(
+                cmd=cmd,
+                center_point_result=pb.CenterPointResult(
+                    result=d.get("result", ""),
+                    x=d.get("x", 0.0),
+                    y=d.get("y", 0.0),
+                )
+            )
+        elif cmd == 201:
+            return pb.LogicBody(
+                cmd=cmd,
+                center_point_set=pb.CenterPointSet(
+                    x=d.get("x", 0.0),
+                    y=d.get("y", 0.0),
+                )
+            )
+
+        elif cmd == 300:
+            return pb.LogicBody(
+                cmd=cmd,
+                ota_result=pb.OtaResult(
+                    result=d.get("result", ""),
+                    url=d.get("wifi", ""),
+                    progress=d.get("progress", 0),
+                    phase=d.get("phase", ""),
+                )
+            )
+        elif cmd == 700:
+            return pb.LogicBody(
+                cmd=cmd,
+                charging_report=pb.ChargingReport(),
+            )
+        else:
+            result_str = d.get("result", "")
+            if isinstance(result_str, dict):
+                import ujson
+                result_str = ujson.dumps(result_str)
+            elif not isinstance(result_str, str):
+                result_str = str(result_str)
+            return pb.LogicBody(
+                cmd=cmd,
+                generic_result=pb.GenericResult(result=result_str),
+            )
+
+    def _make_proto_packet(self, msg_type, data_dict):
+        """使用 protobuf 序列化构造数据包"""
+        if msg_type == 1:
+            msg = pb.LoginRequest(
+                device_id=data_dict.get("deviceId", ""),
+                password=data_dict.get("password", ""),
+                if_admin=data_dict.get("ifAdmin", False),
+                version=data_dict.get("version", ""),
+                vol=data_dict.get("vol", 0),
+                vol_per=data_dict.get("vol_per", 0),
+                iccid=data_dict.get("iccid", ""),
+            )
+        elif msg_type == 4:
+            msg = pb.Heartbeat(
+                t=data_dict.get("t", 0),
+                vol=data_dict.get("vol", 0),
+                vol_per=data_dict.get("vol_per", 0),
+            )
+        elif msg_type == 2:
+            cmd = data_dict.get("cmd", 0)
+            msg = self._build_logic_body(cmd, data_dict)
+        else:
+            return b""
+
+        body_bytes = msg.SerializeToString()
+        return self._netcore.make_packet_pb(msg_type, body_bytes)
+
+    def _parse_recv(self, payload):
+        """解析接收的数据包，返回 (msg_type, body_dict)"""
+        msg_type, body_bytes = self._netcore.parse_packet_raw(payload)
+        if msg_type is None:
+            return None, None
+        try:
+            body_dict = self._parse_proto_body(msg_type, body_bytes)
+            return msg_type, body_dict
+        except Exception as e:
+            self.logger.error(f"[NET] protobuf 反序列化失败: {e}")
+            return None, None
+
+    def _parse_proto_body(self, msg_type, body_bytes):
+        """将 protobuf body bytes 反序列化为 dict"""
+        if msg_type == 1:
+            msg = pb.LoginResponse()
+            msg.ParseFromString(body_bytes)
+            return {"cmd": msg.code, "data": msg.msg}
+        elif msg_type == 4:
+            return {}
+        elif msg_type == 2:
+            msg = pb.LogicBody()
+            msg.ParseFromString(body_bytes)
+            result = {"cmd": msg.cmd}
+            payload_name = msg.WhichOneof('payload')
+            if payload_name:
+                payload_msg = getattr(msg, payload_name)
+                data = {}
+                for field in payload_msg.DESCRIPTOR.fields:
+                    val = getattr(payload_msg, field.name)
+                    if isinstance(val, bytes):
+                        val = val.hex()
+                    data[field.name] = val
+                result["data"] = data
+            return result
+
+        elif msg_type == 100:
+            msg = pb.ImageUploadCommand()
+            msg.ParseFromString(body_bytes)
+            return {"uploadUrl": msg.upload_url, "token": msg.token, "shootId": msg.shoot_id, "outlink": msg.outlink}
+        elif msg_type == 101:
+            msg = pb.LogUploadCommand()
+            msg.ParseFromString(body_bytes)
+            return {"uploadUrl": msg.upload_url, "token": msg.token, "key": msg.key, "outlink": msg.outlink, "archive": msg.archive}
+        else:
+            return {"raw": body_bytes.hex()}
+
     def connect_server(self):
         """
         连接到服务器（自动选择WiFi或4G）
@@ -1833,6 +2062,11 @@ class NetworkManager:
                     time.sleep_ms(200)
                     continue
 
+                # OTA 完成后需要重启，从主循环退出（由 main.py 执行重启）
+                if self.ota_restart_pending:
+                    self.logger.info("[ota] 主循环退出，准备重启...")
+                    break
+
                 if not self.connect_server():
                     time.sleep_ms(1000)
                     continue
@@ -1848,8 +2082,7 @@ class NetworkManager:
                 }
                 iccid_pending_marker = self._maybe_add_iccid_to_login(login_data)
                 print(f"login_data: {login_data}")
-                # if not self.tcp_send_raw(self.make_packet(1, login_data)):
+                if not self.tcp_send_raw(self._make_send_packet(1, login_data)):
-                if not self.tcp_send_raw(self._netcore.make_packet(1, login_data)):
                     self._tcp_connected = False
                     try:
                         self.disconnect_server()
@@ -1880,6 +2113,10 @@ class NetworkManager:
                         time.sleep_ms(200)
                         continue
 
+                    # OTA 完成后需要重启，跳出内层循环
+                    if self.ota_restart_pending:
+                        break
+
                     # 接收数据（根据网络类型选择接收方式）
                     # WiFi 粘包：一次 recv 可能含多条完整包；也可能缓冲里已有完整包但本轮 recv 超时为空
                     rx_items = []
@@ -1926,11 +2163,11 @@ class NetworkManager:
                                     pass
 
                             # msg_type, body = self.parse_packet(payload)
-                            msg_type, body = self._netcore.parse_packet(payload)
+                            msg_type, body = self._parse_recv(payload)
 
                             # 处理登录响应
                             if not logged_in and msg_type == 1:
-                                if body and body.get("cmd") == 1 and body.get("data") == "登录成功":
+                                if body and body.get("cmd") == 0 and body.get("data") == "登录成功":
                                     logged_in = True
                                     last_heartbeat_ack_time = time.ticks_ms()
                                     self.logger.info("登录成功")
@@ -1961,32 +2198,7 @@ class NetworkManager:
                                 last_heartbeat_ack_time = time.ticks_ms()
                                 self.logger.debug("✅ 收到心跳确认")
 
-                            # 处理命令40（分片下载）
+
-                            elif logged_in and msg_type == 40:
-                                if isinstance(body, dict):
-                                    t = body.get('t', 0)
-                                    v = body.get('v')
-                                    # 如果是第一个分片，清空之前的缓存
-                                    if len(self._raw_line_data) == 0 or (
-                                            len(self._raw_line_data) > 0 and self._raw_line_data[0].get('v') != v):
-                                        self._raw_line_data.clear()
-                                    # 或者更简单：每次收到命令40时，如果版本号不同，清空缓存
-                                    if len(self._raw_line_data) > 0:
-                                        first_v = self._raw_line_data[0].get('v')
-                                        if first_v and first_v != v:
-                                            self._raw_line_data.clear()
-                                    self._raw_line_data.append(body)
-                                    if len(self._raw_line_data) >= int(t):
-                                        self.logger.info(f"下载完成")
-                                        from ota_manager import ota_manager
-                                        stock_array = list(map(lambda x: x.get('d'), self._raw_line_data))
-                                        local_filename = config.LOCAL_FILENAME
-                                        with open(local_filename, 'w', encoding='utf-8') as file:
-                                            file.write("\n".join(stock_array))
-                                        ota_manager.apply_ota_and_reboot(None, local_filename)
-                                    else:
-                                        self.safe_enqueue({'data': {'l': len(self._raw_line_data), 'v': v}, 'cmd': 41})
-                                    self.logger.info(f"已下载{len(self._raw_line_data)} 全部：{t} 版本：{v}")
 
                             elif logged_in and msg_type == 100:
                                 self.logger.info(f"[IMAGE_UPLOAD] 收到图片上传命令 {body}")
@@ -2079,215 +2291,151 @@ class NetworkManager:
                                         )
                                         # 立即返回已入队确认
                                         self.safe_enqueue({"result": "log_upload_queued"}, 2)
-                            elif logged_in and msg_type == 201:
+                            # 处理业务指令（纯 proto: cmd 在 body 顶层）
-                                if self.logger:
-                                    self.logger.info(f"[LASER] cmd201:{body}")
-                                raw_x = body.get("x")
-                                raw_y = body.get("y")
-                                try:
-                                    from laser_manager import laser_manager
-                                    ix, iy = laser_manager.set_hardcoded_laser_point(
-                                        raw_x, raw_y
-                                    )
-                                    self.safe_enqueue(
-                                        {
-                                            "cmd": 201,
-                                            "result": "laser_point_set",
-                                            "x": ix,
-                                            "y": iy,
-                                        },
-                                        2,
-                                    )
-                                    self.logger.info(
-                                        f"[LASER] cmd201 硬编码激光点=({ix}, {iy})"
-                                    )
-                                except Exception as e:
-                                    self.logger.error(f"[LASER] cmd201 失败: {e}")
-                                    self.safe_enqueue(
-                                        {
-                                            "cmd": 201,
-                                            "result": "laser_point_set_failed",
-                                            "reason": str(e),
-                                        },
-                                        2,
-                                    )
-                                hardware_manager.start_idle_timer()
-                            # 处理业务指令
                             elif logged_in and isinstance(body, dict):
-                                inner_cmd = None
+                                cmd = body.get("cmd")
-                                data_obj = body.get("data")
+                                data_obj = body.get("data") or {}
-                                if isinstance(data_obj, dict):
+                                if cmd == 2:  # AimRequest 开启激光并校准
-                                    inner_cmd = data_obj.get("cmd")
+                                    from laser_manager import laser_manager
-                                    if inner_cmd == 2:  # 开启激光并校准
+                                    if not laser_manager.calibration_active:
-                                        from laser_manager import laser_manager
+                                        laser_manager.turn_on_laser()
-                                        if not laser_manager.calibration_active:
+                                        time.sleep_ms(100)
-                                            laser_manager.turn_on_laser()
+                                        hardware_manager.stop_idle_timer()
-                                            time.sleep_ms(100)
+                                        if not config.HARDCODE_LASER_POINT:
-                                            hardware_manager.stop_idle_timer()  # 停表
+                                            laser_manager.start_calibration()
-                                            if not config.HARDCODE_LASER_POINT:
+                                            self.safe_enqueue({"result": "calibrating"}, 2)
-                                                laser_manager.start_calibration()
-                                                self.safe_enqueue({"result": "calibrating"}, 2)
-                                            else:
-                                                # 写死的逻辑，不需要校准激光点
-                                                self.safe_enqueue({"result": "laser pos set by hard code"}, 2)
-                                    elif inner_cmd == 3:  # 关闭激光
-                                        from laser_manager import laser_manager
-                                        laser_manager.turn_off_laser()
-                                        laser_manager.stop_calibration()
-                                        hardware_manager.start_idle_timer()  # 开表
-                                        self.safe_enqueue({"result": "laser_off"}, 2)
-                                    elif inner_cmd == 4:  # 上报电量
-                                        voltage = get_bus_voltage()
-                                        battery_percent = voltage_to_percent(voltage)
-                                        battery_data = {
-                                            "battery": battery_percent,
-                                            "voltage": round(float(voltage), 3),
-                                            "netType": self.network_type,
-                                        }
-                                        self.safe_enqueue(battery_data, 2)
-                                        self.logger.info(f"电量上报: {battery_percent}% 充电: {is_charging()}")
-                                        if is_charging():
-                                            self.safe_enqueue(
-                                                {
-                                                    "cmd": 700,
-                                                },
-                                                2,
-                                            )
-                                    elif inner_cmd == 700:
-                                        self.logger.warning("服务器下发关机！！！")
-                                        exit(-1)
-                                    elif inner_cmd == 5:  # OTA 升级
-                                        inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
-                                        ssid = inner_data.get("ssid")
-                                        password = inner_data.get("password")
-                                        ota_url = inner_data.get("url")
-                                        mode = (inner_data.get("mode") or "").strip().lower()
-
-                                        if not ota_url:
-                                            self.logger.error("ota missing_url")
-                                            self.safe_enqueue({"result": "missing_url"}, 2)
-                                            _rx_skip_tcp_iteration = True
-                                            break
-
-                                        from ota_manager import ota_manager
-                                        if ota_manager.update_thread_started:
-                                            self.safe_enqueue({"result": "update_already_started"}, 2)
-                                            _rx_skip_tcp_iteration = True
-                                            break
-
-                                        # 自动判断模式：如果没有明确指定，根据WiFi连接状态和凭证决定
-                                        if mode not in ("4g", "wifi"):
-                                            self.logger.info("ota missing mode, auto-detecting...")
-                                            # 若本次会话已锁定 4G，则 OTA 自动也走 4G，避免后续回切导致体验不一致
-                                            if self._session_force_4g:
-                                                mode = "4g"
-                                                self.logger.info("ota auto-selected: 4g (session locked on 4g)")
-                                            else:
-                                                # 只有同时满足：WiFi已连接 且 提供了WiFi凭证，才使用WiFi
-                                                if self.is_wifi_connected() and ssid and password:
-                                                    mode = "wifi"
-                                                    self.logger.info(
-                                                        "ota auto-selected: wifi (WiFi connected and credentials provided)")
-                                                else:
-                                                    mode = "4g"
-                                                    self.logger.info(
-                                                        "ota auto-selected: 4g (WiFi not available or no credentials)")
-
-                                        hardware_manager.stop_idle_timer()  # 停表，注意OTA停表之后，就没有再开表，因为OTA后面会重启，会重新开表
-
-                                        if mode == "4g":
-                                            ota_manager._set_ota_url(ota_url)  # 记录 OTA URL，供命令7使用
-                                            ota_manager._start_update_thread()
-                                            self._spawn_cmd_thread(ota_manager.direct_ota_download_via_4g, (ota_url,))
-                                        else:  # mode == "wifi"
-                                            if not ssid or not password:
-                                                self.logger.error("ota wifi mode requires ssid and password")
-                                                self.safe_enqueue({"result": "missing_ssid_or_password"}, 2)
-                                            else:
-                                                self.logger.info(f"ssid: {ssid}")
-                                                self.logger.info(f"password: {password}")
-                                                ota_manager._start_update_thread()
-                                                self._spawn_cmd_thread(ota_manager.handle_wifi_and_update,
-                                                                       (ssid, password, ota_url))
-                                    elif inner_cmd == 6:
-                                        try:
-                                            ip = os.popen(
-                                                "ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
-                                            ip = ip if ip else "no_ip"
-                                        except:
-                                            ip = "error_getting_ip"
-                                        self.safe_enqueue({"result": "current_ip", "ip": ip}, 2)
-                                    elif inner_cmd == 44:  # 读 4G 本机号码（AT+CNUM）
-                                        cnum = self.get_4g_phone_number()
-                                        self.logger.info(f"4G 本机号码: {cnum}")
-                                        self.safe_enqueue(
-                                            {"result": "cnum", "number": cnum if cnum is not None else ""}, 2)
-                                    elif inner_cmd == 45:  # 读 MCCID（AT+MCCID）
-                                        mccid = self.get_4g_mccid()
-                                        self.logger.info(f"4G MCCID: {mccid}")
-                                        self.safe_enqueue(
-                                            {"result": "mccid", "mccid": mccid if mccid is not None else ""}, 2)
-                                    elif inner_cmd == 41:
-                                        self.logger.info(f"[TEST] 收到TCP射箭触发命令, {time.time()}")
-                                        self._manual_trigger_flag = True
-                                        self.safe_enqueue({"result": "trigger_ack"}, 2)
-                                        hardware_manager.start_idle_timer()  # 重新计时
-                                    elif inner_cmd == 42:  # 关机命令
-                                        self.logger.info("[SHUTDOWN] 收到TCP关机命令，准备关机...")
-                                        self.safe_enqueue({"result": "shutdown_ack"}, 2)
-                                        time.sleep_ms(1000)
-                                        self.disconnect_server()
-                                        # 尝试关闭4G模块
-                                        try:
-                                            with self.get_uart_lock():
-                                                hardware_manager.at_client.send("AT+CFUN=0", "OK", 5000)
-                                        except:
-                                            pass
-                                        time.sleep_ms(2000)
-                                        os.system("sync")  # 刷新文件系统缓存到磁盘，防止数据丢失
-                                        time.sleep_ms(500)
-                                        # os.system("poweroff")
-                                        hardware_manager.power_off()
-                                        return
-                                    elif inner_cmd == 43:  # 上传日志命令
-                                        # 格式: {"cmd":43, "data":{"ssid":"xxx","password":"xxx","url":"xxx", ...}}
-                                        inner_data = data_obj.get("data", {})
-                                        upload_url = inner_data.get("url")
-                                        wifi_ssid = inner_data.get("ssid")
-                                        wifi_password = inner_data.get("password")
-                                        include_rotated = inner_data.get("include_rotated", True)
-                                        max_files = inner_data.get("max_files")
-                                        archive_format = inner_data.get("archive", "tgz")  # tgz 或 zip
-
-                                        hardware_manager.start_idle_timer()  # 重新计时
-
-                                        if not upload_url:
-                                            self.logger.error("[LOG_UPLOAD] 缺少 url 参数")
-                                            self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_url"},
-                                                              2)
                                         else:
-                                            self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，目标URL: {upload_url}")
+                                            self.safe_enqueue({"result": "laser pos set by hard code"}, 2)
-                                        # 在新线程中执行上传，避免阻塞主循环
+                                elif cmd == 3:  # CloseAimRequest 关闭激光
+                                    from laser_manager import laser_manager
+                                    laser_manager.turn_off_laser()
+                                    laser_manager.stop_calibration()
+                                    hardware_manager.start_idle_timer()
+                                    self.safe_enqueue({"result": "laser_off"}, 2)
+                                elif cmd == 4:  # GetBatteryRequest 上报电量
+                                    voltage = get_bus_voltage()
+                                    battery_percent = voltage_to_percent(voltage)
+                                    charging = is_charging()
+                                    self.safe_enqueue({
+                                        "cmd": 4,
+                                        "battery": battery_percent,
+                                        "voltage": round(float(voltage), 3),
+                                        "netType": self.network_type,
+                                        "charging": charging,
+                                    }, 2)
+                                    self.logger.info(f"电量上报: {battery_percent}% 充电: {charging}")
+                                elif cmd == 700:
+                                    self.logger.warning("服务器下发关机！！！")
+                                    exit(-1)
+                                elif cmd == 5:  # OtaRequest OTA 升级
+                                    ota_url = data_obj.get("url", "")
+
+                                    if not ota_url:
+                                        self.logger.error("ota missing_url")
+                                        self.safe_enqueue({"result": "missing_url"}, 2)
+                                        _rx_skip_tcp_iteration = True
+                                        break
+
+                                    from ota_manager import ota_manager
+                                    if ota_manager.update_thread_started:
+                                        self.safe_enqueue({"result": "update_already_started"}, 2)
+                                        _rx_skip_tcp_iteration = True
+                                        break
+
+                                    if not wifi_manager.is_wifi_connected():
+                                        self.logger.warning("[ota] cmd5 当前未连接WiFi，拒绝OTA")
+                                        self.safe_enqueue({"result": "ota fail", "reason": "wifi not connected"}, 2)
+                                        _rx_skip_tcp_iteration = True
+                                        break
+
+                                    hardware_manager.stop_idle_timer()
+                                    self._spawn_cmd_thread(self._cmd5_ota, (ota_url,))
+                                elif cmd == 41:  # Ota4gSubCodeRequest 射箭触发
+                                    self.logger.info(f"[TEST] 收到TCP射箭触发命令, {time.time()}")
+                                    self._manual_trigger_flag = True
+                                    self.safe_enqueue({"result": "trigger_ack"}, 2)
+                                    hardware_manager.start_idle_timer()
+                                elif cmd == 42:  # ShutdownCommand 关机命令
+                                    self.logger.info("[SHUTDOWN] 收到TCP关机命令，准备关机...")
+                                    self.safe_enqueue({"result": "shutdown_ack"}, 2)
+                                    time.sleep_ms(1000)
+                                    self.disconnect_server()
+                                    try:
+                                        with self.get_uart_lock():
+                                            hardware_manager.at_client.send("AT+CFUN=0", "OK", 5000)
+                                    except:
+                                        pass
+                                    time.sleep_ms(2000)
+                                    os.system("sync")
+                                    time.sleep_ms(500)
+                                    hardware_manager.power_off()
+                                    return
+                                elif cmd == 44:  # 读 4G 本机号码
+                                    cnum = self.get_4g_phone_number()
+                                    self.logger.info(f"4G 本机号码: {cnum}")
+                                    self.safe_enqueue(
+                                        {"result": "cnum", "number": cnum if cnum is not None else ""}, 2)
+                                elif cmd == 45:  # 读 MCCID
+                                    mccid = self.get_4g_mccid()
+                                    self.logger.info(f"4G MCCID: {mccid}")
+                                    self.safe_enqueue(
+                                        {"result": "mccid", "mccid": mccid if mccid is not None else ""}, 2)
+                                elif cmd == 43:  # 上传日志命令
+                                    upload_url = data_obj.get("url")
+                                    wifi_ssid = data_obj.get("ssid")
+                                    wifi_password = data_obj.get("password")
+                                    include_rotated = data_obj.get("include_rotated", True)
+                                    max_files = data_obj.get("max_files")
+                                    archive_format = data_obj.get("archive", "tgz")
+
+                                    hardware_manager.start_idle_timer()
+
+                                    if not upload_url:
+                                        self.logger.error("[LOG_UPLOAD] 缺少 url 参数")
+                                        self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_url"}, 2)
+                                    else:
+                                        self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，目标URL: {upload_url}")
                                         self._spawn_cmd_thread(
                                             self._upload_log_file,
                                             (upload_url, wifi_ssid, wifi_password, include_rotated, max_files,
                                              archive_format)
                                         )
-                                    elif inner_cmd == 200:
+                                elif cmd == 200:  # GenericResult "init_center_point" 触发激光检测
-                                        self.logger.info("[LASER] cmd200 在后台线程执行检测")
+                                    self.logger.info("[LASER] cmd200 在后台线程执行检测")
-                                        self._spawn_cmd_thread(self._cmd200_detect_laser, ())
+                                    self._spawn_cmd_thread(self._cmd200_detect_laser, ())
-                                    elif inner_cmd == 300:
+                                elif cmd == 201:  # SetCenterPointRequest 设置中心点
-                                        self.logger.info("[New Ota] cmd300 在后台线程执行OTA")
+                                    if self.logger:
-                                        self._spawn_cmd_thread(self._cmd300_ota, (data_obj,))
+                                        self.logger.info(f"[LASER] cmd201:{body}")
-                                    elif inner_cmd == 600:
+                                    raw_x = data_obj.get("x")
-                                        self.logger.info("[conn wifi] cmd600 在后台线程执行连接wifi: {data_obj}")
+                                    raw_y = data_obj.get("y")
-                                        self._spawn_cmd_thread(self._cmd600_conn_wifi, (data_obj,))
+                                    try:
-                                    elif inner_cmd == 601:
+                                        from laser_manager import laser_manager
-                                        pass
+                                        ix, iy = laser_manager.set_hardcoded_laser_point(raw_x, raw_y)
-                                else:  # data的结构不是 dict
+                                        self.safe_enqueue({
+                                            "cmd": 201,
+                                            "result": "laser_point_set",
+                                            "x": ix,
+                                            "y": iy,
+                                        }, 2)
+                                        self.logger.info(f"[LASER] cmd201 硬编码激光点=({ix}, {iy})")
+                                    except Exception as e:
+                                        self.logger.error(f"[LASER] cmd201 失败: {e}")
+                                        self.safe_enqueue({
+                                            "cmd": 201,
+                                            "result": "laser_point_set_failed",
+                                            "reason": str(e),
+                                        }, 2)
+                                    hardware_manager.start_idle_timer()
+                                elif cmd == 300:  # OtaRequest 新版OTA
+                                    self.logger.info("[New Ota] cmd300 在后台线程执行OTA")
+                                    self._spawn_cmd_thread(self._cmd300_ota, ({"data": data_obj},))
+                                elif cmd == 600:  # WifiConnectRequest 连接wifi
+                                    self.logger.info(f"[conn wifi] cmd600 在后台线程执行连接wifi: {data_obj}")
+                                    self._spawn_cmd_thread(self._cmd600_conn_wifi, ({"data": data_obj},))
+                                elif cmd == 601:
+                                    pass
+                                else:
                                     self.logger.info(f"[NET] body={body}, {time.time()}")
-                            else:
-                                self.logger.info(f"[NET] 未知数据 {body}, {time.time()}")
                         if _rx_login_fail:
                             break
                         if _rx_skip_tcp_iteration:
@@ -2310,7 +2458,7 @@ class NetworkManager:
 
                         if item:
                             msg_type, data_dict = item
-                            pkt = self._netcore.make_packet(msg_type, data_dict)
+                            pkt = self._make_send_packet(msg_type, data_dict)
                             if not self.tcp_send_raw(pkt):
                                 # 发送失败：将消息放回队首（队列满则丢弃）
                                 with self.get_queue_lock():
@@ -2339,8 +2487,8 @@ class NetworkManager:
                     current_time = time.ticks_ms()
                     if logged_in and current_time - last_heartbeat_send_time > config.HEARTBEAT_INTERVAL * 1000:
                         vol_val = get_bus_voltage()
-                        if not self.tcp_send_raw(
+                        heartbeat_pkt = self._make_send_packet(4, {"vol": vol_val, "vol_per": voltage_to_percent(vol_val)})
-                                self._netcore.make_packet(4, {"vol": vol_val, "vol_per": voltage_to_percent(vol_val)})):
+                        if not self.tcp_send_raw(heartbeat_pkt):
                             # if not self.tcp_send_raw(self.make_packet(4, {"vol": vol_val, "vol_per": voltage_to_percent(vol_val)})):
                             send_hartbeat_fail_count += 1
                             # 短暂波动可能导致一次发送失败：连续失败达到阈值才重连，避免重连风暴

shoot_manager.py

@@ -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
 
 # 缓存相机标定与三角形位置，避免每次射箭重复读磁盘
@@ -54,6 +59,7 @@ def analyze_shot(frame, laser_point=None):
     """
     logger = logger_manager.logger
     from datetime import datetime
+    yellow_algorithm_ms = 0.0
 
     # ── Step 1: 确定激光点 ────────────────────────────────────────────────────
     laser_point_method = None
@@ -69,7 +75,11 @@ def analyze_shot(frame, laser_point=None):
             logger.info(f"[算法] 使用校准值: {laser_manager.laser_point}")
     else:
         # 动态模式：先做一次无激光点检测以估算距离，再推算激光点
-        _, _, _, _, best_radius1_temp, _ = detect_circle_v3(frame, None)
+        _t_yellow = time_std.perf_counter()
+        try:
+            _, _, _, _, best_radius1_temp, _ = detect_circle_v3(frame, None)
+        finally:
+            yellow_algorithm_ms += (time_std.perf_counter() - _t_yellow) * 1000.0
         distance_m_first = estimate_distance(best_radius1_temp) if best_radius1_temp else None
         if distance_m_first and distance_m_first > 0:
             laser_point = laser_manager.calculate_laser_point_from_distance(distance_m_first)
@@ -114,6 +124,7 @@ def analyze_shot(frame, laser_point=None):
             "laser_point": laser_point, "laser_point_method": laser_point_method,
             "offset_method": "yellow_ellipse" if ellipse_params else "yellow_circle",
             "distance_method": "yellow_radius",
+            "yellow_algorithm_ms": float(yellow_algorithm_ms),
         }
         if yolo_roi_xyxy is not None:
             out["yolo_roi_xyxy"] = yolo_roi_xyxy
@@ -121,9 +132,12 @@ def analyze_shot(frame, laser_point=None):
 
     if not use_tri:
         # 三角形未配置，直接跑圆形检测
-        return _build_circle_result(
+        _t_yellow = time_std.perf_counter()
-            detect_circle_v3(frame, laser_point, img_cv=img_cv)
+        try:
-        )
+            cdata = detect_circle_v3(frame, laser_point, img_cv=img_cv)
+        finally:
+            yellow_algorithm_ms += (time_std.perf_counter() - _t_yellow) * 1000.0
+        return _build_circle_result(cdata)
 
     # ── Step 4: 先独占跑三角形，超时或失败后再跑圆形（不与圆心并行，避免抢 CPU）──
     roi_xyxy = None
@@ -276,6 +290,7 @@ def analyze_shot(frame, laser_point=None):
             "laser_point": laser_point, "laser_point_method": laser_point_method,
             "offset_method": tri.get("offset_method") or "triangle_homography",
             "distance_method": tri.get("distance_method") or "pnp_triangle",
+            "yellow_algorithm_ms": float(yellow_algorithm_ms),
             "tri_markers": tri.get("markers", []),
             "tri_markers_completed": tri.get("markers_completed", []),
             "tri_homography": tri.get("homography"),
@@ -296,7 +311,11 @@ def analyze_shot(frame, laser_point=None):
 
     # 三角形超时或失败 → 跑圆心；圆心跑完后再检查三角形是否已结束
     try:
-        cdata = detect_circle_v3(frame, laser_point, img_cv=img_cv)
+        _t_yellow = time_std.perf_counter()
+        try:
+            cdata = detect_circle_v3(frame, laser_point, img_cv=img_cv)
+        finally:
+            yellow_algorithm_ms += (time_std.perf_counter() - _t_yellow) * 1000.0
     except Exception as e:
         logger.error(f"[CIRCLE] 圆形检测异常: {e}")
         cdata = (frame, None, None, None, None, None)
@@ -322,15 +341,25 @@ 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)
 
         # Classify only the current shot frame; never reuse a previous result.
         target_class_result = None
+        yolo_target_ms = 0.0
         try:
             from target_roi_yolo import try_get_target_class_from_yolo
 
-            target_class_result = try_get_target_class_from_yolo(frame, logger=logger)
+            _t_yolo_target = time_std.perf_counter()
+            try:
+                target_class_result = try_get_target_class_from_yolo(frame, logger=logger)
+            finally:
+                yolo_target_ms = (time_std.perf_counter() - _t_yolo_target) * 1000.0
             if logger:
                 logger.info(f"[YOLO-TARGET] 当前箭业务结果: {target_class_result}")
         except Exception as exc:
@@ -360,6 +389,7 @@ def process_shot(adc_val):
         laser_point_method = analysis_result["laser_point_method"]
         offset_method = analysis_result.get("offset_method", "yellow_circle")
         distance_method = analysis_result.get("distance_method", "yellow_radius")
+        yellow_algorithm_ms = float(analysis_result.get("yellow_algorithm_ms", 0.0) or 0.0)
         tri_markers = analysis_result.get("tri_markers", [])
         tri_markers_completed = analysis_result.get("tri_markers_completed", [])
         tri_homography = analysis_result.get("tri_homography")
@@ -380,10 +410,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}")
 
@@ -413,7 +439,9 @@ def process_shot(adc_val):
             "r": 20.0,  # 保留字段（服务端当前忽略，物理外环半径 cm）
             "target_class": target_label,
             "target_class_confidence": (
-                float(target_confidence) if target_confidence is not None else None
+                round(float(target_confidence), 2)
+                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),
@@ -425,6 +453,8 @@ def process_shot(adc_val):
             "target_y": float(y),
             "offset_method": offset_method,
             "distance_method": distance_method,
+            "yellow_algorithm_ms": round(yellow_algorithm_ms, 2),
+            "yolo_target_ms": round(float(yolo_target_ms), 2),
         }
 
         if ellipse_params:
@@ -441,6 +471,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(

target_roi_yolo.py

@@ -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)
+    # YOLO is an optional capability. A broken/incompatible model must not
-    return _detector_by_path[model_path]
+    # 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):
@@ -301,7 +310,10 @@ def _det_obj_score(o):
 
 
 def try_get_target_class_from_yolo(maix_frame, logger=None):
-    """Classify the current target as 20cm or 40cm; return None if unknown."""
+    """调用 YOLO 检测：class_id∈IGNORE_CLASS_IDS(默认 1) 的框在解析时直接过滤；
+    detections 输出其余全部框（不按置信度过滤）；
+    上报以「检出框裁剪放大后的复检结果」为主，复检不可用时用全图结果；
+    有效靶规格(20/40)中取置信度最高的一条，且低于 MIN_REPORT_CONF 不上报类别。"""
     try:
         import config as cfg
     except Exception:
@@ -324,64 +336,239 @@ def try_get_target_class_from_yolo(maix_frame, logger=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))
+
+    # class_id → 靶规格标签，下标即 class_id：0→40, 1→0, 2→20
+    labels = getattr(cfg, "TARGET_CLASS_YOLO_LABELS", (40, 0, 20))
     if isinstance(labels, str):
         labels = tuple(x.strip() for x in labels.split(",") if x.strip())
     labels = tuple(labels)
 
-    def _detect(threshold):
+    # 只有映射到这些标签的检测才算有效靶规格（class 1→label 0 直接忽略）
+    valid_labels = getattr(cfg, "TARGET_CLASS_YOLO_VALID_LABELS", (20, 40))
+    if isinstance(valid_labels, str):
+        valid_labels = tuple(
+            int(float(x.strip())) for x in valid_labels.split(",") if x.strip()
+        )
+    valid_labels = frozenset(int(float(v)) for v in valid_labels)
+
+    # 解析阶段就丢弃这些 class_id（1=靶心），不进入 detections，也不参与上报选择
+    ignore_class_ids = getattr(cfg, "TARGET_CLASS_YOLO_IGNORE_CLASS_IDS", (1,))
+    if isinstance(ignore_class_ids, int):
+        ignore_class_ids = (ignore_class_ids,)
+    if isinstance(ignore_class_ids, str):
+        ignore_class_ids = tuple(
+            int(float(x.strip())) for x in ignore_class_ids.split(",") if x.strip()
+        )
+    ignore_class_ids = frozenset(int(float(v)) for v in ignore_class_ids)
+    n_ignored = [0]
+
+    def _run(threshold):
         try:
-            raw = detector.detect(maix_frame, conf_th=threshold, iou_th=iou_th)
+            return 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 None
-        return _normalize_objs(raw if raw is not None else [])
 
-    def _candidates(objs):
+    def _obj_box(o):
-        found = []
+        try:
-        for obj in objs:
+            return (
-            class_id = _det_obj_class_id(obj)
+                round(float(o.x), 1),
-            if class_id is None or class_id < 0 or class_id >= len(labels):
+                round(float(o.y), 1),
-                continue
+                round(float(o.w), 1),
-            try:
+                round(float(o.h), 1),
-                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]],
             )
-        )
+        except (TypeError, ValueError, AttributeError):
-    if not candidates and bool(
+            return None
-        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:
+    def _parse(raw_result):
+        out = []
+        n_ignored[0] = 0
+        for obj in _normalize_objs(raw_result if raw_result is not None else []):
+            class_id = _det_obj_class_id(obj)
+            if class_id is None or class_id in ignore_class_ids:
+                n_ignored[0] += 1
+                continue
+            label = None
+            if 0 <= class_id < len(labels):
+                try:
+                    label = int(float(labels[class_id]))
+                except (TypeError, ValueError):
+                    label = None
+            out.append(
+                {
+                    "class_id": class_id,
+                    "label": label,
+                    "confidence": round(_det_obj_score(obj), 4),
+                    "box_xywh": _obj_box(obj),
+                }
+            )
+        return out
+
+    # 第 1 次检测
+    raw = _run(conf_th)
+    detections = _parse(raw)
+
+    # 0 结果时重新检测一次（阈值取更小值，仍为 0 则同参数重跑）
+    if not detections:
+        retry_th = float(getattr(cfg, "TARGET_CLASS_YOLO_RETRY_CONF_TH", 0.25) or 0)
+        if not (0 < retry_th < conf_th):
+            retry_th = conf_th
         if logger:
-            logger.warning("[YOLO-TARGET] 当前帧未识别到 20/40，按未知处理")
+            logger.info(f"[YOLO-TARGET] 0 结果，重新检测 conf_th={retry_th}")
-        return None
+        raw_retry = _run(retry_th)
+        detections_retry = _parse(raw_retry)
+        if detections_retry:
+            detections = detections_retry
 
-    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"[YOLO-TARGET] 检测结果={detections}"
-            f"conf={confidence:.3f}"
+            f"（已忽略 class_id∈{tuple(sorted(ignore_class_ids))} 共{n_ignored[0]} 条）"
         )
-    return result
+
+    if not detections:
+        if logger:
+            logger.warning("[YOLO-TARGET] 未检测到目标（已重检 1 次）")
+        return None
+
+    def _best_valid(dets):
+        valid = [d for d in dets if d["label"] in valid_labels]
+        if not valid:
+            return None
+        return max(valid, key=lambda d: d["confidence"])
+
+    def _crop_target(box, margin_frac):
+        """按检出框(带 margin)裁出放大 ROI；返回 Maix image，失败返回 None。"""
+        if not box or len(box) != 4:
+            return None
+        x, y, w, h = [float(v) for v in box]
+        try:
+            fw, fh = int(maix_frame.width()), int(maix_frame.height())
+        except Exception:
+            fw = fh = 0
+        mx, my = w * margin_frac, h * margin_frac
+        x0, y0 = int(max(0, x - mx)), int(max(0, y - my))
+        x1 = int(min(fw, x + w + mx)) if fw else int(x + w + mx)
+        y1 = int(min(fh, y + h + my)) if fh else int(y + h + my)
+        if (x1 - x0) < 32 or (y1 - y0) < 32:
+            if logger:
+                logger.warning(
+                    f"[YOLO-TARGET] 复检裁剪过小 [{x0},{y0},{x1},{y1}]"
+                )
+            return None
+        try:
+            cropped = maix_frame.crop(x0, y0, x1 - x0, y1 - y0)
+            if cropped is not None:
+                return cropped
+        except Exception:
+            pass
+        try:
+            from maix import image as maix_image
+
+            arr = maix_frame.to_numpy()
+            slab = np.ascontiguousarray(arr[y0:y1, x0:x1], dtype=np.uint8)
+            if slab.size == 0:
+                return None
+            return maix_image.cv2image(slab, False, True)
+        except Exception as exc:
+            if logger:
+                logger.warning(f"[YOLO-TARGET] 复检裁剪失败: {exc}")
+            return None
+
+    # 上报口径：只在有效靶规格(20/40)里取置信度最高的一条，class_id=1 已在解析阶段过滤
+    full_best = _best_valid(detections)
+
+    # 二次裁剪复检：把全图检出的靶框裁剪放大后再 detect 一次，与全图结果互相印证
+    # 复检用更低阈值 RECHECK_CONF_TH，置信度大于该值即接受
+    recheck_th = float(getattr(cfg, "TARGET_CLASS_YOLO_RECHECK_CONF_TH", 0.25))
+    crop_best = None
+    recheck_status = "off"
+    if full_best is not None and bool(
+        getattr(cfg, "TARGET_CLASS_YOLO_RECHECK_CROP", True)
+    ):
+        recheck_status = "fail"
+        margin_frac = float(
+            getattr(cfg, "TARGET_CLASS_YOLO_RECHECK_MARGIN_FRAC", 0.1)
+        )
+        crop_img = _crop_target(full_best["box_xywh"], margin_frac)
+        if crop_img is not None:
+            try:
+                raw_crop = detector.detect(
+                    crop_img, conf_th=recheck_th, iou_th=iou_th
+                )
+            except Exception as exc:
+                raw_crop = None
+                if logger:
+                    logger.warning(f"[YOLO-TARGET] 复检 detect 异常: {exc}")
+            if raw_crop is not None:
+                crop_dets = _parse(raw_crop)
+                recheck_status = "ok"
+                crop_best = _best_valid(crop_dets)
+                if crop_best is not None and crop_best["confidence"] <= recheck_th:
+                    crop_best = None
+                if logger:
+                    logger.info(
+                        f"[YOLO-TARGET] 复检(conf_th={recheck_th}, "
+                        f"裁剪框={full_best['box_xywh']}) 结果={crop_dets}"
+                    )
+
+    # 合并：以复检（裁剪放大）结果为主；复检无结果才用全图
+    picked = full_best
+    picked_from_crop = False
+    note = ""
+    if recheck_status == "ok":
+        if crop_best is None:
+            note = f"复检无 conf>{recheck_th} 的有效框，沿用全图"
+        else:
+            picked = crop_best
+            picked_from_crop = True
+            if crop_best["label"] == full_best["label"]:
+                note = f"复检一致({crop_best['label']})，以复检为准"
+            else:
+                note = (
+                    f"复检与全图不一致(全图={full_best['label']}@{full_best['confidence']}, "
+                    f"裁剪={crop_best['label']}@{crop_best['confidence']})，以复检为准"
+                )
+
+    # 置信度门槛：低于阈值不上报类别，避免硬猜（复检结果只需 > RECHECK_CONF_TH）
+    if picked is not None:
+        if picked_from_crop:
+            min_report_conf = recheck_th
+            below = picked["confidence"] <= min_report_conf
+        else:
+            min_report_conf = float(
+                getattr(cfg, "TARGET_CLASS_YOLO_MIN_REPORT_CONF", 0.5)
+            )
+            below = picked["confidence"] < min_report_conf
+        if below:
+            note = (note + "; " if note else "") + (
+                f"conf={picked['confidence']} 未达门槛{min_report_conf}，置信度不足"
+            )
+            picked = None
+
+    if picked is None:
+        label, class_id, confidence = None, None, None
+    else:
+        label = picked["label"]
+        class_id = picked["class_id"]
+        confidence = picked["confidence"]
+
+    if logger:
+        logger.info(
+            f"[YOLO-TARGET] 上报={label}, class_id={class_id}, conf={confidence}, "
+            f"全图候选={[(d['label'], d['confidence']) for d in detections if d['label'] in valid_labels]}, "
+            f"复检={recheck_status} {note}"
+        )
+        if note:
+            logger.warning(f"[YOLO-TARGET] {note}")
+
+    return {
+        "label": label,
+        "class_id": class_id,
+        "confidence": confidence,
+        "detections": detections,
+    }
 
 
 def _det_to_src_xyxy(o, coord_mode: str, src_w: int, src_h: int, net_w: int, net_h: int):

test/test_bluetooth.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+# test_bluetooth.py
+# 参考: https://wiki.sipeed.com/maixpy/doc/zh/modules/bluetooth.html
+#
+# 依赖:
+#   1. 使能蓝牙:  bluetoothctl power on
+#      (可选开机自启: echo "bluetoothctl power on" >> /etc/rc.local)
+#   2. 安装依赖:  pip install bleak
+
+import asyncio
+import sys
+from collections import namedtuple
+
+try:
+    from bleak import BleakScanner, BleakClient
+except ImportError:
+    print("ModuleNotFoundError: No module named 'bleak'")
+    print("请先在设备上安装依赖:")
+    print("  pip install bleak")
+    print("并使能蓝牙:")
+    print("  bluetoothctl power on")
+    sys.exit(1)
+
+MODEL_NBR_UUID = "1A2A"
+
+BLEDev = namedtuple("BLEDev", ["address", "name"])
+
+
+async def scan_devices(timeout=5.0):
+    """扫描周围的蓝牙设备"""
+    print("=" * 60)
+    print("Scanning BLE devices...")
+    print("=" * 60)
+
+    found = {}
+
+    def _cb(device, adv):
+        found[device.address] = adv.local_name or device.name
+
+    try:
+        await BleakScanner.discover(timeout=timeout, detection_callback=_cb)
+    except Exception as e:
+        print(f"Scan failed: {e}")
+        print("Possible causes:")
+        print("  1. Bluetooth not powered on: bluetoothctl power on")
+        print("  2. bleak not installed: pip install bleak")
+        print("  3. BLE hardware/driver not available")
+        return []
+
+    if not found:
+        print("No devices found.")
+        return []
+
+    devices = []
+    for address, name in found.items():
+        devices.append(BLEDev(address, name))
+        print(f"  {address}  {name}")
+    return devices
+
+
+async def connect_device(address, timeout=20.0, retries=3):
+    """连接蓝牙设备并尝试读取特征值"""
+    print("=" * 60)
+    print(f"Connecting to {address} ...")
+    print("=" * 60)
+
+    for attempt in range(1, retries + 1):
+        try:
+            async with BleakClient(address, timeout=timeout) as client:
+                print(f"Connected: {client.is_connected}")
+
+                # 遍历服务和特征值
+                for service in client.services:
+                    print(f"  Service: {service.uuid}")
+                    for char in service.characteristics:
+                        print(f"    Char: {char.uuid}  props={list(char.properties)}")
+
+                # 尝试读取 Model Number 特征
+                try:
+                    value = await client.read_gatt_char(MODEL_NBR_UUID)
+                    print(f"Model Number: {value.decode(errors='replace')}")
+                except Exception as e:
+                    print(f"Read {MODEL_NBR_UUID} failed (device may not have it): {e}")
+
+                return True
+        except Exception as e:
+            print(f"[attempt {attempt}/{retries}] Connect failed: {e}")
+            if attempt < retries:
+                await asyncio.sleep(2)
+
+    print("\nAll attempts failed. 请检查:")
+    print("  1. forrily 是否已连到手机/其它主机（BLE 外设通常只允许一连接），")
+    print("     先在手机上断开它")
+    print("  2. 清除 BlueZ 缓存后重试:")
+    print(f"       bluetoothctl remove {address}")
+    print("       systemctl restart bluetooth")
+    print("       bluetoothctl power on")
+    print("  3. 升级依赖:")
+    print("       pip install -U bleak dbus-fast")
+    return False
+
+
+async def main():
+    target_name = "forrily"
+
+    devices = await scan_devices(timeout=5.0)
+    if not devices:
+        return
+
+    if len(sys.argv) > 1:
+        address = sys.argv[1]
+    else:
+        # 按名字查找目标设备，不连第一个
+        target = None
+        for d in devices:
+            if d.name and d.name.strip().lower() == target_name.lower():
+                target = d
+                break
+        if target is None:
+            for d in devices:
+                if d.name and target_name.lower() in d.name.strip().lower():
+                    target = d
+                    break
+        if target is None:
+            print(f"\nDevice named '{target_name}' not found!")
+            print("Scanned devices are listed above.")
+            return
+        address = target.address
+        print(f"\nFound '{target_name}': {address}")
+
+    await connect_device(address)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test/test_bluetooth_peripheral.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+# test_bluetooth_peripheral.py
+# 设备(MaixCAM)作为 BLE 从机(Peripheral/GATT Server)广播给手机连接。
+# 参考: https://wiki.sipeed.com/maixpy/doc/zh/modules/bluetooth.html
+#
+# 用法:
+#   python test_bluetooth_peripheral.py [广播名字]     # 默认 shoot
+#
+# 依赖:
+#   1. 使能蓝牙:  bluetoothctl power on
+#   2. 安装依赖:  pip install bless
+#   3. 部分 BlueZ 需要 experimental 模式才支持 GATT Server:
+#      编辑 /etc/bluetooth/main.conf 保证有 Experimental = true
+#      然后 systemctl restart bluetooth && bluetoothctl power on
+#
+# 手机端: 用 nRF Connect / 任意 BLE 调试 App 搜索广播名连接，
+#         读写 characteristic 并订阅 notify，每 2 秒会收到计数通知。
+#
+# 实现:
+#   - GATT 服务由 bless 注册到 BlueZ
+#   - 广播由常驻 bluetoothctl 子进程控制 (menu advertise / name / advertise on)，
+#     这是已验证手机能搜到的路径; 若 bluetoothctl 不可用则回退到 btmgmt adv on
+
+import asyncio
+import shutil
+import subprocess
+import sys
+import time
+
+try:
+    from bless import (
+        BlessServer,
+        GATTCharacteristicProperties,
+        GATTAttributePermissions,
+    )
+except ImportError:
+    print("ModuleNotFoundError: No module named 'bless'")
+    print("请先在设备上安装依赖:")
+    print("  pip install bless")
+    print("并使能蓝牙:")
+    print("  bluetoothctl power on")
+    raise SystemExit(1)
+
+DEVICE_NAME = sys.argv[1] if len(sys.argv) > 1 else "shoot"
+SERVICE_UUID = "A07498CA-AD5B-474E-940D-16F1FBE7E8CD"
+CHAR_UUID = "51FF12BB-3ED8-46E5-B4F9-D64E2FEC021B"
+BTCTL_LOG = "/tmp/btctl_adv.log"
+
+# 不同 bless 版本枚举名不同: writeable / writable
+try:
+    WRITE_PERMISSION = GATTAttributePermissions.writeable
+except AttributeError:
+    WRITE_PERMISSION = GATTAttributePermissions.writable
+
+_btctl = None
+
+
+def on_read(characteristic, request):
+    print(f"[GATT] read {characteristic.uuid}")
+    return characteristic.value
+
+
+def on_write(characteristic, value, request):
+    characteristic.value = value
+    print(f"[GATT] write {characteristic.uuid} = {bytes(value)}")
+
+
+def on_subscribe(characteristic, session):
+    print(f"[GATT] client subscribed: {characteristic.uuid}")
+
+
+def on_unsubscribe(characteristic, session):
+    print(f"[GATT] client unsubscribed: {characteristic.uuid}")
+
+
+def _run(cmd):
+    try:
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        out = ((r.stdout or "") + (r.stderr or "")).strip()
+        if out:
+            print(f"[adv] {' '.join(cmd)} -> {out}")
+        return r.returncode == 0
+    except FileNotFoundError:
+        print(f"[adv] {cmd[0]} not found")
+        return False
+    except Exception as e:
+        print(f"[adv] {' '.join(cmd)} failed: {e}")
+        return False
+
+
+def adv_start(name):
+    """启动广播: 优先 bluetoothctl (常驻子进程), 失败则回退 btmgmt"""
+    global _btctl
+
+    if shutil.which("bluetoothctl") is not None:
+        try:
+            log = open(BTCTL_LOG, "w")
+            _btctl = subprocess.Popen(
+                ["bluetoothctl"],
+                stdin=subprocess.PIPE,
+                stdout=log,
+                stderr=subprocess.STDOUT,
+                text=True,
+                bufsize=1,
+            )
+        except Exception as e:
+            print(f"[adv] start bluetoothctl failed: {e}")
+            _btctl = None
+
+        if _btctl is not None:
+            for cmd in (
+                "menu advertise",
+                f"name {name}",
+                "discoverable on",
+                "menu main",
+                "advertise on",
+            ):
+                try:
+                    _btctl.stdin.write(cmd + "\n")
+                    _btctl.stdin.flush()
+                except Exception as e:
+                    print(f"[adv] write '{cmd}' failed: {e}")
+                    break
+                time.sleep(0.3)
+
+            time.sleep(1.0)
+            if _btctl.poll() is None:
+                print(f"[adv] bluetoothctl advertising on (log: {BTCTL_LOG})")
+                return True
+            print("[adv] bluetoothctl exited early, fallback to btmgmt")
+
+    # 回退: mgmt 传统广播
+    if shutil.which("btmgmt") is None:
+        print("[adv] bluetoothctl/btmgmt 均不可用")
+        return False
+    _run(["btmgmt", "le", "on"])
+    _run(["btmgmt", "connectable", "on"])
+    return _run(["btmgmt", "adv", "on"])
+
+
+def adv_stop():
+    global _btctl
+    if _btctl is not None:
+        try:
+            _btctl.stdin.write("advertise off\n")
+            _btctl.stdin.flush()
+            time.sleep(0.3)
+            _btctl.terminate()
+        except Exception:
+            pass
+        _btctl = None
+    elif shutil.which("btmgmt") is not None:
+        _run(["btmgmt", "adv", "off"])
+
+
+async def run():
+    gatt = {
+        SERVICE_UUID: {
+            CHAR_UUID: {
+                "Properties": (
+                    GATTCharacteristicProperties.read
+                    | GATTCharacteristicProperties.write
+                    | GATTCharacteristicProperties.notify
+                ),
+                "Permissions": (
+                    GATTAttributePermissions.readable | WRITE_PERMISSION
+                ),
+                "Value": bytearray(b"hello from maixcam"),
+            }
+        }
+    }
+
+    server = BlessServer(
+        name=DEVICE_NAME,
+        on_read=on_read,
+        on_write=on_write,
+        on_subscribe=on_subscribe,
+        on_unsubscribe=on_unsubscribe,
+    )
+
+    print("Adding GATT ...")
+    await server.add_gatt(gatt)
+
+    print(f"Starting BLE peripheral, name = '{DEVICE_NAME}' ...")
+    try:
+        await server.start()
+    except Exception as e:
+        print(f"start() failed: {e}")
+        print("排查:")
+        print("  1. 蓝牙是否上电:  bluetoothctl power on  (bluetoothctl show 看 Powered)")
+        print("  2. /etc/bluetooth/main.conf 加 Experimental = true 后重启:")
+        print("       systemctl restart bluetooth && bluetoothctl power on")
+        print("  3. 依赖: pip install -U bless dbus-next bleak")
+        return
+
+    adv = await server.is_advertising()
+    print("=" * 60)
+    print(f"'{DEVICE_NAME}' advertising status: {adv}")
+
+    print("Starting advertising ...")
+    if not adv_start(DEVICE_NAME):
+        print("广播启动失败，请手动执行: bluetoothctl -> advertise on")
+
+    print("手机用 nRF Connect 等 App 搜索并连接，")
+    print(f"服务 {SERVICE_UUID}")
+    print(f"特征 {CHAR_UUID}  (read/write/notify)")
+    print("Ctrl+C 退出")
+    print("=" * 60)
+
+    try:
+        # 每 2 秒向订阅者推送一次计数
+        counter = 0
+        while True:
+            counter += 1
+            try:
+                char = server.get_characteristic(CHAR_UUID)
+                if char is not None:
+                    char.value = bytearray(str(counter).encode())
+                    server.update_value(SERVICE_UUID, CHAR_UUID)
+            except Exception as e:
+                # 没有客户端订阅时可能失败，忽略
+                print(f"notify failed: {e}")
+
+            if counter % 10 == 0:
+                try:
+                    print(f"[status] advertising={await server.is_advertising()} "
+                          f"connected={await server.is_connected()}")
+                except Exception:
+                    pass
+            await asyncio.sleep(2)
+    finally:
+        adv_stop()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(run())
+    except KeyboardInterrupt:
+        print("\nStopped.")

test/test_target_yolo_maixvision.py

@@ -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()

test_traditional_target.py

@@ -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()

version.md

@@ -31,3 +31,10 @@
 # 2.15.18 wifi连接成功重新登录
 # 2.16.4  优化射箭延迟
 # 2.17.0  yolo标靶类别识别
+# 2.17.1  26-08-19 17：39 压力传感修改 增量方式 
+
+# 2.17.2 26-08-24 17：56 靶纸识别模型更替
+
+# 2.17.3 26-08-25 9：57 原图拍摄开关
+
+# 2.17.4 26-08-25 14：57 模型修改

version.py

@@ -4,6 +4,6 @@
 应用版本号
 每次 OTA 更新时，只需要更新这个文件中的版本号
 """
-VERSION = '2.17.0'
+VERSION = '2.18.10'
 
 

vision.py

@@ -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):