更新wifi连接的代码，改三角形的连接为1秒超时

4g_download_manager.py

@@ -255,6 +255,10 @@ class DownloadManager4G:
         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)"
 

config.py

@@ -167,7 +167,7 @@ TRIANGLE_SHAPE_COS_TOLERANCE = 0.25    # 直角余弦绝对值上限（原 0.20
 # 三角形检测主超时（毫秒）：join 等待子线程的最长时间。
 # 整段 try_triangle_scoring 含「多路径二值化 + C(n,4) 四角评分 + 单应性 + PnP」，往往比黄心圆检测慢。
 # 建议设为实测最坏耗时的 1.2 倍；超时后圆心检测仍会并行跑完，跑完后若三角形已结束则优先用三角形。
-TRIANGLE_TIMEOUT_MS = 1500
+TRIANGLE_TIMEOUT_MS = 1000
 # True=打印各阶段耗时(ms)，用于定位瓶颈；稳定后可 False 减少日志
 TRIANGLE_TIMING_LOG = True
 # True=Stage2 每个子框内传统三角失败时打一条统计（Otsu/Adaptive 下轮廓数与各拒绝原因计数）

network.py

@@ -239,127 +239,28 @@ class NetworkManager:
         
     def connect_wifi(self, ssid, password, verify_host=None, verify_port=None, persist=True, timeout_s=20):
         """
-        连接 Wi-Fi（委托给 wifi_manager）
+        连接 Wi-Fi：委托 ``wifi_manager.connect_wifi``。
-            
+        未指定 ``verify_host``/``verify_port`` 时，可达性校验使用本管理器配置的 ``_server_ip``/``_server_port``。
-        Returns:
-            (ip, error): IP地址和错误信息（成功时error为None）
         """
-
+        def _verify(ip: str):
-        # 配置文件路径定义
-        conf_path = "/etc/wpa_supplicant.conf"
-        ssid_file = "/boot/wifi.ssid"
-        pass_file = "/boot/wifi.pass"
-
-        def _read_text(path: str):
-            try:
-                if os.path.exists(path):
-                    with open(path, "r", encoding="utf-8") as f:
-                        return f.read()
-            except Exception:
-                return None
-            return None
-
-        def _write_text(path: str, content: str):
-            with open(path, "w", encoding="utf-8") as f:
-                f.write(content)
-
-        def _restore_boot(old_ssid: str | None, old_pass: str | None):
-            # 还原 /boot 凭证：原来没有就删除，原来有就写回
-            try:
-                if old_ssid is None:
-                    if os.path.exists(ssid_file):
-                        os.remove(ssid_file)
-                else:
-                    _write_text(ssid_file, old_ssid)
-            except Exception:
-                pass
-            try:
-                if old_pass is None:
-                    if os.path.exists(pass_file):
-                        os.remove(pass_file)
-                else:
-                    _write_text(pass_file, old_pass)
-            except Exception:
-                pass
-
-        old_conf = _read_text(conf_path)
-        old_boot_ssid = _read_text(ssid_file)
-        old_boot_pass = _read_text(pass_file)
-
-        try:
-            # 生成 wpa_supplicant 配置（写 /etc 作为辅助，具体是否生效取决于 S30wifi 脚本）
-            net_conf = os.popen(f'wpa_passphrase "{ssid}" "{password}"').read()
-            if "network={" not in net_conf:
-                raise RuntimeError("Failed to generate wpa config")
-
-            try:
-                _write_text(
-                    conf_path,
-                    "ctrl_interface=/var/run/wpa_supplicant\n"
-                    "update_config=1\n\n"
-                    + net_conf,
-                )
-            except Exception:
-                # 不强制要求写 /etc 成功（某些系统只用 /boot）
-                pass
-
-            # ====== 临时写入 /boot 凭证，触发 WiFi 服务真正尝试连接新 SSID ======
-            _write_text(ssid_file, ssid.strip())
-            _write_text(pass_file, password.strip())
-
-            # 重启 Wi-Fi 服务
-            os.system("/etc/init.d/S30wifi restart")
-
-            # 等待获取 IP
-            import time as std_time
-            wait_s = int(timeout_s) if timeout_s and timeout_s > 0 else 20
-            wait_s = min(max(wait_s, 5), 60)
-            for _ in range(wait_s):
-                ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
-                if ip:
-                    # 拿到 IP 不代表可上网/可访问目标；继续做可达性验证
-                    self._wifi_connected = True
-                    self._wifi_ip = ip
-                    self.logger.info(f"[WIFI] 已连接，IP: {ip}，开始验证网络可用性...")
-
-                    # 验证能访问指定目标（默认使用 TCP 服务器）
             v_host = verify_host if verify_host is not None else self._server_ip
-                    v_port = int(verify_port) if verify_port is not None else int(self._server_port)
+            v_port = verify_port if verify_port is not None else self._server_port
-                    if v_host and v_port:
-                        if not self.is_server_reachable(v_host, v_port, timeout=5):
-                            raise RuntimeError(f"Target unreachable ({v_host}:{v_port})")
-
-                    # ====== 验证通过 ======
-                    if not persist:
-                        # 不持久化：把 /boot 恢复成旧值（不重启，当前连接保持不变）
-                        _restore_boot(old_boot_ssid, old_boot_pass)
-                        self.logger.info("[WIFI] 网络验证通过，但按 persist=False 回滚 /boot 凭证（不重启）")
-                    else:
-                        self.logger.info("[WIFI] 网络验证通过，/boot 凭证已保留（持久化）")
-
-                    return ip, None
-
-                std_time.sleep(1)
-
-            raise RuntimeError("Timeout: No IP obtained")
-
-        except Exception as e:
-            # 失败：回滚 /boot 和 /etc，重启 WiFi 恢复旧网络
-            _restore_boot(old_boot_ssid, old_boot_pass)
             try:
-                if old_conf is not None:
+                v_port_i = int(v_port) if v_port is not None else None
-                    _write_text(conf_path, old_conf)
+            except (TypeError, ValueError):
-            except Exception:
+                v_port_i = None
-                pass
+            if v_host and v_port_i:
-            try:
+                if not self.is_server_reachable(v_host, v_port_i, timeout=5):
-                os.system("/etc/init.d/S30wifi restart")
+                    return False, f"Target unreachable ({v_host}:{v_port_i})"
-            except Exception:
+            return True, ""
-                pass
 
-            self._wifi_connected = False
+        return wifi_manager.connect_wifi(
-            self._wifi_ip = None
+            ssid,
-            self.logger.error(f"[WIFI] 连接/验证失败，已回滚: {e}")
+            password,
-            return None, str(e)
+            verify_callback=_verify,
+            persist=persist,
+            timeout_s=timeout_s,
+        )
     
     def is_server_reachable(self, host, port=80, timeout=5):
         """检查目标主机端口是否可达（用于网络检测）"""
@@ -1157,7 +1058,6 @@ class NetworkManager:
                         v_port = parsed.port or (443 if parsed.scheme == "https" else 80)
                     except Exception:
                         v_host, v_port = None, None
-
                     ip, error = self.connect_wifi(
                         wifi_ssid,
                         wifi_password,
@@ -2036,6 +1936,8 @@ class NetworkManager:
                                                 self.logger.error("ota wifi mode requires ssid and password")
                                                 self.safe_enqueue({"result": "missing_ssid_or_password"}, 2)
                                             else:
+                                                self.logger.info(f"ssid: {ssid}")
+                                                self.logger.info(f"password: {password}")
                                                 ota_manager._start_update_thread()
                                                 _thread.start_new_thread(ota_manager.handle_wifi_and_update, (ssid, password, ota_url))
                                     elif inner_cmd == 6:

ota_manager.py

@@ -758,7 +758,12 @@ class OTAManager:
 
         parsed = urlparse(url)
         host = parsed.hostname
+        # MHTTPREQUEST 的路径必须包含 query（七牛/ OSS 签名、token 多在 ? 后），否则易 403/HTML，header 无 CL → no_header_or_total
         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)"
 

power.py

@@ -159,7 +159,7 @@ def voltage_to_percent(voltage):
         return 0
     if v <= 0:
         return 0
-    return int(round(_BATTERY_MONITOR.get_soc(v)))
+    return int(int(_BATTERY_MONITOR.get_soc(v) * 10) / 10) # 截断而不是四舍五入
 
 
 class BatteryMonitor:

version.py

@@ -4,7 +4,7 @@
 应用版本号
 每次 OTA 更新时，只需要更新这个文件中的版本号
 """
-VERSION = '1.2.11'
+VERSION = '1.2.12'
 
 # 1.2.0 开始使用C++编译成.so，替换部分代码
 # 1.2.1 ota使用加密包
@@ -18,6 +18,7 @@ VERSION = '1.2.11'
 # 1.2.9 增加电源板的控制和自动关机的功能
 # 1.2.10 config formal
 # 1.2.11 增加三角形的单应性算法，适配对应的靶纸
+# 1.2.110 关掉了黑色三角形算法，只用于测试
 
 
 

wifi.py

@@ -13,6 +13,7 @@ from maix import time
 
 import config
 from logger_manager import logger_manager
+from wpa_supplicant_conf import build_sta_conf_open, build_sta_conf_psk
 
 
 class WiFiManager:
@@ -170,23 +171,25 @@ class WiFiManager:
     
     def connect_wifi(self, ssid, password, verify_callback=None, persist=True, timeout_s=20):
         """
-        连接 Wi-Fi（先用新凭证尝试连接并验证可用性；失败自动回滚；成功后再决定是否落盘）
+        连接 Wi-Fi（唯一实现：写 wpa_supplicant + /boot 凭证，MaixPy Wifi.connect，再等 IP 与可选校验）。
 
-        重要：系统的 /etc/init.d/S30wifi 通常会读取 /boot/wifi.ssid 与 /boot/wifi.pass 来连接 WiFi。
+        ``NetworkManager.connect_wifi`` 仅封装本方法（通过 ``verify_callback`` 传入 host/port 校验）。
-        因此要"真正尝试连接新 WiFi"，必须临时写入 /boot/ 触发重启；若失败则把旧值写回去（回滚）。
+
+        重要：``/boot/wpa_supplicant.conf`` 存在时 S30wifi 会优先 cp，避免 shell 传中文 SSID。
 
         Args:
             ssid: WiFi SSID
             password: WiFi密码
-            verify_callback: 验证回调函数，接收 (ip) 参数，返回 (success: bool, error: str)
+            verify_callback: 可选；``(ip) -> (success: bool, error: str)``，在拿到 IP 后调用
-            persist: 是否持久化保存凭证
+            persist: 是否持久化保存凭证（False 时成功后回滚 /boot 与 /etc 中的本次写入）
-            timeout_s: 连接超时时间（秒）
+            timeout_s: 等待 DHCP / 轮询 IP 的超时基数（秒）；Maix 连接超时亦据此推导
 
         Returns:
             (ip, error): IP地址和错误信息（成功时 error 为 None）
         """
         # 配置文件路径定义
         conf_path = "/etc/wpa_supplicant.conf"
+        boot_wpa_path = "/boot/wpa_supplicant.conf"
         ssid_file = "/boot/wifi.ssid"
         pass_file = "/boot/wifi.pass"
 
@@ -222,33 +225,51 @@ class WiFiManager:
             except Exception:
                 pass
 
+        def _restore_boot_wpa(old_wpa: str | None):
+            try:
+                if old_wpa is None:
+                    if os.path.exists(boot_wpa_path):
+                        os.remove(boot_wpa_path)
+                else:
+                    _write_text(boot_wpa_path, old_wpa)
+            except Exception:
+                pass
+
         old_conf = _read_text(conf_path)
         old_boot_ssid = _read_text(ssid_file)
         old_boot_pass = _read_text(pass_file)
+        old_boot_wpa = _read_text(boot_wpa_path) if os.path.exists(boot_wpa_path) else None
 
         try:
-            # 生成 wpa_supplicant 配置（写 /etc 作为辅助，具体是否生效取决于 S30wifi 脚本）
+            try:
-            net_conf = os.popen(f'wpa_passphrase "{ssid}" "{password}"').read()
+                full_conf = build_sta_conf_psk(ssid.strip(), password.strip())
-            if "network={" not in net_conf:
+            except ValueError as ve:
-                raise RuntimeError("Failed to generate wpa config")
+                raise RuntimeError(str(ve)) from ve
 
             try:
-                _write_text(
+                _write_text(conf_path, full_conf)
-                    conf_path,
-                    "ctrl_interface=/var/run/wpa_supplicant\n"
-                    "update_config=1\n\n"
-                    + net_conf,
-                )
             except Exception:
-                # 不强制要求写 /etc 成功（某些系统只用 /boot）
                 pass
+            _write_text(boot_wpa_path, full_conf)
 
-            # ====== 临时写入 /boot 凭证，触发 WiFi 服务真正尝试连接新 SSID ======
+            # 仍写入 ssid/pass，便于其它脚本/人工查看；S30wifi 优先使用 wpa_supplicant.conf
             _write_text(ssid_file, ssid.strip())
             _write_text(pass_file, password.strip())
 
-            # 重启 Wi-Fi 服务
+            from maix import err as maix_err
-            os.system("/etc/init.d/S30wifi restart")
+            from maix import network as maix_net
+
+            self.logger.info(f"[WIFI] Maix connect start ssid={ssid!r}")
+            w = maix_net.wifi.Wifi()
+            connect_timeout_s = int(timeout_s) if timeout_s and timeout_s > 0 else 60
+            connect_timeout_s = max(10, min(connect_timeout_s, 120))
+            e = w.connect(ssid, password, wait=True, timeout=connect_timeout_s)
+            maix_err.check_raise(e, "connect wifi failed")
+            try:
+                maix_ip = w.get_ip()
+            except Exception:
+                maix_ip = None
+            self.logger.info(f"[WIFI] Maix connect ok ip={maix_ip!r}")
 
             # 等待获取 IP
             wait_s = int(timeout_s) if timeout_s and timeout_s > 0 else 20
@@ -271,6 +292,7 @@ class WiFiManager:
                     if not persist:
                         # 不持久化：把 /boot 恢复成旧值（不重启，当前连接保持不变）
                         _restore_boot(old_boot_ssid, old_boot_pass)
+                        _restore_boot_wpa(old_boot_wpa)
                         self.logger.info("[WIFI] 网络验证通过，但按 persist=False 回滚 /boot 凭证（不重启）")
                     else:
                         self.logger.info("[WIFI] 网络验证通过，/boot 凭证已保留（持久化）")
@@ -284,6 +306,7 @@ class WiFiManager:
         except Exception as e:
             # 失败：回滚 /boot 和 /etc，重启 WiFi 恢复旧网络
             _restore_boot(old_boot_ssid, old_boot_pass)
+            _restore_boot_wpa(old_boot_wpa)
             try:
                 if old_conf is not None:
                     _write_text(conf_path, old_conf)
@@ -301,7 +324,7 @@ class WiFiManager:
 
     def persist_sta_credentials(self, ssid: str, password: str, restart_service: bool = True):
         """
-        仅写入 STA 凭证（/etc/wpa_supplicant.conf + /boot/wifi.ssid|pass），
+        仅写入 STA 凭证（/etc/wpa_supplicant.conf、/boot/wpa_supplicant.conf、/boot/wifi.ssid|pass），
         可选是否立即 /etc/init.d/S30wifi restart。
         不做可达性验证。用于热点配网页提交后切换到连接指定路由器。
         password 为空时按开放网络（key_mgmt=NONE）写入。
@@ -314,6 +337,7 @@ class WiFiManager:
             return False, "SSID 为空"
 
         conf_path = "/etc/wpa_supplicant.conf"
+        boot_wpa_path = "/boot/wpa_supplicant.conf"
         ssid_file = "/boot/wifi.ssid"
         pass_file = "/boot/wifi.pass"
 
@@ -323,23 +347,13 @@ class WiFiManager:
 
         try:
             if password:
-                net_conf = os.popen(f'wpa_passphrase "{ssid}" "{password}"').read()
+                full_conf = build_sta_conf_psk(ssid, password)
-                if "network={" not in net_conf:
-                    return False, "wpa_passphrase 失败"
             else:
-                esc = ssid.replace("\\", "\\\\").replace('"', '\\"')
+                full_conf = build_sta_conf_open(ssid)
-                net_conf = (
+            _write_text(conf_path, full_conf)
-                    "network={\n"
+            _write_text(boot_wpa_path, full_conf)
-                    f'    ssid="{esc}"\n'
+        except ValueError as e:
-                    "    key_mgmt=NONE\n"
+            return False, str(e)
-                    "}\n"
-                )
-            _write_text(
-                conf_path,
-                "ctrl_interface=/var/run/wpa_supplicant\n"
-                "update_config=1\n\n"
-                + net_conf,
-            )
         except Exception as e:
             return False, str(e)
 
@@ -582,7 +596,8 @@ class WiFiManager:
                     reachable = True
                     self._last_wifi_rtt_ms = rtt_ms if reachable else None
                     self._last_wifi_rssi_dbm = rssi_dbm
-                    self.logger.debug(f"[WiFi Monitor] - RTT={rtt_ms:.0f}ms, RSSI={rssi_dbm:.0f}dBm")
+                    _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")
                     
                     # 判断质量是否差（切换前做 2 次快速复测，防止瞬时抖动）
                     def _is_bad_now(_reachable, _rtt, _rssi):
@@ -611,9 +626,14 @@ class WiFiManager:
 
                             bad2 = _is_bad_now(reachable2, rtt2, rssi2)
                             try:
+                                _rtt_disp = (
+                                    rtt2
+                                    if rtt2 is not None and rtt2 != float("inf")
+                                    else -1
+                                )
                                 self.logger.info(
                                     f"[WiFi Monitor] 复测{retry_idx+1}/2: reachable={reachable2}, "
-                                    f"rtt={rtt2 if rtt2 != float('inf') else -1:.0f}ms, rssi={rssi2}, bad={bad2}"
+                                    f"rtt={_rtt_disp:.0f}ms, rssi={rssi2}, bad={bad2}"
                                 )
                             except Exception:
                                 pass