wat

4g_download_manager.py

@@ -1,403 +0,0 @@
-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

@@ -1,450 +0,0 @@
-#!/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,12 +1,10 @@
 id: t11
 name: t11
-version: 2.18.2
+version: 3.0.3
 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

@@ -8,15 +8,6 @@ import threading
 import config
 from logger_manager import logger_manager
 
-_USE_CV = False
-try:
-    import cv2
-    import numpy as np
-    from maix import image as _maix_image
-    _USE_CV = True
-except ImportError:
-    pass
-
 
 class CameraManager:
     """相机管理器（单例）"""
@@ -110,23 +101,7 @@ class CameraManager:
         with self._camera_lock:
             if self._camera is None:
                 self.init_camera()
-            frame = self._camera.read()
-            if frame is not None and _USE_CV:
-                try:
-                    v_flip = getattr(config, 'CAMERA_V_FLIP', False)
-                    h_mirror = getattr(config, 'CAMERA_H_MIRROR', False)
-                    if v_flip or h_mirror:
-                        img_cv = _maix_image.image2cv(frame, False, False)
-                        if v_flip and h_mirror:
-                            img_cv = cv2.flip(img_cv, -1)
-                        elif v_flip:
-                            img_cv = cv2.flip(img_cv, 0)
-                        elif h_mirror:
-                            img_cv = cv2.flip(img_cv, 1)
-                        frame = _maix_image.cv2image(img_cv, False, False)
-                except Exception:
-                    pass
-            return frame
+            return self._camera.read()
     
     def show(self, image):
         """

config.py

@@ -15,8 +15,6 @@ LOCAL_FILENAME = APP_DIR + "/main_tmp.py"
 # 相机初始化分辨率（CameraManager / main.py 使用）
 CAMERA_WIDTH = 640
 CAMERA_HEIGHT = 480
-CAMERA_V_FLIP = True   # 摄像头垂直翻转（上下颠倒时设为 True）
-CAMERA_H_MIRROR = True  # 摄像头水平镜像（左右反了时设为 True）
 
 # 三角形检测缩图比例：默认按相机最长边缩到 1/2（性能更稳；可按需调整）
 # 取值范围建议 (0.25 ~ 1.0]；1.0 表示不缩图
@@ -241,10 +239,10 @@ TRIANGLE_BLACKHAT_KERNEL_FRAC = 0.018  # 核大小 ≈ min(h,w)*frac，取奇数
 # ── YOLO(NPU) 靶环 ROI → 裁剪后再跑三角形（减小 CPU 处理面积）──────────────────
 # 日志里 net_in=W×H 来自 .mud 模型（det.input_width/height），不是这里配置的。
 TRIANGLE_YOLO_ROI_ENABLE = True
-TRIANGLE_YOLO_MODEL_PATH = APP_DIR + "/model_317211.mud"
+TRIANGLE_YOLO_MODEL_PATH = APP_DIR + "/model_270139.mud"
 # 参与 ROI 的类别：多类时只填「整靶/靶环」的 id；不要填角标类，否则 union 仍可对，但 largest 会偏小。
 TRIANGLE_YOLO_RING_CLASS_IDS = (0,)
-TRIANGLE_YOLO_CONF_TH = 0.9
+TRIANGLE_YOLO_CONF_TH = 0.7
 TRIANGLE_YOLO_IOU_TH = 0.45
 # YOLO 首次/临界帧可能在高阈值下 0 框；启用后仅在 0 候选时用较低阈值重试一次。
 # 后续仍会经过 min_box_side、ROI aspect、三角形几何校验，避免直接放大假阳性。
@@ -273,7 +271,7 @@ TRIANGLE_YOLO_PRELOAD_ON_BOOT = False
 TARGET_CLASS_YOLO_ENABLE = True
 TARGET_CLASS_YOLO_MODEL_PATH = APP_DIR + "/model_317828.mud"
 TARGET_CLASS_YOLO_LABELS = (20, 40)
-TARGET_CLASS_YOLO_CONF_TH = 0.66
+TARGET_CLASS_YOLO_CONF_TH = 0.50
 TARGET_CLASS_YOLO_IOU_TH = 0.45
 TARGET_CLASS_YOLO_RETRY_ON_EMPTY = False
 TARGET_CLASS_YOLO_RETRY_CONF_TH = 0.25
@@ -333,17 +331,14 @@ LOG_QUEUE_MAXSIZE = 10000       # 日志队列上限
 MAX_CMD_THREADS = 10            # 并发命令线程上限（防止服务器下发命令时无限创建线程）
 
 # ==================== 图像保存配置 ====================
-SAVE_IMAGE_ENABLED = True  # 是否保存图像（True=保存，False=不保存）
-SAVE_IMAGE_ON_FAILURE = False  # 检测失败时是否强制保存图像（供调试测试用）
+SAVE_IMAGE_ENABLED = False  # 是否保存图像（True=保存，False=不保存）
+SAVE_IMAGE_ON_FAILURE = True  # 检测失败时是否强制保存图像（供调试测试用）
 PHOTO_DIR = "/root/phot"    # 照片存储目录
 MAX_IMAGES = 1000
-SAVE_RAW_IMAGE_ENABLED = False  # 原图保存功能保留，但当前关闭
-RAW_IMAGE_DIR = PHOTO_DIR + "/raw"
-RAW_IMAGE_MAX_IMAGES = MAX_IMAGES
 # Stage2 调试目录（默认 PHOTO_DIR/stage2_roi）内 JPEG 最多保留张数；None 表示与 MAX_IMAGES 相同
 TRIANGLE_BLACK_YOLO_STAGE2_ROI_MAX_IMAGES = None
 
-SHOW_CAMERA_PHOTO_WHILE_SHOOTING = False  # 关闭拍摄实时显示
+SHOW_CAMERA_PHOTO_WHILE_SHOOTING = False  # 是否在拍摄时显示摄像头图像（True=显示，False=不显示），建议在连着USB测试过程中打开
 
 # ==================== OTA配置 ====================
 MAX_BACKUPS = 5

main.py

@@ -136,7 +136,6 @@ def cmd_str():
     sync_system_time_from_4g()
 
     # 2.1 WiFi 热点配网兜底：仅当 STA 与 4G 均不可用时起 AP + HTTP；提交后删 /boot/wifi.ap、建 wifi.sta 并 reboot
-    _ota_pending_path = f"{config.APP_DIR}/ota_pending.json"
     try:
         from wifi_config_httpd import maybe_start_wifi_ap_fallback
 
@@ -172,10 +171,8 @@ def cmd_str():
             and bool(getattr(config, "TARGET_CLASS_YOLO_PRELOAD_ON_BOOT", True))
         )
         _preload_yolo = _preload_yolo or _need_black_preload or _need_target_preload
-        if _preload_yolo and not os.path.exists(f"{config.APP_DIR}/ota_pending.json"):
+        if _preload_yolo:
             preload_yolo_detector(logger)
-        elif _preload_yolo and logger:
-            logger.warning("[YOLO] ota_pending.json found; skip model preload until rollback check")
     except Exception as e:
         if logger:
             logger.warning(f"[YOLO-ROI] 启动预加载异常（不影响后续射箭）: {e}")
@@ -257,11 +254,7 @@ def cmd_str():
     network_manager.read_device_id()
 
     # 5. 创建照片存储目录（如果启用图像保存或检测失败时强制保存）
-    if (
-        config.SAVE_IMAGE_ENABLED
-        or getattr(config, "SAVE_IMAGE_ON_FAILURE", False)
-        or getattr(config, "SAVE_RAW_IMAGE_ENABLED", False)
-    ):
+    if config.SAVE_IMAGE_ENABLED or getattr(config, "SAVE_IMAGE_ON_FAILURE", False):
         photo_dir = config.PHOTO_DIR
         if photo_dir not in os.listdir("/root"):
             try:
@@ -293,13 +286,12 @@ def cmd_str():
         logger.info("系统准备完成...")
 
     last_adc_trigger = 0
-    trigger_adc_val = 0  # 触发时的气压值，气压需降回此值以下才能再次触发
     # 读取一次ADC初始值，防止开机时传感器已有压力导致误触发
-    enable_check = True
     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
 
@@ -389,16 +381,22 @@ def cmd_str():
                 pressure_max = adc_val
             if len(pressure_buf) >= PRESSURE_BATCH_SIZE:
                 _flush_pressure_buf("batch")
-            # 突变增量检测：压力增量大于300时触发
-            # 触发后需等气压降到触发值以下才重新检测增量
-            if adc_val < trigger_adc_val :
-                enable_check = True
-            if (adc_val - last_adc_val) > 500 and enable_check:
+            # 峰值检测：压力从峰值下降时触发，确保捕获到最大冲击时刻
+            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位置
                 hardware_manager.start_idle_timer()  # 重新计时
+                diff_ms = current_time - last_adc_trigger
+                if diff_ms < 3000:
+                    peak_adc_val = 0  # 去抖期间重置峰值
+                    time.sleep_ms(5)
+                    continue
                 last_adc_trigger = current_time
-                trigger_adc_val = adc_val  # 记录触发时的气压值
-                last_adc_val = adc_val  # 更新基准值，防止连续增量误触发
-                enable_check = False
+                peak_adc_val = 0  # 触发后重置峰值
+                # 触发前先把缓存刷出来，避免波形被长耗时处理截断
                 _flush_pressure_buf("before_trigger")
 
                 try:
@@ -417,7 +415,7 @@ def cmd_str():
                         camera_manager.show(camera_manager.read_frame())
                     except Exception as e:
                         pass
-                time.sleep_ms(1)
+                time.sleep_ms(5)
             last_adc_val = adc_val
 
         except Exception as e:

model_270139.mud

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

model_270820.mud

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

model_285484.mud

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

model_317211.mud

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

shoot_manager.py

@@ -8,7 +8,7 @@ from laser_manager import laser_manager
 from logger_manager import logger_manager
 from network import network_manager
 from triangle_target import load_camera_from_xml, load_triangle_positions, try_triangle_scoring
-from vision import estimate_distance, detect_circle_v3, enqueue_save_shot, enqueue_save_raw_shot
+from vision import estimate_distance, detect_circle_v3, enqueue_save_shot
 from maix import image, time
 
 # 缓存相机标定与三角形位置，避免每次射箭重复读磁盘
@@ -322,11 +322,6 @@ def process_shot(adc_val):
     try:
         frame = camera_manager.read_frame()
 
-        # 在任何检测和绘图之前复制原始帧；默认由配置关闭，不增加量产开销。
-        from shot_id_generator import shot_id_generator
-        shot_id = shot_id_generator.generate_id()
-        enqueue_save_raw_shot(frame, shot_id)
-
         # 网络事件移到拍照之后，避免阻塞拍照
         network_manager.safe_enqueue({"shoot_event": "start"}, msg_type=2, high=True)
 
@@ -385,6 +380,10 @@ def process_shot(adc_val):
         if dx is None and dy is None and logger:
             logger.warning("[MAIN] 未检测到偏移量（三角形与圆形均失败），但会保存图像")
 
+        # 生成射箭ID
+        from shot_id_generator import shot_id_generator
+        shot_id = shot_id_generator.generate_id()
+
         if logger:
             logger.info(f"[MAIN] 射箭ID: {shot_id}")
 
@@ -442,13 +441,6 @@ def process_shot(adc_val):
             inner_data["ellipse_center_x"] = None
             inner_data["ellipse_center_y"] = None
 
-        upload_time_ms = int(time_std.time() * 1000)
-        upload_time_sec, upload_time_millis = divmod(upload_time_ms, 1000)
-        inner_data["upload_time"] = (
-            time_std.strftime("%Y-%m-%d %H:%M:%S", time_std.localtime(upload_time_sec))
-            + f".{upload_time_millis:03d}"
-        )
-
         report_data = {"cmd": 1, "data": inner_data}
         if logger:
             logger.info(

test/gpio_high.py

@@ -3,8 +3,8 @@
 from maix import app, gpio, pinmap, time
 
 
-PIN = "A17"
-GPIO_NAME = "GPIOA17"
+PIN = "P19"
+GPIO_NAME = "GPIOP19"
 
 
 def main():

version.md

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

version.py

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

vision.py

@@ -902,15 +902,12 @@ def _save_shot_image_impl(img_cv, center, radius, method, ellipse_params,
 
 
 def _save_worker_loop():
-    """存图 worker：处理标注图和可选的纯原图任务。"""
+    """存图 worker：从队列取任务并调用 _save_shot_image_impl。"""
     while True:
         try:
             item = _save_queue.get()
             if item is None:
                 break
-            if isinstance(item, dict) and item.get("kind") == "raw":
-                _save_raw_image_impl(item["img_cv"], item["shot_id"], item["photo_dir"])
-            else:
             _save_shot_image_impl(*item)
         except Exception as e:
             logger = logger_manager.logger
@@ -939,52 +936,6 @@ def start_save_shot_worker():
         logger.info("[VISION] 存图 worker 线程已启动")
 
 
-def _save_raw_image_impl(img_cv, shot_id, photo_dir):
-    """保存未标注、未裁剪的完整原始帧。"""
-    logger = logger_manager.logger
-    try:
-        os.makedirs(photo_dir, exist_ok=True)
-        filename = os.path.join(photo_dir, f"shot_{shot_id}_raw.jpg")
-        image.cv2image(img_cv, False, False).save(filename)
-        prune_old_images_in_dir(
-            photo_dir,
-            getattr(config, "RAW_IMAGE_MAX_IMAGES", config.MAX_IMAGES),
-            logger,
-            "[VISION-RAW]",
-        )
-        if logger:
-            logger.info(f"[VISION-RAW] 已保存纯原图: {filename}")
-        return filename
-    except Exception as e:
-        if logger:
-            logger.error(f"[VISION-RAW] 保存纯原图失败: {e}")
-        return None
-
-
-def enqueue_save_raw_shot(frame, shot_id, photo_dir=None):
-    """复制并异步保存原始帧；由 SAVE_RAW_IMAGE_ENABLED 控制是否启用。"""
-    if not getattr(config, "SAVE_RAW_IMAGE_ENABLED", False):
-        return
-    if photo_dir is None:
-        photo_dir = getattr(config, "RAW_IMAGE_DIR", os.path.join(config.PHOTO_DIR, "raw"))
-    try:
-        img_copy = np.copy(image.image2cv(frame, False, False))
-        _save_queue.put_nowait({
-            "kind": "raw",
-            "img_cv": img_copy,
-            "shot_id": shot_id,
-            "photo_dir": photo_dir,
-        })
-    except queue.Full:
-        logger = logger_manager.logger
-        if logger:
-            logger.warning("[VISION-RAW] 存图队列已满，跳过本次纯原图保存")
-    except Exception as e:
-        logger = logger_manager.logger
-        if logger:
-            logger.error(f"[VISION-RAW] 复制纯原图失败: {e}")
-
-
 def enqueue_save_shot(result_img, center, radius, method, ellipse_params,
                       laser_point, distance_m, shot_id=None, photo_dir=None,
                       yolo_roi_xyxy=None, force_save=False):