pref: 20 cm adapter

app.yaml

@@ -1,6 +1,6 @@
 id: t11
 name: t11
-version: 2.14.1
+version: 2.15.9
 author: t11
 icon: ''
 desc: t11
@@ -14,12 +14,14 @@ files:
   - cameraParameters.xml
   - config.py
   - hardware.py
+  - laser_detector.py
   - laser_manager.py
   - logger_manager.py
   - main.py
   - model_270139.cvimodel
   - model_270139.mud
   - network.py
+  - ota_curl.sh
   - ota_manager.py
   - power.py
   - server.pem

config.py

@@ -134,7 +134,7 @@ IMAGE_CENTER_Y = 240  # 图像中心 Y 坐标
 # ==================== 三角形四角标记：单应性偏移 + PnP 估距 ====================
 # 依赖 cameraParameters.xml（相机内参）与 triangle_positions.json（四角物方坐标，厘米或毫米见 JSON 约定）。
 # 部署时请把这两个文件放到 APP_DIR（与 main 同应用目录），或改下面路径为设备上的实际绝对路径。
-USE_TRIANGLE_OFFSET = True  # False 时仅走黄心圆/椭圆 + 半径估距，不使用三角形路径
+USE_TRIANGLE_OFFSET = False  # False 时仅走黄心圆/椭圆 + 半径估距，不使用三角形路径
 CAMERA_CALIB_XML = APP_DIR + "/cameraParameters.xml"
 TRIANGLE_POSITIONS_JSON = APP_DIR + "/triangle_positions.json"
 # 检测到的三角形边长在图像中的像素范围，分辨率或靶纸占比变化时可微调
@@ -260,7 +260,7 @@ TRIANGLE_SAMPLE_RADIUS_CM = 15.0
 TRIANGLE_SAMPLE_ANGLES_DEG = (0, 90, 180, 270)
 TRIANGLE_SAMPLE_PATCH_HALF_PX = 2
 # 开机阶段预加载 YOLO detector；detect 使用 dual_buff=False，避免返回上一帧结果。
-TRIANGLE_YOLO_PRELOAD_ON_BOOT = True
+TRIANGLE_YOLO_PRELOAD_ON_BOOT = False
 
 # ── 第二段 YOLO：仅在 Stage1 裁切出的靶环图上推理（与合成 stage2 训练数据一致）→ 子框内传统算法取直角点 ──
 # Stage1 靶环裁切内如何找黑三角标记（对比耗时时可切换）：

laser_detector.py

@@ -1,10 +1,12 @@
 from maix import image, time
+from logger_manager import logger_manager
 from camera_manager import camera_manager
 
 _USE_CV = False
 try:
     import cv2
     import numpy as np
+
     _USE_CV = True
 except ImportError:
     pass
@@ -12,9 +14,38 @@ except ImportError:
 WIDTH = 640
 HEIGHT = 480
 THRESHOLD = 100
-RED_RATIO = 1.3
+RED_RATIO = 1.5
-SEARCH_RADIUS = 60
+SEARCH_RADIUS = 80
-STABLE_COUNT = 5
+TRACK_RADIUS = 30
+MIN_PIXELS = 3
+COARSE_STEP = 2
+STABLE_COUNT = 2
+MAX_SKIP_FRAMES = 5
+
+# Temporal smoothing
+_EMA_ALPHA = 0.35
+_GATE_PX = 10
+_FRAME_INTERVAL_MS = 50
+
+_prev_smoothed = None
+
+
+def _red_weighted_centroid(r_ch, g_ch, b_ch, mask, x0, y0):
+    y_ids, x_ids = np.where(mask)
+    if len(y_ids) == 0:
+        return None
+    r_vals = r_ch[y_ids, x_ids].astype(np.float64)
+    g_vals = g_ch[y_ids, x_ids].astype(np.float64)
+    b_vals = b_ch[y_ids, x_ids].astype(np.float64)
+    w = r_vals - np.maximum(g_vals, b_vals)
+    w = np.clip(w, 0, None)
+    w = w * w
+    total_w = w.sum()
+    if total_w < 1e-6:
+        return None
+    cx = (x_ids.astype(np.float64) * w).sum() / total_w + x0
+    cy = (y_ids.astype(np.float64) * w).sum() / total_w + y0
+    return (float(cx), float(cy))
 
 
 def find_ellipse(img_cv, cx, cy, roi_r, th, ratio):
@@ -41,106 +72,177 @@ def find_ellipse(img_cv, cx, cy, roi_r, th, ratio):
     for pt in cnt:
         pt[0][0] += x1
         pt[0][1] += y1
-    if len(cnt) >= 5:
+    ellipse_valid = len(cnt) >= 5
+    if ellipse_valid:
         (ex, ey), (ew, eh), ang = cv2.fitEllipse(cnt)
         mask_ellipse = np.zeros((HEIGHT, WIDTH), dtype=np.uint8)
         cv2.ellipse(mask_ellipse, (int(ex), int(ey)), (int(ew / 2), int(eh / 2)), ang, 0, 360, 255, -1)
-        brightness = img_cv[:, :, 0].astype(np.int32) + img_cv[:, :, 1].astype(np.int32) + img_cv[:, :, 2].astype(np.int32)
+        return _red_weighted_centroid(
-        masked = np.where(mask_ellipse > 0, brightness, 0)
+            img_cv[:, :, 0], img_cv[:, :, 1], img_cv[:, :, 2],
-        vals = masked[masked > 0]
+            mask_ellipse > 0, 0, 0
-        if len(vals) > 0:
+        )
-            bth = np.percentile(vals, 90)
-            bmask = (masked >= bth).astype(np.uint8) * 255
-            bcontours, _ = cv2.findContours(bmask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
-            if bcontours:
-                blargest = max(bcontours, key=cv2.contourArea)
-                if cv2.contourArea(blargest) >= 3 and len(blargest) >= 5:
-                    (ix, iy), _, _ = cv2.fitEllipse(blargest)
-                    return (float(ix), float(iy))
-                M = cv2.moments(blargest)
-                if M["m00"] > 0:
-                    return (float(M["m10"] / M["m00"]), float(M["m01"] / M["m00"]))
-        return (float(ex), float(ey))
     M = cv2.moments(cnt)
     if M["m00"] > 0:
         return (float(M["m10"] / M["m00"]), float(M["m01"] / M["m00"]))
     return None
 
 
+def is_red(r, g, b, th, ratio):
+    if r > th and r > g * ratio and r > b * ratio:
+        return True
+    if (r > 200 and g > 200 and b > 200 and r >= g and r >= b
+            and (r - g) > 10 and (r - b) > 10):
+        return True
+    return False
+
+
 def find_brightest_bytes(frame, cx, cy, roi_r, th, ratio):
     x1 = max(0, cx - roi_r)
     x2 = min(WIDTH, cx + roi_r)
     y1 = max(0, cy - roi_r)
     y2 = min(HEIGHT, cy + roi_r)
     data = frame.to_bytes()
+
     best_score = 0
-    best_pos = None
+    best_x = (x1 + x2) // 2
-    for y in range(y1, y2, 2):
+    best_y = (y1 + y2) // 2
-        for x in range(x1, x2, 2):
+    found_any = False
+    for y in range(y1, y2, COARSE_STEP):
+        for x in range(x1, x2, COARSE_STEP):
             idx = (y * WIDTH + x) * 3
-            r = data[idx]; g = data[idx+1]; b = data[idx+2]
+            r = data[idx]
-            if (r > th and r > g * ratio and r > b * ratio) or \
+            g = data[idx + 1]
-               (r > 200 and g > 200 and b > 200 and r >= g and r >= b and (r - g) > 10 and (r - b) > 10):
+            b = data[idx + 2]
+            if is_red(r, g, b, th, ratio):
                 score = r + g + b
-                dx = x - cx; dy = y - cy
+                dx = x - cx
-                score *= max(0.5, 1.0 - ((dx*dx + dy*dy) ** 0.5 / roi_r) * 0.5)
+                dy = y - cy
+                dist_decay = max(0.5, 1.0 - ((dx * dx + dy * dy) ** 0.5 / roi_r) * 0.5)
+                score *= dist_decay
                 if score > best_score:
                     best_score = score
-                    best_pos = (x, y)
+                    best_x = x
-    if best_pos is None:
+                    best_y = y
+                    found_any = True
+
+    if not found_any:
         return None
-    fx, fy = best_pos
+
-    x1f = max(0, fx - 3); x2f = min(WIDTH, fx + 4)
+    sf = 4
-    y1f = max(0, fy - 3); y2f = min(HEIGHT, fy + 4)
+    fx1 = max(x1, best_x - sf)
-    best_bright = 0
+    fx2 = min(x2, best_x + sf + 1)
-    final_pos = best_pos
+    fy1 = max(y1, best_y - sf)
-    for y in range(y1f, y2f):
+    fy2 = min(y2, best_y + sf + 1)
-        for x in range(x1f, x2f):
+
+    sum_x = 0.0
+    sum_y = 0.0
+    total_w = 0.0
+    count = 0
+    for y in range(fy1, fy2):
+        for x in range(fx1, fx2):
             idx = (y * WIDTH + x) * 3
-            r = data[idx]; g = data[idx+1]; b = data[idx+2]
+            r = data[idx]
-            if (r > th and r > g * ratio and r > b * ratio) or \
+            g = data[idx + 1]
-               (r > 200 and g > 200 and b > 200 and r >= g and r >= b and (r - g) > 10 and (r - b) > 10):
+            b = data[idx + 2]
-                rgb_sum = r + g + b
+            if is_red(r, g, b, th, ratio):
-                if rgb_sum > best_bright:
+                w = r + g + b
-                    best_bright = rgb_sum
+                sum_x += x * w
-                    final_pos = (float(x), float(y))
+                sum_y += y * w
-    return final_pos
+                total_w += w
+                count += 1
+
+    if count < MIN_PIXELS:
+        return (float(best_x), float(best_y))
+
+    return (float(sum_x / total_w), float(sum_y / total_w))
+
+
+def _ema_filter(pos, alpha=_EMA_ALPHA):
+    global _prev_smoothed
+    if _prev_smoothed is None:
+        _prev_smoothed = pos
+        return pos
+    sx = alpha * pos[0] + (1 - alpha) * _prev_smoothed[0]
+    sy = alpha * pos[1] + (1 - alpha) * _prev_smoothed[1]
+    _prev_smoothed = (sx, sy)
+    return _prev_smoothed
+
+
+def _gated(pos, gate_px=_GATE_PX):
+    global _prev_smoothed
+    if _prev_smoothed is None:
+        return True
+    dx = pos[0] - _prev_smoothed[0]
+    dy = pos[1] - _prev_smoothed[1]
+    return (dx * dx + dy * dy) <= gate_px * gate_px
 
 
 def get_stable_laser_point(timeout_ms=15000, stable_count=STABLE_COUNT):
-    own_cam = False
+    global _prev_smoothed
+    _prev_smoothed = None
     try:
-        last_pos = None
+        last_raw = None
         stable = 0
         start = time.ticks_ms()
         cx, cy = WIDTH // 2, HEIGHT // 2
+        track_count = 0
+        skip_count = 0
         while True:
             if abs(time.ticks_diff(time.ticks_ms(), start)) > timeout_ms:
+                _prev_smoothed = None
                 return None
             frame = camera_manager.read_frame()
             if frame is None:
                 time.sleep_ms(10)
                 continue
-            pos_bright = find_brightest_bytes(frame, cx, cy, SEARCH_RADIUS, THRESHOLD, RED_RATIO)
+
+            if track_count > 0 and _prev_smoothed is not None:
+                search_cx = int(_prev_smoothed[0])
+                search_cy = int(_prev_smoothed[1])
+                search_r = TRACK_RADIUS
+            else:
+                search_cx = cx
+                search_cy = cy
+                search_r = SEARCH_RADIUS
+
+            pos_bright = find_brightest_bytes(frame, search_cx, search_cy, search_r, THRESHOLD, RED_RATIO)
             pos = pos_bright
-            print(f"pos:{pos},stable:{stable}")
             if _USE_CV:
                 img_cv = image.image2cv(frame, False, False)
-                pos_ellipse = find_ellipse(img_cv, cx, cy, SEARCH_RADIUS, THRESHOLD, RED_RATIO)
+                pos_ellipse = find_ellipse(img_cv, search_cx, search_cy, search_r, THRESHOLD, RED_RATIO)
                 if pos_ellipse is not None:
                     pos = pos_ellipse
+
             if pos is not None:
-                if last_pos and abs(pos[0] - last_pos[0]) < 1 and abs(pos[1] - last_pos[1]) < 1:
+                skip_count = 0
+                track_count += 1
+                filtered = _ema_filter(pos)
+                if last_raw is not None:
+                    dx = abs(filtered[0] - last_raw[0])
+                    dy = abs(filtered[1] - last_raw[1])
+                    if dx <= 2 and dy <= 2:
                         stable += 1
                     else:
                         stable = 1
-                last_pos = pos
+                else:
+                    stable = 1
+                last_raw = filtered
+                if logger_manager.logger:
+                    logger_manager.logger.info(f"pos:{pos},filtered:{filtered},stable:{stable}")
                 if stable >= stable_count:
-                    return (int(pos[0]), int(pos[1]))
+                    result = (int(filtered[0]), int(filtered[1]))
-            time.sleep_ms(500)
+                    _prev_smoothed = None
+                    return result
+            else:
+                skip_count += 1
+                if logger_manager.logger:
+                    logger_manager.logger.info(f"find_brightest_bytes None, skip={skip_count}, track={track_count}, search_center=({search_cx},{search_cy}), search_r={search_r}")
+                if skip_count > MAX_SKIP_FRAMES:
+                    _prev_smoothed = None
+                    track_count = 0
+                    stable = 0
+                    last_raw = None
+
+            time.sleep_ms(_FRAME_INTERVAL_MS)
     finally:
-        if own_cam:
+        _prev_smoothed = None
-            try:
-                cam.close()
-            except:
-                pass

laser_manager.py

@@ -54,8 +54,8 @@ class LaserManager:
     @property
     def laser_point(self):
         """当前激光点（如果启用硬编码，则返回硬编码值）"""
-        if config.HARDCODE_LASER_POINT:
+        # if config.HARDCODE_LASER_POINT:
-            return config.HARDCODE_LASER_POINT_VALUE
+        #     return config.HARDCODE_LASER_POINT_VALUE
         return self._laser_point
     
     def get_last_frame_with_ellipse(self):

network.py

@@ -8,7 +8,7 @@ import json
 import re
 from math import e
 import struct
-from maix import time
+from maix import time, network, err
 import hmac
 import hashlib
 import ujson
@@ -21,8 +21,7 @@ from hardware import hardware_manager
 from power import get_bus_voltage, voltage_to_percent
 from logger_manager import logger_manager
 from wifi import wifi_manager
-
+import subprocess
-
 
 
 def _wifi_tls_would_block(exc):
@@ -84,7 +83,8 @@ class NetworkManager:
         try:
             import archery_netcore as _netcore
             self._netcore = _netcore
-            if hasattr(self._netcore, "parse_packet") and hasattr(self._netcore, "make_packet") and hasattr(self._netcore, "actions_for_inner_cmd"):
+            if hasattr(self._netcore, "parse_packet") and hasattr(self._netcore, "make_packet") and hasattr(
+                    self._netcore, "actions_for_inner_cmd"):
                 print("[NET] archery_netcore found")
             else:
                 print("[NET] archery_netcore not found parse_packet or make_packet")
@@ -147,7 +147,6 @@ class NetworkManager:
 
     # ==================== 内部状态管理方法 ====================
 
-    
     def set_manual_trigger(self, value=True):
         """设置手动触发标志（公共方法）"""
         self._manual_trigger_flag = value
@@ -203,6 +202,7 @@ class NetworkManager:
 
     def read_device_id(self):
         """从 /device_key 文件读取设备唯一 ID，失败则使用默认值"""
+
         def _set_password_for_device_id(device_id):
             if getattr(config, "USE_TCP_SSL", False):
                 iccid = self.get_4g_mccid()
@@ -242,6 +242,7 @@ class NetworkManager:
         连接 Wi-Fi：委托 ``wifi_manager.connect_wifi``。
         未指定 ``verify_host``/``verify_port`` 时，可达性校验使用本管理器配置的 ``_server_ip``/``_server_port``。
         """
+
         def _verify(ip: str):
             v_host = verify_host if verify_host is not None else self._server_ip
             v_port = verify_port if verify_port is not None else self._server_port
@@ -299,7 +300,10 @@ class NetworkManager:
             if atc is None:
                 return False
 
-            with self.get_uart_lock():
+            if not self._uart4g_lock.acquire(timeout=3000):
+                self.logger.warning("[4G] 获取 uart4g_lock 超时，跳过 4G 可用性检查")
+                return False
+            try:
                 # 1) SIM 就绪
                 r = atc.send("AT+CPIN?", "READY", 3000)
                 if "READY" not in r:
@@ -340,6 +344,8 @@ class NetworkManager:
                 if ip2:
                     return True
                 return False
+            finally:
+                self._uart4g_lock.release()
         except Exception:
             return False
 
@@ -355,8 +361,13 @@ class NetworkManager:
             atc = hardware_manager.at_client
             if atc is None:
                 return None
-            with self.get_uart_lock():
+            if not self._uart4g_lock.acquire(timeout=3000):
+                self.logger.warning("[4G] get_4g_phone_number 获取锁超时")
+                return None
+            try:
                 resp = atc.send("AT+CNUM", "OK", 3000)
+            finally:
+                self._uart4g_lock.release()
             if not resp:
                 return None
             # 可能多行 +CNUM，取第一个非空号码
@@ -379,8 +390,13 @@ class NetworkManager:
             atc = hardware_manager.at_client
             if atc is None:
                 return None
-            with self.get_uart_lock():
+            if not self._uart4g_lock.acquire(timeout=3000):
+                self.logger.warning("[4G] get_4g_mccid 获取锁超时")
+                return None
+            try:
                 resp = atc.send("AT+MCCID", "OK", 3000)
+            finally:
+                self._uart4g_lock.release()
             if not resp or "ERROR" in resp.upper():
                 return None
             m = re.search(r"\+MCCID:\s*(.+)", resp, re.IGNORECASE)
@@ -537,14 +553,84 @@ class NetworkManager:
             self._session_force_4g = False
             return False
 
+    def _cmd200_detect_laser(self):
+        """后台线程执行 cmd200 激光检测，避免阻塞主循环"""
+        from laser_manager import laser_manager
+        try:
+            laser_manager.turn_on_laser()
+            self.logger.info("[LASER] cmd200 已发送开激光指令")
+        except Exception as e:
+            self.logger.warning(f"[LASER] cmd200 开激光异常: {e}")
+
+        try:
+            from laser_detector import get_stable_laser_point
+            time.sleep_ms(500)
+            result = get_stable_laser_point(timeout_ms=60000)
+            if result:
+                x, y = result
+                self.safe_enqueue({
+                    "cmd": 200,
+                    "result": "laser_detect_ok",
+                    "x": x,
+                    "y": y,
+                }, 2)
+                self.logger.info(f"[LASER] cmd200 检测结果: ({x}, {y})")
+            else:
+                self.safe_enqueue({
+                    "cmd": 200,
+                    "result": "laser_detect_failed",
+                }, 2)
+                self.logger.warning("[LASER] cmd200 检测失败")
+        except Exception as e:
+            self.logger.error(f"[LASER] cmd200 检测异常: {e}")
+
+    def _cmd300_ota(self, data_obj):
+        """后台线程执行 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}")
+        ssid = inner_data.get("ssid")
+        password = inner_data.get("password")
+        ota_res_url = inner_data.get("url")
+        try:
+            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": 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 safe_enqueue(self, data_dict, msg_type=2, high=False):
         """线程安全地将消息加入队列（公共方法）"""
         self._enqueue((msg_type, data_dict), high)
 
-
-    
-
     def connect_server(self):
         """
         连接到服务器（自动选择WiFi或4G）
@@ -672,7 +758,10 @@ class NetworkManager:
         host = self._server_ip
         port = getattr(config, "TCP_SSL_PORT", 443) if use_ssl else config.SERVER_PORT
         tail = getattr(config, "MIPOPEN_TAIL", "")
-        with self.get_uart_lock():
+        if not self._uart4g_lock.acquire(timeout=15000):
+            self.logger.warning("[4G-TCP] 连接：获取 uart4g_lock 超时")
+            return False
+        try:
             resp = hardware_manager.at_client.send(f"AT+MIPCLOSE={link_id}", "OK", 1000)
             self.logger.info(f"[4G-TCP] AT+MIPCLOSE={link_id} response: {resp}")
 
@@ -686,6 +775,8 @@ class NetworkManager:
             cmd = f'AT+MIPOPEN={link_id},"TCP","{host}",{port}'
             res = hardware_manager.at_client.send(cmd, "+MIPOPEN", 8000)
             self.logger.info(f"[4G-TCP] {cmd} response: {res}")
+        finally:
+            self._uart4g_lock.release()
         if f"+MIPOPEN: {link_id},0" in res:
             self._tcp_connected = True
             return True
@@ -808,9 +899,13 @@ class NetworkManager:
 
     def _disconnect_tcp_via_4g(self):
         link_id = getattr(config, "TCP_LINK_ID", 0)
-        with self.get_uart_lock():
+        if not self._uart4g_lock.acquire(timeout=2000):
+            self.logger.warning("[4G-TCP] 断开连接：获取 uart4g_lock 超时")
+            return
+        try:
             hardware_manager.at_client.send(f"AT+MIPCLOSE={link_id}", "OK", 1000)
-
+        finally:
+            self._uart4g_lock.release()
 
     def tcp_send_raw(self, data: bytes, max_retries=2) -> bool:
         """
@@ -859,7 +954,7 @@ class NetworkManager:
                             raise
                         if sent == 0:
                             # socket连接已断开
-                            self.logger.warning(f"[WIFI-TCP] 发送失败，socket已断开（尝试 {attempt+1}/{max_retries}）")
+                            self.logger.warning(f"[WIFI-TCP] 发送失败，socket已断开（尝试 {attempt + 1}/{max_retries}）")
                             raise OSError("wifi socket closed (send returned 0)")
                         total_sent += sent
 
@@ -870,7 +965,7 @@ class NetworkManager:
                     time.sleep_ms(50)
 
                 except OSError as e:
-                    self.logger.error(f"[WIFI-TCP] 发送异常: {e}（尝试 {attempt+1}/{max_retries}）")
+                    self.logger.error(f"[WIFI-TCP] 发送异常: {e}（尝试 {attempt + 1}/{max_retries}）")
                     # 发送异常通常意味着连接已不可用，主动关闭以触发重连
                     try:
                         wifi_manager.wifi_socket.close()
@@ -880,7 +975,7 @@ class NetworkManager:
                     self._tcp_connected = False
                     return False
                 except Exception as e:
-                    self.logger.error(f"[WIFI-TCP] 未知错误: {e}（尝试 {attempt+1}/{max_retries}）")
+                    self.logger.error(f"[WIFI-TCP] 未知错误: {e}（尝试 {attempt + 1}/{max_retries}）")
                     try:
                         wifi_manager.wifi_socket.close()
                     except:
@@ -893,7 +988,10 @@ class NetworkManager:
 
     def _tcp_send_raw_via_4g(self, data: bytes, max_retries=2) -> bool:
         link_id = getattr(config, "TCP_LINK_ID", 0)
-        with self.get_uart_lock():
+        if not self._uart4g_lock.acquire(timeout=2000):
+            self.logger.warning("[4G-TCP] 获取 uart4g_lock 超时（其他线程持有），跳过本次发送")
+            return False
+        try:
             for _ in range(max_retries):
                 cmd = f'AT+MIPSEND={link_id},{len(data)}'
                 if ">" not in hardware_manager.at_client.send(cmd, ">", 2000):
@@ -914,6 +1012,8 @@ class NetworkManager:
                     return True
                 time.sleep_ms(50)
             return False
+        finally:
+            self._uart4g_lock.release()
 
     def _configure_ssl_before_connect(self, link_id: int) -> bool:
         """按手册：MSSLCFG(auth) -> (可选) MSSLCERTWR -> MSSLCFG(cert) -> MIPCFG(ssl)"""
@@ -966,7 +1066,6 @@ class NetworkManager:
             r = hardware_manager.at_client.send(f'AT+MSSLCERTRD="{cert_filename}"', "OK", 3000)
             self.logger.info(f"[4G-TCP] AT+MSSLCERTRD=\"{cert_filename}\" response: {r}")
 
-
             # 3) 引用根证书
             r = hardware_manager.at_client.send(f'AT+MSSLCFG="cert",{ssl_id},"{cert_filename}"', "OK", 3000)
             if "OK" not in r:
@@ -1024,7 +1123,8 @@ class NetworkManager:
             self.logger.error(f"[WIFI-TCP] 接收数据异常: {e}")
             return b""
 
-    def _upload_log_file(self, upload_url, wifi_ssid=None, wifi_password=None, include_rotated=True, max_files=None, archive_format="tgz"):
+    def _upload_log_file(self, upload_url, wifi_ssid=None, wifi_password=None, include_rotated=True, max_files=None,
+                         archive_format="tgz"):
         """上传日志文件到指定URL
         
         Args:
@@ -1157,7 +1257,8 @@ class NetworkManager:
                     staged_paths.append(dst)
             except Exception as e:
                 self.logger.error(f"[LOG_UPLOAD] 复制日志快照失败: {e}")
-                self.safe_enqueue({"result": "log_upload_failed", "reason": "snapshot_failed", "detail": str(e)[:100]}, 2)
+                self.safe_enqueue({"result": "log_upload_failed", "reason": "snapshot_failed", "detail": str(e)[:100]},
+                                  2)
                 try:
                     shutil.rmtree(staging_dir)
                 except:
@@ -1185,7 +1286,8 @@ class NetworkManager:
                 self.logger.info(f"[LOG_UPLOAD] 日志压缩包已生成: {archive_path}")
             except Exception as e:
                 self.logger.error(f"[LOG_UPLOAD] 打包压缩失败: {e}")
-                self.safe_enqueue({"result": "log_upload_failed", "reason": "archive_failed", "detail": str(e)[:100]}, 2)
+                self.safe_enqueue({"result": "log_upload_failed", "reason": "archive_failed", "detail": str(e)[:100]},
+                                  2)
                 try:
                     shutil.rmtree(staging_dir)
                 except:
@@ -1234,7 +1336,8 @@ class NetworkManager:
                         "status_code": response.status_code
                     }, 2)
                 else:
-                    self.logger.error(f"[LOG_UPLOAD] 上传失败! 状态码: {response.status_code}, 响应: {response.text[:200]}")
+                    self.logger.error(
+                        f"[LOG_UPLOAD] 上传失败! 状态码: {response.status_code}, 响应: {response.text[:200]}")
                     self.safe_enqueue({
                         "result": "log_upload_failed",
                         "reason": f"http_{response.status_code}",
@@ -1370,7 +1473,8 @@ class NetworkManager:
         except Exception as e:
             return None, f"prepare_exception: {e}"
 
-    def _upload_log_file_v2(self, upload_url, upload_token, key, outlink="", include_rotated=True, max_files=None, archive_format="tgz"):
+    def _upload_log_file_v2(self, upload_url, upload_token, key, outlink="", include_rotated=True, max_files=None,
+                            archive_format="tgz"):
         """上传日志到 Qiniu（支持 WiFi 和 4G 双路径）
 
         流程：准备日志归档 -> 自动检测网络 -> WiFi(requests) 或 4G(AT命令) 上传
@@ -1705,7 +1809,8 @@ class NetworkManager:
 
                             if not logged_in:
                                 try:
-                                    self.logger.debug(f"[TCP] rx link={link_id} len={len(payload)} head={payload[:12].hex()}")
+                                    self.logger.debug(
+                                        f"[TCP] rx link={link_id} len={len(payload)} head={payload[:12].hex()}")
                                 except:
                                     pass
 
@@ -1731,7 +1836,8 @@ class NetworkManager:
                                                     pending_obj = json.load(f)
                                             except:
                                                 pending_obj = {}
-                                            self.safe_enqueue({"result": "ota_ok", "url": pending_obj.get("url", "")}, 2)
+                                            self.safe_enqueue({"result": "ota_ok", "url": pending_obj.get("url", "")},
+                                                              2)
                                             self.logger.info("[OTA] 已上报 ota_ok，等待心跳确认后删除 pending")
                                     except Exception as e:
                                         self.logger.error(f"[OTA] ota_ok 上报失败: {e}")
@@ -1750,7 +1856,8 @@ class NetworkManager:
                                     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):
+                                    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:
@@ -1767,7 +1874,7 @@ class NetworkManager:
                                             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.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:
@@ -1784,7 +1891,8 @@ class NetworkManager:
                                     # 验证必需字段
                                     if not upload_url or not upload_token or not shoot_id:
                                         self.logger.error("[IMAGE_UPLOAD] 缺少必需参数: uploadUrl, token 或 shootId")
-                                        self.safe_enqueue({"result": "image_upload_failed", "reason": "missing_params"}, 2)
+                                        self.safe_enqueue({"result": "image_upload_failed", "reason": "missing_params"},
+                                                          2)
                                     else:
                                         self.logger.info(f"[IMAGE_UPLOAD] 收到图片上传命令，shootId: {shoot_id}")
                                         # 查找文件名中包含 shoot_id 的图片文件（文件名格式：shot_{shoot_id}_*.bmp）
@@ -1805,15 +1913,19 @@ class NetworkManager:
                                                         reverse=True
                                                     )
                                                     target_image = os.path.join(photo_dir, matched_images[0])
-                                                    self.logger.info(f"[IMAGE_UPLOAD] 找到匹配shootId的图片: {matched_images[0]}")
+                                                    self.logger.info(
+                                                        f"[IMAGE_UPLOAD] 找到匹配shootId的图片: {matched_images[0]}")
                                                 else:
-                                                    self.logger.warning(f"[IMAGE_UPLOAD] 未找到包含shootId={shoot_id}的图片文件")
+                                                    self.logger.warning(
+                                                        f"[IMAGE_UPLOAD] 未找到包含shootId={shoot_id}的图片文件")
                                         except Exception as e:
                                             self.logger.error(f"[IMAGE_UPLOAD] 查找图片失败: {e}")
 
                                         if not target_image:
                                             self.logger.error(f"[IMAGE_UPLOAD] 未找到shootId={shoot_id}对应的图片文件")
-                                            self.safe_enqueue({"result": "image_upload_failed", "reason": "no_image_found", "shootId": shoot_id}, 2)
+                                            self.safe_enqueue(
+                                                {"result": "image_upload_failed", "reason": "no_image_found",
+                                                 "shootId": shoot_id}, 2)
                                         else:
                                             # 构建上传key
                                             ext = os.path.splitext(target_image)[1].lower()
@@ -1845,14 +1957,16 @@ class NetworkManager:
                                     # 验证必需字段
                                     if not upload_url or not upload_token or not key:
                                         self.logger.error("[LOG_UPLOAD] 缺少必需参数: uploadUrl, token 或 key")
-                                        self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_params"}, 2)
+                                        self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_params"},
+                                                          2)
                                     else:
                                         self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，key: {key}")
                                         # 在新线程中执行上传，避免阻塞主循环
                                         import _thread
                                         _thread.start_new_thread(
                                             self._upload_log_file_v2,
-                                            (upload_url, upload_token, key, outlink, include_rotated, max_files, archive_format)
+                                            (upload_url, upload_token, key, outlink, include_rotated, max_files,
+                                             archive_format)
                                         )
                                         # 立即返回已入队确认
                                         self.safe_enqueue({"result": "log_upload_queued"}, 2)
@@ -1952,10 +2066,12 @@ class NetworkManager:
                                                 # 只有同时满足：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)")
+                                                    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)")
+                                                    self.logger.info(
+                                                        "ota auto-selected: 4g (WiFi not available or no credentials)")
 
                                         hardware_manager.stop_idle_timer()  # 停表，注意OTA停表之后，就没有再开表，因为OTA后面会重启，会重新开表
 
@@ -1971,10 +2087,12 @@ class NetworkManager:
                                                 self.logger.info(f"ssid: {ssid}")
                                                 self.logger.info(f"password: {password}")
                                                 ota_manager._start_update_thread()
-                                                _thread.start_new_thread(ota_manager.handle_wifi_and_update, (ssid, password, ota_url))
+                                                _thread.start_new_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 = 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"
@@ -1982,11 +2100,13 @@ class NetworkManager:
                                     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)
+                                        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)
+                                        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
@@ -2023,50 +2143,25 @@ class NetworkManager:
 
                                         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}")
                                             # 在新线程中执行上传，避免阻塞主循环
                                             import _thread
                                             _thread.start_new_thread(
                                                 self._upload_log_file,
-                                                (upload_url, wifi_ssid, wifi_password, include_rotated, max_files, archive_format)
+                                                (upload_url, wifi_ssid, wifi_password, include_rotated, max_files,
+                                                 archive_format)
                                             )
                                     elif inner_cmd == 200:
-                                        from laser_manager import laser_manager
+                                        self.logger.info("[LASER] cmd200 在后台线程执行检测")
-                                        try:
+                                        import _thread
-                                            laser_manager.turn_on_laser()
+                                        _thread.start_new_thread(self._cmd200_detect_laser, ())
-                                            if self.logger:
+                                    elif inner_cmd == 300:
-                                                self.logger.info("[LASER] cmd200 已发送开激光指令")
+                                        self.logger.info("[New Ota] cmd300 在后台线程执行OTA")
-                                        except Exception as e:
+                                        import _thread
-                                            if self.logger:
+                                        _thread.start_new_thread(self._cmd300_ota, (data_obj,))
-                                                self.logger.warning(f"[LASER] cmd200 开激光异常: {e}")
-                                        try:
-                                            from laser_detector import get_stable_laser_point
-                                            time.sleep_ms(500)
-                                            result = get_stable_laser_point(timeout_ms=60000)
-                                            if result:
-                                                x, y = result
-                                                self.safe_enqueue({
-                                                    "cmd": 200,
-                                                    "result": "laser_detect_ok",
-                                                    "x": x,
-                                                    "y": y,
-                                                }, 2)
-                                                if self.logger:
-                                                    self.logger.info(f"[LASER] cmd200 检测结果: ({x}, {y})")
-                                            else:
-                                                self.safe_enqueue({
-                                                    "cmd": 200,
-                                                    "result": "laser_detect_failed",
-                                                }, 2)
-                                                if self.logger:
-                                                    self.logger.warning("[LASER] cmd200 检测失败")
-                                        except Exception as e:
-                                            if self.logger:
-                                                self.logger.error(f"[LASER] cmd200 检测异常: {e}")
-                                        time.sleep_ms(500)
-
                                 else:  # data的结构不是 dict
                                     self.logger.info(f"[NET] body={body}, {time.time()}")
                             else:
@@ -2120,7 +2215,8 @@ class NetworkManager:
                     current_time = time.ticks_ms()
                     if logged_in and current_time - last_heartbeat_send_time > config.HEARTBEAT_INTERVAL * 1000:
                         vol_val = get_bus_voltage()
-                        if not self.tcp_send_raw(self._netcore.make_packet(4, {"vol": vol_val, "vol_per": voltage_to_percent(vol_val)})):
+                        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
                             # 短暂波动可能导致一次发送失败：连续失败达到阈值才重连，避免重连风暴
@@ -2180,38 +2276,44 @@ class NetworkManager:
 # 创建全局单例实例
 network_manager = NetworkManager()
 
+
 # ==================== 向后兼容的函数接口 ====================
 
 def tcp_main():
     """TCP主循环（向后兼容接口）"""
     return network_manager.tcp_main()
 
+
 def read_device_id():
     """读取设备ID（向后兼容接口）"""
     return network_manager.read_device_id()
 
+
 def safe_enqueue(data_dict, msg_type=2, high=False):
     """线程安全地加入队列（向后兼容接口）"""
     return network_manager.safe_enqueue(data_dict, msg_type, high)
 
+
 def connect_server():
     """连接服务器（向后兼容接口）"""
     return network_manager.connect_server()
 
+
 def disconnet_server():
     """断开服务器连接（向后兼容接口）"""
     return network_manager.disconnect_server()
 
+
 def is_wifi_connected():
     """检查WiFi是否已连接（向后兼容接口）"""
     return network_manager.is_wifi_connected()
 
+
 def connect_wifi(ssid, password):
     """连接WiFi（向后兼容接口）"""
     return network_manager.connect_wifi(ssid, password)
 
+
 def is_server_reachable(host, port=80, timeout=5):
     """检查服务器是否可达（向后兼容接口）"""
     return network_manager.is_server_reachable(host, port, timeout)
-
-

ota_curl.sh

@@ -0,0 +1,57 @@
+#!/bin/sh
+# OTA 更新脚本 - 使用 curl 断点下载
+# 用法: sh ota_curl.sh <下载URL>
+# 示例: sh ota_curl.sh http://example.com/maix-t11-v2.15.1.zip
+
+set -e
+
+APP_DIR="/maixapp/apps/t11"
+BACKUP_BASE="$APP_DIR/backups"
+TMP_DIR="/tmp/ota_curl"
+PENDING_FILE="$APP_DIR/ota_pending.json"
+
+if [ $# -lt 1 ]; then
+    echo "用法: $0 <下载URL>"
+    exit 1
+fi
+
+OTA_URL="$1"
+FILENAME=$(basename "$OTA_URL" | sed 's/?.*//')
+[ -z "$FILENAME" ] && FILENAME="update.zip"
+
+mkdir -p "$TMP_DIR" "$BACKUP_BASE"
+
+# 1. 断点下载
+echo "[OTA] 开始下载: $OTA_URL"
+echo "[OTA] 保存到: $TMP_DIR/$FILENAME"
+curl -C - -L --retry 3 --retry-delay 5 -o "$TMP_DIR/$FILENAME" "$OTA_URL"
+echo "[OTA] 下载完成"
+
+# 2. 备份当前目录
+TIMESTAMP=$(date +%Y%m%d_%H%M%S 2>/dev/null || echo "00000000_000000")
+BACKUP_DIR="$BACKUP_BASE/backup_$TIMESTAMP"
+mkdir -p "$BACKUP_DIR"
+echo "[OTA] 备份到: $BACKUP_DIR"
+for f in "$APP_DIR"/*.py "$APP_DIR"/*.json "$APP_DIR"/*.xml "$APP_DIR"/*.yaml "$APP_DIR"/*.pem "$APP_DIR"/*.mud "$APP_DIR"/*.so "$APP_DIR"/S99archery; do
+    [ -f "$f" ] && cp "$f" "$BACKUP_DIR/"
+done
+echo "[OTA] 备份完成"
+
+# 3. 解压并替换文件
+echo "[OTA] 开始更新..."
+if echo "$FILENAME" | grep -qi '\.zip$'; then
+    unzip -q -o "$TMP_DIR/$FILENAME" -d "$APP_DIR/"
+else
+    cp "$TMP_DIR/$FILENAME" "$APP_DIR/"
+fi
+sync
+
+# 4. 写入 pending 文件（用于崩溃恢复）
+echo '{"ts":0,"url":"'"$OTA_URL"'","backup_dir":"'"$BACKUP_DIR"'","restart_count":0,"max_restarts":3}' > "$PENDING_FILE"
+sync
+
+echo "[OTA] 更新完成，准备重启..."
+
+# 5. 重启
+sleep 1
+reboot

test/test_decect.py

@@ -0,0 +1,330 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+离线测试脚本：直接复用 detect_circle 逻辑进行测试
+运行环境：MaixPy (Sipeed MAIX)
+"""
+import sys
+import os
+# import time
+from maix import image, time
+import cv2
+import numpy as np
+import math
+
+# ==================== 全局配置 (与 test_main.py 保持一致) ====================
+REAL_RADIUS_CM = 20  # 靶心实际半径（厘米）
+
+def detect_circle_v3(frame, laser_point=None, img_cv=None):
+    """检测图像中的靶心（优先清晰轮廓，其次黄色区域）- 返回椭圆参数版本
+    增加红色圆圈检测，验证黄色圆圈是否为真正的靶心
+    如果提供 laser_point，会选择最接近激光点的目标
+    优化：
+      1. 缩图到 MAX_DET_DIM 后再做 HSV/形态学，最长边 640->320 可获得 ~4x 加速
+      2. 红色掩码在黄色轮廓循环外只计算一次，避免 N 次重复计算
+      3. img_cv 可由外部传入（与其他线程共享转换结果），为 None 时自动转换
+    Args:
+        frame: 图像帧（img_cv 为 None 时使用）
+        laser_point: 激光点坐标 (x, y)，用于多目标场景下的目标选择
+        img_cv: 已转换的 numpy BGR/RGB 图像；不为 None 时跳过 image2cv 转换
+    Returns:
+        (result_img, best_center, best_radius, method, best_radius1, ellipse_params)
+    """
+    if img_cv is None:
+        img_cv = image.image2cv(frame, False, False)
+    from datetime import datetime
+    print(f"[detect_circle_v3] begin {datetime.now()}")
+    # -- 1. 缩图加速（与三角形路径保持一致）
+    h_orig, w_orig = img_cv.shape[:2]
+    MAX_DET_DIM = 480
+    long_side = max(h_orig, w_orig)
+    if long_side > MAX_DET_DIM:
+        det_scale = MAX_DET_DIM / long_side
+        img_det = cv2.resize(img_cv, (int(w_orig * det_scale), int(h_orig * det_scale)),
+                             interpolation=cv2.INTER_LINEAR)
+        inv_scale = 1.0 / det_scale  # 检测坐标 -> 原始坐标的倍率
+    else:
+        img_det = img_cv
+        inv_scale = 1.0
+
+    # 激光点映射到检测分辨率
+    lp_det = None
+    if laser_point is not None:
+        lp_det = (laser_point[0] / inv_scale, laser_point[1] / inv_scale)
+    best_center = best_radius = best_radius1 = method = None
+    ellipse_params = None
+
+    print(f"[detect_circle_v3] step 1 fin {datetime.now()}")
+
+    # -- 2. HSV + 黄色掩码
+    hsv = cv2.cvtColor(img_det, cv2.COLOR_RGB2HSV)
+    h, s, v = cv2.split(hsv)
+    s = np.clip(s * 1.1, 0, 255).astype(np.uint8)
+    hsv = cv2.merge((h, s, v))
+    lower_yellow = np.array([7, 80, 0])
+    upper_yellow = np.array([32, 255, 255])
+    mask_yellow = cv2.inRange(hsv, lower_yellow, upper_yellow)
+    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+    mask_yellow = cv2.morphologyEx(mask_yellow, cv2.MORPH_CLOSE, kernel)
+
+    print(f"[detect_circle_v3] step 2 fin {datetime.now()}")
+
+    # -- 3. 红色掩码：在循环外只算一次
+    mask_red = cv2.bitwise_or(
+        cv2.inRange(hsv, np.array([0, 50, 40]), np.array([10, 255, 255])),
+        cv2.inRange(hsv, np.array([170, 50, 40]), np.array([180, 255, 255])),
+    )
+    kernel_red = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+    mask_red = cv2.morphologyEx(mask_red, cv2.MORPH_CLOSE, kernel_red)
+    contours_red, _ = cv2.findContours(mask_red, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    # 预先把红色轮廓筛选成 (center, radius) 列表，后续直接查表
+    red_candidates = []
+    for cnt_r in contours_red:
+        ar = cv2.contourArea(cnt_r)
+        if ar <= 10:
+            continue
+        pr = cv2.arcLength(cnt_r, True)
+        if pr <= 0 or (4 * np.pi * ar) / (pr * pr) <= 0.3:
+            continue
+        if len(cnt_r) >= 5:
+            (xr, yr), (wr, hr), _ = cv2.fitEllipse(cnt_r)
+            red_candidates.append({"center": (int(xr), int(yr)), "radius": int(min(wr, hr) / 2)})
+        else:
+            (xr, yr), rr = cv2.minEnclosingCircle(cnt_r)
+            red_candidates.append({"center": (int(xr), int(yr)), "radius": int(rr)})
+
+    print(f"[detect_circle_v3] step 3 fin {datetime.now()}")
+
+    # -- 4. 黄色轮廓循环（复用上面的红色候选列表）
+    contours_yellow, _ = cv2.findContours(mask_yellow, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    valid_targets = []
+    for cnt_yellow in contours_yellow:
+        area = cv2.contourArea(cnt_yellow)
+        if area <= 15:
+            continue
+        perimeter = cv2.arcLength(cnt_yellow, True)
+        if perimeter <= 0:
+            continue
+        circularity = (4 * np.pi * area) / (perimeter * perimeter)
+        if circularity <= 0.5:
+            continue
+        print(f"[target] -> 面积:{area:.1f}, 圆度:{circularity:.2f}")
+        if len(cnt_yellow) >= 5:
+            (x, y), (width, height), angle = cv2.fitEllipse(cnt_yellow)
+            yellow_ellipse = ((x, y), (width, height), angle)
+            yellow_center = (int(x), int(y))
+            yellow_radius = int(min(width, height) / 2)
+        else:
+            (x, y), radius = cv2.minEnclosingCircle(cnt_yellow)
+            yellow_center = (int(x), int(y))
+            yellow_radius = int(radius)
+            yellow_ellipse = None
+        # 在预筛好的红色候选中匹配
+        matched = False
+        for rc in red_candidates:
+            ddx = yellow_center[0] - rc["center"][0]
+            ddy = yellow_center[1] - rc["center"][1]
+            dist_centers = math.hypot(ddx, ddy)
+            if dist_centers < yellow_radius * 1.5 and rc["radius"] > yellow_radius * 0.7:
+                print(f"[target] -> 找到匹配的红圈: 黄心({yellow_center}), "
+                                f"红心({rc['center']}), 距离:{dist_centers:.1f}, "
+                                f"黄半径:{yellow_radius}, 红半径:{rc['radius']}")
+                valid_targets.append({
+                    "center": yellow_center,
+                    "radius": yellow_radius,
+                    "ellipse": yellow_ellipse,
+                    "area": area,
+                })
+                matched = True
+                break
+        if not matched :
+            print("Debug -> 未找到匹配的红色圆圈，可能是误识别")
+
+    print(f"[detect_circle_v3] step 4 fin {datetime.now()}")
+
+    # -- 5. 选最佳目标，坐标还原到原始分辨率
+    if valid_targets:
+        if lp_det:
+            best_target = min(valid_targets,
+                              key=lambda t: (t["center"][0] - lp_det[0]) ** 2
+                                            + (t["center"][1] - lp_det[1]) ** 2)
+            method = "v3_ellipse_red_validated_laser_selected"
+        else:
+            best_target = max(valid_targets, key=lambda t: t["area"])
+            method = "v3_ellipse_red_validated"
+        bc = best_target["center"]
+        br = best_target["radius"]
+        be = best_target["ellipse"]
+        if inv_scale != 1.0:
+            best_center = (int(bc[0] * inv_scale), int(bc[1] * inv_scale))
+            best_radius = int(br * inv_scale)
+            if be is not None:
+                (ex, ey), (ew, eh), ea = be
+                be = ((ex * inv_scale, ey * inv_scale),
+                      (ew * inv_scale, eh * inv_scale), ea)
+        else:
+            best_center = bc
+            best_radius = br
+        ellipse_params = be
+        best_radius1 = best_radius * 5
+    result_img = image.cv2image(img_cv, False, False)
+    print(f"[detect_circle_v3] step 5 fin {datetime.now()}")
+    return result_img, best_center, best_radius, method, best_radius1, ellipse_params
+
+
+def run_offline_test(image_path):
+    """读取图片，检测圆，绘制结果，保存图片"""
+
+    # 1. 检查文件是否存在
+    if not os.path.exists(image_path):
+        print(f"[ERROR] 找不到图片文件: {image_path}")
+        return
+
+    # 2. 使用 maix.image 读取图片 (适配 MaixPy v4)
+    try:
+        # 使用 image.load 读取文件，返回 Image 对象
+        img = image.load(image_path)
+        print(f"[INFO] 成功读取图片: {image_path} (尺寸: {img.width()}x{img.height()})")
+    except Exception as e:
+        print(f"[ERROR] 读取图片失败: {e}")
+        print("提示：请确认 MaixPy 版本是否为 v4，且图片路径正确。")
+        return
+
+    # 3. 调用 detect_circle_v3 函数
+    print("[INFO] 正在调用 detect_circle_v3 进行检测...")
+    start_time = time.ticks_ms()
+
+    result_img, center, radius, method, radius1, ellipse_params = detect_circle_v3(img)
+
+    cost_time = time.ticks_ms() - start_time
+    print(f"[INFO] 检测完成，耗时: {cost_time}ms")
+    print(f"       结果 -> 圆心: {center}, 半径: {radius}, 方法: {method}")
+    if ellipse_params:
+        (ell_center, (width, height), angle) = ellipse_params
+        print(
+            f"       椭圆 -> 中心: ({ell_center[0]:.1f}, {ell_center[1]:.1f}), 长轴: {max(width, height):.1f}, 短轴: {min(width, height):.1f}, 角度: {angle:.1f}°")
+
+    # 4. 绘制辅助线（可选，用于调试）
+    if center and radius:
+        # 为了绘制椭圆，需要转换回 cv2 图像
+        img_cv = image.image2cv(result_img, False, False)
+
+        cx, cy = center
+
+        # 如果有椭圆参数，绘制椭圆
+        if ellipse_params:
+            (ell_center, (width, height), angle) = ellipse_params
+            cx_ell, cy_ell = int(ell_center[0]), int(ell_center[1])
+
+            # 确定长轴和短轴
+            if width >= height:
+                # width 是长轴，height 是短轴
+                axes_major = width
+                axes_minor = height
+                major_angle = angle  # 长轴角度就是 angle
+                minor_angle = angle + 90  # 短轴角度 = 长轴角度 + 90度
+            else:
+                # height 是长轴，width 是短轴
+                axes_major = height
+                axes_minor = width
+                major_angle = angle + 90  # 长轴角度 = width角度 + 90度
+                minor_angle = angle  # 短轴角度就是 angle
+
+            # 使用 OpenCV 绘制椭圆（绿色，线宽2）
+            cv2.ellipse(img_cv,
+                        (cx_ell, cy_ell),  # 中心点
+                        (int(width / 2), int(height / 2)),  # 半宽、半高
+                        angle,  # 旋转角度（OpenCV需要原始angle）
+                        0, 360,  # 起始和结束角度
+                        (0, 255, 0),  # 绿色 (RGB格式)
+                        2)  # 线宽
+
+            # 绘制椭圆中心点（红色）
+            cv2.circle(img_cv, (cx_ell, cy_ell), 3, (255, 0, 0), -1)
+
+            import math
+            # 绘制短轴（蓝色线条）
+            minor_length = axes_minor / 2
+            minor_angle_rad = math.radians(minor_angle)
+            dx_minor = minor_length * math.cos(minor_angle_rad)
+            dy_minor = minor_length * math.sin(minor_angle_rad)
+            pt1_minor = (int(cx_ell - dx_minor), int(cy_ell - dy_minor))
+            pt2_minor = (int(cx_ell + dx_minor), int(cy_ell + dy_minor))
+            cv2.line(img_cv, pt1_minor, pt2_minor, (0, 0, 255), 2)  # 蓝色 (RGB格式)
+        else:
+            # 如果没有椭圆参数，绘制圆形（红色）
+            cv2.circle(img_cv, (cx, cy), radius, (0, 0, 255), 2)
+            cv2.circle(img_cv, (cx, cy), 2, (0, 0, 255), -1)
+
+        # 转换回 maix image
+        result_img = image.cv2image(img_cv, False, False)
+
+        # 定义颜色对象用于文字
+        try:
+            color_black = image.Color.from_rgb(0, 0, 0)
+        except AttributeError:
+            color_black = image.Color(0, 0, 0)
+
+        # D. 添加文字信息
+        FOCAL_LENGTH_PIX = 1900
+        d = (REAL_RADIUS_CM * FOCAL_LENGTH_PIX) / radius1 / 100.0
+        info_str = f"R:{radius} M:{method} D:{d:.2f}"
+        print(info_str)
+
+        # 计算文字位置，防止超出图片边界
+        r_outer = int(radius * 11.0) if radius else 100
+        text_y = cy - r_outer - 20 if cy > r_outer + 20 else cy + r_outer + 20
+
+        # 调用 draw_string
+        result_img.draw_string(0, 0, info_str, color=color_black, scale=1.0)
+
+    # 5. 保存结果图片
+    base, ext = os.path.splitext(image_path)
+    output_path = f"{base}_result{ext}"
+    try:
+        result_img.save(output_path, quality=100)
+        print(f"[SUCCESS] 结果已保存至: {output_path}")
+    except Exception as e:
+        print(f"[ERROR] 保存图片失败: {e}")
+
+
+if __name__ == "__main__":
+    # ================= 配置区域 =================
+
+    # 1. 设置要测试的图片路径
+    # 建议将图片放在与脚本同级目录，或者使用绝对路径
+    TARGET_IMAGE = "/root/phot/None_314_258_0_0041.bmp"
+
+    TARGET_DIR = "/root/phot"  # 修改为你想要读取的目录路径
+
+    # 支持的图片格式
+    IMAGE_EXTENSIONS = ['.jpg', '.jpeg', '.png', '.bmp']
+
+    # ================= 执行区域 =================
+    if 'TARGET_DIR' in locals():
+        # 读取目录下所有图片文件，过滤掉 _result.jpg 后缀的文件
+        image_files = []
+        if os.path.exists(TARGET_DIR) and os.path.isdir(TARGET_DIR):
+            for filename in os.listdir(TARGET_DIR):
+                # 检查文件扩展名
+                if any(filename.lower().endswith(ext) for ext in IMAGE_EXTENSIONS):
+                    # 过滤掉 _result.jpg 后缀的文件
+                    if not filename.endswith('_result.jpg'):
+                        filepath = os.path.join(TARGET_DIR, filename)
+                        if os.path.isfile(filepath):
+                            image_files.append(filepath)
+
+            # 按文件名排序（可选）
+            image_files.sort()
+
+            print(f"[INFO] 在目录 {TARGET_DIR} 中找到 {len(image_files)} 张图片")
+
+            # 处理每张图片
+            for img_path in image_files:
+                print(f"\n{'=' * 10} 开始处理: {img_path} {'=' * 10}")
+                run_offline_test(img_path)
+        else:
+            print(f"[ERROR] 目录不存在或不是有效目录: {TARGET_DIR}")
+
+    else:
+        run_offline_test(TARGET_IMAGE)

version.md

@@ -0,0 +1,22 @@
+# 1.2.0 开始使用C++编译成.so，替换部分代码
+# 1.2.1 ota使用加密包
+# 1.2.2 支持wifi ota，并且设定时区，并使用单独线程保存图片
+# 1.2.3 修改ADC_TRIGGER_THRESHOLD 为2300，支持上传日志到服务器
+# 1.2.4 修改ADC_TRIGGER_THRESHOLD 为3000，并默认关闭摄像头的显示，并把ADC的采样间隔从50ms降低到10ms
+# 1.2.5 支持空气传感器采样，并默认关闭日志。优化断网时的发送队列丢消息问题，解决 WiFi 断线检测不可靠问题。
+# 1.2.6 在链接 wifi 前先判断 wifi 的可用性，假如不可用，则不落盘。增加日志批量压缩上传功能
+# 1.2.7 修复OTA失败的bug, 空气压力传感器的阈值是2500
+# 1.2.8 （1） 加快 wifi 下数据传输的速度。（2） 调整射箭时处理的逻辑，优先上报数据，再存照片之类的操作。（3）假如是用户打开激光的，射箭触发后不再关闭激光，因为是调瞄阶段
+# 1.2.9 增加电源板的控制和自动关机的功能
+# 1.2.10 config formal
+# 1.2.11 增加三角形的单应性算法，适配对应的靶纸
+# 1.2.110 关掉了黑色三角形算法，只用于测试
+# 1.2.13 修改wifi连接
+# 1.2.14 修改了icc登录部分
+# 2.15.3 新版本ota，去除ai算环数方法
+# 2.15.4 更新版本号
+# 2.15.5 打印ota进度
+# 2.15.6 更新版本号
+# 2.15.7 更新版本号
+# 2.15.8 启动不加载预加载yolo
+# 2.15.9 20cm

version.py

@@ -4,28 +4,6 @@
 应用版本号
 每次 OTA 更新时，只需要更新这个文件中的版本号
 """
-VERSION = '2.14.1'
+VERSION = '2.15.9'
-
-
-# 1.2.0 开始使用C++编译成.so，替换部分代码
-# 1.2.1 ota使用加密包
-# 1.2.2 支持wifi ota，并且设定时区，并使用单独线程保存图片
-# 1.2.3 修改ADC_TRIGGER_THRESHOLD 为2300，支持上传日志到服务器
-# 1.2.4 修改ADC_TRIGGER_THRESHOLD 为3000，并默认关闭摄像头的显示，并把ADC的采样间隔从50ms降低到10ms
-# 1.2.5 支持空气传感器采样，并默认关闭日志。优化断网时的发送队列丢消息问题，解决 WiFi 断线检测不可靠问题。
-# 1.2.6 在链接 wifi 前先判断 wifi 的可用性，假如不可用，则不落盘。增加日志批量压缩上传功能
-# 1.2.7 修复OTA失败的bug, 空气压力传感器的阈值是2500
-# 1.2.8 （1） 加快 wifi 下数据传输的速度。（2） 调整射箭时处理的逻辑，优先上报数据，再存照片之类的操作。（3）假如是用户打开激光的，射箭触发后不再关闭激光，因为是调瞄阶段
-# 1.2.9 增加电源板的控制和自动关机的功能
-# 1.2.10 config formal
-# 1.2.11 增加三角形的单应性算法，适配对应的靶纸
-# 1.2.110 关掉了黑色三角形算法，只用于测试
-# 1.2.13 修改wifi连接
-# 1.2.14 修改了icc登录部分
-
-
-
-
-
 
 

vision.py

@@ -535,7 +535,7 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
     logger.debug(f"[detect_circle_v3] begin {datetime.now()}")
     # -- 1. 缩图加速（与三角形路径保持一致）
     h_orig, w_orig = img_cv.shape[:2]
-    MAX_DET_DIM = 320
+    MAX_DET_DIM = 480
     long_side = max(h_orig, w_orig)
     if long_side > MAX_DET_DIM:
         det_scale = MAX_DET_DIM / long_side
@@ -570,8 +570,8 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
 
     # -- 3. 红色掩码：在循环外只算一次
     mask_red = cv2.bitwise_or(
-        cv2.inRange(hsv, np.array([0, 80, 0]),   np.array([10,  255, 255])),
+        cv2.inRange(hsv, np.array([0, 50, 40]),   np.array([10,  255, 255])),
-        cv2.inRange(hsv, np.array([170, 80, 0]), np.array([180, 255, 255])),
+        cv2.inRange(hsv, np.array([170, 50, 40]), np.array([180, 255, 255])),
     )
     kernel_red = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
     mask_red = cv2.morphologyEx(mask_red, cv2.MORPH_CLOSE, kernel_red)
@@ -580,10 +580,10 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
     red_candidates = []
     for cnt_r in contours_red:
         ar = cv2.contourArea(cnt_r)
-        if ar <= 50:
+        if ar <= 10:
             continue
         pr = cv2.arcLength(cnt_r, True)
-        if pr <= 0 or (4 * np.pi * ar) / (pr * pr) <= 0.6:
+        if pr <= 0 or (4 * np.pi * ar) / (pr * pr) <= 0.3:
             continue
         if len(cnt_r) >= 5:
             (xr, yr), (wr, hr), _ = cv2.fitEllipse(cnt_r)
@@ -599,13 +599,13 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
     valid_targets = []
     for cnt_yellow in contours_yellow:
         area = cv2.contourArea(cnt_yellow)
-        if area <= 50:
+        if area <= 15:
             continue
         perimeter = cv2.arcLength(cnt_yellow, True)
         if perimeter <= 0:
             continue
         circularity = (4 * np.pi * area) / (perimeter * perimeter)
-        if circularity <= 0.7:
+        if circularity <= 0.5:
             continue
         if logger:
             logger.info(f"[target] -> 面积:{area:.1f}, 圆度:{circularity:.2f}")
@@ -625,7 +625,7 @@ def detect_circle_v3(frame, laser_point=None, img_cv=None):
             ddx = yellow_center[0] - rc["center"][0]
             ddy = yellow_center[1] - rc["center"][1]
             dist_centers = math.hypot(ddx, ddy)
-            if dist_centers < yellow_radius * 1.5 and rc["radius"] > yellow_radius * 0.8:
+            if dist_centers < yellow_radius * 1.5 and rc["radius"] > yellow_radius * 0.7:
                 if logger:
                     logger.info(f"[target] -> 找到匹配的红圈: 黄心({yellow_center}), "
                                 f"红心({rc['center']}), 距离:{dist_centers:.1f}, "