400 lines
15 KiB
Python
400 lines
15 KiB
Python
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 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()
|