diff --git a/4g_download_manager.py b/4g_download_manager.py deleted file mode 100644 index f171d87..0000000 --- a/4g_download_manager.py +++ /dev/null @@ -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() diff --git a/4g_upload_manager.py b/4g_upload_manager.py deleted file mode 100644 index 32e5c35..0000000 --- a/4g_upload_manager.py +++ /dev/null @@ -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= // Close old socket (ignore error) - AT+MIPOPEN=,"TCP","",80 // Open TCP socket - // Wait for +MIPOPEN: ,0 (success) - AT+MIPSEND=, // Send data - // Wait for ">" prompt, then write raw bytes - // Repeat MIPSEND for all chunks - // Wait for +MIPURC: "rtcp" response - AT+MIPCLOSE= // 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: ,0 (success) or +MIPOPEN: , - # 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",,, - 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='',") - print(" key='shootPic/device01/shoot01.png'") - print(" )") - print(" print('Upload result:', result)") diff --git a/app.yaml b/app.yaml index bb6b38d..c3fb77d 100644 --- a/app.yaml +++ b/app.yaml @@ -5,8 +5,6 @@ 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 @@ -18,6 +16,8 @@ files: - laser_manager.py - logger_manager.py - main.py + - model_317828.cvimodel + - model_317828.mud - network.py - ota_curl.sh - ota_manager.py diff --git a/config.py b/config.py index 4ce06cd..416ec3b 100644 --- a/config.py +++ b/config.py @@ -356,7 +356,7 @@ PIN_MAPPINGS = { "A14": "GPIOA14", # 激光开关：低开、高关 "A24": "GPIOA24", # 电源板关机控制 "A25": "GPIOA25", # 电源状态绿灯 - "P19": "GPIOP19", # 电源状态红灯 + "A23": "GPIOA23", # 电源状态红灯 } # ==================== 电源配置 ==================== @@ -368,7 +368,7 @@ CHARGING_AUTO_POWER_OFF_ENABLED = False # 电源状态指示灯 STATUS_LED_GREEN_GPIO = "GPIOA25" -STATUS_LED_RED_GPIO = "GPIOP19" +STATUS_LED_RED_GPIO = "GPIOA23" STATUS_LED_GREEN_ENABLED = True STATUS_LED_RED_ENABLED = True STATUS_LED_ACTIVE_LEVEL = 1