yolo

.gitattributes

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

.gitignore

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

.vscode/settings.json

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

4g.py

@@ -109,7 +109,6 @@
 from maix import app, uart, pinmap, time
 import hashlib
 import hmac
-import re
 import ujson
 
 # ========== 配置 ==========
@@ -131,109 +130,53 @@ def generate_token(device_id):
     return "Arrow_" + hmac.new((SALT + device_id).encode(), SALT2.encode(), hashlib.sha256).hexdigest()
 
 def send_cmd(cmd_str, timeout_ms=3000):
-    """发送 AT 指令并返回完整响应；超时返回已收到的内容。"""
+    """发送 AT 指令并等待 OK / ERROR"""
     print("[AT] =>", cmd_str)
     http_serial.write((cmd_str + "\r\n").encode())
     buffer = b""
     start = time.ticks_ms()
-    while time.ticks_diff(time.ticks_ms(), start) < timeout_ms:
+    while time.ticks_ms() - start < timeout_ms:
         data = http_serial.read(128)
         if data:
             buffer += data
             try:
-                decoded = buffer.decode("utf-8", "ignore")
-                if "OK" in decoded or "+CME ERROR" in decoded or "ERROR" in decoded:
-                    print("[AT] <=", decoded.strip())
-                    return decoded
+                decoded = buffer.decode()
+                print("<= ", decoded.strip())
+                if "OK" in decoded:
+                    return True
+                if "+CME ERROR" in decoded or "ERROR" in decoded:
+                    return False
             except:
                 pass
         time.sleep_ms(10)
-    decoded = buffer.decode("utf-8", "ignore")
-    print("[AT] !! timeout", timeout_ms, "ms, response:", decoded.strip() or "<empty>")
-    return decoded
-
-
-def response_ok(response):
-    return "OK" in response and "ERROR" not in response
-
-
-def wait_modem_ready():
-    """等待模组响应，并确认 PDP 上下文已经获得 IP。"""
-    for attempt in range(15):
-        if response_ok(send_cmd("AT", 1000)):
-            break
-        print("[4G] 等待模组启动", attempt + 1, "/15")
-        time.sleep_ms(1000)
-    else:
-        print("[4G] UART2 无 AT 响应，请检查模组供电、A28/A29 接线和串口占用")
-        return False
-
-    send_cmd("ATE0", 1000)
-    cpin = send_cmd("AT+CPIN?", 3000)
-    if "READY" not in cpin:
-        print("[4G] SIM 卡未就绪:", cpin.strip())
-        return False
-
-    addr = send_cmd("AT+CGPADDR=1", 3000)
-    match = re.search(r'\+CGPADDR:\s*1,"([^\"]+)"', addr)
-    if match and match.group(1) != "0.0.0.0":
-        print("[4G] PDP ready, IP:", match.group(1))
-        return True
-
-    send_cmd("AT+MIPCALL=1,1", 15000)
-    for _ in range(20):
-        addr = send_cmd("AT+CGPADDR=1", 3000)
-        match = re.search(r'\+CGPADDR:\s*1,"([^\"]+)"', addr)
-        if match and match.group(1) != "0.0.0.0":
-            print("[4G] PDP ready, IP:", match.group(1))
-            return True
-        time.sleep_ms(1000)
-
-    print("[4G] PDP 未获得 IP，请检查 SIM 流量、信号和 APN")
     return False
 
-
-def clear_http_instances():
-    for instance_id in range(6):
-        send_cmd(f"AT+MHTTPDEL={instance_id}", 1200)
-
 def create_http_instance(url):
     cmd = f'AT+MHTTPCREATE="{url}"'
-    response = send_cmd(cmd, 8000)
-    match = re.search(r"\+MHTTPCREATE:\s*(\d+)", response)
-    if not response_ok(response) or not match:
-        print("❌ 创建 HTTP 实例失败，模组响应:", response.strip() or "<empty>")
-        return None
-    return int(match.group(1))
+    if send_cmd(cmd):
+        # 尝试提取 instance ID（如果模块返回）
+        # 注意：部分模块不会返回 ID，可忽略，直接用 0 或 1
+        return True
+    return False
 
 def send_http_request(url, api_path, token, device_id, json_data):
     # 1. 创建 HTTP 实例
-    instance_id = create_http_instance(url)
-    if instance_id is None:
+    if not create_http_instance(url):
+        print("❌ 创建 HTTP 实例失败")
         return False
 
-    # 2. 设置 Headers
-    commands = (
-        f'AT+MHTTPCFG="header",{instance_id},"Content-Type: application/json"',
-        f'AT+MHTTPCFG="header",{instance_id},"Authorization: {token}"',
-        f'AT+MHTTPCFG="header",{instance_id},"DeviceId: {device_id}"',
-    )
-    for command in commands:
-        if not response_ok(send_cmd(command)):
-            print("❌ HTTP Header 配置失败")
-            send_cmd(f"AT+MHTTPDEL={instance_id}", 2000)
-            return False
+    # 2. 设置 Headers（假设实例 ID 为 0，或根据模块默认）
+    instance_id = 0  # 大多数模块默认实例为 0；若支持多实例，需解析返回值
+    send_cmd(f'AT+MHTTPCFG="header",{instance_id},"Content-Type: application/json"')
+    send_cmd(f'AT+MHTTPCFG="header",{instance_id},"Authorization: {token}"')
+    send_cmd(f'AT+MHTTPCFG="header",{instance_id},"DeviceId: {device_id}"')
 
     # 3. 发送 Body
     json_str = ujson.dumps(json_data)
-    at_json = json_str.replace("\\", "\\\\").replace('"', '\\"')
-    if not response_ok(send_cmd(f'AT+MHTTPCONTENT={instance_id},0,0,"{at_json}"', 8000)):
-        print("❌ HTTP Body 配置失败")
-        send_cmd(f"AT+MHTTPDEL={instance_id}", 2000)
-        return False
+    send_cmd(f'AT+MHTTPCONTENT={instance_id},0,0,"{json_str}"')
 
     # 4. 发起 POST 请求
-    if response_ok(send_cmd(f'AT+MHTTPREQUEST={instance_id},2,0,"{api_path}"', 15000)):
+    if send_cmd(f'AT+MHTTPREQUEST={instance_id},2,0,"{api_path}"'):
         print("✅ HTTP 请求已发送")
         return True
     else:
@@ -256,7 +199,7 @@ def read_response(timeout_ms=5000):
 print("🚀 启动直接上传流程...")
 
 token = generate_token(device_id)
-print("🔑 Token 已生成:", token[:12] + "...")
+print("🔑 Token:", token)
 
 # 构造模拟数据
 timestamp = int(time.time() * 1000)
@@ -273,14 +216,7 @@ json_data = {
 }
 
 # 执行上传
-upload_ok = False
-if not wait_modem_ready():
-    print("💥 4G 模组未就绪")
-else:
-    clear_http_instances()
-    upload_ok = send_http_request(url, api_path, token, device_id, json_data)
-
-if upload_ok:
+if send_http_request(url, api_path, token, device_id, json_data):
     read_response()
 else:
     print("💥 上传流程失败")

4g_download_manager.py

@@ -0,0 +1,403 @@
+import re
+import hashlib
+import binascii
+from maix import time
+from power import get_bus_voltage, voltage_to_percent
+from urllib.parse import urlparse
+from hardware import hardware_manager
+
+
+class DownloadManager4G:
+    """4g下载管理器（单例）"""
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super(DownloadManager4G, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        # 私有状态
+        self.FRAG_SIZE = 1024
+        self.FRAG_DELAY = 10
+        self._initialized = True
+
+    def _log(self, *a):
+        if debug:
+            self.logger.debug(" ".join(str(x) for x in a))
+
+    def _pwr_log(self, prefix=""):
+        """debug 用：输出电压/电量"""
+        if not debug:
+            return
+        try:
+            v = get_bus_voltage()
+            p = voltage_to_percent(v)
+            self.logger.debug(f"[PWR]{prefix} v={v:.3f}V p={p}%")
+        except Exception as e:
+            try:
+                self.logger.debug(f"[PWR]{prefix} read_failed: {e}")
+            except:
+                pass
+
+    def _clear_http_events(self):
+        if hardware_manager.at_client:
+            while hardware_manager.at_client.pop_http_event() is not None:
+                pass
+
+    def _parse_httpid(self, resp: str):
+        m = re.search(r"\+MHTTPCREATE:\s*(\d+)", resp)
+        return int(m.group(1)) if m else None
+
+    def _get_ip(self, ):
+        r = hardware_manager.at_client.send("AT+CGPADDR=1", "OK", 3000)
+        m = re.search(r'\+CGPADDR:\s*1,"([^"]+)"', r)
+        return m.group(1) if m else ""
+
+    def _ensure_pdp(self, ):
+        ip = self._get_ip()
+        if ip and ip != "0.0.0.0":
+            return True, ip
+        hardware_manager.at_client.send("AT+MIPCALL=1,1", "OK", 15000)
+        for _ in range(10):
+            ip = self._get_ip()
+            if ip and ip != "0.0.0.0":
+                return True, ip
+            time.sleep(1)
+        return False, ip
+
+    def _extract_hdr_fields(self, hdr_text: str):
+        mlen = re.search(r"Content-Length:\s*(\d+)", hdr_text, re.IGNORECASE)
+        clen = int(mlen.group(1)) if mlen else None
+        mmd5 = re.search(r"Content-Md5:\s*([A-Za-z0-9+/=]+)", hdr_text, re.IGNORECASE)
+        md5_b64 = mmd5.group(1).strip() if mmd5 else None
+        return clen, md5_b64
+
+    def _extract_content_range(self, hdr_text: str):
+        m = re.search(r"Content-Range:\s*bytes\s*(\d+)\s*-\s*(\d+)\s*/\s*(\d+)", hdr_text, re.IGNORECASE)
+        if not m:
+            return None, None, None
+        try:
+            return int(m.group(1)), int(m.group(2)), int(m.group(3))
+        except:
+            return None, None, None
+
+    def _hard_reset_http(self, ):
+        """模块进入"坏状态"时的保守清场"""
+        self._clear_http_events()
+        for i in range(0, 6):
+            try:
+                hardware_manager.at_client.send(f"AT+MHTTPDEL={i}", "OK", 1200)
+            except:
+                pass
+        self._clear_http_events()
+
+    def _create_httpid(self, full_reset=False):
+        self._clear_http_events()
+        if hardware_manager.at_client:
+            hardware_manager.at_client.flush()
+        if full_reset:
+            self._hard_reset_http()
+        resp = hardware_manager.at_client.send(f'AT+MHTTPCREATE="{base_url}"', "OK", 8000)
+        hid = self._parse_httpid(resp)
+        if self._is_https:
+            resp = hardware_manager.at_client.send(f'AT+MHTTPCFG="ssl",{hid},1,1', "OK", 2000)
+            if "ERROR" in resp or "CME ERROR" in resp:
+                self.logger.error(f"MHTTPCFG SSL failed: {resp}")
+                # 尝试https 降级到http
+                downgraded_base_url = base_url.replace("https://", "http://")
+                resp = hardware_manager.at_client.send(f'AT+MHTTPCREATE="{downgraded_base_url}"', "OK", 8000)
+                hid = self._parse_httpid(resp)
+
+        return hid, resp
+
+    def _fetch_range_into_buf(self, start, want_len, out_buf, path, full_reset=False):
+        """
+        请求 Range [start, start+want_len)，写入 out_buf（bytearray，长度=want_len）
+        返回 (ok, msg, total_len, md5_b64, got_len)
+        """
+        end_incl = start + want_len - 1
+        hid, cresp = self._create_httpid(full_reset=full_reset)
+        if hid is None:
+            return False, f"MHTTPCREATE failed: {cresp}", None, None, 0
+
+        # 降低 URC 压力（分片/延迟）
+        hardware_manager.at_client.send(f'AT+MHTTPCFG="fragment",{hid},{self.FRAG_SIZE},{self.FRAG_DELAY}', "OK", 1500)
+        # 设置 Range header（inclusive）
+        hardware_manager.at_client.send(f'AT+MHTTPCFG="header",{hid},"Range: bytes={start}-{end_incl}"', "OK", 3000)
+
+        req = hardware_manager.at_client.send(f'AT+MHTTPREQUEST={hid},1,0,"{path}"', "OK", 15000)
+        if "ERROR" in req or "CME ERROR" in req:
+            hardware_manager.at_client.send(f"AT+MHTTPDEL={hid}", "OK", 2000)
+            return False, f"MHTTPREQUEST failed: {req}", None, None, 0
+
+        # 等 header + content
+        hdr_text = None
+        hdr_accum = ""
+        code = None
+        resp_total = None
+        total_len = None
+        md5_b64 = None
+
+        got_ranges = set()
+        last_sum = 0
+        t0 = time.ticks_ms()
+        timeout_ms = 9000
+        logged_hdr = False
+
+        while time.ticks_ms() - t0 < timeout_ms:
+            ev = hardware_manager.at_client.pop_http_event() if hardware_manager.at_client else None
+            if not ev:
+                time.sleep_ms(5)
+                continue
+
+            if ev[0] == "header":
+                _, ehid, ecode, ehdr = ev
+                if ehid != hid:
+                    continue
+                code = ecode
+                hdr_text = ehdr
+                if ehdr:
+                    hdr_accum = (hdr_accum + "\n" + ehdr) if hdr_accum else ehdr
+
+                resp_total_tmp, md5_tmp = self._extract_hdr_fields(hdr_accum)
+                if md5_tmp:
+                    md5_b64 = md5_tmp
+                cr_s, cr_e, cr_total = self._extract_content_range(hdr_accum)
+                if cr_total is not None:
+                    total_len = cr_total
+                if resp_total_tmp is not None:
+                    resp_total = resp_total_tmp
+                elif resp_total is None and (cr_s is not None) and (cr_e is not None) and (cr_e >= cr_s):
+                    resp_total = (cr_e - cr_s + 1)
+                if (not logged_hdr) and (resp_total is not None or total_len is not None):
+                    self._log(f"[HDR] id={hid} code={code} clen={resp_total} cr={cr_s}-{cr_e}/{cr_total}")
+                    logged_hdr = True
+                continue
+
+            if ev[0] == "content":
+                _, ehid, _total, _sum, _cur, payload = ev
+                if ehid != hid:
+                    continue
+                if resp_total is None:
+                    resp_total = _total
+                if resp_total is None or resp_total <= 0:
+                    continue
+                start_rel = _sum - _cur
+                end_rel = _sum
+                if start_rel < 0 or start_rel >= resp_total:
+                    continue
+                if end_rel > resp_total:
+                    end_rel = resp_total
+                actual_len = min(len(payload), end_rel - start_rel)
+                if actual_len <= 0:
+                    continue
+                out_buf[start_rel:start_rel + actual_len] = payload[:actual_len]
+                got_ranges.add((start_rel, start_rel + actual_len))
+                if _sum > last_sum:
+                    last_sum = _sum
+                if debug and (last_sum >= resp_total or (last_sum % 512 == 0)):
+                    self._log(f"[CHUNK] {start}+{last_sum}/{resp_total}")
+
+                if last_sum >= resp_total:
+                    break
+
+        # 清理实例（快路径：只删当前 hid）
+        try:
+            hardware_manager.at_client.send(f"AT+MHTTPDEL={hid}", "OK", 2000)
+        except:
+            pass
+
+        if resp_total is None:
+            return False, "no_header_or_total", total_len, md5_b64, 0
+
+        # 计算实际填充长度
+        merged = sorted(got_ranges)
+        merged2 = []
+        for s, e in merged:
+            if not merged2 or s > merged2[-1][1]:
+                merged2.append((s, e))
+            else:
+                merged2[-1] = (merged2[-1][0], max(merged2[-1][1], e))
+        filled = sum(e - s for s, e in merged2)
+
+        if filled < resp_total:
+            return False, f"incomplete_chunk got={filled} expected={resp_total} code={code}", total_len, md5_b64, filled
+
+        got_len = resp_total
+        return True, "OK", total_len, md5_b64, got_len
+
+    def download_file_via_4g(self, url, filename,
+                             total_timeout_ms=600000,
+                             retries=3,
+                             debug=False):
+        """
+        ML307R HTTP 下载（更稳的"固定小块 Range 顺序下载"，基于main109.py）：
+        - 只依赖 +MHTTPURC:"header"/"content"（不依赖 MHTTPREAD/cached）
+        - 每次只请求一个小块 Range（默认 10240B），失败就重试同一块，必要时缩小块大小
+        - 每个 chunk 都重新 MHTTPCREATE/MHTTPREQUEST，避免卡在"206 header 但不吐 content"的坏状态
+        - 使用二进制模式下载，确保文件完整性
+        """
+
+
+        # 小块策略（与main109.py保持一致）
+        CHUNK_MAX = 10240
+        CHUNK_MIN = 128
+        CHUNK_RETRIES = 12
+
+
+        t_func0 = time.ticks_ms()
+
+        parsed = urlparse(url)
+        host = parsed.hostname
+        path = parsed.path or "/"
+        if parsed.query:
+            path = f"{path}?{parsed.query}"
+        if parsed.fragment:
+            path = f"{path}#{parsed.fragment}"
+        if not host:
+            return False, "bad_url (no host)"
+
+        if isinstance(url, str) and url.startswith("https://static.shelingxingqiu.com/"):
+            base_url = "https://static.shelingxingqiu.com"
+            # TODO：使用https，看看是否能成功
+            self._is_https = True
+        else:
+            base_url = f"http://{host}"
+            self._is_https = False
+
+
+        try:
+            self._begin_ota()
+        except:
+            pass
+
+        from network import network_manager
+        with network_manager.get_uart_lock():
+            try:
+                ok_pdp, ip = self._ensure_pdp()
+                if not ok_pdp:
+                    return False, f"PDP not ready (ip={ip})"
+
+                # 先清空旧事件，避免串台
+                self._clear_http_events()
+
+                # 为了支持随机写入，先创建空文件
+                try:
+                    with open(filename, "wb") as f:
+                        f.write(b"")
+                except Exception as e:
+                    return False, f"open_file_failed: {e}"
+
+                total_len = None
+                expect_md5_b64 = None
+
+                offset = 0
+                chunk = CHUNK_MAX
+                t_start = time.ticks_ms()
+                last_progress_ms = t_start
+                STALL_TIMEOUT_MS = 60000
+                last_pwr_ms = t_start
+                self._pwr_log(prefix=" ota_start")
+                bad_http_state = 0
+
+                while True:
+                    now = time.ticks_ms()
+                    if debug and time.ticks_diff(now, last_pwr_ms) >= 5000:
+                        last_pwr_ms = now
+                        self._pwr_log(prefix=f" off={offset}/{total_len or '?'}")
+                    if time.ticks_diff(now, t_start) > total_timeout_ms:
+                        return False, f"timeout overall after {total_timeout_ms}ms offset={offset} total={total_len}"
+
+                    if time.ticks_diff(now, last_progress_ms) > STALL_TIMEOUT_MS:
+                        return False, f"timeout stalled {STALL_TIMEOUT_MS}ms offset={offset} total={total_len}"
+
+                    if total_len is not None and offset >= total_len:
+                        break
+
+                    want = chunk
+                    if total_len is not None:
+                        remain = total_len - offset
+                        if remain <= 0:
+                            break
+                        if want > remain:
+                            want = remain
+
+                    # 本 chunk 的 buffer（长度=want）
+                    buf = bytearray(want)
+
+                    success = False
+                    last_err = "unknown"
+                    md5_seen = None
+                    got_len = 0
+                    for k in range(1, CHUNK_RETRIES + 1):
+                        do_full_reset = (bad_http_state >= 2)
+                        ok, msg, tlen, md5_b64, got = self._fetch_range_into_buf(offset, want, buf, base_url, path, full_reset=do_full_reset)
+                        last_err = msg
+                        if tlen is not None and total_len is None:
+                            total_len = tlen
+                        if md5_b64 and not expect_md5_b64:
+                            expect_md5_b64 = md5_b64
+                        if ok:
+                            success = True
+                            got_len = got
+                            bad_http_state = 0
+                            break
+
+                        try:
+                            if ("no_header_or_total" in msg) or ("MHTTPREQUEST failed" in msg) or (
+                                    "MHTTPCREATE failed" in msg):
+                                bad_http_state += 1
+                            else:
+                                bad_http_state = max(0, bad_http_state - 1)
+                        except:
+                            pass
+
+                        if chunk > CHUNK_MIN:
+                            chunk = max(CHUNK_MIN, chunk // 2)
+                            want = min(chunk, want)
+                            buf = bytearray(want)
+                        self._log(f"[RETRY] off={offset} want={want} try={k} err={msg}")
+                        self._pwr_log(prefix=f" retry{k} off={offset}")
+                        time.sleep_ms(120)
+
+                    if not success:
+                        return False, f"chunk_failed off={offset} want={want} err={last_err} total={total_len}"
+
+                    # 写入文件（二进制模式）
+                    try:
+                        with open(filename, "r+b") as f:
+                            f.seek(offset)
+                            f.write(bytes(buf))
+                    except Exception as e:
+                        return False, f"write_failed off={offset}: {e}"
+
+                    offset += len(buf)
+                    last_progress_ms = time.ticks_ms()
+                    chunk = CHUNK_MAX
+                    if debug:
+                        self._log(f"[OK] offset={offset}/{total_len or '?'}")
+
+                # MD5 校验
+                if expect_md5_b64 and hashlib is not None:
+                    try:
+                        with open(filename, "rb") as f:
+                            data = f.read()
+                        digest = hashlib.md5(data).digest()
+                        got_b64 = binascii.b2a_base64(digest).decode().strip()
+                        if got_b64 != expect_md5_b64:
+                            return False, f"md5_mismatch got={got_b64} expected={expect_md5_b64}"
+                        self.logger.debug(f"[4G-DL] MD5 verified: {got_b64}")
+                    except Exception as e:
+                        return False, f"md5_check_failed: {e}"
+
+                t_cost = time.ticks_diff(time.ticks_ms(), t_func0)
+                self.logger.info(f"[4G-DL] download complete: size={offset} ip={ip} cost_ms={t_cost}")
+                return True, f"OK size={offset} ip={ip} cost_ms={t_cost}"
+
+            finally:
+                self._end_ota()

4g_upload_manager.py

@@ -0,0 +1,450 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+4G Image Upload Manager
+Uploads images to Qiniu cloud via ML307R 4G module TCP socket (MIPOPEN + MIPSEND).
+
+AT Command Sequence (ML307R TCP socket POST):
+  AT+MIPCALL=1,1                                  // Ensure PDP context active
+  AT+MIPCLOSE=<id>                                // Close old socket (ignore error)
+  AT+MIPOPEN=<id>,"TCP","<host>",80              // Open TCP socket
+  // Wait for +MIPOPEN: <id>,0 (success)
+  AT+MIPSEND=<id>,<len>                           // Send data
+  // Wait for ">" prompt, then write raw bytes
+  // Repeat MIPSEND for all chunks
+  // Wait for +MIPURC: "rtcp" response
+  AT+MIPCLOSE=<id>                                // Close socket
+"""
+
+import re
+import os
+import json
+from maix import time
+from urllib.parse import urlparse
+from logger_manager import logger_manager
+from hardware import hardware_manager
+
+# Multipart form boundary (simple alphanumeric to avoid AT command parser issues)
+BOUNDARY = "QiniuFormBoundary" + hex(int(time.time()))[2:]
+# Chunk size for MIPSEND (max 1024 to avoid AT line buffer limits)
+SEND_CHUNK = 1024
+# Socket ID for upload (dedicated to avoid conflict with main app TCP)
+UPLOAD_SOCK_ID = 3
+
+
+class FourGUploadManager:
+    """4G image upload manager using ML307R TCP socket (MIPOPEN + MIPSEND)"""
+
+    def __init__(self, at_client):
+        """Initialize with AT client instance"""
+        self.at = at_client
+        self.logger = logger_manager.logger
+
+    # ------------------------------------------------------------------ logging
+    def _log(self, msg):
+        try:
+            self.logger.debug("[4G-UL] " + msg)
+        except Exception:
+            print("[4G-UL] " + msg)
+
+    def _log_info(self, msg):
+        try:
+            self.logger.info("[4G-UL] " + msg)
+        except Exception:
+            print("[4G-UL] " + msg)
+
+    def _log_error(self, msg):
+        try:
+            self.logger.error("[4G-UL] " + msg)
+        except Exception:
+            print("[4G-UL] " + msg)
+
+    # --------------------------------------------------------------- helpers
+    def _ensure_pdp(self):
+        """Ensure PDP context is active; returns (ok, ip)"""
+        r = self.at.send("AT+CGPADDR=1", "OK", 3000)
+        m = re.search(r'\+CGPADDR:\s*1,"([^"]+)"', r)
+        ip = m.group(1) if m else ""
+        if ip and ip != "0.0.0.0":
+            return True, ip
+        self.at.send("AT+MIPCALL=1,1", "OK", 15000)
+        for _ in range(10):
+            r = self.at.send("AT+CGPADDR=1", "OK", 3000)
+            m = re.search(r'\+CGPADDR:\s*1,"([^"]+)"', r)
+            ip = m.group(1) if m else ""
+            if ip and ip != "0.0.0.0":
+                return True, ip
+            time.sleep(1)
+        return False, ip
+
+    def _is_error(self, resp):
+        """Check AT response for any error indicators"""
+        return "ERROR" in resp or "CME ERROR" in resp
+
+    # --------------------------------------------------------- multipart body
+    def _build_multipart_body(self, image_path, upload_token, key):
+        """
+        Build multipart/form-data body as bytes for Qiniu upload.
+
+        Fields:
+          - token : Qiniu upload token
+          - key   : object key in bucket
+          - file  : binary image data
+        """
+        boundary = BOUNDARY.encode()
+
+        with open(image_path, "rb") as f:
+            file_data = f.read()
+
+        filename = os.path.basename(image_path)
+        ext = os.path.splitext(image_path)[1].lower()
+        ct_map = {
+            ".png": "image/png",
+            ".jpg": "image/jpeg",
+            ".jpeg": "image/jpeg",
+            ".bmp": "image/bmp",
+            ".webp": "image/webp",
+        }
+        content_type = ct_map.get(ext, "application/octet-stream")
+
+        body = bytearray()
+
+        # -- token field --
+        body += b"--" + boundary + b"\r\n"
+        body += b'Content-Disposition: form-data; name="token"\r\n'
+        body += b"\r\n"
+        body += upload_token.encode("utf-8") + b"\r\n"
+
+        # -- key field --
+        body += b"--" + boundary + b"\r\n"
+        body += b'Content-Disposition: form-data; name="key"\r\n'
+        body += b"\r\n"
+        body += key.encode("utf-8") + b"\r\n"
+
+        # -- file field --
+        body += b"--" + boundary + b"\r\n"
+        body += (
+            b'Content-Disposition: form-data; name="file"; filename="'
+            + filename.encode("utf-8")
+            + b'"\r\n'
+        )
+        body += b"Content-Type: " + content_type.encode("utf-8") + b"\r\n"
+        body += b"\r\n"
+        body += file_data + b"\r\n"
+
+        # -- closing boundary --
+        body += b"--" + boundary + b"--\r\n"
+
+        return bytes(body)
+
+    # --------------------------------------------------- TCP socket helpers
+    def _close_socket(self, sock_id):
+        """Close socket, ignore CME ERROR 55 (already closed)"""
+        try:
+            resp = self.at.send("AT+MIPCLOSE=" + str(sock_id), "OK", 5000)
+            self._log("socket " + str(sock_id) + " closed: " + resp)
+        except Exception as e:
+            # Ignore CME ERROR 55 (socket not open)
+            self._log("socket close (may already be closed): " + str(e))
+
+    def _open_socket(self, sock_id, host, port):
+        """
+        Open TCP socket to host:port.
+        Returns (success, error_msg)
+        """
+        cmd = 'AT+MIPOPEN=' + str(sock_id) + ',"TCP","' + host + '",' + str(port)
+        resp = self.at.send(cmd, "OK", 15000)
+
+        if self._is_error(resp):
+            return False, "MIPOPEN failed: " + resp
+
+        # Wait for +MIPOPEN: <id>,0 (success) or +MIPOPEN: <id>,<error_code>
+        # The URC may come in the same response or separately
+        mipopen_pattern = r"\+MIPOPEN:\s*" + str(sock_id) + r",(\d+)"
+        m = re.search(mipopen_pattern, resp)
+
+        if m:
+            result_code = int(m.group(1))
+            if result_code == 0:
+                return True, ""
+            else:
+                return False, "MIPOPEN error code: " + str(result_code)
+
+        # If not in initial response, wait for URC
+        try:
+            urc_resp = self.at.send("", "+MIPOPEN:", 15000)
+            m = re.search(mipopen_pattern, urc_resp)
+            if m:
+                result_code = int(m.group(1))
+                if result_code == 0:
+                    return True, ""
+                else:
+                    return False, "MIPOPEN error code: " + str(result_code)
+        except Exception as e:
+            return False, "MIPOPEN URC timeout: " + str(e)
+
+        return False, "MIPOPEN no response"
+
+    def _send_chunk(self, sock_id, chunk):
+        """
+        Send a single chunk via MIPSEND.
+        Thread safety is provided by the outer network_manager.get_uart_lock().
+        NOTE: Do NOT add self.at._cmd_lock here — self.at.send() already
+        acquires it internally and threading.Lock is not reentrant.
+        Returns (success, error_msg)
+        """
+        chunk_len = len(chunk)
+
+        # Step 1: Send AT+MIPSEND command and wait for ">" prompt
+        cmd = "AT+MIPSEND=" + str(sock_id) + "," + str(chunk_len)
+        try:
+            resp = self.at.send(cmd, ">", 3000)
+            if ">" not in resp:
+                return False, "MIPSEND no > prompt: " + resp
+        except Exception as e:
+            return False, "MIPSEND > prompt error: " + str(e)
+
+        # Step 2: Write raw binary bytes directly to UART
+        # Must be done immediately after ">" prompt, no lock re-acquisition
+        try:
+            self.at.uart.write(chunk)
+        except Exception as e:
+            return False, "MIPSEND write error: " + str(e)
+
+        # Step 3: Wait for OK or SEND OK confirmation
+        try:
+            confirm_resp = self.at.send("", "OK", 8000)
+            if self._is_error(confirm_resp):
+                return False, "MIPSEND confirmation error: " + confirm_resp
+        except Exception as e:
+            return False, "MIPSEND confirmation timeout: " + str(e)
+
+        return True, ""
+
+    def _send_data(self, sock_id, data):
+        """
+        Send data in chunks via MIPSEND.
+        Returns (success, error_msg)
+        """
+        total_len = len(data)
+        offset = 0
+        chunk_num = 0
+
+        while offset < total_len:
+            end = min(offset + SEND_CHUNK, total_len)
+            chunk = data[offset:end]
+
+            ok, err = self._send_chunk(sock_id, chunk)
+            if not ok:
+                return False, "Chunk " + str(chunk_num) + " failed: " + err
+
+            chunk_num += 1
+            offset = end
+
+            if chunk_num % 10 == 0 or offset >= total_len:
+                self._log(
+                    "send progress: "
+                    + str(offset) + "/" + str(total_len)
+                    + " bytes (" + str(chunk_num) + " chunks)"
+                )
+
+        self._log("all data sent: " + str(chunk_num) + " chunks, " + str(total_len) + " bytes")
+        return True, ""
+
+    def _wait_for_response(self, sock_id, timeout_ms=30000):
+        """
+        Wait for +MIPURC: "rtcp" response.
+        Returns (success, status_code, body, error_msg)
+        """
+        pattern = r'\+MIPURC:\s*"rtcp",\s*' + str(sock_id) + r',\s*(\d+),'
+        t0 = time.ticks_ms()
+
+        while time.ticks_diff(time.ticks_ms(), t0) < timeout_ms:
+            try:
+                # Try to get response with short timeout
+                resp = self.at.send("", "+MIPURC:", 1000)
+                m = re.search(pattern, resp)
+                if m:
+                    data_len = int(m.group(1))
+                    # Extract HTTP response data after the URC header
+                    # Format: +MIPURC: "rtcp",<sock_id>,<len>,<data>
+                    urc_end = resp.find("+MIPURC:")
+                    if urc_end >= 0:
+                        # Find the data after the length field
+                        match_end = m.end()
+                        http_data = resp[match_end:match_end + data_len]
+
+                        # Parse HTTP status line
+                        status_match = re.search(r"HTTP/\d\.\d\s+(\d+)", http_data)
+                        status_code = int(status_match.group(1)) if status_match else None
+
+                        # Extract body (after headers)
+                        header_end = http_data.find("\r\n\r\n")
+                        if header_end >= 0:
+                            body = http_data[header_end + 4:]
+                        else:
+                            body = http_data
+
+                        return True, status_code, body, ""
+            except Exception:
+                pass
+
+            time.sleep_ms(100)
+
+        return False, None, "", "Response timeout"
+
+    def _build_http_request(self, host, body_bytes):
+        """
+        Build full HTTP POST request as bytes.
+        """
+        headers = (
+            "POST / HTTP/1.1\r\n"
+            "Host: " + host + "\r\n"
+            "Content-Type: multipart/form-data; boundary=" + BOUNDARY + "\r\n"
+            "Content-Length: " + str(len(body_bytes)) + "\r\n"
+            "Connection: close\r\n"
+            "\r\n"
+        )
+        return headers.encode("utf-8") + body_bytes
+
+    # ============================================================ public API
+    def upload_file(self, file_path, upload_url, upload_token, key):
+        """Generic file upload to Qiniu cloud via 4G TCP socket POST.
+
+        Args:
+            file_path: Local path to any file
+            upload_url: Qiniu upload URL
+            upload_token: Qiniu upload token
+            key: File key in Qiniu bucket
+
+        Returns:
+            dict with 'success' bool and 'key'/'error' fields
+        """
+        return self.upload_image(file_path, upload_url, upload_token, key)
+
+    def upload_image(self, image_path, upload_url, upload_token, key):
+        """
+        Upload image to Qiniu cloud via 4G TCP socket POST.
+
+        Args:
+            image_path: Local path to image file
+            upload_url: Qiniu upload URL (e.g., "https://upload.qiniup.com")
+            upload_token: Qiniu upload token
+            key: File key in Qiniu (e.g., "shootPic/device01/shoot01.png")
+
+        Returns:
+            dict with 'success' bool and 'key'/'error' fields
+        """
+        if not self.at:
+            return {"success": False, "error": "AT client not available"}
+
+        if not os.path.exists(image_path):
+            return {"success": False, "error": "Image file not found: " + image_path}
+
+        # Force HTTP for 4G module (extract hostname, use port 80)
+        parsed = urlparse(upload_url)
+        host = parsed.hostname
+        if not host:
+            return {"success": False, "error": "Invalid upload URL: " + upload_url}
+
+        if upload_url.lower().startswith("https://"):
+            self._log_info("Converted HTTPS->HTTP for 4G module")
+
+        file_size = os.path.getsize(image_path)
+        self._log_info(
+            "upload: " + image_path + " (" + str(file_size) + "B) -> "
+            + host + " key=" + key
+        )
+
+        from network import network_manager
+        with network_manager.get_uart_lock():
+            try:
+                # ---- Step 1: Ensure PDP context ----
+                ok_pdp, ip = self._ensure_pdp()
+                if not ok_pdp:
+                    return {"success": False, "error": "PDP not ready (ip=" + str(ip) + ")"}
+
+                # ---- Step 2: Close old socket ----
+                self._close_socket(UPLOAD_SOCK_ID)
+
+                # ---- Step 3: Open TCP socket ----
+                ok, err = self._open_socket(UPLOAD_SOCK_ID, host, 80)
+                if not ok:
+                    return {"success": False, "error": "Socket open failed: " + err}
+
+                try:
+                    # ---- Step 4: Build multipart body and HTTP request ----
+                    body = self._build_multipart_body(image_path, upload_token, key)
+                    http_request = self._build_http_request(host, body)
+                    self._log("HTTP request size: " + str(len(http_request)) + " bytes")
+
+                    # ---- Step 5: Send data via MIPSEND ----
+                    ok, err = self._send_data(UPLOAD_SOCK_ID, http_request)
+                    if not ok:
+                        return {"success": False, "error": "Send failed: " + err}
+
+                    # ---- Step 6: Wait for response ----
+                    ok, status_code, resp_body, err = self._wait_for_response(UPLOAD_SOCK_ID)
+                    if not ok:
+                        return {"success": False, "error": "Response error: " + err}
+
+                    # ---- Step 7: Parse response ----
+                    if status_code is None:
+                        return {"success": False, "error": "No HTTP status in response"}
+
+                    if 200 <= status_code < 300:
+                        try:
+                            resp_json = json.loads(resp_body)
+                            resp_key = resp_json.get("key", key)
+                            self._log_info("upload success: key=" + resp_key + " code=" + str(status_code))
+                            return {"success": True, "key": resp_key}
+                        except Exception as e:
+                            self._log_error("response parse error: " + str(e))
+                            return {
+                                "success": True,
+                                "key": key,
+                                "raw": resp_body,
+                            }
+                    else:
+                        self._log_error(
+                            "HTTP error: code=" + str(status_code) + " body=" + resp_body[:200]
+                        )
+                        return {
+                            "success": False,
+                            "error": "HTTP " + str(status_code),
+                            "response": resp_body,
+                        }
+
+                finally:
+                    # ---- Step 8: Always close socket ----
+                    self._close_socket(UPLOAD_SOCK_ID)
+
+            except Exception as e:
+                self._log_error("upload exception: " + str(e))
+                return {"success": False, "error": str(e)}
+
+
+# ====================================================================== demo
+if __name__ == "__main__":
+    # Demo usage — requires actual ML307R 4G module hardware to run.
+    print("FourGUploadManager - requires ML307R 4G module hardware")
+    print()
+    print("Usage:")
+    print("  from hardware import hardware_manager")
+    print("  from at_client import ATClient")
+    print("  from maix import uart")
+    print()
+    print("  # Initialize UART and AT client (normally done in hardware init)")
+    print("  uart4g = uart.UART('/dev/ttyS1', 115200, ...)")
+    print("  at_client = ATClient(uart4g)")
+    print("  at_client.start()")
+    print()
+    print("  # Upload image to Qiniu")
+    print("  uploader = FourGUploadManager(at_client)")
+    print("  result = uploader.upload_image(")
+    print("      image_path='/maixapp/apps/t11/shoot.png',")
+    print("      upload_url='https://upload.qiniup.com',")
+    print("      upload_token='<qiniu_upload_token>',")
+    print("      key='shootPic/device01/shoot01.png'")
+    print("  )")
+    print("  print('Upload result:', result)")

app.yaml

@@ -1,29 +1,31 @@
 id: t11
 name: t11
-version: 3.0.3
+version: 2.15.35
 author: t11
 icon: ''
 desc: t11
 files:
+  - 4g_download_manager.py
+  - 4g_upload_manager.py
   - app.yaml
   - archery_netcore.cpython-311-riscv64-linux-gnu.so
   - 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
+  - model_285484.cvimodel
+  - model_285484.mud
   - network.py
   - ota_curl.sh
   - ota_manager.py
   - power.py
   - server.pem
-  - set_autostart.py
   - shoot_manager.py
   - shot_id_generator.py
   - target_roi_yolo.py

at_client.py

@@ -69,7 +69,7 @@ class ATClient:
         # 同上：避免在 _reader_loop 持锁期间二次 acquire
         self._http_events.append(ev)
 
-    def send(self, cmd: str, expect: str = "OK", timeout_ms: int = 2000):
+    def send(self, cmd: str, expect: str = "OK", timeout_ms: int = 2000, abort_event=None):
         """
         发送 AT 命令并等待 expect（子串匹配）。
         注意：expect=">" 用于等待 prompt。
@@ -90,6 +90,9 @@ class ATClient:
 
             t0 = time.ticks_ms()
             while abs(time.ticks_diff(time.ticks_ms(), t0)) < timeout_ms:
+                if abort_event is not None and abort_event.is_set():
+                    self._waiting = False
+                    break
                 if (not self._waiting) or (self._expect in self._resp):
                     self._waiting = False
                     break
@@ -102,6 +105,39 @@ class ATClient:
             except:
                 return str(self._resp)
 
+    def send_raw_and_wait(self, data: bytes, expect: str = "OK", timeout_ms: int = 1000,
+                          suffix: bytes = b""):
+        """Register the response waiter before writing raw UART data."""
+        expect_b = expect.encode() if isinstance(expect, str) else expect
+        with self._cmd_lock:
+            with self._q_lock:
+                self._waiting = True
+                self._expect = expect_b
+                self._resp = b""
+
+            total = 0
+            while total < len(data):
+                n = self.uart.write(data[total:])
+                if not n or n < 0:
+                    time.sleep_ms(1)
+                    continue
+                total += n
+            if suffix:
+                self.uart.write(suffix)
+
+            t0 = time.ticks_ms()
+            while abs(time.ticks_diff(time.ticks_ms(), t0)) < timeout_ms:
+                if (not self._waiting) or (self._expect in self._resp):
+                    self._waiting = False
+                    break
+                time.sleep_ms(5)
+
+            self._waiting = False
+            try:
+                return self._resp.decode(errors="ignore")
+            except:
+                return str(self._resp)
+
     def _find_urc_tag(self, tag: bytes):
         """
         只在"真正的 URC 边界"查找 tag，避免误命中 HTTP payload 内容。

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

@@ -96,11 +96,6 @@ ADC_LASER_THRESHOLD = 3000
 
 # ==================== 激光配置 ====================
 MODULE_ADDR = 0x00
-# 激光开关改由 A14 GPIO 控制：低电平开启，高电平关闭。
-LASER_CONTROL_PIN = "A14"
-LASER_CONTROL_GPIO = "GPIOA14"
-LASER_CONTROL_ON_LEVEL = 0
-LASER_CONTROL_OFF_LEVEL = 1
 LASER_ON_CMD = bytes([0xAA, MODULE_ADDR, 0x01, 0xBE, 0x00, 0x01, 0x00, 0x01, 0xC1])
 LASER_OFF_CMD = bytes([0xAA, MODULE_ADDR, 0x01, 0xBE, 0x00, 0x01, 0x00, 0x00, 0xC0])
 DISTANCE_QUERY_CMD = bytes([0xAA, MODULE_ADDR, 0x00, 0x20, 0x00, 0x01, 0x00, 0x00, 0x21])  # 激光测距查询命令
@@ -267,9 +262,9 @@ TRIANGLE_SAMPLE_PATCH_HALF_PX = 2
 # 开机阶段预加载 YOLO detector；detect 使用 dual_buff=False，避免返回上一帧结果。
 TRIANGLE_YOLO_PRELOAD_ON_BOOT = False
 
-# YOLO target size classification: class 0=20cm, class 1=40cm.
+# YOLO 靶规格识别：class 0=20cm，class 1=40cm。
 TARGET_CLASS_YOLO_ENABLE = True
-TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_317828.mud"
+TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_285484.mud"
 TARGET_CLASS_YOLO_LABELS = (20, 40)
 TARGET_CLASS_YOLO_CONF_TH = 0.50
 TARGET_CLASS_YOLO_IOU_TH = 0.45
@@ -332,7 +327,6 @@ MAX_CMD_THREADS = 10            # 并发命令线程上限（防止服务器下
 
 # ==================== 图像保存配置 ====================
 SAVE_IMAGE_ENABLED = False  # 是否保存图像（True=保存，False=不保存）
-SAVE_IMAGE_ON_FAILURE = True  # 检测失败时是否强制保存图像（供调试测试用）
 PHOTO_DIR = "/root/phot"    # 照片存储目录
 MAX_IMAGES = 1000
 # Stage2 调试目录（默认 PHOTO_DIR/stage2_roi）内 JPEG 最多保留张数；None 表示与 MAX_IMAGES 相同
@@ -353,29 +347,23 @@ PIN_MAPPINGS = {
     "A28": "UART2_TX",
     "A15": "I2C5_SCL",
     "A27": "I2C5_SDA",
-    "A14": "GPIOA14",  # 激光开关：低开、高关
     "A24": "GPIOA24",  # 电源板关机控制
-    "A25": "GPIOA25",  # 电源状态绿灯
-    "A23": "GPIOA23",  # 电源状态红灯
 }
 
 # ==================== 电源配置 ====================
 AUTO_POWER_OFF_IN_SECONDS = 10 * 60  # 自动关机时间（秒），0表示不自动关机
-# 充电时自动关机暂时禁用；需要恢复时改为 True。
-CHARGING_AUTO_POWER_OFF_ENABLED = False
 
-# 一代电源控制：A24 由电源板负责按键/关机信号，软件关机时输出高电平。
-
-# 电源状态指示灯
-STATUS_LED_GREEN_GPIO = "GPIOA25"
-STATUS_LED_RED_GPIO = "GPIOA23"
-STATUS_LED_GREEN_ENABLED = True
-STATUS_LED_RED_ENABLED = True
-STATUS_LED_ACTIVE_LEVEL = 1
-STATUS_LED_LOW_BATTERY_PERCENT = 10
-STATUS_LED_FULL_BATTERY_PERCENT = 90
-STATUS_LED_CHARGING_BLINK_MS = 500
-STATUS_LED_POLL_MS = 1000
+# 实机数据：正常放电约为正电流，插入充电线后约为负电流。
+CHARGING_SHUTDOWN_ENABLED = True  # True=充电时退出应用，False=关闭充电关机功能
+CHARGING_DIAGNOSTIC_LOG_ENABLED = False
+CHARGING_CHECK_INTERVAL_MS = 3000
+CHARGING_CURRENT_THRESHOLD_MA = 100.0
+CHARGING_CONFIRM_COUNT = 2
+CHARGING_NOTIFY_TIMEOUT_MS = 30000
+CHARGING_4G_UART_LOCK_TIMEOUT_SEC = 2.5
+CHARGING_4G_PROMPT_TIMEOUT_MS = 1500
+CHARGING_4G_CONFIRM_TIMEOUT_MS = 1000
+CHARGING_EXIT_SCRIPT = APP_DIR + "/charging_exit.sh"
 
 BATTERY_SOC_LPF_ALPHA = 0.5
 BATTERY_SOC_AVG_WINDOW = 5

deploy/index.html

@@ -0,0 +1,20 @@
+<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
+  <meta name="description" content="Weusing · 思考、记录与实践">
+  <title>Weusing · 思考与记录</title>
+  <link rel="stylesheet" href="/style.css">
+</head>
+<body>
+  <header class="top"><a class="brand" href="/">WEUSING<span>.</span></a><nav><a href="#posts">文章</a><a href="#about">关于</a></nav></header>
+  <main>
+    <section class="intro"><p class="kicker">A SMALL NOTEBOOK ON THE WEB</p><h1>把想法写下来，<br><em>让时间看得见。</em></h1><p class="lede">这里记录技术、产品和生活里的小发现。保持好奇，持续构建。</p></section>
+    <section id="posts" class="posts"><div class="section-head"><h2>最新文章</h2><span>2026 / 08</span></div>
+      <article><div class="date">08.13<br><small>2026</small></div><div><p class="tag">TECHNOLOGY</p><h3>让静态网站快起来：从请求到首屏的几件小事</h3><p class="excerpt">更少的依赖、更短的路径，以及一些值得长期坚持的工程习惯。</p></div><a class="arrow" href="#">↗</a></article>
+      <article><div class="date">08.06<br><small>2026</small></div><div><p class="tag">NOTES</p><h3>在复杂系统里，保留一条清晰的路</h3><p class="excerpt">记录一次调试过程，也记录那些最终留下来的判断。</p></div><a class="arrow" href="#">↗</a></article>
+      <article><div class="date">07.28<br><small>2026</small></div><div><p class="tag">LIFE</p><h3>慢一点，观察风从哪里来</h3><p class="excerpt">日常、远方和一些不急着得到答案的问题。</p></div><a class="arrow" href="#">↗</a></article>
+    </section>
+    <section id="about" class="about"><p class="kicker">ABOUT THIS SPACE</p><p>Weusing 是一个个人写作空间。愿每一次发布，都比上一次更接近真实。</p></section>
+  </main><footer><span>© 2026 WEUSING</span><span>BUILT WITH HTML &amp; CSS</span></footer>
+</body></html>

deploy/style.css

@@ -0,0 +1 @@
+:root{--ink:#202326;--muted:#747a80;--line:#d9d7d1;--accent:#d95d39;--paper:#f5f3ee}*{box-sizing:border-box}html{scroll-behavior:smooth}body{margin:0;background:var(--paper);color:var(--ink);font-family:Arial,"Noto Sans SC",sans-serif}.top,main,footer{max-width:1080px;margin:auto}.top{height:88px;display:flex;align-items:center;justify-content:space-between;border-bottom:1px solid var(--line)}.brand{font-size:18px;letter-spacing:2px;font-weight:700;color:var(--ink);text-decoration:none}.brand span{color:var(--accent)}nav{display:flex;gap:30px}nav a{font-size:13px;color:var(--muted);text-decoration:none}nav a:hover{color:var(--accent)}.intro{padding:105px 0 115px;border-bottom:1px solid var(--line)}.kicker,.tag{font-size:11px;letter-spacing:2px;color:var(--accent);font-weight:700}.intro h1{font-size:clamp(42px,7vw,78px);line-height:1.1;letter-spacing:-2px;margin:25px 0}.intro em{font-family:Georgia,serif;font-weight:400;color:#555;font-style:italic}.lede{max-width:390px;color:var(--muted);font-size:16px;line-height:1.8}.posts{padding:72px 0}.section-head{display:flex;align-items:baseline;justify-content:space-between;border-bottom:2px solid var(--ink);padding-bottom:16px}.section-head h2{font-size:25px;margin:0}.section-head span{font-size:12px;color:var(--muted)}article{display:grid;grid-template-columns:90px 1fr 32px;gap:28px;padding:32px 0;border-bottom:1px solid var(--line);align-items:start}.date{font-size:14px;color:var(--accent);line-height:1.45;font-weight:700}.date small{font-size:11px;color:var(--muted);font-weight:400}article h3{font-size:23px;line-height:1.35;margin:9px 0 8px;font-weight:500}article .tag{margin:0;color:var(--muted);font-size:10px}.excerpt{color:var(--muted);font-size:14px;line-height:1.7;margin:0}.arrow{color:var(--accent);font-size:22px;text-decoration:none}.about{padding:30px 0 110px;display:grid;grid-template-columns:1fr 2fr;gap:30px;border-top:1px solid var(--line)}.about p:last-child{font:italic 27px/1.5 Georgia,serif;max-width:620px;margin:0}footer{border-top:1px solid var(--line);padding:25px 0 35px;display:flex;justify-content:space-between;color:var(--muted);font-size:10px;letter-spacing:1px}@media(max-width:700px){.top,main,footer{margin:0 22px}.top{height:70px}.intro{padding:70px 0 75px}.intro h1{letter-spacing:-1px}article{grid-template-columns:58px 1fr 20px;gap:14px}article h3{font-size:18px}.about{grid-template-columns:1fr;padding-bottom:75px}.about p:last-child{font-size:23px}footer{margin:0;padding:22px}.section-head{margin-top:0}}

hardware.py

@@ -5,7 +5,6 @@
 提供硬件对象的统一管理和访问
 """
 from maix import time
-import _thread
 import config
 from at_client import ATClient
 
@@ -29,7 +28,6 @@ class HardwareManager:
         self._bus = None                 # I2C总线
         self._adc_obj = None             # ADC对象
         self._at_client = None            # AT客户端
-        self._status_led_monitor_started = False
 
         self._last_active_time = 0       # 用于记录用户的最后一次活跃的时间
         self._stop_timer = False         # 用于停止定时器的标志
@@ -106,87 +104,11 @@ class HardwareManager:
             # 物理引脚是 A24，对应 GPIO 功能是 GPIOA24
             # 注意：这里需要先在 config.PIN_MAPPINGS 中配置好 "A24": "GPIOA24"
             from maix import gpio
-            # 一代电源板关机信号为高电平
+            # 输出高电平关闭
             gpio.GPIO("GPIOA24", gpio.Mode.OUT).value(1)
         except Exception as e:
             print(f"关机失败: {e}")
 
-    def start_status_led_monitor(self):
-        """后台更新状态灯：正常/充满绿常亮、充电绿闪烁、低电量红常亮。"""
-        if self._status_led_monitor_started:
-            return
-        self._status_led_monitor_started = True
-        _thread.start_new_thread(self._status_led_loop, ())
-
-    def _status_led_loop(self):
-        from maix import gpio
-        from power import get_bus_voltage, is_charging, voltage_to_percent
-
-        try:
-            green = None
-            if getattr(config, "STATUS_LED_GREEN_ENABLED", True):
-                green = gpio.GPIO(config.STATUS_LED_GREEN_GPIO, gpio.Mode.OUT)
-            red = None
-            if getattr(config, "STATUS_LED_RED_ENABLED", True):
-                red = gpio.GPIO(config.STATUS_LED_RED_GPIO, gpio.Mode.OUT)
-            active = int(config.STATUS_LED_ACTIVE_LEVEL)
-            inactive = 0 if active else 1
-            if green is not None:
-                green.value(inactive)
-            if red is not None:
-                red.value(inactive)
-            last_state = None
-            blink_on = False
-            blink_period = max(100, int(config.STATUS_LED_CHARGING_BLINK_MS))
-            poll_ms = max(100, int(config.STATUS_LED_POLL_MS))
-            tick_ms = min(blink_period, poll_ms)
-            sensor_elapsed = poll_ms
-            blink_elapsed = blink_period
-            state = "normal"
-            percent = None
-            charging = False
-
-            while self._status_led_monitor_started:
-                if sensor_elapsed >= poll_ms:
-                    voltage = get_bus_voltage()
-                    percent = voltage_to_percent(voltage) if voltage > 0 else None
-                    charging = is_charging()
-                    low = percent is not None and percent <= int(config.STATUS_LED_LOW_BATTERY_PERCENT)
-                    full = percent is not None and percent >= int(config.STATUS_LED_FULL_BATTERY_PERCENT)
-                    if charging:
-                        state = "full" if full else "charging"
-                    else:
-                        state = "low" if low else "normal"
-                    sensor_elapsed = 0
-
-                if state == "low":
-                    if green is not None:
-                        green.value(inactive)
-                    if red is not None:
-                        red.value(active)
-                elif state == "charging":
-                    if blink_elapsed >= blink_period:
-                        blink_on = not blink_on
-                        blink_elapsed = 0
-                    if green is not None:
-                        green.value(active if blink_on else inactive)
-                    if red is not None:
-                        red.value(inactive)
-                else:  # normal or full
-                    if green is not None:
-                        green.value(active)
-                    if red is not None:
-                        red.value(inactive)
-                if state != last_state:
-                    print(f"[STATUS_LED] state={state} percent={percent} charging={charging}")
-                    last_state = state
-                time.sleep_ms(tick_ms)
-                sensor_elapsed += tick_ms
-                blink_elapsed += tick_ms
-        except Exception as e:
-            self._status_led_monitor_started = False
-            print(f"[STATUS_LED] monitor failed: {e}")
-
     def start_idle_timer(self):
         self._stop_timer = False
         self._last_active_time = time.time()

laser_manager.py

@@ -31,7 +31,6 @@ class LaserManager:
         
         # 私有状态
         self._serial = None  # 激光串口，由 laser_manager 自己持有
-        self._laser_gpio = None  # A14 激光开关，低电平开启、高电平关闭
         self._calibration_active = False
         self._calibration_result = None
         self._calibration_lock = threading.Lock()
@@ -70,21 +69,10 @@ class LaserManager:
     
     # ==================== 初始化方法 ====================
 
-    def init_control_gpio(self):
-        """尽早初始化 A14，并拉高确保激光关闭。"""
-        from maix import gpio, pinmap
-
-        pinmap.set_pin_function(config.LASER_CONTROL_PIN, config.LASER_CONTROL_GPIO)
-        if self._laser_gpio is None:
-            self._laser_gpio = gpio.GPIO(config.LASER_CONTROL_GPIO, gpio.Mode.OUT)
-        self._laser_gpio.value(config.LASER_CONTROL_OFF_LEVEL)
-        self._laser_turned_on = False
-        print(f"[LASER] {config.LASER_CONTROL_PIN}=HIGH，激光已关闭")
-
     def init(self, serial_device=None, baudrate=None):
         """
-        初始化激光模块（A14 开关 + 测距串口）
-        初始化时先将 A14 拉高关闭激光，防止开机误触发
+        初始化激光模块（包括串口）
+        初始化完成后主动发送关闭命令，防止 UART 初始化噪声误触发激光
         
         Args:
             serial_device: 串口设备路径，默认使用 config.DISTANCE_SERIAL_DEVICE
@@ -94,11 +82,23 @@ class LaserManager:
         device = serial_device or config.DISTANCE_SERIAL_DEVICE
         baud = baudrate or config.DISTANCE_SERIAL_BAUDRATE
 
-        self.init_control_gpio()
-
         self._serial = uart.UART(device, baud)
         print(f"[LASER] 激光串口初始化完成: device={device}, baudrate={baud}")
 
+        # 等待串口稳定后主动关闭激光，防止初始化噪声误触发
+        time.sleep_ms(100)
+        try:
+            self._serial.read(-1)  # 清空接收缓冲区
+        except Exception:
+            pass
+        self._serial.write(config.LASER_OFF_CMD)
+        time.sleep_ms(60)
+        try:
+            self._serial.read(-1)  # 清空回包
+        except Exception:
+            pass
+        print("[LASER] 已发送关闭命令（防止开机误触发）")
+
     # ==================== 业务方法 ====================
     
     def load_laser_point(self):
@@ -147,38 +147,66 @@ class LaserManager:
             return False
     
     def turn_on_laser(self):
-        """A14 输出低电平，开启激光。"""
-        if self._laser_gpio is None:
-            if self.logger:
-                self.logger.error("[LASER] A14 GPIO 未初始化，请先调用 init()")
-            return False
+        """发送指令开启激光，并读取回包（部分模块支持）"""
+        if self._serial is None:
+            self.logger.error("[LASER] 激光串口未初始化，请先调用 init()")
+            return None
+        
+        # 打印调试信息
+        self.logger.info(f"[LASER] 发送开启命令: {config.LASER_ON_CMD.hex()}")
+        
+        # 清空接收缓冲区
         try:
-            self._laser_gpio.value(config.LASER_CONTROL_ON_LEVEL)
-            self._laser_turned_on = True
-            if self.logger:
-                self.logger.info("[LASER] A14=LOW，激光开启")
-            return True
-        except Exception as e:
-            if self.logger:
-                self.logger.error(f"[LASER] A14 开启激光失败: {e}")
-            return False
+            self._serial.read(-1)  # 清空缓冲区
+        except:
+            pass
+        
+        # 发送命令
+        written = self._serial.write(config.LASER_ON_CMD)
+        self.logger.info(f"[LASER] 写入字节数: {written}")
+        
+        time.sleep_ms(60)
+
+        # 读取回包
+        resp = self._serial.read(len=20, timeout=10)
+        if resp:
+            self.logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
+            if resp == config.LASER_ON_CMD:
+                self.logger.info("✅ 激光开启指令已确认")
+        else:
+            self.logger.warning("🔇 无回包（可能正常或模块不支持回包）")
+        self._laser_turned_on = True
+        return resp
     
     def turn_off_laser(self):
-        """A14 输出高电平，关闭激光。"""
-        if self._laser_gpio is None:
-            if self.logger:
-                self.logger.error("[LASER] A14 GPIO 未初始化，请先调用 init()")
-            return False
+        """发送指令关闭激光"""
+        if self._serial is None:
+            self.logger.error("[LASER] 激光串口未初始化，请先调用 init()")
+            return None
+        
+        # 打印调试信息
+        self.logger.info(f"[LASER] 发送关闭命令: {config.LASER_OFF_CMD.hex()}")
+        
+        # 清空接收缓冲区
         try:
-            self._laser_gpio.value(config.LASER_CONTROL_OFF_LEVEL)
-            self._laser_turned_on = False
-            if self.logger:
-                self.logger.info("[LASER] A14=HIGH，激光关闭")
-            return True
-        except Exception as e:
-            if self.logger:
-                self.logger.error(f"[LASER] A14 关闭激光失败: {e}")
-            return False
+            self._serial.read(-1)
+        except:
+            pass
+        
+        # 发送命令
+        written = self._serial.write(config.LASER_OFF_CMD)
+        self.logger.info(f"[LASER] 写入字节数: {written}")
+
+        time.sleep_ms(60)
+        
+        # 读取回包
+        resp = self._serial.read(20)
+        if resp:
+            self.logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
+        else:
+            self.logger.warning("🔇 无回包")
+        self._laser_turned_on = False
+        return resp
 
     def flash_laser(self, duration_ms=1000):
         """闪一下激光（非阻塞版本）"""

main.py

@@ -76,14 +76,12 @@ def laser_calibration_worker():
                 import traceback
                 traceback.print_exc()
             time.sleep_ms(1000)  # 等待1秒后继续
+
 def cmd_str():
     """主程序入口"""
     # ==================== 第一阶段：硬件初始化 ====================
     # 按照 main104.py 的顺序，先完成所有硬件初始化
 
-    # 开机第一步先拉高 A14 关闭激光，避免其他硬件初始化期间误亮。
-    laser_manager.init_control_gpio()
-
     # 1. 引脚功能映射
     for pin, func in config.PIN_MAPPINGS.items():
         try:
@@ -105,8 +103,6 @@ def cmd_str():
     print(f"[BOOT] init_ina226 开始 wall_s={_w_boot:.3f}")
     init_ina226()
     print(f"[BOOT] init_ina226 结束 wall +{int(round((wall_time.time() - _w_boot) * 1000))} ms")
-    # 启动 A25 绿灯和 A23 红灯状态指示。
-    hardware_manager.start_status_led_monitor()
 
     # 4. 初始化显示和相机
     _w_boot = wall_time.time()
@@ -124,7 +120,7 @@ def cmd_str():
 
     # ==================== 第二阶段：软件初始化 ====================
 
-    # 1. 初始化日志系统（WARNING级别，不打印/写入INFO和DEBUG日志，提高执行流畅度）
+    # 1. 初始化日志系统
     import logging
     logger_manager.init_logging(log_level=logging.WARNING)
     logger = logger_manager.logger
@@ -253,8 +249,8 @@ def cmd_str():
     # 4. 初始化设备ID（network_manager 内部会自动设置 device_id 和 password）
     network_manager.read_device_id()
 
-    # 5. 创建照片存储目录（如果启用图像保存或检测失败时强制保存）
-    if config.SAVE_IMAGE_ENABLED or getattr(config, "SAVE_IMAGE_ON_FAILURE", False):
+    # 5. 创建照片存储目录（如果启用图像保存）
+    if config.SAVE_IMAGE_ENABLED:
         photo_dir = config.PHOTO_DIR
         if photo_dir not in os.listdir("/root"):
             try:
@@ -286,12 +282,6 @@ def cmd_str():
         logger.info("系统准备完成...")
 
     last_adc_trigger = 0
-    # 读取一次ADC初始值，防止开机时传感器已有压力导致误触发
-    try:
-        last_adc_val = hardware_manager.adc_obj.read()
-    except Exception:
-        last_adc_val = 0
-    peak_adc_val = 0  # 当前周期内的压力峰值
     # 气压采样：减少日志频率（每 N 个点输出一条），避免 logger.debug 拖慢采样
     PRESSURE_BATCH_SIZE = 100
 
@@ -343,6 +333,7 @@ def cmd_str():
                 time.sleep_ms(250)
                 continue
 
+            # todo 去除或者不在这里检测
             # 不在 OTA 状态下，检测是否空闲足够长，自动关机
             # print(f"[MAIN] 空闲时间: {hardware_manager.get_idle_time_in_sec() }秒")
             # print(f"配置关机时间：{config.AUTO_POWER_OFF_IN_SECONDS} 秒")
@@ -381,21 +372,15 @@ def cmd_str():
                 pressure_max = adc_val
             if len(pressure_buf) >= PRESSURE_BATCH_SIZE:
                 _flush_pressure_buf("batch")
-            # 峰值检测：压力从峰值下降时触发，确保捕获到最大冲击时刻
-            if adc_val > peak_adc_val:
-                peak_adc_val = adc_val  # 更新峰值
-            if (peak_adc_val >= config.ADC_TRIGGER_THRESHOLD
-                    and adc_val < peak_adc_val
-                    and last_adc_val >= peak_adc_val):
-                # 封顶后下降沿触发：peak是最大值，当前值开始下降，且上次值还在peak位置
+            # 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:
-                    peak_adc_val = 0  # 去抖期间重置峰值
-                    time.sleep_ms(5)
+                    logger.info(f"[MAIN] 扳机触发过于频繁, {diff_ms}ms")
                     continue
                 last_adc_trigger = current_time
-                peak_adc_val = 0  # 触发后重置峰值
                 # 触发前先把缓存刷出来，避免波形被长耗时处理截断
                 _flush_pressure_buf("before_trigger")
 
@@ -414,9 +399,10 @@ def cmd_str():
                     try:
                         camera_manager.show(camera_manager.read_frame())
                     except Exception as e:
-                        pass
+                        logger = logger_manager.logger
+                        if logger:
+                            logger.error(f"[MAIN] 显示异常: {e}")
                 time.sleep_ms(5)
-            last_adc_val = adc_val
 
         except Exception as e:
             # 主循环的顶层异常捕获，防止程序静默退出

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

@@ -0,0 +1,13 @@
+
+[basic]
+type = cvimodel
+model = model_270820.cvimodel
+
+[extra]
+model_type = yolov5
+input_type = rgb
+mean = 0, 0, 0
+scale = 0.00392156862745098, 0.00392156862745098, 0.00392156862745098
+anchors = 10, 13, 16, 30, 33, 23, 30, 61, 62, 45, 59, 119, 116, 90, 156, 198, 373, 326
+labels = triangle
+

model_285484.mud

@@ -1,7 +1,7 @@
 
 [basic]
 type = cvimodel
-model = model_317828.cvimodel
+model = model_285484.cvimodel
 
 [extra]
 model_type = yolov5

network.py

@@ -67,6 +67,7 @@ class NetworkManager:
         self._queue_lock = threading.Lock()
         self._send_event = threading.Event()
         self._uart4g_lock = threading.Lock()
+        self._terminal_send_event = threading.Event()
         self._device_id = None
         self._password = None
         self._raw_line_data = []
@@ -678,7 +679,7 @@ class NetworkManager:
                 except OSError:
                     pass
             w = network.wifi.Wifi()
-            e = w.connect(ssid, password, wait=True, timeout=15)
+            e = w.connect(ssid, password, wait=True, timeout=10)
             err.check_raise(e, "connect wifi failed")
             if self.logger:
                 self.logger.info(f"[ota] Connect success, got ip{w.get_ip()}")
@@ -711,6 +712,35 @@ class NetworkManager:
         """线程安全地将消息加入队列（公共方法）"""
         self._enqueue((msg_type, data_dict), high)
 
+    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 safe_replace_queue_and_wait(self, data_dict, msg_type=2, timeout_ms=30000):
+        """Drop queued messages, enqueue one terminal message, and wait for its TCP write."""
+        sent_event = threading.Event()
+        with self._queue_lock:
+            self._high_send_queue.clear()
+            self._normal_send_queue.clear()
+            self._high_send_queue.append((msg_type, data_dict, sent_event))
+        self._send_event.set()
+        return bool(sent_event.wait(max(0, int(timeout_ms)) / 1000.0))
+
+    def safe_terminal_send_and_wait(self, data_dict, msg_type=2, timeout_ms=30000):
+        """Cancel ordinary 4G waits and replace queued work with one terminal message."""
+        sent_event = threading.Event()
+        result = {"sent": False}
+        self._terminal_send_event.set()
+        with self._queue_lock:
+            self._high_send_queue.clear()
+            self._normal_send_queue.clear()
+            self._high_send_queue.append((msg_type, data_dict, sent_event, "terminal", result))
+        self._send_event.set()
+        completed = sent_event.wait(max(0, int(timeout_ms)) / 1000.0)
+        return bool(completed and result["sent"])
+
     def connect_server(self):
         """
         连接到服务器（自动选择WiFi或4G）
@@ -897,6 +927,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:
@@ -1104,8 +1140,12 @@ class NetworkManager:
             return False
         try:
             for _ in range(max_retries):
+                if self._terminal_send_event.is_set():
+                    return False
                 cmd = f'AT+MIPSEND={link_id},{len(data)}'
                 if ">" not in hardware_manager.at_client.send(cmd, ">", 2000):
+                    if self._terminal_send_event.is_set():
+                        return False
                     time.sleep_ms(50)
                     continue
 
@@ -1120,14 +1160,73 @@ class NetworkManager:
                 hardware_manager.uart4g.write(b"\x1A")
                 with hardware_manager.at_client._q_lock:
                     hardware_manager.at_client._rx = b""
-                r = hardware_manager.at_client.send("", "OK", 8000)
+                r = hardware_manager.at_client.send(
+                    "", "OK", 8000, abort_event=self._terminal_send_event
+                )
                 if ("SEND OK" in r) or ("OK" in r) or ("+MIPSEND" in r):
                     return True
+                if self._terminal_send_event.is_set():
+                    return False
                 time.sleep_ms(50)
             return False
         finally:
             self._uart4g_lock.release()
 
+    def _tcp_send_terminal_raw(self, data: bytes) -> bool:
+        if not self._tcp_connected:
+            return False
+        if self._network_type == "wifi":
+            return self._tcp_send_raw_via_wifi(data, max_retries=1)
+        if self._network_type != "4g":
+            return False
+
+        link_id = getattr(config, "TCP_LINK_ID", 0)
+        lock_timeout_sec = float(
+            getattr(config, "CHARGING_4G_UART_LOCK_TIMEOUT_SEC", 2.5)
+        )
+        prompt_timeout_ms = int(
+            getattr(config, "CHARGING_4G_PROMPT_TIMEOUT_MS", 1500)
+        )
+        confirm_timeout_ms = int(
+            getattr(config, "CHARGING_4G_CONFIRM_TIMEOUT_MS", 1000)
+        )
+        lock_start_ms = time.ticks_ms()
+        if not self._uart4g_lock.acquire(timeout=max(0.0, lock_timeout_sec)):
+            self.logger.warning(
+                f"[CHARGE-4G] uart_lock timeout timeout_sec={lock_timeout_sec}"
+            )
+            return False
+        try:
+            lock_elapsed_ms = abs(time.ticks_diff(time.ticks_ms(), lock_start_ms))
+            cmd = f'AT+MIPSEND={link_id},{len(data)}'
+            prompt_start_ms = time.ticks_ms()
+            if ">" not in hardware_manager.at_client.send(
+                    cmd, ">", max(0, prompt_timeout_ms)):
+                prompt_elapsed_ms = abs(time.ticks_diff(time.ticks_ms(), prompt_start_ms))
+                self.logger.warning(
+                    f"[CHARGE-4G] prompt failed lock_ms={lock_elapsed_ms} "
+                    f"prompt_ms={prompt_elapsed_ms}"
+                )
+                return False
+            prompt_elapsed_ms = abs(time.ticks_diff(time.ticks_ms(), prompt_start_ms))
+            confirm_start_ms = time.ticks_ms()
+            r = hardware_manager.at_client.send_raw_and_wait(
+                data,
+                expect="OK",
+                timeout_ms=max(0, confirm_timeout_ms),
+                suffix=b"\x1A",
+            )
+            confirm_elapsed_ms = abs(time.ticks_diff(time.ticks_ms(), confirm_start_ms))
+            sent = ("SEND OK" in r) or ("OK" in r) or ("+MIPSEND" in r)
+            self.logger.warning(
+                f"[CHARGE-4G] send_done lock_ms={lock_elapsed_ms} "
+                f"prompt_ms={prompt_elapsed_ms} confirm_ms={confirm_elapsed_ms} "
+                f"sent={sent}"
+            )
+            return sent
+        finally:
+            self._uart4g_lock.release()
+
     def _configure_ssl_before_connect(self, link_id: int) -> bool:
         """按手册：MSSLCFG(auth) -> (可选) MSSLCERTWR -> MSSLCFG(cert) -> MIPCFG(ssl)"""
         ssl_id = getattr(config, "SSL_ID", 1)
@@ -1212,6 +1311,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:
@@ -1812,16 +1919,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
@@ -1863,12 +1963,25 @@ class NetworkManager:
                 pending_cleared = False
                 last_heartbeat_ack_time = time.ticks_ms()
                 last_heartbeat_send_time = time.ticks_ms()
+                last_wifi_sta_check_time = time.ticks_ms()
 
                 while True:
                     # 如果底层连接已断开，尽快跳出内层循环触发重连/重选网络
                     if not self._tcp_connected:
                         break
 
+                    if self._network_type == "wifi":
+                        now_ms = time.ticks_ms()
+                        if abs(time.ticks_diff(now_ms, last_wifi_sta_check_time)) >= 1000:
+                            last_wifi_sta_check_time = now_ms
+                            if not wifi_manager.is_sta_associated():
+                                self.logger.warning(
+                                    "[WIFI-TCP] STA disconnected; leave WiFi session and reselect network"
+                                )
+                                wifi_manager.disconnect_wifi()
+                                self._tcp_connected = False
+                                break
+
                     # OTA 期间暂停 TCP 活动
                     try:
                         from ota_manager import ota_manager
@@ -2146,7 +2259,7 @@ class NetworkManager:
                                         }
                                         self.safe_enqueue(battery_data, 2)
                                         self.logger.info(f"电量上报: {battery_percent}% 充电: {is_charging()}")
-                                        if getattr(config, "CHARGING_AUTO_POWER_OFF_ENABLED", False) and is_charging():
+                                        if is_charging():
                                             self.safe_enqueue(
                                                 {
                                                     "cmd": 700,
@@ -2309,9 +2422,36 @@ class NetworkManager:
                                 item_is_high = False
 
                         if item:
-                            msg_type, data_dict = item
+                            msg_type, data_dict = item[:2]
+                            sent_event = item[2] if len(item) > 2 else None
+                            item_is_terminal = len(item) > 3 and item[3] == "terminal"
+                            terminal_result = item[4] if item_is_terminal and len(item) > 4 else None
+                            if (
+                                isinstance(data_dict, dict)
+                                and data_dict.get("cmd") == 1
+                                and isinstance(data_dict.get("data"), dict)
+                            ):
+                                shot_data = data_dict["data"]
+                                self.logger.info(
+                                    f"[REPORT-TX] shot_id={shot_data.get('shot_id')}, "
+                                    f"target_class={shot_data.get('target_class')}, "
+                                    f"confidence={shot_data.get('target_class_confidence')}"
+                                )
                             pkt = self._netcore.make_packet(msg_type, data_dict)
-                            if not self.tcp_send_raw(pkt):
+                            send_ok = (
+                                self._tcp_send_terminal_raw(pkt)
+                                if item_is_terminal
+                                else self.tcp_send_raw(pkt)
+                            )
+                            if not send_ok:
+                                if item_is_terminal:
+                                    if terminal_result is not None:
+                                        terminal_result["sent"] = False
+                                    if sent_event is not None:
+                                        sent_event.set()
+                                    break
+                                if self._terminal_send_event.is_set():
+                                    continue
                                 # 发送失败：将消息放回队首（队列满则丢弃）
                                 with self.get_queue_lock():
                                     if item_is_high:
@@ -2326,6 +2466,10 @@ class NetworkManager:
                                 except:
                                     pass
                                 break
+                            if sent_event is not None:
+                                if terminal_result is not None:
+                                    terminal_result["sent"] = True
+                                sent_event.set()
 
                     # 发送激光校准结果
                     if logged_in:

official_pack_fip.py

@@ -1,603 +0,0 @@
-#!/usr/bin/env python3
-# PYTHON_ARGCOMPLETE_OK
-
-import sys
-import logging
-import os
-import re
-import os.path
-import collections
-import uuid
-import argparse
-import tarfile
-import io
-from struct import pack, unpack
-
-
-PYTHON_MIN_VERSION = (3, 5, 2)  # Ubuntu 16.04 LTS contains Python v3.5.2 by default
-
-
-if sys.version_info < PYTHON_MIN_VERSION:
-    print("Python >= %r is required" % (PYTHON_MIN_VERSION,))
-    sys.exit(-1)
-
-
-try:
-    import coloredlogs
-except ImportError:
-    coloredlogs = None
-
-try:
-    import argcomplete
-except ImportError:
-    argcomplete = None
-
-TOC_HEADER_NAME = 0xAA640001
-FIP_MAX_SIZE = 0xA0000
-FIP_ALIGN_SIZE = 2 * 1024
-ENTRY_SIZE = 0x28
-
-IV_ZERO = b"\0" * 16
-
-
-class FIP_HEADER_FLAG:
-    BitRange = collections.namedtuple("BitRange", "shift, bits")
-
-    REE_SCS = BitRange(0, 2)
-    REE_ENCRYPTION = BitRange(2, 2)
-
-    @classmethod
-    def test(cls, value, flag):
-        v = value >> flag.shift
-        v &= (1 << flag.bits) - 1
-        return v
-
-    @classmethod
-    def value(cls, flag):
-        v = (1 << flag.bits) - 1
-        v <<= flag.shift
-        return v
-
-
-class FIP_UUID:
-    # from arm-trusted-firmware/include/tools_share/firmware_image_package.h
-    uuid_c_define = """
-/* ToC Entry UUIDs */
-#define UUID_LICENSE_FILE \
-    {0x25360c62, 0x5151, 0x48ad, 0xb5, 0x91, {0x2d, 0x35, 0x67, 0x26, 0x85, 0xa5} }
-#define UUID_TRUSTED_UPDATE_FIRMWARE_SCP_BL2U \
-    {0x03279265, 0x742f, 0x44e6, 0x8d, 0xff, {0x57, 0x9a, 0xc1, 0xff, 0x06, 0x10} }
-#define UUID_TRUSTED_UPDATE_FIRMWARE_BL2U \
-    {0x37ebb360, 0xe5c1, 0x41ea, 0x9d, 0xf3, {0x19, 0xed, 0xa1, 0x1f, 0x68, 0x01} }
-#define UUID_TRUSTED_UPDATE_FIRMWARE_NS_BL2U \
-    {0x111d514f, 0xe52b, 0x494e, 0xb4, 0xc5, {0x83, 0xc2, 0xf7, 0x15, 0x84, 0x0a} }
-#define UUID_TRUSTED_FWU_CERT \
-    {0xb28a4071, 0xd618, 0x4c87, 0x8b, 0x2e, {0xc6, 0xdc, 0xcd, 0x50, 0xf0, 0x96} }
-#define UUID_TRUSTED_BOOT_FIRMWARE_BL2 \
-    {0x0becf95f, 0x224d, 0x4d3e, 0xa5, 0x44, {0xc3, 0x9d, 0x81, 0xc7, 0x3f, 0x0a} }
-#define UUID_BLD \
-    {0x3dfd6697, 0xbe89, 0x49e8, 0xae, 0x5d, {0x78, 0xa1, 0x40, 0x60, 0x82, 0x13} }
-#define UUID_EL3_RUNTIME_FIRMWARE_BL31 \
-    {0x6d08d447, 0xfe4c, 0x4698, 0x9b, 0x95, {0x29, 0x50, 0xcb, 0xbd, 0x5a, 0x00} }
-#define UUID_SECURE_PAYLOAD_BL32 \
-    {0x89e1d005, 0xdc53, 0x4713, 0x8d, 0x2b, {0x50, 0x0a, 0x4b, 0x7a, 0x3e, 0x38} }
-#define UUID_NON_TRUSTED_FIRMWARE_BL33 \
-    {0xa7eed0d6, 0xeafc, 0x4bd5, 0x97, 0x82, {0x99, 0x34, 0xf2, 0x34, 0xb6, 0xe4} }
-/* Key certificates */
-#define UUID_ROT_KEY_CERT \
-    {0x721d2d86, 0x60f8, 0x11e4, 0x92, 0x0b, {0x8b, 0xe7, 0x62, 0x16, 0x0f, 0x24} }
-#define UUID_BLD1_KEY_CERT \
-    {0x90e87e82, 0x60f8, 0x11e4, 0xa1, 0xb4, {0x77, 0x7a, 0x21, 0xb4, 0xf9, 0x4c} }
-#define UUID_BLD2_KEY_CERT \
-    {0xa1214202, 0x60f8, 0x11e4, 0x8d, 0x9b, {0xf3, 0x3c, 0x0e, 0x15, 0xa0, 0x14} }
-#define UUID_SOC_FW_KEY_CERT \
-    {0xccbeb88a, 0x60f9, 0x11e4, 0x9a, 0xd0, {0xeb, 0x48, 0x22, 0xd8, 0xdc, 0xf8} }
-#define UUID_TRUSTED_OS_FW_KEY_CERT \
-    {0x03d67794, 0x60fb, 0x11e4, 0x85, 0xdd, {0xb7, 0x10, 0x5b, 0x8c, 0xee, 0x04} }
-#define UUID_BL33_KEY_CERT \
-    {0x2a83d58a, 0x60fb, 0x11e4, 0x8a, 0xaf, {0xdf, 0x30, 0xbb, 0xc4, 0x98, 0x59} }
-/* Content certificates */
-#define UUID_TRUSTED_BOOT_FW_CERT \
-    {0xea69e2d6, 0x635d, 0x11e4, 0x8d, 0x8c, {0x9f, 0xba, 0xbe, 0x99, 0x56, 0xa5} }
-#define UUID_BLD_CONTENT_CERT \
-    {0x046fbe44, 0x635e, 0x11e4, 0xb2, 0x8b, {0x73, 0xd8, 0xea, 0xae, 0x96, 0x56} }
-#define UUID_SOC_FW_CONTENT_CERT \
-    {0x200cb2e2, 0x635e, 0x11e4, 0x9c, 0xe8, {0xab, 0xcc, 0xf9, 0x2b, 0xb6, 0x66} }
-#define UUID_TRUSTED_OS_FW_CONTENT_CERT \
-    {0x11449fa4, 0x635e, 0x11e4, 0x87, 0x28, {0x3f, 0x05, 0x72, 0x2a, 0xf3, 0x3d} }
-#define UUID_BL33_CONTENT_CERT \
-    {0xf3c1c48e, 0x635d, 0x11e4, 0xa7, 0xa9, {0x87, 0xee, 0x40, 0xb2, 0x3f, 0xa7} }
-/* CV keys */
-#define UUID_CV_TRUSTED_KEY_CERT \
-    {0x64fbfc49, 0x4b8c, 0x4ad3, 0xb9, 0x92, {0x93, 0x55, 0x89, 0xee, 0xf0, 0x12} }
-#define UUID_CV_NON_TRUSTED_KEY_CERT \
-    {0xcb48bf0d, 0x7012, 0x4201, 0xbc, 0x35, {0x8a, 0x51, 0xc4, 0x90, 0x90, 0x94} }
-
-/* DDR init*/
-#define UUID_CV_DDRINIT_KEY_CERT \
-    {0xa61c53c9, 0x886c, 0x484f, 0x96, 0x5d, {0xd2, 0xda, 0xd7, 0xc3, 0xeb, 0x13} }
-#define UUID_CV_DDRINIT_CONTENT_CERT \
-    {0x9dfaabd2, 0x7f1b, 0x47e6, 0xa8, 0xa6, {0x6a, 0xc3, 0x10, 0xcc, 0xac, 0x91} }
-#define UUID_CV_DDRINIT \
-    {0x5888a5cd, 0x38fc, 0x4f66, 0xae, 0x3d, {0x2e, 0x18, 0x6d, 0x69, 0x41, 0xfb} }
-
-/* Fast boot */
-#define UUID_CV_FASTBOOT_KEY_CERT \
-    {0x285df54e, 0x7b50, 0x4309, 0x9b, 0x52, {0x4b, 0xc4, 0x92, 0x82, 0x60, 0xdd} }
-#define UUID_CV_FASTBOOT_CONTENT_CERT \
-    {0x61f7595b, 0x8d77, 0x4e13, 0x91, 0x2a, {0x63, 0x6e, 0x58, 0xda, 0x5b, 0x69} }
-#define UUID_CV_FASTBOOT \
-    {0x43766198, 0xc363, 0x48db, 0xa9, 0x97, {0xf1, 0x0e, 0x93, 0x80, 0x4f, 0xea} }
-"""
-
-    @classmethod
-    def cls_init(cls):
-        txt = cls.uuid_c_define
-        txt = txt.replace("\r\n", "\n")
-        txt = txt.replace("\\\n", "\n")
-        rx = r"""
-            \#define\s+
-            (?P<name>\S+)\s+
-            {
-                \s*(?P<u0>0x\S+)\s*,\s*
-                \s*(?P<u1>0x\S+)\s*,\s*
-                \s*(?P<u2>0x\S+)\s*,\s*
-                \s*(?P<u3>0x\S+)\s*,\s*
-                \s*(?P<u4>0x\S+)\s*,\s*
-                {
-                    \s*(?P<u5>0x\S+)\s*,\s*
-                    \s*(?P<u6>0x\S+)\s*,\s*
-                    \s*(?P<u7>0x\S+)\s*,\s*
-                    \s*(?P<u8>0x\S+)\s*,\s*
-                    \s*(?P<u9>0x\S+)\s*,\s*
-                    \s*(?P<u10>0x\S+)\s*
-                }\s*,?\s*
-            }
-        """
-        for m in re.finditer(rx, txt, flags=re.X):
-            name = m.group("name")
-            u = m.group(*["u%d" % i for i in range(11)])
-            u = [int(i, 0) for i in u]
-            u = pack("<IHHBBBBBBBB", *u)
-            u = uuid.UUID(bytes=u)
-            setattr(cls, name, u)
-
-
-class Entry:
-    __slots__ = ["name", "loc", "uuid", "address", "flag", "content"]
-
-    def __init__(self):
-        self.loc = 0
-        self.uuid = uuid.UUID(int=0)
-        self.address = 0
-        self.flag = 0
-        self.content = b""
-
-    @classmethod
-    def make(cls, uuid, content):
-        entry = cls()
-        entry.uuid = uuid
-        entry.content = content
-        return entry
-
-    @classmethod
-    def from_fip(cls, name, loc, fip_bin):
-        data = fip_bin[loc : loc + ENTRY_SIZE]
-        uuid_bytes, address, size, flag = unpack("<16sQQQ", data)
-        content = fip_bin[address : address + size]
-
-        entry = cls()
-        entry.name = name
-        entry.loc = loc
-        entry.uuid = uuid.UUID(bytes=uuid_bytes)
-        entry.address = address
-        entry.flag = flag
-        entry.content = content
-        return entry
-
-    def to_bytes(self):
-        return pack("<16sQQQ", self.uuid.bytes, self.address, self.size, self.flag)
-
-    @property
-    def size(self):
-        return len(self.content)
-
-    @property
-    def end(self):
-        return self.address + self.size
-
-    def __str__(self):
-        return "<%-31s loc=0x%03x U=%s a=0x%05x,0x%05x,0x%05x f=0x%x>" % (
-            self.name,
-            self.loc,
-            self.uuid.hex[:8],
-            self.address,
-            self.end,
-            self.size,
-            self.flag,
-        )
-
-
-class FIP:
-    ENTRY_NAMES = collections.OrderedDict(
-        [
-            ("LICENSE_FILE", "UUID_LICENSE_FILE"),
-            ("BL2", "UUID_TRUSTED_BOOT_FIRMWARE_BL2"),
-            ("BLD", "UUID_BLD"),
-            ("BL31", "UUID_EL3_RUNTIME_FIRMWARE_BL31"),
-            ("BL32", "UUID_SECURE_PAYLOAD_BL32"),
-            ("BL33", "UUID_NON_TRUSTED_FIRMWARE_BL33"),
-            ("BLD1_KEY_CERT", "UUID_BLD1_KEY_CERT"),
-            ("BLD2_KEY_CERT", "UUID_BLD2_KEY_CERT"),
-            ("CV_TRUSTED_KEY_CERT", "UUID_CV_TRUSTED_KEY_CERT"),
-            ("SOC_FW_KEY_CERT", "UUID_SOC_FW_KEY_CERT"),
-            ("TRUSTED_OS_FW_KEY_CERT", "UUID_TRUSTED_OS_FW_KEY_CERT"),
-            ("CV_NON_TRUSTED_KEY_CERT", "UUID_CV_NON_TRUSTED_KEY_CERT"),
-            ("BL33_KEY_CERT", "UUID_BL33_KEY_CERT"),
-            ("TRUSTED_BOOT_FW_CERT", "UUID_TRUSTED_BOOT_FW_CERT"),
-            ("BLD_CONTENT_CERT", "UUID_BLD_CONTENT_CERT"),
-            ("SOC_FW_CONTENT_CERT", "UUID_SOC_FW_CONTENT_CERT"),
-            ("TRUSTED_OS_FW_CONTENT_CERT", "UUID_TRUSTED_OS_FW_CONTENT_CERT"),
-            ("BL33_CONTENT_CERT", "UUID_BL33_CONTENT_CERT"),
-            ("CV_DDRINIT", "UUID_CV_DDRINIT"),
-            ("CV_FASTBOOT", "UUID_CV_FASTBOOT"),
-        ]
-    )
-
-    TOC_Header = collections.namedtuple(
-        "TOC_Header", "name, serial, flag_res, flag_plat, flag_res2"
-    )
-
-    def __init__(self, path):
-        logging.info("FIP_BIN: %s", path)
-        self.path = path
-
-    def load(self):
-        with open(self.path, "rb") as fp:
-            self.binary = fp.read(FIP_MAX_SIZE)
-        logging.info("%s is %d bytes", self.path, len(self.binary))
-
-        self.header = self.TOC_Header(*unpack("<IIIHH", self.binary[0x00:0x10]))
-        if self.header.name != TOC_HEADER_NAME:
-            raise ValueError(
-                "FIP header is 0x%08x but should be 0x%08x"
-                % (self.header[0], TOC_HEADER_NAME)
-            )
-
-        logging.info("TOC header: flag_plat=0x%04x", self.header.flag_plat)
-        logging.info(
-            "  REE_SCS: %r",
-            FIP_HEADER_FLAG.test(self.header.flag_plat, FIP_HEADER_FLAG.REE_SCS),
-        )
-        logging.info(
-            "  REE_ENCRYPTION: %r",
-            FIP_HEADER_FLAG.test(self.header.flag_plat, FIP_HEADER_FLAG.REE_ENCRYPTION),
-        )
-
-        ents = []
-        for k, v in self.ENTRY_NAMES.items():
-            try:
-                ents.append((k, self.find_entry(v)))
-            except ValueError as err:
-                logging.warning("%s", err)
-
-        ents.sort(key=lambda x: x[1].address)
-        for n, (k, v) in enumerate(ents):
-            logging.debug("%s", v)
-            if n > 0:
-                pk, pv = ents[n - 1]
-                if v.loc != pv.loc + ENTRY_SIZE or v.address != pv.address + pv.size:
-                    raise Exception("Invalid FIP")
-
-        rest = self.binary[ents[-1][1].end :]
-        loc = rest.find(b"APLB")
-        if loc < 0:
-            raise Exception("No BLD/DDRC")
-        self.blp_ddrc_binary = rest[loc:]
-        logging.debug("blp_ddrc: 0x%04x at 0x%08x", len(self.blp_ddrc_binary), loc)
-
-        self.ents = collections.OrderedDict(ents)
-
-    def make_fip(self, output_path=None):
-        logging.info("New TOC header: flag_plat=0x%04x", self.header.flag_plat)
-        header_bin = pack("<IIIHH", *self.header)
-        fip_bin = header_bin
-
-        # Sort self.ents by the order of FIP.ENTRY_NAMES
-        sorted_ents = collections.OrderedDict()
-        for name in self.ENTRY_NAMES:
-            try:
-                sorted_ents[name] = self.ents[name]
-            except KeyError:
-                pass
-        self.ents = sorted_ents
-
-        offset = (len(self.ents) + 1) * ENTRY_SIZE + 0x10
-
-        for k, v in self.ents.items():
-            v.address = offset
-            fip_bin += v.to_bytes()
-            offset += v.size
-
-        null_entry = Entry()
-        null_entry.address = offset
-        fip_bin += null_entry.to_bytes()
-
-        for k, v in self.ents.items():
-            fip_bin += v.content
-
-        if (len(fip_bin) % FIP_ALIGN_SIZE) > 0:
-            fip_bin += b"\x00" * (FIP_ALIGN_SIZE - len(fip_bin) % FIP_ALIGN_SIZE)
-        fip_bin += self.blp_ddrc_binary
-
-        if output_path:
-            path = output_path
-        else:
-            path = os.path.splitext(self.path)
-            path = path[0] + "_signed_encrypted" + path[1]
-        logging.info("Save new FIP image to %s", path)
-        with open(path, "wb") as fp:
-            fp.write(fip_bin)
-
-    def dump_uuids(self):
-        for k, v in vars(FIP_UUID).items():
-            if k.startswith("UUID_"):
-                print("%-38s" % k, v.hex)
-
-    def find_entry(self, name):
-        # UUID=0, offset=any, size=0, flags=0
-        nullm = re.search(rb"\0{16}.{8}\0{16}", self.binary, flags=re.DOTALL)
-        if nullm is None:
-            raise Exception("NULL TOC entry is not found")
-
-        max_toc_size = nullm.start(0)
-        uuid = getattr(FIP_UUID, name)
-        loc = self.binary.find(uuid.bytes, 0, max_toc_size)
-        if loc < 0:
-            raise ValueError("%s is not found" % name)
-        return Entry.from_fip(name, loc, self.binary)
-
-
-def entry(args):
-    logging.debug("cmd_fip")
-
-
-def init_logging(log_file=None, file_level="DEBUG", stdout_level="WARNING"):
-
-    root_logger = logging.getLogger()
-    root_logger.setLevel(logging.NOTSET)
-
-    fmt = "%(asctime)s %(levelname)8s:%(name)s:%(message)s"
-
-    if log_file is not None:
-        file_handler = logging.FileHandler(log_file, encoding="utf-8")
-        file_handler.setFormatter(logging.Formatter(fmt))
-        file_handler.setLevel(file_level)
-        root_logger.addHandler(file_handler)
-
-    if coloredlogs:
-        os.environ["COLOREDLOGS_DATE_FORMAT"] = "%H:%M:%S"
-
-        field_styles = {
-            "asctime": {"color": "green"},
-            "hostname": {"color": "magenta"},
-            "levelname": {"color": "black", "bold": True},
-            "name": {"color": "blue"},
-            "programname": {"color": "cyan"},
-        }
-
-        level_styles = coloredlogs.DEFAULT_LEVEL_STYLES
-        level_styles["debug"]["color"] = "cyan"
-
-        coloredlogs.install(
-            level=stdout_level,
-            fmt=fmt,
-            field_styles=field_styles,
-            level_styles=level_styles,
-            milliseconds=True,
-        )
-
-
-def parse_fip(fip_path):
-    logging.debug("parse_fip: %s", fip_path)
-    fip = FIP(fip_path)
-    fip.load()
-
-
-def unpack_fip(fip_path):
-    logging.debug("unpack_fip: %s", fip_path)
-    fip = FIP(fip_path)
-    fip.load()
-
-    def save(name, content):
-        fn = os.path.splitext(fip_path)
-        fn = "%s_%s%s" % (fn[0], name, fn[1])
-        logging.info("Save %s", fn)
-        with open(fn, "wb") as fp:
-            fp.write(content)
-
-    for k, v in fip.ents.items():
-        save(k, v.content)
-
-    save("BLP_DDRC", fip.blp_ddrc_binary)
-
-
-def tar_bld(fip_path, output_path, multibin):
-    logging.debug("tar_bld: %s multibin=%r", fip_path, multibin)
-    fip = FIP(fip_path)
-    fip.load()
-
-    members = [
-        "BLD_CONTENT_CERT",
-        "BLD2_KEY_CERT",
-        "BLD1_KEY_CERT",
-        "CV_DDRINIT" if multibin else "BLD",
-    ]
-
-    if not output_path:
-        output_path = os.path.join(os.path.dirname(fip_path), "bld.tar")
-
-    logging.info("bld_tar_path=%s", output_path)
-
-    with tarfile.open(output_path, "w") as tf:
-        for m in members:
-            logging.debug("Tar %s", m)
-            try:
-                fp = io.BytesIO(fip.ents[m].content)
-            except KeyError:
-                logging.warning("%s doesn't exist", m)
-                continue
-            info = tarfile.TarInfo(name=m + ".bin")
-            info.size = len(fp.getbuffer())
-            tf.addfile(tarinfo=info, fileobj=fp)
-
-
-def merge_fip(fip_path, inputs, output_path):
-    logging.debug("merge_fip: %s", fip_path)
-    fip = FIP(fip_path)
-    fip.load()
-
-    for name in FIP.ENTRY_NAMES:
-        binary = inputs.get(name)
-        if not binary:
-            continue
-
-        logging.debug("merge %s", name)
-        ent = fip.ents.get(name)
-        if ent:
-            ent.content = binary
-        else:
-            ent = Entry.make(getattr(FIP_UUID, "UUID_" + name), binary)
-            fip.ents[name] = ent
-
-    binary = inputs.get("BLP_DDRC")
-    if binary:
-        fip.blp_ddrc_binary = binary
-
-    if not output_path:
-        fn = os.path.splitext(fip_path)
-        fn = "%s_%s%s" % (fn[0], "merged", fn[1])
-        output_path = fn
-
-    fip.make_fip(output_path)
-
-
-def round_up(n, k):
-    return (n + k - 1) // k * k
-
-
-def read_blp_and_ddrc(inputs, blp_path, ddrc_path):
-    logging.info("Open %s and %s", blp_path, ddrc_path)
-    with open(blp_path, "rb") as fp:
-        blp_bin = fp.read()
-
-    logging.info("Open %s", ddrc_path)
-    with open(ddrc_path, "rb") as fp:
-        ddrc_bin = fp.read()
-
-    blp_bin += b"\0" * (round_up(len(blp_bin), FIP_ALIGN_SIZE) - len(blp_bin))
-    ddrc_bin += b"\0" * (round_up(len(ddrc_bin), FIP_ALIGN_SIZE) - len(ddrc_bin))
-
-    inputs["BLP_DDRC"] = blp_bin + ddrc_bin
-
-
-def read_bld_tar(inputs, bld_tar_path, multibin):
-    logging.info("Open %s multibin=%r", bld_tar_path, multibin)
-    members = [
-        "BLD_CONTENT_CERT.bin",
-        "BLD2_KEY_CERT.bin",
-        "BLD1_KEY_CERT.bin",
-        "CV_DDRINIT.bin" if multibin else "BLD.bin",
-    ]
-
-    with tarfile.open(bld_tar_path, "r") as tf:
-        for member in members:
-            try:
-                fp = tf.extractfile(member)
-                inputs[os.path.splitext(member)[0]] = fp.read()
-            except KeyError:
-                logging.warning("%s does not exist", member)
-
-
-def main():
-    parser = argparse.ArgumentParser(description="FIP packer")
-
-    for name in FIP.ENTRY_NAMES:
-        parser.add_argument(
-            "--add-%s" % name.lower(),
-            dest=name,
-            type=str,
-            help="Merge %s into FIP" % name,
-        )
-    parser.add_argument(
-        "--add-blp-ddrc", dest="BLP_DDRC", type=str, help="Merge BLP+DDRC into FIP"
-    )
-    parser.add_argument("--add-blp", dest="BLP", type=str, help="Merge BLP into FIP")
-    parser.add_argument("--add-ddrc", dest="DDRC", type=str, help="Merge DDRC into FIP")
-    parser.add_argument(
-        "--add-bld-tar", dest="BLD_TAR", type=str, help="Merge BLD.tar into FIP"
-    )
-
-    parser.add_argument("--multibin", action="store_true", help="Use multibin")
-
-    parser.add_argument("FIP_BIN", type=str, nargs=1, help="Input FIP binary")
-    parser.add_argument("--output", type=str, help="Output filename")
-
-    parser.add_argument(
-        "--version", action="store_true", help="Output version information and exit"
-    )
-    parser.add_argument(
-        "--verbose",
-        help="Increase output verbosity",
-        action="store_const",
-        const=logging.DEBUG,
-        default=logging.DEBUG,
-    )
-
-    parser.add_argument("--unpack", action="store_true", help="Unpack FIP.bin")
-    parser.add_argument("--parse", action="store_true", help="Parse FIP.bin")
-    parser.add_argument(
-        "--tar-bld", action="store_true", help="Extrace BLD.bin and tar"
-    )
-
-    if argcomplete:
-        argcomplete.autocomplete(parser)
-
-    args = parser.parse_args()
-
-    init_logging(stdout_level=args.verbose)
-
-    logging.debug("args=%r", args)
-
-    FIP_UUID.cls_init()
-
-    if args.parse:
-        parse_fip(args.FIP_BIN[0])
-
-    if args.unpack:
-        unpack_fip(args.FIP_BIN[0])
-
-    if args.tar_bld:
-        tar_bld(args.FIP_BIN[0], args.output, args.multibin)
-
-    inputs = collections.OrderedDict()
-    for name in list(FIP.ENTRY_NAMES) + ["BLP_DDRC"]:
-        fn = getattr(args, name)
-        if not fn:
-            continue
-        logging.info("Open %s", fn)
-        with open(fn, "rb") as fp:
-            inputs[name] = fp.read()
-
-    if args.BLP or args.DDRC:
-        read_blp_and_ddrc(inputs, args.BLP, args.DDRC)
-
-    if args.BLD_TAR:
-        read_bld_tar(inputs, args.BLD_TAR, args.multibin)
-
-    if len(inputs):
-        merge_fip(args.FIP_BIN[0], inputs, args.output)
-
-
-if __name__ == "__main__":
-    main()

power.py

@@ -7,12 +7,11 @@
 import config
 import os
 import subprocess
-import _thread
 from logger_manager import logger_manager
 from maix import time as maix_time
 
+
 _INA226_PRESENT = None
-_INA226_LOCK = _thread.allocate_lock()
 
 
 def _ina226_ready() -> bool:
@@ -34,11 +33,7 @@ def write_register(reg, value):
     data = [(value >> 8) & 0xFF, value & 0xFF]
     # 某些底层驱动在失败时只打印 “write failed” 并返回 -1，而不是抛异常；
     # 为避免误判“初始化成功”导致后续 readfrom_mem SIGSEGV，这里把失败显式转成异常。
-    _INA226_LOCK.acquire()
-    try:
-        ret = hardware_manager.bus.writeto_mem(config.INA226_ADDR, reg, bytes(data))
-    finally:
-        _INA226_LOCK.release()
+    ret = hardware_manager.bus.writeto_mem(config.INA226_ADDR, reg, bytes(data))
     if isinstance(ret, int) and ret < 0:
         if logger:
             logger.error(f"[INA226] writeto_mem 失败: addr=0x{config.INA226_ADDR:02X} reg=0x{reg:02X} ret={ret}")
@@ -48,11 +43,7 @@ def write_register(reg, value):
 def read_register(reg):
     """读取INA226寄存器"""
     from hardware import hardware_manager
-    _INA226_LOCK.acquire()
-    try:
-        data = hardware_manager.bus.readfrom_mem(config.INA226_ADDR, reg, 2)
-    finally:
-        _INA226_LOCK.release()
+    data = hardware_manager.bus.readfrom_mem(config.INA226_ADDR, reg, 2)
     return (data[0] << 8) | data[1]
 
 
@@ -170,7 +161,7 @@ def voltage_to_percent(voltage):
         return 0
     if v <= 0:
         return 0
-    return int(int(_BATTERY_MONITOR.get_soc(v) * 10) / 10)  # 截断而不是四舍五入
+    return int(int(_BATTERY_MONITOR.get_soc(v) * 10) / 10) # 截断而不是四舍五入
 
 
 class BatteryMonitor:

shoot_manager.py

@@ -321,11 +321,9 @@ def process_shot(adc_val):
 
     try:
         frame = camera_manager.read_frame()
-
-        # 网络事件移到拍照之后，避免阻塞拍照
         network_manager.safe_enqueue({"shoot_event": "start"}, msg_type=2, high=True)
 
-        # Classify only the current shot frame; never reuse a previous result.
+        # 每箭只识别当前帧，不复用上一箭的靶规格结果。
         target_class_result = None
         try:
             from target_roi_yolo import try_get_target_class_from_yolo
@@ -551,7 +549,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,
@@ -561,9 +558,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

@@ -285,7 +285,7 @@ def _normalize_objs(objs):
 
 
 def _det_obj_score(o):
-    """Return confidence across supported Maix YOLO result formats."""
+    """兼容 Maix YOLO 不同版本的置信度字段。"""
     for key in ("score", "confidence", "conf", "prob"):
         if hasattr(o, key):
             try:
@@ -301,7 +301,7 @@ def _det_obj_score(o):
 
 
 def try_get_target_class_from_yolo(maix_frame, logger=None):
-    """Classify the current target as 20cm or 40cm; return None if unknown."""
+    """识别当前帧的 20/40 靶规格，失败返回 None。"""
     try:
         import config as cfg
     except Exception:

test/gpio_high.py

@@ -1,27 +0,0 @@
-"""Run independently and keep A24 at a high logic level."""
-
-from maix import app, gpio, pinmap, time
-
-
-PIN = "P19"
-GPIO_NAME = "GPIOP19"
-
-
-def main():
-    pinmap.set_pin_function(PIN, GPIO_NAME)
-    output = gpio.GPIO(GPIO_NAME, gpio.Mode.OUT)
-    output.value(1)
-    print(f"{PIN} is HIGH. Stop the script to set it LOW.")
-
-    try:
-        while not app.need_exit():
-            # Refresh the output in case another component changes its state.
-            output.value(1)
-            time.sleep_ms(100)
-    finally:
-        output.value(0)
-        print(f"{PIN} is LOW.")
-
-
-if __name__ == "__main__":
-    main()

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_gpio_a14.py

@@ -1,62 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""Interactive GPIO test for physical pin A14."""
-
-import sys
-
-from maix import gpio, pinmap
-
-
-PIN = "A14"
-GPIO_NAME = "GPIOA14"
-
-
-def set_level(output, command):
-    if command == "1":
-        output.value(1)
-        print("A14 = HIGH, laser OFF")
-        return True
-    if command == "0":
-        output.value(0)
-        print("A14 = LOW, laser ON")
-        return True
-    return False
-
-
-def main():
-    pinmap.set_pin_function(PIN, GPIO_NAME)
-    output = gpio.GPIO(GPIO_NAME, gpio.Mode.OUT)
-
-    # One-shot mode for SSH/serial shells: python3 test_gpio_a14.py 1|0
-    if len(sys.argv) > 1:
-        command = sys.argv[1].strip()
-        if not set_level(output, command):
-            print("Invalid argument. Use 1 or 0.")
-            return
-        return
-
-    output.value(1)
-    print("A14 laser test: input 0 for ON, 1 for OFF, q to quit.")
-
-    try:
-        while True:
-            try:
-                command = input("A14> ").strip().lower()
-            except EOFError:
-                print("This runner has no stdin. Run from an SSH/serial shell with argument 1 or 0.")
-                return
-            if set_level(output, command):
-                continue
-            elif command in ("q", "quit", "exit"):
-                break
-            elif command:
-                print("Invalid input. Use 1, 0, or q.")
-    except KeyboardInterrupt:
-        print()
-    finally:
-        output.value(1)
-        print("A14 = HIGH, laser OFF, test stopped.")
-
-
-if __name__ == "__main__":
-    main()

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_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/wifi_gpio_ina_isolation.py

@@ -1,122 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""Standalone WiFi/GPIO/INA226 isolation test for the official MaixPy tool.
-
-This file intentionally does not import project modules or start project
-threads. Select TEST_MODE below, then run the file directly.
-"""
-
-import time
-
-from maix import gpio, i2c, network, pinmap
-
-
-# Change only this value before each run.
-# wifi       WiFi only
-# a25        A25 only
-# a23        A23 only
-# gpio       A23/A26 only
-# ina        INA226 only
-# gpio_ina   GPIOs, then INA226
-# a25_wifi   A25, then WiFi
-# a23_wifi   A23, then WiFi
-# all        GPIOs, INA226, then WiFi
-TEST_MODE = "a25_wifi"
-
-WIFI_SSID = "sheling4b02-5G"
-WIFI_PASSWORD = "Aa12345678"
-WIFI_TIMEOUT_S = 20
-I2C_BUS_NUM = 5
-INA226_ADDR = 0x40
-
-
-def init_leds():
-    return init_selected_leds(True, True)
-
-
-def init_selected_leds(use_a26, use_a23):
-    outputs = []
-    if use_a26:
-        print("Initializing A25 -> GPIOA25")
-        pinmap.set_pin_function("A25", "GPIOA25")
-        green = gpio.GPIO("GPIOA25", gpio.Mode.OUT)
-        green.value(0)
-        outputs.append(("GPIOA25", green))
-        print("GPIOA25 initialized LOW")
-    if use_a23:
-        print("Initializing A23 -> GPIOA23")
-        pinmap.set_pin_function("A23", "GPIOA23")
-        red = gpio.GPIO("GPIOA23", gpio.Mode.OUT)
-        red.value(0)
-        outputs.append(("GPIOA23", red))
-        print("GPIOA23 initialized LOW")
-    return outputs
-
-
-def test_ina226():
-    # Match the board mapping used by the application before opening I2C5.
-    pinmap.set_pin_function("A15", "I2C5_SCL")
-    pinmap.set_pin_function("A27", "I2C5_SDA")
-    print("A15/A27 configured for I2C5")
-    print("Initializing I2C bus", I2C_BUS_NUM)
-    bus = i2c.I2C(I2C_BUS_NUM, i2c.Mode.MASTER)
-    print("Reading INA226 at 0x%02X" % INA226_ADDR)
-    config = bus.readfrom_mem(INA226_ADDR, 0x00, 2)
-    voltage_raw = bus.readfrom_mem(INA226_ADDR, 0x02, 2)
-    voltage = ((voltage_raw[0] << 8) | voltage_raw[1]) * 1.25 / 1000
-    print("INA226 config=0x%02X%02X voltage=%.3fV" % (config[0], config[1], voltage))
-    return bus
-
-
-def test_wifi():
-    print("Starting MaixPy WiFi connection...")
-    wifi = network.wifi.Wifi()
-    result = wifi.connect(WIFI_SSID, WIFI_PASSWORD, wait=True, timeout=WIFI_TIMEOUT_S)
-    print("WiFi connect result:", result)
-    print("WiFi connected:", wifi.is_connected())
-    try:
-        print("WiFi IP:", wifi.get_ip())
-    except Exception as exc:
-        print("WiFi status query failed:", exc)
-
-
-def main():
-    valid = ("wifi", "a25", "a23", "gpio", "ina", "gpio_ina", "a25_wifi", "a23_wifi", "all")
-    mode = TEST_MODE.lower()
-    if mode not in valid:
-        print("TEST_MODE must be one of:", ", ".join(valid))
-        return 1
-
-    leds = []
-    try:
-        print("=== Standalone WiFi/GPIO/INA226 isolation ===")
-        print("mode:", mode)
-        if mode in ("a25", "a25_wifi"):
-            leds = init_selected_leds(True, False)
-            time.sleep(1)
-        elif mode in ("a23", "a23_wifi"):
-            leds = init_selected_leds(False, True)
-            time.sleep(1)
-        elif mode in ("gpio", "gpio_ina", "all"):
-            leds = init_leds()
-            time.sleep(1)
-        if mode in ("ina", "gpio_ina", "all"):
-            test_ina226()
-            time.sleep(1)
-        if mode in ("wifi", "a25_wifi", "a23_wifi", "all"):
-            test_wifi()
-        print("TEST COMPLETE")
-        return 0
-    except Exception as exc:
-        print("TEST FAILED:", repr(exc))
-        return 1
-    finally:
-        for name, led in leds:
-            try:
-                led.value(0)
-                print(name, "LOW")
-            except Exception as exc:
-                print(name, "cleanup failed:", exc)
-
-
-main()

test/wifi_gpio_isolation.py

@@ -1,88 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""Standalone WiFi/GPIO isolation test.
-
-This script intentionally does not import any project module. It only tests
-MaixPy WiFi startup with optional A23/A26 GPIO initialization.
-"""
-
-import time
-
-from maix import gpio, network, pinmap
-
-
-GREEN_PIN = "A26"
-GREEN_GPIO = "GPIOA26"
-RED_PIN = "A23"
-RED_GPIO = "GPIOA23"
-
-# Run this file directly from the official MaixPy tool.
-# Change only TEST_MODE between runs: none -> a26 -> a23 -> both.
-TEST_MODE = "none"
-WIFI_SSID = "sheling4b02-5G"
-WIFI_PASSWORD = "Aa12345678"
-WIFI_TIMEOUT_S = 20
-
-
-def init_gpio(mode):
-    outputs = []
-    if mode in ("a26", "both"):
-        pinmap.set_pin_function(GREEN_PIN, GREEN_GPIO)
-        green = gpio.GPIO(GREEN_GPIO, gpio.Mode.OUT)
-        green.value(1)
-        outputs.append((GREEN_GPIO, green))
-        print("GPIOA26 initialized HIGH")
-    if mode in ("a23", "both"):
-        pinmap.set_pin_function(RED_PIN, RED_GPIO)
-        red = gpio.GPIO(RED_GPIO, gpio.Mode.OUT)
-        red.value(1)
-        outputs.append((RED_GPIO, red))
-        print("GPIOA23 initialized HIGH")
-    return outputs
-
-
-def connect_wifi(ssid, password, timeout_s):
-    print("Starting MaixPy WiFi connection...")
-    wifi = network.wifi.Wifi()
-    result = wifi.connect(ssid, password, wait=True, timeout=timeout_s)
-    print("WiFi connect result:", result)
-    try:
-        print("WiFi connected:", wifi.is_connected())
-        print("WiFi IP:", wifi.get_ip())
-    except Exception as exc:
-        print("WiFi status query failed:", exc)
-    return result
-
-
-def main():
-    mode = TEST_MODE.lower()
-    if mode not in ("none", "a26", "a23", "both"):
-        print("TEST_MODE must be none, a26, a23, or both")
-        return 1
-    ssid = WIFI_SSID
-    password = WIFI_PASSWORD
-    timeout_s = WIFI_TIMEOUT_S
-
-    print("=== Standalone WiFi/GPIO isolation ===")
-    print("mode:", mode)
-    print("ssid:", ssid)
-    outputs = []
-    try:
-        outputs = init_gpio(mode)
-        time.sleep(1)
-        connect_wifi(ssid, password, timeout_s)
-        return 0
-    except Exception as exc:
-        print("TEST FAILED:", repr(exc))
-        return 1
-    finally:
-        for gpio_name, output in outputs:
-            try:
-                output.value(0)
-                print(gpio_name, "LOW")
-            except Exception as exc:
-                print(gpio_name, "cleanup failed:", exc)
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

version.md

@@ -29,6 +29,11 @@
 # 2.15.16 修复wifi连接问题
 # 2.15.17 修复wifi连接问题
 # 2.15.18 wifi连接成功重新登录
-# 2.16.4  优化射箭延迟
-# 2.17.0  yolo标靶类别识别
-# 3.0.4 26-09-03 9:36 引脚修改：A23 -> P19 red light 
+# 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 
+# 2.15.33 26-8-12 14:03 修改充4g电关机时间 修复切换网络卡住bug

version.py

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

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
@@ -938,12 +938,11 @@ def start_save_shot_worker():
 
 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
@@ -966,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)
@@ -978,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
@@ -1000,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:
@@ -549,15 +549,20 @@ class WiFiManager:
             on_poor_quality_callback: WiFi质量差时的回调函数
         """
         with self._wifi_quality_lock:
-            if self._wifi_quality_monitor_thread is not None and self._wifi_quality_monitor_thread.is_alive():
+            current_thread = self._wifi_quality_monitor_thread
+            current_stop_event = self._wifi_quality_stop_event
+            if (current_thread is not None and current_thread.is_alive()
+                    and not current_stop_event.is_set()):
                 self.logger.warning("[WiFi Monitor] 监测线程已在运行")
                 return
             
             self._network_type_callback = network_type_callback
             self._on_poor_quality_callback = on_poor_quality_callback
-            self._wifi_quality_stop_event.clear()
+            stop_event = threading.Event()
+            self._wifi_quality_stop_event = stop_event
             self._wifi_quality_monitor_thread = threading.Thread(
                 target=self._quality_monitor_loop,
+                args=(stop_event,),
                 daemon=True,
                 name="wifi_quality_monitor"
             )
@@ -568,13 +573,14 @@ class WiFiManager:
         """停止 WiFi 质量监测线程"""
         with self._wifi_quality_lock:
             t = self._wifi_quality_monitor_thread
+            stop_event = self._wifi_quality_stop_event
             if t is None:
                 return
             if not t.is_alive():
                 self._wifi_quality_monitor_thread = None
                 return
         
-        self._wifi_quality_stop_event.set()
+        stop_event.set()
         try:
             t.join(timeout=2.0)
         except Exception as e:
@@ -588,37 +594,41 @@ class WiFiManager:
                     self._wifi_quality_monitor_thread = None
                     self.logger.info("[WiFi Monitor] 已停止后台监测线程")
     
-    def _quality_monitor_loop(self):
+    def _quality_monitor_loop(self, stop_event):
         """
         WiFi 质量监测循环（后台线程）
-        每 5 秒测量一次 RTT 和 RSSI，发现质量差则触发切换
+        每 5 秒检查 STA 关联状态和 RSSI，发现断链或质量差则触发切换
         """
-        while not self._wifi_quality_stop_event.is_set():
+        while not 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)
 
@@ -627,14 +637,10 @@ class WiFiManager:
                         self.logger.warning("[WiFi Monitor] 质量差,切换前快速重试 2 次（每次间隔1秒）")
 
                         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
-                            # )
+                            if stop_event.wait(1.0):
+                                return
+                            reachable2 = self.is_sta_associated()
+                            rtt2 = None
                             rssi2 = self._get_wifi_rssi_dbm()
 
                             # 更新缓存，便于外部查看最新状态
@@ -643,14 +649,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
@@ -665,7 +667,7 @@ class WiFiManager:
                             self._on_poor_quality_callback()
                 
                 # 休眠 5 秒
-                time.sleep(5)
+                stop_event.wait(5.0)
                 
             except Exception as e:
                 self.logger.error(f"[WiFi Monitor] 监测异常：{e}")