ZZH000829/archery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove rtt from wifi monitoring

Browse Source
This commit is contained in:
gcw_4spBpAfv
2026-04-03 11:24:29 +08:00
parent ec80107128
commit 685dce2519
4 changed files with 186 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
config.py
View File

@@ -17,6 +17,13 @@ SERVER_IP = "www.shelingxingqiu.com"
SERVER_PORT = 50005
HEARTBEAT_INTERVAL = 15 # 心跳间隔（秒）
# WiFi 质量评估（开机先尝试 WiFi；质量差且 4G 可用则切到 4G，本次上电直至关机锁定 4G）
WIFI_QUALITY_RTT_SAMPLES = 3 # 到业务服务器 TCP 建连耗时采样次数，取中位数
WIFI_QUALITY_RTT_BAD_MS = 600.0 # 中位数超过此值认为延迟过高
WIFI_QUALITY_RTT_WARN_MS = 350.0 # 与 RSSI 联合：超过此值且信号弱也判为差
WIFI_QUALITY_RSSI_BAD_DBM = -80.0 # 低于此 dBm（更负更差）视为信号弱
WIFI_QUALITY_USE_RSSI = True # 是否把 RSSI 纳入综合判定（False 则仅看 RTT）
# ===== TCP over SSL(TLS) 配置 =====
USE_TCP_SSL = False # True=按手册走 MSSLCFG/MIPCFG 绑定 SSL
TCP_LINK_ID = 2 #
@@ -24,11 +31,11 @@ TCP_SSL_PORT = 443 # TLS 端口（不一定必须 443，以服务器为
# SSL profile
SSL_ID = 1 # ssl_id=1
SSL_AUTH_MODE = 1 # 1=单向认证（验证服务器），2=双向
SSL_VERIFY_MODE = 1 # 1=写入并使用 CA 证书；0=不验（仅测试加密，风险大）
SSL_AUTH_MODE = 0 # 1=单向认证（验证服务器），2=双向
SSL_VERIFY_MODE = 1 # 0=不验（仅测试用）；1=写入并使用 CA 证书
SSL_CERT_FILENAME = "test.cer" # 模组里证书名（MSSLCERTWR / MSSLCFG="cert" 用）
SSL_CERT_PATH = "/root/test.cer" # 设备文件系统里 CA 证书路径（你自己放进去）
SSL_CERT_FILENAME = "www.shelingxingqiu.com.crt" # 模组里证书名（MSSLCERTWR / MSSLCFG="cert" 用）
SSL_CERT_PATH = "/root/www.shelingxingqiu.com.crt" # 设备文件系统里 CA 证书路径（你自己放进去）
# MIPOPEN 末尾的参数在不同固件里含义可能不同；按你手册例子保留
MIPOPEN_TAIL = ",,0"
@@ -117,7 +124,7 @@ SAVE_IMAGE_ENABLED = True # 是否保存图像（True=保存，False=不保存
PHOTO_DIR = "/root/phot" # 照片存储目录
MAX_IMAGES = 1000
SHOW_CAMERA_PHOTO_WHILE_SHOOTING = False # 是否在拍摄时显示摄像头图像（True=显示，False=不显示），建议在连着USB测试过程中打开
SHOW_CAMERA_PHOTO_WHILE_SHOOTING = True # 是否在拍摄时显示摄像头图像（True=显示，False=不显示），建议在连着USB测试过程中打开
# ==================== OTA配置 ====================
MAX_BACKUPS = 5

1
main.py
View File

@@ -323,6 +323,7 @@ def cmd_str():
hardware_manager.start_idle_timer() # 重新计时
diff_ms = current_time - last_adc_trigger
if diff_ms < 3000:
logger.info(f"[MAIN] 扳机触发过于频繁, {diff_ms}ms")
continue
last_adc_trigger = current_time
# 触发前先把缓存刷出来，避免波形被长耗时处理截断

307
network.py
View File

@@ -440,20 +440,19 @@ class NetworkManager:
# 1) 开机先尝试 WiFi，并评估质量
if prefer_wifi and self.is_wifi_connected():
wifi_rssi_dbm = self._get_wifi_rssi_dbm()
wifi_rtt_ms, wifi_reachable = self._measure_wifi_tcp_rtt_ms(
host, port,
samples=getattr(config, "WIFI_QUALITY_RTT_SAMPLES", 3),
per_sample_timeout_ms=900,
)
wifi_rtt_ms = 0
wifi_reachable = True
# wifi_rtt_ms, wifi_reachable = self._measure_wifi_tcp_rtt_ms(
# host, port,
# samples=getattr(config, "WIFI_QUALITY_RTT_SAMPLES", 3),
# per_sample_timeout_ms=900,
# )
wifi_bad = self._is_wifi_quality_bad(wifi_rtt_ms, wifi_rssi_dbm)
try:
self.logger.info(
f"[NET] WiFi质量评估：rtt_ms(median)={wifi_rtt_ms:.1f}, rssi_dbm={wifi_rssi_dbm}, "
f"reachable={wifi_reachable}, bad={wifi_bad}"
)
except Exception:
pass
self.logger.info(
f"[NET] WiFi质量评估：rtt_ms(median)={wifi_rtt_ms:.1f}, rssi_dbm={wifi_rssi_dbm}, "
f"reachable={wifi_reachable}, bad={wifi_bad}"
)
# 如果质量差且 4G 可用 -> 切换 4G 并锁定本次会话
if wifi_bad and self.is_4g_available():
@@ -623,7 +622,7 @@ class NetworkManager:
self._tcp_connected = True
self.logger.info("[WIFI-TCP] TCP 连接已建立")
# 启动 WiFi 质量后台监测
# 启动 WiFi 质量后台检测
self._start_wifi_quality_monitor()
return True
@@ -755,7 +754,7 @@ class NetworkManager:
def _disconnect_tcp_via_wifi(self):
"""断开 WiFi TCP 连接并停止监测"""
# 先停止监测线程
# 关闭wifi检测
self._stop_wifi_quality_monitor()
# 再关闭 socket
@@ -1298,6 +1297,7 @@ class NetworkManager:
data = self.receive_tcp_data_via_wifi(timeout_ms=50)
if data:
# 将数据添加到缓冲区
self.logger.info(f"[NET] 接收WiFi数据, {time.time()}")
wifi_manager.recv_buffer += data
# 尝试从缓冲区解析完整的数据包
@@ -1315,10 +1315,12 @@ class NetworkManager:
break
else:
# 数据包不完整，等待更多数据
self.logger.info(f"[NET] 接收WiFi数据不完整, {time.time()}")
break
except:
# 解析失败，清空缓冲区
wifi_manager.recv_buffer = b""
self.logger.info(f"[NET] 接收WiFi数据解析失败, {time.time()}")
break
elif self._network_type == "4g":
# 4G接收数据
@@ -1399,152 +1401,155 @@ class NetworkManager:
data_obj = body.get("data")
if isinstance(data_obj, dict):
inner_cmd = data_obj.get("cmd")
if inner_cmd == 2: # 开启激光并校准
from laser_manager import laser_manager
if not laser_manager.calibration_active:
laser_manager.turn_on_laser()
time.sleep_ms(100)
hardware_manager.stop_idle_timer() # 停表
if not config.HARDCODE_LASER_POINT:
laser_manager.start_calibration()
self.safe_enqueue({"result": "calibrating"}, 2)
else:
# 写死的逻辑，不需要校准激光点
self.safe_enqueue({"result": "laser pos set by hard code"}, 2)
elif inner_cmd == 3: # 关闭激光
from laser_manager import laser_manager
laser_manager.turn_off_laser()
laser_manager.stop_calibration()
hardware_manager.start_idle_timer() # 开表
self.safe_enqueue({"result": "laser_off"}, 2)
elif inner_cmd == 4: # 上报电量
voltage = get_bus_voltage()
battery_percent = voltage_to_percent(voltage)
battery_data = {"battery": battery_percent, "voltage": round(voltage, 3)}
self.safe_enqueue(battery_data, 2)
self.logger.info(f"电量上报: {battery_percent}%")
elif inner_cmd == 5: # OTA 升级
inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
ssid = inner_data.get("ssid")
password = inner_data.get("password")
ota_url = inner_data.get("url")
mode = (inner_data.get("mode") or "").strip().lower()
if not ota_url:
self.logger.error("ota missing_url")
self.safe_enqueue({"result": "missing_url"}, 2)
continue
from ota_manager import ota_manager
if ota_manager.update_thread_started:
self.safe_enqueue({"result": "update_already_started"}, 2)
continue
# 自动判断模式：如果没有明确指定，根据WiFi连接状态和凭证决定
if mode not in ("4g", "wifi"):
self.logger.info("ota missing mode, auto-detecting...")
# 若本次会话已锁定 4G，则 OTA 自动也走 4G，避免后续回切导致体验不一致
if self._session_force_4g:
mode = "4g"
self.logger.info("ota auto-selected: 4g (session locked on 4g)")
else:
# 只有同时满足：WiFi已连接 且 提供了WiFi凭证，才使用WiFi
if self.is_wifi_connected() and ssid and password:
mode = "wifi"
self.logger.info("ota auto-selected: wifi (WiFi connected and credentials provided)")
if inner_cmd == 2: # 开启激光并校准
from laser_manager import laser_manager
if not laser_manager.calibration_active:
laser_manager.turn_on_laser()
time.sleep_ms(100)
hardware_manager.stop_idle_timer() # 停表
if not config.HARDCODE_LASER_POINT:
laser_manager.start_calibration()
self.safe_enqueue({"result": "calibrating"}, 2)
else:
# 写死的逻辑，不需要校准激光点
self.safe_enqueue({"result": "laser pos set by hard code"}, 2)
elif inner_cmd == 3: # 关闭激光
from laser_manager import laser_manager
laser_manager.turn_off_laser()
laser_manager.stop_calibration()
hardware_manager.start_idle_timer() # 开表
self.safe_enqueue({"result": "laser_off"}, 2)
elif inner_cmd == 4: # 上报电量
voltage = get_bus_voltage()
battery_percent = voltage_to_percent(voltage)
battery_data = {"battery": battery_percent, "voltage": round(voltage, 3)}
self.safe_enqueue(battery_data, 2)
self.logger.info(f"电量上报: {battery_percent}%")
elif inner_cmd == 5: # OTA 升级
inner_data = data_obj.get("data", {}) if isinstance(data_obj, dict) else {}
ssid = inner_data.get("ssid")
password = inner_data.get("password")
ota_url = inner_data.get("url")
mode = (inner_data.get("mode") or "").strip().lower()
if not ota_url:
self.logger.error("ota missing_url")
self.safe_enqueue({"result": "missing_url"}, 2)
continue
from ota_manager import ota_manager
if ota_manager.update_thread_started:
self.safe_enqueue({"result": "update_already_started"}, 2)
continue
# 自动判断模式：如果没有明确指定，根据WiFi连接状态和凭证决定
if mode not in ("4g", "wifi"):
self.logger.info("ota missing mode, auto-detecting...")
# 若本次会话已锁定 4G，则 OTA 自动也走 4G，避免后续回切导致体验不一致
if self._session_force_4g:
mode = "4g"
self.logger.info("ota auto-selected: 4g (WiFi not available or no credentials)")
self.logger.info("ota auto-selected: 4g (session locked on 4g)")
else:
# 只有同时满足：WiFi已连接 且 提供了WiFi凭证，才使用WiFi
if self.is_wifi_connected() and ssid and password:
mode = "wifi"
self.logger.info("ota auto-selected: wifi (WiFi connected and credentials provided)")
else:
mode = "4g"
self.logger.info("ota auto-selected: 4g (WiFi not available or no credentials)")
hardware_manager.stop_idle_timer() # 停表，注意OTA停表之后，就没有再开表，因为OTA后面会重启，会重新开表
hardware_manager.stop_idle_timer() # 停表，注意OTA停表之后，就没有再开表，因为OTA后面会重启，会重新开表
if mode == "4g":
ota_manager._set_ota_url(ota_url) # 记录 OTA URL，供命令7使用
ota_manager._start_update_thread()
_thread.start_new_thread(ota_manager.direct_ota_download_via_4g, (ota_url,))
else: # mode == "wifi"
if not ssid or not password:
self.logger.error("ota wifi mode requires ssid and password")
self.safe_enqueue({"result": "missing_ssid_or_password"}, 2)
else:
if mode == "4g":
ota_manager._set_ota_url(ota_url) # 记录 OTA URL，供命令7使用
ota_manager._start_update_thread()
_thread.start_new_thread(ota_manager.handle_wifi_and_update, (ssid, password, ota_url))
elif inner_cmd == 6:
try:
ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
ip = ip if ip else "no_ip"
except:
ip = "error_getting_ip"
self.safe_enqueue({"result": "current_ip", "ip": ip}, 2)
# elif inner_cmd == 7:
# from ota_manager import ota_manager
# if ota_manager.update_thread_started:
# self.safe_enqueue({"result": "update_already_started"}, 2)
# continue
_thread.start_new_thread(ota_manager.direct_ota_download_via_4g, (ota_url,))
else: # mode == "wifi"
if not ssid or not password:
self.logger.error("ota wifi mode requires ssid and password")
self.safe_enqueue({"result": "missing_ssid_or_password"}, 2)
else:
ota_manager._start_update_thread()
_thread.start_new_thread(ota_manager.handle_wifi_and_update, (ssid, password, ota_url))
elif inner_cmd == 6:
try:
ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
ip = ip if ip else "no_ip"
except:
ip = "error_getting_ip"
self.safe_enqueue({"result": "current_ip", "ip": ip}, 2)
# elif inner_cmd == 7:
# from ota_manager import ota_manager
# if ota_manager.update_thread_started:
# self.safe_enqueue({"result": "update_already_started"}, 2)
# continue
# try:
# ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
# except:
# ip = None
# try:
# ip = os.popen("ifconfig wlan0 2>/dev/null | grep 'inet ' | awk '{print $2}'").read().strip()
# except:
# ip = None
# if not ip:
# self.safe_enqueue({"result": "ota_rejected", "reason": "no_wifi_ip"}, 2)
# else:
# # 注意：direct_ota_download 需要 ota_url 参数
# # 如果 ota_manager.ota_url 为 None，需要从其他地方获取
# ota_url_to_use = ota_manager.ota_url
# if not ota_url_to_use:
# self.logger.error("[OTA] cmd=7 但 OTA_URL 未设置")
# self.safe_enqueue({"result": "ota_failed", "reason": "ota_url_not_set"}, 2)
# else:
# ota_manager._start_update_thread()
# _thread.start_new_thread(ota_manager.direct_ota_download, (ota_url_to_use,))
elif inner_cmd == 41:
self.logger.info("[TEST] 收到TCP射箭触发命令")
self._manual_trigger_flag = True
self.safe_enqueue({"result": "trigger_ack"}, 2)
hardware_manager.start_idle_timer() # 重新计时
elif inner_cmd == 42: # 关机命令
self.logger.info("[SHUTDOWN] 收到TCP关机命令，准备关机...")
self.safe_enqueue({"result": "shutdown_ack"}, 2)
time.sleep_ms(1000)
self.disconnect_server()
# 尝试关闭4G模块
try:
with self.get_uart_lock():
hardware_manager.at_client.send("AT+CFUN=0", "OK", 5000)
except:
pass
time.sleep_ms(2000)
os.system("sync") # 刷新文件系统缓存到磁盘，防止数据丢失
time.sleep_ms(500)
# os.system("poweroff")
hardware_manager.power_off()
return
elif inner_cmd == 43: # 上传日志命令
# 格式: {"cmd":43, "data":{"ssid":"xxx","password":"xxx","url":"xxx", ...}}
inner_data = data_obj.get("data", {})
upload_url = inner_data.get("url")
wifi_ssid = inner_data.get("ssid")
wifi_password = inner_data.get("password")
include_rotated = inner_data.get("include_rotated", True)
max_files = inner_data.get("max_files")
archive_format = inner_data.get("archive", "tgz") # tgz 或 zip
# if not ip:
# self.safe_enqueue({"result": "ota_rejected", "reason": "no_wifi_ip"}, 2)
# else:
# # 注意：direct_ota_download 需要 ota_url 参数
# # 如果 ota_manager.ota_url 为 None，需要从其他地方获取
# ota_url_to_use = ota_manager.ota_url
# if not ota_url_to_use:
# self.logger.error("[OTA] cmd=7 但 OTA_URL 未设置")
# self.safe_enqueue({"result": "ota_failed", "reason": "ota_url_not_set"}, 2)
# else:
# ota_manager._start_update_thread()
# _thread.start_new_thread(ota_manager.direct_ota_download, (ota_url_to_use,))
elif inner_cmd == 41:
self.logger.info(f"[TEST] 收到TCP射箭触发命令, {time.time()}")
self._manual_trigger_flag = True
self.safe_enqueue({"result": "trigger_ack"}, 2)
hardware_manager.start_idle_timer() # 重新计时
elif inner_cmd == 42: # 关机命令
self.logger.info("[SHUTDOWN] 收到TCP关机命令，准备关机...")
self.safe_enqueue({"result": "shutdown_ack"}, 2)
time.sleep_ms(1000)
self.disconnect_server()
# 尝试关闭4G模块
try:
with self.get_uart_lock():
hardware_manager.at_client.send("AT+CFUN=0", "OK", 5000)
except:
pass
time.sleep_ms(2000)
os.system("sync") # 刷新文件系统缓存到磁盘，防止数据丢失
time.sleep_ms(500)
# os.system("poweroff")
hardware_manager.power_off()
return
elif inner_cmd == 43: # 上传日志命令
# 格式: {"cmd":43, "data":{"ssid":"xxx","password":"xxx","url":"xxx", ...}}
inner_data = data_obj.get("data", {})
upload_url = inner_data.get("url")
wifi_ssid = inner_data.get("ssid")
wifi_password = inner_data.get("password")
include_rotated = inner_data.get("include_rotated", True)
max_files = inner_data.get("max_files")
archive_format = inner_data.get("archive", "tgz") # tgz 或 zip
hardware_manager.start_idle_timer() # 重新计时
hardware_manager.start_idle_timer() # 重新计时
if not upload_url:
self.logger.error("[LOG_UPLOAD] 缺少 url 参数")
self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_url"}, 2)
else:
self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，目标URL: {upload_url}")
# 在新线程中执行上传，避免阻塞主循环
import _thread
_thread.start_new_thread(
self._upload_log_file,
(upload_url, wifi_ssid, wifi_password, include_rotated, max_files, archive_format)
)
if not upload_url:
self.logger.error("[LOG_UPLOAD] 缺少 url 参数")
self.safe_enqueue({"result": "log_upload_failed", "reason": "missing_url"}, 2)
else:
self.logger.info(f"[LOG_UPLOAD] 收到日志上传命令，目标URL: {upload_url}")
# 在新线程中执行上传，避免阻塞主循环
import _thread
_thread.start_new_thread(
self._upload_log_file,
(upload_url, wifi_ssid, wifi_password, include_rotated, max_files, archive_format)
)
else: # data的结构不是 dict
self.logger.info(f"[NET] body={body}, {time.time()}")
else:
self.logger.info(f"[NET] 未知数据 {body}, {time.time()}")
else:
time.sleep_ms(5)

28
wifi.py
View File

@@ -121,8 +121,8 @@ class WiFiManager:
# 优先用 MaixPy network（如果可用）
try:
from maix import network
wlan = network.WLAN(network.TYPE_WIFI)
if wlan.isconnected():
wifi = network.wifi.Wifi()
if wifi.is_connected():
self._wifi_connected = True
return True
except:
@@ -480,16 +480,19 @@ class WiFiManager:
# 只在 WiFi 连接时才测量
network_type = self._network_type_callback()
if network_type == "wifi" and self._wifi_socket:
# 测量 RTT（1 个样本，快速测量）
rtt_ms, reachable = self._measure_wifi_tcp_rtt_ms(
self._server_ip, self._server_port,
samples=1, per_sample_timeout_ms=600
)
# # 测量 RTT（1 个样本，快速测量）
# rtt_ms, reachable = self._measure_wifi_tcp_rtt_ms(
# self._server_ip, self._server_port,
# samples=1, per_sample_timeout_ms=600
# )
# 获取 RSSI
rssi_dbm = self._get_wifi_rssi_dbm()
# 更新缓存
# 不使用 RTT 测量
rtt_ms = 0
reachable = True
self._last_wifi_rtt_ms = rtt_ms if reachable else None
self._last_wifi_rssi_dbm = rssi_dbm
self.logger.debug(f"[WiFi Monitor] - RTT={rtt_ms:.0f}ms, RSSI={rssi_dbm:.0f}dBm")
@@ -506,10 +509,13 @@ class WiFiManager:
for retry_idx in range(2):
time.sleep_ms(1000)
rtt2, reachable2 = self._measure_wifi_tcp_rtt_ms(
self._server_ip, self._server_port,
samples=1, per_sample_timeout_ms=600
)
# 不使用 RTT 测量
rtt2 = 0
reachable2 = True
# rtt2, reachable2 = self._measure_wifi_tcp_rtt_ms(
# self._server_ip, self._server_port,
# samples=1, per_sample_timeout_ms=600
# )
rssi2 = self._get_wifi_rssi_dbm()
# 更新缓存，便于外部查看最新状态