fix: 拍照优先

.gitattributes

@@ -0,0 +1 @@
+*.sh text eol=lf

.gitignore

@@ -1,3 +1,4 @@
 /cpp_ext/build/
 /.cursor/
 /dist/
+.idea

.idea/.gitignore

@@ -0,0 +1,8 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/archery.iml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="jdk" jdkName="yolov8" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PyDocumentationSettings">
+    <option name="format" value="PLAIN" />
+    <option name="myDocStringFormat" value="Plain" />
+  </component>
+</module>

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/misc.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="yolov8" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="yolov8" project-jdk-type="Python SDK" />
+</project>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/archery.iml" filepath="$PROJECT_DIR$/.idea/archery.iml" />
+    </modules>
+  </component>
+</project>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="" vcs="Git" />
+  </component>
+</project>

.vscode/settings.json

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

app.yaml

@@ -1,6 +1,6 @@
 id: t11
 name: t11
-version: 2.18.0
+version: 2.15.31
 author: t11
 icon: ''
 desc: t11
@@ -12,14 +12,13 @@ files:
   - at_client.py
   - camera_manager.py
   - cameraParameters.xml
+  - charging_exit.sh
   - config.py
   - hardware.py
   - laser_detector.py
   - laser_manager.py
   - logger_manager.py
   - main.py
-  - model_317828.cvimodel
-  - model_317828.mud
   - network.py
   - ota_curl.sh
   - ota_manager.py
@@ -28,7 +27,6 @@ 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

charging_exit.sh

@@ -0,0 +1,47 @@
+#!/bin/sh
+
+# The application supplies its own PID. Refuse broad or malformed targets.
+TARGET_PID="$1"
+LASER_DEVICE="${2:-/dev/ttyS1}"
+LASER_BAUD="${3:-9600}"
+
+turn_off_laser() {
+    if [ ! -c "$LASER_DEVICE" ]; then
+        echo "[CHARGE] laser serial device not found: $LASER_DEVICE" >&2
+        return 1
+    fi
+
+    stty -F "$LASER_DEVICE" "$LASER_BAUD" raw -echo 2>/dev/null || return 1
+    printf '\252\000\001\276\000\001\000\000\300' > "$LASER_DEVICE"
+}
+
+case "$TARGET_PID" in
+    ''|*[!0-9]*)
+        echo "[CHARGE] invalid application pid: $TARGET_PID" >&2
+        exit 2
+        ;;
+esac
+
+if [ "$TARGET_PID" -le 1 ]; then
+    echo "[CHARGE] refusing to terminate pid: $TARGET_PID" >&2
+    exit 2
+fi
+
+# First request laser-off while the application still owns the initialized UART.
+turn_off_laser || true
+
+kill -TERM "$TARGET_PID" 2>/dev/null || true
+
+# Wait up to two seconds for a graceful exit, then force termination.
+WAIT_COUNT=0
+while kill -0 "$TARGET_PID" 2>/dev/null && [ "$WAIT_COUNT" -lt 20 ]; do
+    sleep 0.1
+    WAIT_COUNT=$((WAIT_COUNT + 1))
+done
+if kill -0 "$TARGET_PID" 2>/dev/null; then
+    kill -KILL "$TARGET_PID" 2>/dev/null || true
+    sleep 0.1
+fi
+
+# Send laser-off again after the application releases the UART.
+turn_off_laser || true

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 = 50007          # TLS 端口（不一定必须 443，以服务器为准）
+TCP_SSL_PORT = 50006          # 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_317211.mud"
+TRIANGLE_YOLO_MODEL_PATH = APP_DIR + "/model_270139.mud"
 # 参与 ROI 的类别：多类时只填「整靶/靶环」的 id；不要填角标类，否则 union 仍可对，但 largest 会偏小。
 TRIANGLE_YOLO_RING_CLASS_IDS = (0,)
-TRIANGLE_YOLO_CONF_TH = 0.9
+TRIANGLE_YOLO_CONF_TH = 0.7
 TRIANGLE_YOLO_IOU_TH = 0.45
 # YOLO 首次/临界帧可能在高阈值下 0 框；启用后仅在 0 候选时用较低阈值重试一次。
 # 后续仍会经过 min_box_side、ROI aspect、三角形几何校验，避免直接放大假阳性。
@@ -262,16 +262,6 @@ TRIANGLE_SAMPLE_PATCH_HALF_PX = 2
 # 开机阶段预加载 YOLO detector；detect 使用 dual_buff=False，避免返回上一帧结果。
 TRIANGLE_YOLO_PRELOAD_ON_BOOT = False
 
-# YOLO target size classification: class 0=20cm, class 1=40cm.
-TARGET_CLASS_YOLO_ENABLE = True
-TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_317828.mud"
-TARGET_CLASS_YOLO_LABELS = (20, 40)
-TARGET_CLASS_YOLO_CONF_TH = 0.66
-TARGET_CLASS_YOLO_IOU_TH = 0.45
-TARGET_CLASS_YOLO_RETRY_ON_EMPTY = False
-TARGET_CLASS_YOLO_RETRY_CONF_TH = 0.25
-TARGET_CLASS_YOLO_PRELOAD_ON_BOOT = True
-
 # ── 第二段 YOLO：仅在 Stage1 裁切出的靶环图上推理（与合成 stage2 训练数据一致）→ 子框内传统算法取直角点 ──
 # Stage1 靶环裁切内如何找黑三角标记（对比耗时时可切换）：
 # "yolo" — 调 Stage2 黑三角模型得子框，再子框内传统提取（需 TRIANGLE_BLACK_YOLO_ENABLE=True）。
@@ -326,13 +316,9 @@ LOG_QUEUE_MAXSIZE = 10000       # 日志队列上限
 MAX_CMD_THREADS = 10            # 并发命令线程上限（防止服务器下发命令时无限创建线程）
 
 # ==================== 图像保存配置 ====================
-SAVE_IMAGE_ENABLED = True  # 是否保存图像（True=保存，False=不保存）
-SAVE_IMAGE_ON_FAILURE = True  # 检测失败时是否强制保存图像（供调试测试用）
+SAVE_IMAGE_ENABLED = False  # 是否保存图像（True=保存，False=不保存）
 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
 
@@ -355,22 +341,16 @@ PIN_MAPPINGS = {
 }
 
 # ==================== 电源配置 ====================
-AUTO_POWER_OFF_IN_SECONDS = 100 * 60  # 自动关机时间（秒），0表示不自动关机
-# 充电时自动关机暂时禁用；需要恢复时改为 True。
-CHARGING_AUTO_POWER_OFF_ENABLED = False
+AUTO_POWER_OFF_IN_SECONDS = 10 * 60  # 自动关机时间（秒），0表示不自动关机
 
-# 一代电源控制：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
+# 实机数据：正常放电约为正电流，插入充电线后约为负电流。
+CHARGING_SHUTDOWN_ENABLED = True  # True=充电时退出应用，False=关闭充电关机功能
+CHARGING_DIAGNOSTIC_LOG_ENABLED = False
+CHARGING_CHECK_INTERVAL_MS = 5000
+CHARGING_CURRENT_THRESHOLD_MA = 100.0
+CHARGING_CONFIRM_COUNT = 2
+CHARGING_NOTIFY_TIMEOUT_MS = 30000
+CHARGING_EXIT_SCRIPT = APP_DIR + "/charging_exit.sh"
 
 BATTERY_SOC_LPF_ALPHA = 0.5
 BATTERY_SOC_AVG_WINDOW = 5

main.py

@@ -13,9 +13,7 @@ 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
 
 # 导入新模块
@@ -24,7 +22,7 @@ from version import VERSION
 # from logger import init_logging, get_logger, stop_logging
 from logger_manager import logger_manager
 from time_sync import sync_system_time_from_4g
-from power import init_ina226
+from power import charging_shutdown_monitor, init_ina226
 from laser_manager import laser_manager
 from vision import start_save_shot_worker
 from network import network_manager
@@ -122,11 +120,18 @@ def cmd_str():
 
     # ==================== 第二阶段：软件初始化 ====================
 
-    # 1. 初始化日志系统（WARNING级别，不打印/写入INFO和DEBUG日志，提高执行流畅度）
+    # 1. 初始化日志系统
     import logging
-    logger_manager.init_logging(log_level=logging.DEBUG)
+    logger_manager.init_logging(log_level=logging.WARNING)
     logger = logger_manager.logger
 
+    # 充电关机独立读取 INA226，不依赖 TCP 连接或心跳流程。
+    try:
+        _thread.start_new_thread(charging_shutdown_monitor, ())
+    except Exception as e:
+        if logger:
+            logger.error(f"[CHARGE] 启动独立监测线程失败: {e}")
+
     # 补充：因为初始化的时候，激光会亮，先关了它
     # laser_manager.turn_off_laser()
 
@@ -134,7 +139,6 @@ 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
 
@@ -165,21 +169,15 @@ def cmd_str():
             and _loc_black == "yolo"
             and bool(getattr(config, "TRIANGLE_BLACK_YOLO_PRELOAD_ON_BOOT", True))
         )
-        _need_target_preload = (
-            bool(getattr(config, "TARGET_CLASS_YOLO_ENABLE", False))
-            and bool(getattr(config, "TARGET_CLASS_YOLO_PRELOAD_ON_BOOT", True))
-        )
-        _preload_yolo = _preload_yolo or _need_black_preload or _need_target_preload
-        if _preload_yolo and not os.path.exists(_ota_pending_path):
+        _preload_yolo = _preload_yolo or _need_black_preload
+        if _preload_yolo:
             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 = _ota_pending_path
+    pending_path = f"{config.APP_DIR}/ota_pending.json"
     if os.path.exists(pending_path):
         try:
             with open(pending_path, 'r', encoding='utf-8') as f:
@@ -254,61 +252,8 @@ 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)
-        or getattr(config, "SAVE_RAW_IMAGE_ENABLED", False)
-    ):
+    # 5. 创建照片存储目录（如果启用图像保存）
+    if config.SAVE_IMAGE_ENABLED:
         photo_dir = config.PHOTO_DIR
         if photo_dir not in os.listdir("/root"):
             try:
@@ -340,14 +285,6 @@ 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
     # 气压采样：减少日志频率（每 N 个点输出一条），避免 logger.debug 拖慢采样
     PRESSURE_BATCH_SIZE = 100
 
@@ -399,14 +336,7 @@ 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
-
+            # todo 去除或者不在这里检测
             # 不在 OTA 状态下，检测是否空闲足够长，自动关机
             # print(f"[MAIN] 空闲时间: {hardware_manager.get_idle_time_in_sec() }秒")
             # print(f"配置关机时间：{config.AUTO_POWER_OFF_IN_SECONDS} 秒")
@@ -445,16 +375,16 @@ def cmd_str():
                 pressure_max = adc_val
             if len(pressure_buf) >= PRESSURE_BATCH_SIZE:
                 _flush_pressure_buf("batch")
-            # 突变增量检测：压力增量大于400时触发
-            # 触发后需等气压降到触发值以下才重新检测增量
-            if adc_val < trigger_adc_val :
-                enable_check = True
-            if (adc_val - last_adc_val) > 200 and enable_check:
+            # if adc_val >= 2000:
+            # print(f"adc :{adc_val}")
+            if adc_val >= config.ADC_TRIGGER_THRESHOLD:
                 hardware_manager.start_idle_timer()  # 重新计时
+                diff_ms = current_time - last_adc_trigger
+                if diff_ms < 3000:
+                    logger.info(f"[MAIN] 扳机触发过于频繁, {diff_ms}ms")
+                    continue
                 last_adc_trigger = current_time
-                trigger_adc_val = adc_val  # 记录触发时的气压值
-                last_adc_val = adc_val  # 更新基准值，防止连续增量误触发
-                enable_check = False
+                # 触发前先把缓存刷出来，避免波形被长耗时处理截断
                 _flush_pressure_buf("before_trigger")
 
                 try:
@@ -472,9 +402,10 @@ def cmd_str():
                     try:
                         camera_manager.show(camera_manager.read_frame())
                     except Exception as e:
-                        pass
-                time.sleep_ms(1)
-            last_adc_val = adc_val
+                        logger = logger_manager.logger
+                        if logger:
+                            logger.error(f"[MAIN] 显示异常: {e}")
+                time.sleep_ms(5)
 
         except Exception as e:
             # 主循环的顶层异常捕获，防止程序静默退出
@@ -492,62 +423,10 @@ def cmd_str():
                 _flush_pressure_buf("exception")
             except:
                 pass
-            time.sleep_ms(1000)  # 等待1秒后 continue
+            time.sleep_ms(1000)  # 等待1秒后继续
+
+
 
-    # 主循环退出后，如果是由 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

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

network.py

@@ -13,22 +13,15 @@ import hmac
 import hashlib
 import ujson
 import os
-import sys
 import threading
 import socket
 import config
 
 from hardware import hardware_manager
-from power import get_bus_voltage, voltage_to_percent, is_charging
+from power import get_bus_voltage, voltage_to_percent
 from logger_manager import logger_manager
 from wifi import wifi_manager
-
-# protobuf 支持（纯 proto 协议，必须可用）
-try:
-    import tcp_messages_pb2 as pb
-except ImportError:
-    print("[NET] tcp_messages_pb2 not found, protobuf disabled")
-    raise
+import subprocess
 
 
 def _wifi_tls_would_block(exc):
@@ -76,18 +69,9 @@ 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
@@ -206,7 +190,13 @@ 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语句）"""
@@ -625,123 +615,14 @@ 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，避免阻塞主循环
-        流程：检查WiFi → 下载ZIP → 解压覆盖项目 → 重启程序
-        """
+        """后台线程执行 cmd300 OTA，避免阻塞主循环"""
         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}")
-        ota_res_url = inner_data.get("url")
-
-        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:
-                    pkt = self._make_send_packet(2, {"cmd": 300, "result": "ota skip", "reason": f"cooldown {remaining}s"})
-                    self.tcp_send_raw(pkt)
-                except Exception:
-                    pass
-                return
-
-        if not wifi_manager.is_wifi_connected():
-            self.logger.warning("[ota] cmd300 当前未连接WiFi，拒绝OTA")
-            self.safe_enqueue({"cmd": 300, "result": "ota fail", "reason": "wifi not connected"}, 2)
-            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": f"ota {phase}",
-                    "progress": progress,
-                    "phase": phase,
-                })
-                ok = self.tcp_send_raw(pkt)
-                if not ok:
-                    self.logger.warning(f"[ota] 进度发送失败: phase={phase} progress={progress}")
-            except Exception as e:
-                self.logger.error(f"[ota] 发送进度异常: {e}")
-
-        try:
-            from ota_manager import ota_manager
-            ok, msg = ota_manager.perform_ota(ota_res_url, progress_callback=_ota_progress)
-            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._last_ota_time = time.time()
-            try:
-                pkt = self._make_send_packet(2, {"cmd": 300, "result": "ota fail", "reason": str(e)})
-                self.tcp_send_raw(pkt)
-            except Exception as ex:
-                self.logger.error(f"[ota] 发送失败结果异常: {ex}")
-
-    def _cmd600_conn_wifi(self, data_obj):
-        hardware_manager.start_idle_timer()
-        inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
-        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()
+        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"):
                 try:
@@ -749,10 +630,64 @@ class NetworkManager:
                 except OSError:
                     pass
             w = network.wifi.Wifi()
-            e = w.connect(ssid, password, wait=True, timeout=5)
+            e = w.connect(ssid, password, wait=True, timeout=15)
             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": 300,
+                    "result": "ota start...",
+                    "wifi": w.get_ip(),
+                },
+                2,
+            )
+            subprocess.run(
+                ["sh", "/maixapp/apps/t11/ota_curl.sh", ota_res_url])
+            self.safe_enqueue(
+                {
+                    "cmd": 300,
+                    "result": "success",
+                    "wifi": w.get_ip(),
+                },
+                2,
+            )
+        except Exception as e:
+            self.logger.error(f"[ota] cmd300 失败: {e}")
+            self.safe_enqueue(
+                {
+                    "cmd": 300,
+                    "result": "ota fail",
+                    "reason": str(e),
+                },
+                2,
+            )
+
+    def _cmd600_conn_wifi(self, data_obj):
+        hardware_manager.start_idle_timer()
+        inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
+        self.logger.info(f"[conn wifi] cmd600 , data: {inner_data}")
+        ssid = inner_data.get("ssid")
+        password = inner_data.get("password")
+        try:
+            for _f in ("/etc/wpa_supplicant.conf", "/boot/wpa_supplicant.conf", "/boot/wifi.ssid", "/boot/wifi.pass"):
+                try:
+                    os.remove(_f)
+                except OSError:
+                    pass
+            w = network.wifi.Wifi()
+            e = w.connect(ssid, password, wait=True, timeout=15)
+            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
@@ -760,11 +695,6 @@ class NetworkManager:
             self.logger.info("[conn wifi] WiFi已连接，等待主循环重新登录")
         except Exception as e:
             self.logger.error(f"cmd600 失败: {e}")
-            # 同步发送失败结果（旧连接仍存活时直接发送）
-            if prev_network_type == "4g":
-                pkt = self._make_send_packet(2, {"cmd": 600, "result": "conn fail", "reason": str(e)})
-                self.tcp_send_raw(pkt)
-            else:
             self.safe_enqueue(
                 {
                     "cmd": 600,
@@ -773,172 +703,17 @@ class NetworkManager:
                 },
                 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 safe_enqueue_and_wait(self, data_dict, msg_type=2, high=False, timeout_ms=30000):
+        """将消息加入队列，并等待网络线程确认已写入 TCP 连接。"""
+        sent_event = threading.Event()
+        self._enqueue((msg_type, data_dict, sent_event), high)
+        return bool(sent_event.wait(max(0, int(timeout_ms)) / 1000.0))
 
     def connect_server(self):
         """
@@ -1126,6 +901,12 @@ class NetworkManager:
         """检查WiFi TCP连接是否仍然有效"""
         if not wifi_manager.wifi_socket:
             return False
+        # TLS socket 无法可靠使用 MSG_PEEK，但物理 WiFi 链路仍可通过 STA 关联状态判断。
+        if not wifi_manager.is_sta_associated():
+            self.logger.warning("[WIFI-TCP] STA 已断开，关闭 WiFi TCP 并重新选网")
+            wifi_manager.disconnect_wifi()
+            self._tcp_connected = False
+            return False
         # TLS(ssl.wrap_socket/SSLContext.wrap_socket) 后的 socket 往往不支持 MSG_PEEK/MSG_DONTWAIT。
         # 这种情况下“主动探测”反而容易误报断线；让真正的 send/recv 去判定更稳。
         try:
@@ -1441,6 +1222,14 @@ class NetworkManager:
             # 这里保持 socket 为非阻塞模式（连接时已 setblocking(False)）。
             # 不要反复 settimeout()，否则会把 socket 切回"阻塞+超时"，并导致 conncheck 误报 timed out。
             data = wifi_manager.wifi_socket.recv(4096)  # 每次最多接收4KB（无数据会抛 BlockingIOError）
+            if data == b"":
+                self.logger.warning("[WIFI-TCP] 对端已关闭连接")
+                try:
+                    wifi_manager.wifi_socket.close()
+                except Exception:
+                    pass
+                wifi_manager.wifi_socket = None
+                self._tcp_connected = False
             return data
 
         except BlockingIOError:
@@ -2041,16 +1830,9 @@ class NetworkManager:
         self.logger.info("[NET] TCP主线程启动")
 
         send_hartbeat_fail_count = 0
-        last_charging_check = 0
-        CHARGING_CHECK_INTERVAL = 5000  # 5秒检查一次充电状态
 
         while True:
             try:
-                # 检查充电状态（每5秒检查一次）
-                current_time = time.ticks_ms()
-                if current_time - last_charging_check > CHARGING_CHECK_INTERVAL:
-                    last_charging_check = current_time
-
                 # OTA 期间不要 connect/登录/心跳/发送
                 try:
                     from ota_manager import ota_manager
@@ -2062,11 +1844,6 @@ 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
@@ -2082,7 +1859,8 @@ 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_send_packet(1, login_data)):
+                # if not self.tcp_send_raw(self.make_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()
@@ -2113,10 +1891,6 @@ class NetworkManager:
                         time.sleep_ms(200)
                         continue
 
-                    # OTA 完成后需要重启，跳出内层循环
-                    if self.ota_restart_pending:
-                        break
-
                     # 接收数据（根据网络类型选择接收方式）
                     # WiFi 粘包：一次 recv 可能含多条完整包；也可能缓冲里已有完整包但本轮 recv 超时为空
                     rx_items = []
@@ -2163,11 +1937,11 @@ class NetworkManager:
                                     pass
 
                             # msg_type, body = self.parse_packet(payload)
-                            msg_type, body = self._parse_recv(payload)
+                            msg_type, body = self._netcore.parse_packet(payload)
 
                             # 处理登录响应
                             if not logged_in and msg_type == 1:
-                                if body and body.get("cmd") == 0 and body.get("data") == "登录成功":
+                                if body and body.get("cmd") == 1 and body.get("data") == "登录成功":
                                     logged_in = True
                                     last_heartbeat_ack_time = time.ticks_ms()
                                     self.logger.info("登录成功")
@@ -2198,7 +1972,32 @@ 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}")
@@ -2291,44 +2090,79 @@ class NetworkManager:
                                         )
                                         # 立即返回已入队确认
                                         self.safe_enqueue({"result": "log_upload_queued"}, 2)
-                            # 处理业务指令（纯 proto: cmd 在 body 顶层）
+                            elif logged_in and msg_type == 201:
+                                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):
-                                cmd = body.get("cmd")
-                                data_obj = body.get("data") or {}
-                                if cmd == 2:  # AimRequest 开启激光并校准
+                                inner_cmd = None
+                                data_obj = body.get("data")
+                                if isinstance(data_obj, dict):
+                                    inner_cmd = data_obj.get("cmd")
+                                    if inner_cmd == 2:  # 开启激光并校准
                                         from laser_manager import laser_manager
                                         if not laser_manager.calibration_active:
                                             laser_manager.turn_on_laser()
                                             time.sleep_ms(100)
-                                        hardware_manager.stop_idle_timer()
+                                            hardware_manager.stop_idle_timer()  # 停表
                                             if not config.HARDCODE_LASER_POINT:
                                                 laser_manager.start_calibration()
                                                 self.safe_enqueue({"result": "calibrating"}, 2)
                                             else:
+                                                # 写死的逻辑，不需要校准激光点
                                                 self.safe_enqueue({"result": "laser pos set by hard code"}, 2)
-                                elif cmd == 3:  # CloseAimRequest 关闭激光
+                                    elif inner_cmd == 3:  # 关闭激光
                                         from laser_manager import laser_manager
                                         laser_manager.turn_off_laser()
                                         laser_manager.stop_calibration()
-                                    hardware_manager.start_idle_timer()
+                                        hardware_manager.start_idle_timer()  # 开表
                                         self.safe_enqueue({"result": "laser_off"}, 2)
-                                elif cmd == 4:  # GetBatteryRequest 上报电量
+                                    elif inner_cmd == 4:  # 上报电量
                                         voltage = get_bus_voltage()
                                         battery_percent = voltage_to_percent(voltage)
-                                    charging = is_charging()
-                                    self.safe_enqueue({
-                                        "cmd": 4,
+                                        battery_data = {
                                             "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", "")
+                                        }
+                                        self.safe_enqueue(battery_data, 2)
+                                        self.logger.info(f"电量上报: {battery_percent}%")
+                                    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")
@@ -2342,100 +2176,119 @@ class NetworkManager:
                                             _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
+                                        # 自动判断模式：如果没有明确指定，根据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()
-                                    self._spawn_cmd_thread(self._cmd5_ota, (ota_url,))
-                                elif cmd == 41:  # Ota4gSubCodeRequest 射箭触发
+                                        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 cmd == 42:  # ShutdownCommand 关机命令
+                                        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")
+                                        os.system("sync")  # 刷新文件系统缓存到磁盘，防止数据丢失
                                         time.sleep_ms(500)
+                                        # os.system("poweroff")
                                         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")
+                                    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()
+                                        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)
+                                            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 cmd == 200:  # GenericResult "init_center_point" 触发激光检测
+                                    elif inner_cmd == 200:
                                         self.logger.info("[LASER] cmd200 在后台线程执行检测")
                                         self._spawn_cmd_thread(self._cmd200_detect_laser, ())
-                                elif cmd == 201:  # SetCenterPointRequest 设置中心点
-                                    if self.logger:
-                                        self.logger.info(f"[LASER] cmd201:{body}")
-                                    raw_x = data_obj.get("x")
-                                    raw_y = data_obj.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 cmd == 300:  # OtaRequest 新版OTA
+                                    elif inner_cmd == 300:
                                         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:
+                                        self._spawn_cmd_thread(self._cmd300_ota, (data_obj,))
+                                    elif inner_cmd == 600:
+                                        self.logger.info("[conn wifi] cmd600 在后台线程执行连接wifi: {data_obj}")
+                                        self._spawn_cmd_thread(self._cmd600_conn_wifi, (data_obj,))
+                                    elif inner_cmd == 601:
                                         pass
-                                else:
+                                else:  # data的结构不是 dict
                                     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:
@@ -2457,8 +2310,9 @@ class NetworkManager:
                                 item_is_high = False
 
                         if item:
-                            msg_type, data_dict = item
-                            pkt = self._make_send_packet(msg_type, data_dict)
+                            msg_type, data_dict = item[:2]
+                            sent_event = item[2] if len(item) > 2 else None
+                            pkt = self._netcore.make_packet(msg_type, data_dict)
                             if not self.tcp_send_raw(pkt):
                                 # 发送失败：将消息放回队首（队列满则丢弃）
                                 with self.get_queue_lock():
@@ -2474,6 +2328,8 @@ class NetworkManager:
                                 except:
                                     pass
                                 break
+                            if sent_event is not None:
+                                sent_event.set()
 
                     # 发送激光校准结果
                     if logged_in:
@@ -2487,8 +2343,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()
-                        heartbeat_pkt = self._make_send_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._netcore.make_packet(4, {"vol": vol_val, "vol_per": voltage_to_percent(vol_val)})):
                             # 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
                             # 短暂波动可能导致一次发送失败：连续失败达到阈值才重连，避免重连风暴

power.py

@@ -10,6 +10,7 @@ import subprocess
 from logger_manager import logger_manager
 from maix import time as maix_time
 
+
 _INA226_PRESENT = None
 
 
@@ -141,6 +142,115 @@ def is_charging(threshold_ma=10.0):
         return False
 
 
+def charging_shutdown_monitor():
+    """独立监测 INA226；连续确认充电后通知服务器并退出应用。"""
+    logger = logger_manager.logger
+    shutdown_enabled = bool(getattr(config, "CHARGING_SHUTDOWN_ENABLED", False))
+    diagnostic_enabled = bool(getattr(config, "CHARGING_DIAGNOSTIC_LOG_ENABLED", False))
+    if not shutdown_enabled and not diagnostic_enabled:
+        if logger:
+            logger.info("[CHARGE] 充电退出监测已禁用")
+        return
+
+    interval_ms = max(100, int(getattr(config, "CHARGING_CHECK_INTERVAL_MS", 5000)))
+    threshold_ma = float(getattr(config, "CHARGING_CURRENT_THRESHOLD_MA", 10.0))
+    confirm_required = max(1, int(getattr(config, "CHARGING_CONFIRM_COUNT", 2)))
+    confirm_count = 0
+
+    if logger:
+        logger.info(
+            f"[CHARGE] 独立监测线程启动: interval={interval_ms}ms, "
+            f"threshold={threshold_ma:.1f}mA, confirm={confirm_required}, "
+            f"shutdown={'on' if shutdown_enabled else 'off'}"
+        )
+
+    while True:
+        current_ma = get_current()
+        if diagnostic_enabled and logger:
+            voltage = get_bus_voltage()
+            logger.info(
+                f"[CHARGE-DIAG] INA226 voltage={voltage:.3f}V, "
+                f"current={current_ma:.1f}mA"
+            )
+
+        if not shutdown_enabled:
+            maix_time.sleep_ms(interval_ms)
+            continue
+
+        if current_ma < -abs(threshold_ma):
+            confirm_count += 1
+            if logger:
+                logger.info(
+                    f"[CHARGE] INA226 充电电流 {current_ma:.1f}mA "
+                    f"({confirm_count}/{confirm_required})"
+                )
+        else:
+            confirm_count = 0
+
+        if confirm_count >= confirm_required:
+            script_path = getattr(
+                config,
+                "CHARGING_EXIT_SCRIPT",
+                config.APP_DIR + "/charging_exit.sh",
+            )
+            if not os.path.isfile(script_path):
+                if logger:
+                    logger.error(f"[CHARGE] 退出脚本不存在: {script_path}")
+                confirm_count = 0
+            else:
+                if logger:
+                    logger.warning(
+                        f"[CHARGE] 已连续确认充电，通知服务器后退出应用: current={current_ma:.1f}mA"
+                    )
+                try:
+                    from network import network_manager
+
+                    notify_timeout_ms = max(
+                        0,
+                        int(getattr(config, "CHARGING_NOTIFY_TIMEOUT_MS", 30000)),
+                    )
+                    notification_sent = network_manager.safe_enqueue_and_wait(
+                        {"poweroff": "充电中"},
+                        2,
+                        high=True,
+                        timeout_ms=notify_timeout_ms,
+                    )
+                    if notification_sent:
+                        if logger:
+                            logger.info("[CHARGE] 充电状态已发送到服务器")
+                    elif logger:
+                        logger.warning(
+                            f"[CHARGE] 等待服务器发送超时({notify_timeout_ms}ms)，继续执行退出"
+                        )
+                except Exception as e:
+                    if logger:
+                        logger.error(f"[CHARGE] 充电状态上报失败，继续执行退出: {e}")
+                try:
+                    from laser_manager import laser_manager
+
+                    laser_manager.turn_off_laser()
+                    if logger:
+                        logger.info("[CHARGE] 激光关闭命令已发送")
+                except Exception as e:
+                    if logger:
+                        logger.error(f"[CHARGE] Python 关闭激光失败，交由退出脚本兜底: {e}")
+                try:
+                    subprocess.Popen([
+                        "/bin/sh",
+                        script_path,
+                        str(os.getpid()),
+                        str(getattr(config, "DISTANCE_SERIAL_DEVICE", "/dev/ttyS1")),
+                        str(getattr(config, "DISTANCE_SERIAL_BAUDRATE", 9600)),
+                    ])
+                    return
+                except Exception as e:
+                    if logger:
+                        logger.error(f"[CHARGE] 调用退出脚本失败: {e}")
+                    confirm_count = 0
+
+        maix_time.sleep_ms(interval_ms)
+
+
 def voltage_to_percent(voltage):
     """
     根据电压估算电池百分比（高密度查表插值 + 滤波）。

shoot_manager.py

@@ -8,12 +8,7 @@ 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,
-    enqueue_save_raw_shot,
-)
+from vision import estimate_distance, detect_circle_v3, enqueue_save_shot
 from maix import image, time
 
 # 缓存相机标定与三角形位置，避免每次射箭重复读磁盘
@@ -59,7 +54,6 @@ 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
@@ -75,11 +69,7 @@ def analyze_shot(frame, laser_point=None):
             logger.info(f"[算法] 使用校准值: {laser_manager.laser_point}")
     else:
         # 动态模式：先做一次无激光点检测以估算距离，再推算激光点
-        _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)
@@ -124,7 +114,6 @@ 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
@@ -132,12 +121,9 @@ def analyze_shot(frame, laser_point=None):
 
     if not use_tri:
         # 三角形未配置，直接跑圆形检测
-        _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
-        return _build_circle_result(cdata)
+        return _build_circle_result(
+            detect_circle_v3(frame, laser_point, img_cv=img_cv)
+        )
 
     # ── Step 4: 先独占跑三角形，超时或失败后再跑圆形（不与圆心并行，避免抢 CPU）──
     roi_xyxy = None
@@ -290,7 +276,6 @@ 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"),
@@ -310,12 +295,8 @@ def analyze_shot(frame, laser_point=None):
         logger.warning(f"[TRI] 超时 {tri_timeout_s:.2f}s 仍未结束，启动圆心算法（三角形仍在后台）")
 
     # 三角形超时或失败 → 跑圆心；圆心跑完后再检查三角形是否已结束
-    try:
-        _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)
@@ -340,32 +321,8 @@ 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
-
-            _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:
-            if logger:
-                logger.warning(f"[YOLO-TARGET] 当前箭分类失败，按未知处理: {exc}")
-
         # 调用算法分析
         analysis_result = analyze_shot(frame)
 
@@ -389,7 +346,6 @@ 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")
@@ -410,6 +366,10 @@ 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}")
 
@@ -422,27 +382,11 @@ def process_shot(adc_val):
         srv_y = round(float(dy), 4) if dy is not None else 200.0
 
         # 构造上报数据
-        target_label = (
-            target_class_result.get("label")
-            if isinstance(target_class_result, dict)
-            else None
-        )
-        target_confidence = (
-            target_class_result.get("confidence")
-            if isinstance(target_class_result, dict)
-            else None
-        )
         inner_data = {
             "shot_id": shot_id,
             "x": srv_x,
             "y": srv_y,
             "r": 20.0,  # 保留字段（服务端当前忽略，物理外环半径 cm）
-            "target_class": target_label,
-            "target_class_confidence": (
-                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),
             "d_laser_quality": laser_signal_quality,
@@ -453,8 +397,6 @@ 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:
@@ -471,19 +413,7 @@ 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(
-                f"[REPORT-TARGET] enqueue shot_id={shot_id}, "
-                f"target_class={target_label}, confidence={target_confidence}"
-            )
         network_manager.safe_enqueue(report_data, msg_type=2, high=True)
 
         # 数据上报后再画标注，不干扰检测阶段的原始画面
@@ -588,7 +518,6 @@ def process_shot(adc_val):
             laser_manager.flash_laser(config.FLASH_LASER_DURATION_MS)
 
         # 保存图像（异步队列，与 main.py 一致）
-        _force_save = (dx is None and dy is None) and getattr(config, "SAVE_IMAGE_ON_FAILURE", False)
         enqueue_save_shot(
             result_img,
             center,
@@ -598,9 +527,8 @@ def process_shot(adc_val):
             (x, y),
             distance_m,
             shot_id=shot_id,
-            photo_dir=config.PHOTO_DIR if (config.SAVE_IMAGE_ENABLED or _force_save) else None,
+            photo_dir=config.PHOTO_DIR if config.SAVE_IMAGE_ENABLED else None,
             yolo_roi_xyxy=yolo_roi_xyxy if draw_yolo_roi else None,
-            force_save=_force_save,
         )
 
         if logger:

target_roi_yolo.py

@@ -89,29 +89,6 @@ def _stage2_roi_crop_save_worker(
 _detector_by_path = {}
 
 
-def _resolve_model_path(model_path: str):
-    """Resolve a model in either the installed app or MaixVision run directory."""
-    model_path = (model_path or "").strip()
-    if model_path and os.path.isfile(model_path):
-        return model_path
-    if not model_path:
-        return ""
-    name = os.path.basename(model_path)
-    module_dir = os.path.dirname(os.path.abspath(__file__))
-    candidates = (
-        os.path.join(module_dir, name),
-        os.path.join(module_dir, "test", name),
-        os.path.join("/tmp/maixpy_run", name),
-        os.path.join("/tmp/maixpy_run", "test", name),
-        os.path.join(os.getcwd(), name),
-        os.path.join(os.getcwd(), "test", name),
-    )
-    for candidate in candidates:
-        if os.path.isfile(candidate):
-            return candidate
-    return model_path
-
-
 def reset_yolo_detector_cache():
     """切换模型路径时可调用（通常不必）。"""
     global _detector_by_path
@@ -126,19 +103,10 @@ def _get_detector(model_path: str):
         return _detector_by_path[model_path]
     try:
         from maix import nn
-    except Exception:
+    except ImportError:
         return None
-    # YOLO is an optional capability. A broken/incompatible model must not
-    # abort boot (especially before the OTA rollback check).
-    try:
-        detector = nn.YOLOv5(model=model_path, dual_buff=False)
-    except Exception:
-        # Cache the failure to avoid retrying a broken native load every frame.
-        # reset_yolo_detector_cache() clears this after a model replacement.
-        _detector_by_path[model_path] = None
-        return None
-    _detector_by_path[model_path] = detector
-    return detector
+    _detector_by_path[model_path] = nn.YOLOv5(model=model_path, dual_buff=False)
+    return _detector_by_path[model_path]
 
 
 def preload_yolo_detector(logger=None):
@@ -207,23 +175,6 @@ def preload_yolo_detector(logger=None):
                 % (_loc_black,)
             )
 
-    if bool(getattr(cfg, "TARGET_CLASS_YOLO_ENABLE", False)) and bool(
-        getattr(cfg, "TARGET_CLASS_YOLO_PRELOAD_ON_BOOT", True)
-    ):
-        class_model_path = _resolve_model_path(
-            getattr(cfg, "TARGET_CLASS_YOLO_MODEL_PATH", "") or ""
-        )
-        class_detector = _get_detector(class_model_path)
-        if class_detector is None:
-            if logger:
-                logger.warning(
-                    f"[YOLO-TARGET] 预加载失败：无法加载模型 {class_model_path}"
-                )
-        else:
-            ok = True
-            if logger:
-                logger.info(f"[YOLO-TARGET] 靶规格模型已预加载: {class_model_path}")
-
     return ok
 
 
@@ -255,10 +206,8 @@ def _det_obj_class_id(o):
             if v is None:
                 continue
             try:
-                if callable(v):
-                    v = v()
                 return int(float(v))
-            except (TypeError, ValueError, AttributeError):
+            except (TypeError, ValueError):
                 continue
     return None
 
@@ -293,106 +242,6 @@ def _normalize_objs(objs):
     return out
 
 
-def _det_obj_score(o):
-    """Return confidence across supported Maix YOLO result formats."""
-    for key in ("score", "confidence", "conf", "prob"):
-        if hasattr(o, key):
-            try:
-                value = getattr(o, key)
-                if callable(value):
-                    value = value()
-                value = float(value)
-                if value == value:
-                    return value
-            except (TypeError, ValueError, AttributeError):
-                pass
-    return 0.0
-
-
-def try_get_target_class_from_yolo(maix_frame, logger=None):
-    """Classify the current target as 20cm or 40cm; return None if unknown."""
-    try:
-        import config as cfg
-    except Exception:
-        return None
-
-    if not bool(getattr(cfg, "TARGET_CLASS_YOLO_ENABLE", False)):
-        return None
-    model_path = _resolve_model_path(
-        getattr(cfg, "TARGET_CLASS_YOLO_MODEL_PATH", "") or ""
-    )
-    if not os.path.isfile(model_path):
-        if logger:
-            logger.warning(f"[YOLO-TARGET] 模型文件不存在: {model_path}")
-        return None
-    detector = _get_detector(model_path)
-    if detector is None:
-        if logger:
-            logger.warning("[YOLO-TARGET] 无法加载 nn.YOLOv5")
-        return None
-
-    conf_th = float(getattr(cfg, "TARGET_CLASS_YOLO_CONF_TH", 0.5))
-    iou_th = float(getattr(cfg, "TARGET_CLASS_YOLO_IOU_TH", 0.45))
-    labels = getattr(cfg, "TARGET_CLASS_YOLO_LABELS", (20, 40))
-    if isinstance(labels, str):
-        labels = tuple(x.strip() for x in labels.split(",") if x.strip())
-    labels = tuple(labels)
-
-    def _detect(threshold):
-        try:
-            raw = detector.detect(maix_frame, conf_th=threshold, iou_th=iou_th)
-        except Exception as exc:
-            if logger:
-                logger.warning(f"[YOLO-TARGET] detect 异常: {exc}")
-            return []
-        return _normalize_objs(raw if raw is not None else [])
-
-    def _candidates(objs):
-        found = []
-        for obj in objs:
-            class_id = _det_obj_class_id(obj)
-            if class_id is None or class_id < 0 or class_id >= len(labels):
-                continue
-            try:
-                label = int(float(labels[class_id]))
-            except (TypeError, ValueError):
-                continue
-            if label in (20, 40):
-                found.append((label, class_id, _det_obj_score(obj)))
-        return found
-
-    objects = _detect(conf_th)
-    candidates = _candidates(objects)
-    if logger and objects:
-        logger.info(
-            "[YOLO-TARGET] 原始框=%d, 解析类别=%s"
-            % (
-                len(objects),
-                [(_det_obj_class_id(o), _det_obj_score(o)) for o in objects[:8]],
-            )
-        )
-    if not candidates and bool(
-        getattr(cfg, "TARGET_CLASS_YOLO_RETRY_ON_EMPTY", False)
-    ):
-        retry_th = float(getattr(cfg, "TARGET_CLASS_YOLO_RETRY_CONF_TH", conf_th))
-        if 0 < retry_th < conf_th:
-            candidates = _candidates(_detect(retry_th))
-
-    if not candidates:
-        if logger:
-            logger.warning("[YOLO-TARGET] 当前帧未识别到 20/40，按未知处理")
-        return None
-
-    label, class_id, confidence = max(candidates, key=lambda item: item[2])
-    result = {"label": label, "class_id": class_id, "confidence": confidence}
-    if logger:
-        logger.info(
-            f"[YOLO-TARGET] 当前帧分类={label}, class_id={class_id}, "
-            f"conf={confidence:.3f}"
-        )
-    return result
-
-
 def _det_to_src_xyxy(o, coord_mode: str, src_w: int, src_h: int, net_w: int, net_h: int):
     """把单个检测框转为全图坐标系下的 xyxy（半开区间语义与后续 clip 一致）。"""
     x, y, w, h = float(o.x), float(o.y), float(o.w), float(o.h)

test/test_charging_shutdown.py

@@ -0,0 +1,144 @@
+import importlib.util
+from pathlib import Path
+import sys
+import types
+import unittest
+from unittest import mock
+
+
+class _StopMonitor(Exception):
+    pass
+
+
+class _FakeTime:
+    now_ms = 0
+    stop_at_ms = None
+
+    @classmethod
+    def reset(cls, stop_at_ms=None):
+        cls.now_ms = 0
+        cls.stop_at_ms = stop_at_ms
+
+    @classmethod
+    def ticks_ms(cls):
+        return cls.now_ms
+
+    @classmethod
+    def sleep_ms(cls, milliseconds):
+        cls.now_ms += milliseconds
+        if cls.stop_at_ms is not None and cls.now_ms >= cls.stop_at_ms:
+            raise _StopMonitor()
+
+
+def _load_power_module():
+    module_path = Path(__file__).resolve().parents[1] / "power.py"
+    module_name = "power_charging_shutdown_test"
+    maix_module = types.ModuleType("maix")
+    maix_module.time = _FakeTime
+
+    previous_maix = sys.modules.get("maix")
+    sys.modules["maix"] = maix_module
+    try:
+        spec = importlib.util.spec_from_file_location(module_name, module_path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        if previous_maix is None:
+            sys.modules.pop("maix", None)
+        else:
+            sys.modules["maix"] = previous_maix
+
+
+power = _load_power_module()
+
+
+class ChargingShutdownTests(unittest.TestCase):
+    def setUp(self):
+        self.config_patch = mock.patch.multiple(
+            power.config,
+            CHARGING_SHUTDOWN_ENABLED=True,
+            CHARGING_DIAGNOSTIC_LOG_ENABLED=False,
+            CHARGING_CHECK_INTERVAL_MS=5000,
+            CHARGING_CURRENT_THRESHOLD_MA=100.0,
+            CHARGING_CONFIRM_COUNT=2,
+            CHARGING_NOTIFY_TIMEOUT_MS=30000,
+            CHARGING_EXIT_SCRIPT="/tmp/charging_exit.sh",
+        )
+        self.config_patch.start()
+        self.network_manager = mock.Mock()
+        self.network_manager.safe_enqueue_and_wait.return_value = True
+        network_module = types.ModuleType("network")
+        network_module.network_manager = self.network_manager
+        self.network_module_patch = mock.patch.dict(
+            sys.modules,
+            {"network": network_module},
+        )
+        self.network_module_patch.start()
+        _FakeTime.reset()
+
+    def tearDown(self):
+        self.network_module_patch.stop()
+        self.config_patch.stop()
+
+    def test_two_charging_samples_notify_server_and_exit(self):
+        popen_calls = []
+        with (
+            mock.patch.object(power, "get_current", return_value=-200.0),
+            mock.patch.object(power.os.path, "isfile", return_value=True),
+            mock.patch.object(
+                power.subprocess,
+                "Popen",
+                side_effect=lambda args: popen_calls.append(args),
+            ),
+        ):
+            power.charging_shutdown_monitor()
+
+        self.assertEqual(_FakeTime.now_ms, 5000)
+        self.assertEqual(len(popen_calls), 1)
+        self.network_manager.safe_enqueue_and_wait.assert_called_once_with(
+            {"poweroff": "充电中"}, 2, high=True, timeout_ms=30000
+        )
+
+    def test_discharging_does_not_notify_or_exit(self):
+        _FakeTime.reset(stop_at_ms=10000)
+        popen_calls = []
+
+        with (
+            mock.patch.object(power, "get_current", return_value=200.0),
+            mock.patch.object(power.os.path, "isfile", return_value=True),
+            mock.patch.object(
+                power.subprocess,
+                "Popen",
+                side_effect=lambda args: popen_calls.append(args),
+            ),
+            self.assertRaises(_StopMonitor),
+        ):
+            power.charging_shutdown_monitor()
+
+        self.assertEqual(popen_calls, [])
+        self.network_manager.safe_enqueue_and_wait.assert_not_called()
+
+    def test_failed_sample_resets_confirmation_count(self):
+        popen_calls = []
+        currents = iter((-200.0, 0.0, -200.0, -200.0))
+        with (
+            mock.patch.object(power, "get_current", side_effect=lambda: next(currents)),
+            mock.patch.object(power.os.path, "isfile", return_value=True),
+            mock.patch.object(
+                power.subprocess,
+                "Popen",
+                side_effect=lambda args: popen_calls.append(args),
+            ),
+        ):
+            power.charging_shutdown_monitor()
+
+        self.assertEqual(_FakeTime.now_ms, 15000)
+        self.assertEqual(len(popen_calls), 1)
+        self.network_manager.safe_enqueue_and_wait.assert_called_once_with(
+            {"poweroff": "充电中"}, 2, high=True, timeout_ms=30000
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_decect_circle_v4.py

@@ -154,11 +154,11 @@ def detect_circle_v3(frame, laser_point=None):
                                     max_r = max(red_radius, yellow_radius)
                                     size_ratio = min_r / max_r if max_r > 0 else 0
                                     print(f"Debug -> 圆心距={distance:.1f}(阈值={max_distance:.1f}), "
-                                          f"大小比={size_ratio:.2f}(阈值=0.4), "
-                                          f"距离OK={distance < max_distance}, 大小OK={size_ratio >= 0.4}")
+                                          f"大小比={size_ratio:.2f}(阈值=0.5), "
+                                          f"距离OK={distance < max_distance}, 大小OK={size_ratio > 0.5}")
 
                                     # 允许红圈在黄圈外侧或内侧，只要大小相近（较小/较大 >= 0.5）
-                                    if distance < max_distance and size_ratio >= 0.4:
+                                    if distance < max_distance and size_ratio > 0.5:
                                         found_valid_red = True
                                         print(
                                             f"[target] -> 找到匹配的红圈: 黄心({yellow_center}), 红心({red_center}), 距离:{distance:.1f}, 黄半径:{yellow_radius}, 红半径:{red_radius}")
@@ -598,7 +598,7 @@ if __name__ == "__main__":
 
     # 1. 设置要测试的图片路径
     # 建议将图片放在与脚本同级目录，或者使用绝对路径
-    TARGET_IMAGE = "/root/phot/shot_1830921_0_no_target.jpg"
+    TARGET_IMAGE = "/root/phot/None_314_258_0_0041.bmp"
 
     TARGET_DIR = "/root/phot"  # 修改为你想要读取的目录路径
 

test/test_p21_voltage.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Read the digital voltage level on the MaixCAM P21 pin.
+
+P21 is a digital GPIO pin, not the MaixCAM analog ADC input. Therefore this
+script can only distinguish LOW and HIGH. For a continuous voltage value,
+connect the signal to the board's B3/ADC pin and use ADC channel 0 instead.
+
+Do not apply more than 3.3 V to P21. Always connect the signal ground to the
+MaixCAM ground.
+"""
+
+from maix import app, gpio, pinmap, time
+
+
+PIN = "P21"
+IO_HIGH_VOLTAGE = 3.3
+SAMPLE_INTERVAL_MS = 200
+
+
+def find_gpio_function(pin):
+    """Return the GPIO function supported by the requested physical pin."""
+    functions = pinmap.get_pin_functions(pin)
+    gpio_functions = [name for name in functions if name.startswith("GPIO")]
+
+    print(f"{pin} supported functions: {', '.join(functions)}")
+    if not gpio_functions:
+        raise RuntimeError(f"{pin} does not provide a GPIO input function")
+
+    return gpio_functions[0]
+
+
+def main():
+    gpio_function = find_gpio_function(PIN)
+    pinmap.set_pin_function(PIN, gpio_function)
+    voltage_input = gpio.GPIO(gpio_function, gpio.Mode.IN)
+
+    print(f"Reading {PIN} through {gpio_function}")
+    print("P21 only reports LOW/HIGH; displayed voltage is an estimate.")
+    print("Press the MaixCAM exit key to stop.")
+
+    while not app.need_exit():
+        level = voltage_input.value()
+        estimated_voltage = IO_HIGH_VOLTAGE if level else 0.0
+        state = "HIGH" if level else "LOW"
+        print(
+            f"{PIN}: level={level}, state={state}, "
+            f"estimated_voltage={estimated_voltage:.1f} V"
+        )
+        time.sleep_ms(SAMPLE_INTERVAL_MS)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        print(f"P21 voltage detection failed: {error}")
+        print("Check that this MaixCAM model exposes P21 as a GPIO pin.")
+        raise

test/test_target_yolo_maixvision.py

@@ -1,108 +0,0 @@
-#!/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/test_wifi_failover.py

@@ -0,0 +1,139 @@
+import json
+import sys
+import types
+import unittest
+
+
+class _FakeTime:
+    @staticmethod
+    def sleep(_seconds):
+        pass
+
+    @staticmethod
+    def sleep_ms(_milliseconds):
+        pass
+
+    @staticmethod
+    def ticks_ms():
+        return 0
+
+    @staticmethod
+    def ticks_diff(left, right):
+        return left - right
+
+
+class _FakeLogger:
+    def debug(self, *_args, **_kwargs):
+        pass
+
+    def info(self, *_args, **_kwargs):
+        pass
+
+    def warning(self, *_args, **_kwargs):
+        pass
+
+    def error(self, *_args, **_kwargs):
+        pass
+
+
+class _FakeSocket:
+    def __init__(self, recv_data=b""):
+        self.recv_data = recv_data
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+    def recv(self, _size, *_flags):
+        return self.recv_data
+
+
+class _StopAfterCallback:
+    def __init__(self):
+        self.stopped = False
+
+    def is_set(self):
+        return self.stopped
+
+
+maix_module = types.ModuleType("maix")
+maix_module.time = _FakeTime
+maix_module.network = types.SimpleNamespace()
+maix_module.err = types.SimpleNamespace()
+sys.modules.setdefault("maix", maix_module)
+sys.modules.setdefault("ujson", json)
+
+netcore_module = types.ModuleType("archery_netcore")
+netcore_module.get_config = lambda: {"SERVER_IP": "127.0.0.1", "SERVER_PORT": 1234}
+netcore_module.parse_packet = lambda _packet: (0, {})
+netcore_module.make_packet = lambda *_args, **_kwargs: b""
+netcore_module.actions_for_inner_cmd = lambda *_args, **_kwargs: []
+sys.modules["archery_netcore"] = netcore_module
+
+hardware_module = types.ModuleType("hardware")
+hardware_module.hardware_manager = types.SimpleNamespace()
+sys.modules["hardware"] = hardware_module
+
+power_module = types.ModuleType("power")
+power_module.get_bus_voltage = lambda: 0
+power_module.voltage_to_percent = lambda _voltage: 0
+sys.modules["power"] = power_module
+
+import logger_manager
+import wifi
+import network
+
+
+class WiFiFailoverTests(unittest.TestCase):
+    def setUp(self):
+        logger_manager.logger_manager._logger = _FakeLogger()
+
+    def test_monitor_switches_when_sta_association_is_lost(self):
+        manager = wifi.wifi_manager
+        stop_event = _StopAfterCallback()
+        callbacks = []
+
+        manager._wifi_socket = _FakeSocket()
+        manager._wifi_quality_stop_event = stop_event
+        manager._network_type_callback = lambda: "wifi"
+        manager.is_sta_associated = lambda: False
+        manager._get_wifi_rssi_dbm = lambda: None
+
+        def on_poor_quality():
+            callbacks.append(True)
+            stop_event.stopped = True
+
+        manager._on_poor_quality_callback = on_poor_quality
+        manager._quality_monitor_loop()
+
+        self.assertEqual(callbacks, [True])
+        self.assertIsNone(manager.last_wifi_rtt_ms)
+
+    def test_tls_connection_check_rejects_lost_sta_association(self):
+        manager = network.network_manager
+        sock = _FakeSocket()
+        wifi.wifi_manager._wifi_socket = sock
+        wifi.wifi_manager._wifi_connected = True
+        wifi.wifi_manager._wifi_ip = "192.168.1.2"
+        wifi.wifi_manager.is_sta_associated = lambda: False
+        manager._tcp_connected = True
+
+        self.assertFalse(manager._check_wifi_connection())
+        self.assertTrue(sock.closed)
+        self.assertIsNone(wifi.wifi_manager.wifi_socket)
+        self.assertFalse(manager.tcp_connected)
+
+    def test_receive_eof_marks_wifi_tcp_disconnected(self):
+        manager = network.network_manager
+        sock = _FakeSocket(recv_data=b"")
+        wifi.wifi_manager._wifi_socket = sock
+        manager._tcp_connected = True
+
+        self.assertEqual(manager.receive_tcp_data_via_wifi(), b"")
+        self.assertTrue(sock.closed)
+        self.assertIsNone(wifi.wifi_manager.wifi_socket)
+        self.assertFalse(manager.tcp_connected)
+
+
+if __name__ == "__main__":
+    unittest.main()

test_traditional_target.py

@@ -1,184 +0,0 @@
-#!/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

@@ -29,12 +29,10 @@
 # 2.15.16 修复wifi连接问题
 # 2.15.17 修复wifi连接问题
 # 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 模型修改
+# 2.15.20 加了充电关机，激光也同时关闭
+# 2.15.21 测试4g 扩大了缓存池和改了心跳时间
+# 2.15.22 修复了4g网络和wifi切换问题
+# 2.15.23 合并充电关机与稳定版网络修复
+# 2.15.24 空改测试
+# 2.15.25 修复整合后关机失败和ota格式更新问题
+# 2.15.26 

version.py

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

vision.py

@@ -631,7 +631,7 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
             min_r = min(rc["radius"], yellow_radius)
             max_r = max(rc["radius"], yellow_radius)
             size_ratio = min_r / max_r if max_r > 0 else 0
-            if dist_centers < max_dist and size_ratio >= 0.4:
+            if dist_centers < max_dist and size_ratio >= 0.3:
                 if logger:
                     logger.info(f"[target] -> 找到匹配的红圈: 黄心({yellow_center}), "
                                 f"红心({rc['center']}), 距离:{dist_centers:.1f}, "
@@ -797,12 +797,12 @@ def estimate_pixel(physical_distance_cm, target_distance_m):
 
 def _save_shot_image_impl(img_cv, center, radius, method, ellipse_params,
                           laser_point, distance_m, shot_id=None, photo_dir=None,
-                          yolo_roi_xyxy=None, force_save=False):
+                          yolo_roi_xyxy=None):
     """
     内部实现：在 img_cv (numpy HWC RGB) 上绘制标注并保存。
     由 save_shot_image（同步）和存图 worker（异步）调用。
     """
-    if not config.SAVE_IMAGE_ENABLED and not force_save:
+    if not config.SAVE_IMAGE_ENABLED:
         return None
     if photo_dir is None:
         photo_dir = config.PHOTO_DIR
@@ -908,11 +908,6 @@ def _save_worker_loop():
             item = _save_queue.get()
             if item is None:
                 break
-            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
@@ -941,64 +936,13 @@ 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):
+                      yolo_roi_xyxy=None):
     """
     将存图任务放入队列，由 worker 异步保存。主线程传入 result_img 的复制，不阻塞。
-    force_save=True 时，忽略 SAVE_IMAGE_ENABLED 配置强制保存（用于检测失败时的调试图像）。
     """
-    if not config.SAVE_IMAGE_ENABLED and not force_save:
+    if not config.SAVE_IMAGE_ENABLED:
         return
     if photo_dir is None:
         photo_dir = config.PHOTO_DIR
@@ -1021,7 +965,6 @@ def enqueue_save_shot(result_img, center, radius, method, ellipse_params,
         shot_id,
         photo_dir,
         yolo_roi_xyxy,
-        force_save,
     )
     try:
         _save_queue.put_nowait(task)
@@ -1033,12 +976,12 @@ def enqueue_save_shot(result_img, center, radius, method, ellipse_params,
 
 def save_shot_image(result_img, center, radius, method, ellipse_params,
                     laser_point, distance_m, shot_id=None, photo_dir=None,
-                    yolo_roi_xyxy=None, force_save=False):
+                    yolo_roi_xyxy=None):
     """
     保存射击图像（带标注）。同步调用，会阻塞。
     主流程建议使用 enqueue_save_shot；此处保留供校准、测试等场景使用。
     """
-    if not config.SAVE_IMAGE_ENABLED and not force_save:
+    if not config.SAVE_IMAGE_ENABLED:
         return None
     if photo_dir is None:
         photo_dir = config.PHOTO_DIR
@@ -1055,7 +998,6 @@ def save_shot_image(result_img, center, radius, method, ellipse_params,
             shot_id,
             photo_dir,
             yolo_roi_xyxy,
-            force_save,
         )
     except Exception as e:
         logger = logger_manager.logger

wifi.py

@@ -541,7 +541,7 @@ class WiFiManager:
     
     def start_quality_monitor(self, network_type_callback, on_poor_quality_callback):
         """
-        启动 WiFi 质量后台监测线程（每 5 秒测量一次 RTT 和 RSSI）
+        启动 WiFi 质量后台监测线程（每 5 秒检查 STA 关联状态和 RSSI）
         只在 WiFi 连接时运行，不影响业务发送性能
         
         Args:
@@ -591,34 +591,38 @@ class WiFiManager:
     def _quality_monitor_loop(self):
         """
         WiFi 质量监测循环（后台线程）
-        每 5 秒测量一次 RTT 和 RSSI，发现质量差则触发切换
+        每 5 秒检查 STA 关联状态和 RSSI，发现断链或质量差则触发切换
         """
         while not self._wifi_quality_stop_event.is_set():
             try:
                 # 只在 WiFi 连接时才测量
                 network_type = self._network_type_callback()
                 if network_type == "wifi" and self._wifi_socket:
-                    # # 测量 RTT（1 个样本，快速测量）
-                    # rtt_ms, reachable = self._measure_wifi_tcp_rtt_ms(
-                    #     self._server_ip, self._server_port,
-                    #     samples=1, per_sample_timeout_ms=600
-                    # )
+                    # RTT 测量当前禁用；STA 关联状态用于判断物理 WiFi 链路是否仍存在。
+                    # 不能把禁用的 RTT 伪装成 0ms，否则关闭热点后会一直被判为正常。
+                    reachable = self.is_sta_associated()
+                    rtt_ms = None
 
                     # 获取 RSSI
                     rssi_dbm = self._get_wifi_rssi_dbm()
 
                     # 更新缓存
-                    # 不使用 RTT 测量
-                    rtt_ms = 0
-                    reachable = True
-                    self._last_wifi_rtt_ms = rtt_ms if reachable else None
+                    self._last_wifi_rtt_ms = rtt_ms
                     self._last_wifi_rssi_dbm = rssi_dbm
+                    _rtt_s = f"{rtt_ms:.0f}ms" if rtt_ms is not None else "n/a"
                     _rssi_s = f"{rssi_dbm:.0f}" if rssi_dbm is not None else "n/a"
-                    self.logger.debug(f"[WiFi Monitor] - RTT={rtt_ms:.0f}ms, RSSI={_rssi_s}dBm")
+                    self.logger.debug(
+                        f"[WiFi Monitor] - associated={reachable}, RTT={_rtt_s}, RSSI={_rssi_s}dBm"
+                    )
                     
                     # 判断质量是否差（切换前做 2 次快速复测，防止瞬时抖动）
                     def _is_bad_now(_reachable, _rtt, _rssi):
-                        if (not _reachable) or (_rtt is None) or (_rtt == float("inf")):
+                        if not _reachable:
+                            return True
+                        # RTT 未启用时不参与质量判断；链路状态仍由 STA 关联保证。
+                        if _rtt is None:
+                            return False
+                        if _rtt == float("inf"):
                             return True
                         return self._is_wifi_quality_bad(_rtt, _rssi)
 
@@ -628,13 +632,8 @@ class WiFiManager:
 
                         for retry_idx in range(2):
                             time.sleep_ms(1000)
-                            # 不使用 RTT 测量
-                            rtt2 = 0
-                            reachable2 = True
-                            # rtt2, reachable2 = self._measure_wifi_tcp_rtt_ms(
-                            #     self._server_ip, self._server_port,
-                            #     samples=1, per_sample_timeout_ms=600
-                            # )
+                            reachable2 = self.is_sta_associated()
+                            rtt2 = None
                             rssi2 = self._get_wifi_rssi_dbm()
 
                             # 更新缓存，便于外部查看最新状态
@@ -643,14 +642,10 @@ class WiFiManager:
 
                             bad2 = _is_bad_now(reachable2, rtt2, rssi2)
                             try:
-                                _rtt_disp = (
-                                    rtt2
-                                    if rtt2 is not None and rtt2 != float("inf")
-                                    else -1
-                                )
+                                _rtt_disp = f"{rtt2:.0f}ms" if rtt2 is not None else "n/a"
                                 self.logger.info(
                                     f"[WiFi Monitor] 复测{retry_idx+1}/2: reachable={reachable2}, "
-                                    f"rtt={_rtt_disp:.0f}ms, rssi={rssi2}, bad={bad2}"
+                                    f"rtt={_rtt_disp}, rssi={rssi2}, bad={bad2}"
                                 )
                             except Exception:
                                 pass