wsremote.py ESP32 远程调试解决方案
✅ 文件分块传输（1024字节）✅ "WA"/"WB" 固定头。✅ 7个状态的二进制状态机。✅ 78字节请求头格式。
·
开发一个ESP32的WebREPL客户端工具，一直没有稳定可靠的感觉，直到思路打开，用谷歌浏览器F12观察客户端如何与micropython设备对话。
Request URL
http://192.168.1.19:8266/
Request Method
GET
Status Code
200 OK
Remote Address
192.168.1.19:8266
Referrer Policy
strict-origin-when-cross-origin
它揭示了浏览器和 MicroPython 设备之间，在 WebSocket 连接建立之前，发生的第一次HTTP 交互。
第一步：HTTP 初始请求
当你在浏览器地址栏输入 ws://192.168.1.19:8266 并尝试连接时，浏览器实际上首先发送了一个普通的 HTTP GET 请求到相同的地址 http://192.168.1.19:8266/。
请求方法：GET
状态码：200 OK
请求头：这是一个标准的网页请求头，包含了 Accept、User-Agent 等，特别注意，这里还没有 Upgrade: websocket 头。
这说明，MicroPython 的 WebREPL 服务器在端口 8266 上，首先扮演的是一个最简单的 HTTP 文件服务器的角色。它收到这个 GET 请求后，返回了一个 HTML 页面。你之前贴出的那个包含多个 <script src="..."> 标签的 HTML 内容，就是这次响应的结果。
第二步：加载客户端
浏览器接收到这个 HTML 页面后，开始解析并依次加载页面里引用的资源文件：
webrepl_content.js
webrepl.css
term.js
FileSaver.js
webrepl.js
这些文件同样是通过普通的 HTTP GET 请求从 MicroPython 设备（192.168.1.19:8266）获取的。--这里是重大误解
第三步：建立 WebSocket
只有当所有这些 JavaScript 代码都加载并执行完毕后，其中一个脚本（很可能是 webrepl.js 或 webrepl_content.js）才会真正发起 WebSocket 连接请求。这个请求的 Request Headers 里，才会出现我们之前讨论的关键字段 Upgrade: websocket 和 Connection: Upgrade。
这些 FileSaver.js、term.js、webrepl.js 等文件，不是从你的 MicroPython 设备（192.168.1.19）发出的，而是从 https://micropython.org/webrepl/ 加载的。
如何判断的？看 Sources 面板的目录结构：
192.168.1.19:8266 下面只有 (index)——这是设备返回的 HTML 外壳
micropython.org/webrepl 下面才是所有的 .js 和 .css 文件——这些是从官网加载的
这正是你之前看到 Console 警告的根本原因：(index) 这个 HTML 文件里，通过 document.write 动态插入了 <script src="https://micropython.org/webrepl/xxx.js"> 标签，导致浏览器跨域加载脚本。
为什么这样设计？
这种"设备只给 HTML 外壳，JS 从官网加载"的设计，其实很有道理：
节省设备存储空间：MicroPython 设备的 Flash 很宝贵，省下几十 KB 的 JS 文件空间，可以放更多 Python 代码
统一维护：官方可以随时更新 JS 客户端（修复 bug、改进界面），用户每次用都是最新版
跨平台兼容：浏览器直接从官网加载，可以充分利用 CDN 加速，而且避免设备处理复杂的 HTTP 范围请求等
这正是 WebREPL 客户端有时候“难打开”的根本原因——它把最重要的交互逻辑放在遥远的官网上，而你的设备只给了一个空壳。
这一点对你要做的事情其实没有实质性影响。因为：
WebSocket 协议完全由设备决定：无论 JS 从哪里加载，它最终连接的是你的设备 (ws://192.168.1.19:8266/)，必须遵循设备固件实现的协议
JS 文件只是"说明书"：这些 JS 文件里写的代码，正是"如何与设备通信"的说明书。你可以在 Sources 面板里仔细阅读这些 JS 代码，里面包含了：
密码认证的具体实现（MD5 计算、随机数处理）
文件传输的二进制包格式
所有操作码的定义
所以你现在可以直接在浏览器里调试这些 JS 文件，看到最真实的协议实现——这正是你之前想要的"浏览器客户端作为参考答案"。
你从官网下载的这 5 个文件，恰好构成了 WebREPL 客户端的完整源码。这简直是“参考答案”中的“标准答案”——比单纯监视网络帧要深入得多，因为你现在能看到每一帧数据是如何构造出来的。
文件角色全景图
文件 角色 关键作用
webrepl_content.js 页面生成器 动态写入 HTML 结构（解释了之前看到的 document.write 警告）
webrepl.css 样式表 控制界面外观（不重要，可忽略）
term.js 终端模拟器 处理 ANSI 转义序列、光标移动、颜色等（来自第三方库，非核心）
FileSaver.js 文件保存器 浏览器端保存文件的库（来自第三方，非核心）
webrepl.js 核心协议实现 WebSocket 连接、认证、文件传输协议的全部逻辑
核心文件：webrepl.js 协议分析
这是你最需要关注的文件，里面包含了 WebREPL 协议的完整实现。
对你的IDE集成的关键启示
状态机是关键：浏览器用 binary_state 管理不同阶段，你的 Python 实现也需要类似的状态管理
数据包格式完全明确：
请求头固定78字节，格式为 WA + 操作码 + 保留 + 8字节0 + 4字节大小 + 2字节长度 + 64字节文件名
响应头固定4字节，格式为 WB + 2字节状态码
文件数据块：上传时是裸数据，下载时是 [2字节长度][数据]
引导字节机制：下载时，客户端必须主动发送 0x00 请求下一块，这是流控的核心
错误处理：响应中的状态码 0 表示成功，非0表示错误，你的实现需要正确解析
文件名处理：文件名固定用64字节缓冲区，不足补0，这是 C 风格的字符串处理
你现在有了最权威的“参考答案”。接下来可以：
对照着 webrepl.js，用 Python 重新实现每个函数：
connect() → WebSocket 连接建立
put_file() → 构造78字节请求头
文件上传循环 → 分块发送
get_file() → 引导字节循环
重点关注二进制状态机，用 Python 的 asyncio 状态模式实现
测试每个阶段，用真实的 MicroPython 设备验证
webrepl.js是从micropythong.org下载到浏览器的核心文件，里面包含了 WebREPL 协议的完整实现。把 webrepl.js 的核心逻辑融入 Python与micropython设备（单片机）对话。
改进文件传输协议的核心部分，根据webrepl.js的完整实现来改造文件传输功能（文件上传和下载功能）
关键问题是需要实现正确的协议结构和数据传输流程。
“完全掌控”策略的价值所在：基于一个停滞多年的、老旧的客户端（如 webreplcmd）去开发新工具会遇到问题。而基于现代的 websockets 16.0 库，用 Python 去实现这个协议，正是要建立一个全新的、符合现代标准的客户端，彻底摆脱这些历史包袱。
关键设计决策
1. 状态驱动而非事件驱动
webrepl.js 使用全局 binary_state 变量，在单线程事件循环中工作。Python 的 asyncio 允许我们用线性代码表达状态机，每个 await 点就是状态转换点，代码更清晰。
2. 缓冲区管理
下载时需要处理“粘包”问题——设备可能一次性发送多个数据块。参考 webrepl.js 的累积方式，用缓冲区暂存未处理完的数据。
3. 错误处理层次
协议层错误（魔术字错误、状态码非0）
网络层错误（连接断开、超时）
应用层错误（文件不存在、权限问题）
4. 超时控制
每个等待操作都应设置超时，避免永久阻塞。webrepl.js 没有超时是因为浏览器环境不同，我们的工具需要更健壮。
总结：为什么这四大决策如此重要？
状态驱动：让代码的逻辑流与协议的状态机自然对应，而不是用一个全局变量和一个巨大的 switch 来手动管理状态。这是从“手动机床”到“数控机床”的跃迁。
缓冲区管理：网络是不可靠的，数据可能分片、粘包。健壮的缓冲区管理是可靠传输的基石，而不是假设“每次消息都是完整的”。
错误处理层次：网络错误、协议错误、设备错误，各有各的处理方式。分层设计让每个层级的错误都能被恰当地捕获、转换、报告，而不是用一个 Exception 通吃。
超时控制：网络环境千变万化，没有超时的程序就像没有刹车的汽车。合理的超时策略让程序在异常情况下能及时恢复，而不是永久阻塞。
这四大决策共同构成了一个工业化级的 WebREPL 客户端，而不是一个“能用就行”的脚本。这正是你追求的“完全掌控”——不仅掌控协议细节，更掌控代码的质量、健壮性和可维护性。
#!/usr/bin/env python3
"""
WSRemote - ESP32 远程调试工具 v3.2.1
专注于 WebSocket 连接，统一接口操作 ESP32 单片机
版本 3.2.1 改进：
- 新增现代化客户端模式（基于 webrepl.js 完整协议实现）
- 支持完整的 WebREPL 协议，包括挑战-响应认证
- 新增 --modern 命令行参数启用现代化客户端
- 新增 use_modern_client() API 方法
- 保持向后兼容性，传统模式仍然可用
现代化客户端特性：
- 基于 webrepl.js 的完整协议实现
- 正确的挑战-响应认证机制
- 正确的请求头格式（<2sBB8sLH64s）
- 正确的下载数据格式（前2字节表示长度）
- IDE友好的结构化输出
- 更健壮的错误处理和超时控制
REPL 模式说明：
1. 正常 REPL 模式 (Normal REPL / WebREPL):
- 提示符: >>>
- 进入方式: 直接连接后自动进入
- 命令执行: 发送命令 + \r\n，立即执行
- 响应格式: <命令>\r\n<输出>\r\n>>>
2. 原始 REPL 模式 (Raw REPL):
- 提示符: >
- 进入方式: Ctrl+A (0x01)
- 命令执行: 发送命令 + \r\n + Ctrl+D (0x04)
- 响应格式: OK<输出>\r\n\x04\x04>
依赖库:
- websockets (16.0): WebSocket 协议实现
- 标准库: sys, os, argparse, asyncio, struct, socket, time, pathlib
"""
import sys
import os
import argparse
import asyncio
import struct
import socket
import time
import hashlib
from enum import Enum
from dataclasses import dataclass
from pathlib import Path
try:
import websockets
except ImportError:
print("错误: 需要安装 websockets 库")
print("安装命令: pip install websockets")
sys.exit(1)
# WebREPL 协议常量
WEBREPL_REQ_S = "<2sBBQLH64s"
WEBREPL_PUT_FILE = 1
WEBREPL_GET_FILE = 2
WEBREPL_GET_VER = 3
# REPL 模式常量
REPL_MODE_NORMAL = "normal" # 正常 REPL 模式 (WebREPL)
REPL_MODE_RAW = "raw" # 原始 REPL 模式 (Raw REPL)
REPL_MODE_UNKNOWN = "unknown" # 未知模式
# 现代化 WebREPL 客户端相关常量
class WebREPLOpCode(Enum):
"""WebREPL 操作码枚举"""
PUT_FILE = 1
GET_FILE = 2
GET_VER = 3
class ReceiveBuffer:
"""接收缓冲区 - 处理下载数据格式"""
def __init__(self):
self.buffer = b""
def append(self, data: bytes) -> None:
"""添加数据到缓冲区"""
self.buffer += data
def extract_chunks(self) -> tuple:
"""提取数据块和EOF标记"""
# WebREPL下载数据格式：前2字节是长度，然后是数据
chunks = []
eof = False
while len(self.buffer) >= 2:
# 解析长度（小端序）
length = self.buffer[0] | (self.buffer[1] << 8)
if length == 0:
# EOF标记
eof = True
self.buffer = self.buffer[2:]
break
if len(self.buffer) >= 2 + length:
# 提取完整的数据块
chunk = self.buffer[2:2+length]
chunks.append(chunk)
self.buffer = self.buffer[2+length:]
else:
# 数据不完整，等待更多数据
break
return chunks, eof
class ProtocolError(Exception):
"""协议错误"""
pass
class DeviceError(Exception):
"""设备错误"""
pass
@dataclass
class WebREPLConfig:
"""WebREPL 配置类"""
host: str
port: int = 8266
password: str = "123456"
timeout: float = 15.0
verbose: bool = True
class ModernWebREPLClient:
"""现代化 WebREPL 客户端 - 最终版"""
def __init__(self, config: WebREPLConfig):
self.config = config
self.ws = None
self.buffer = ReceiveBuffer() # 仅用于下载
# IDE 友好的输出格式
self.ide_output = {
'success': True,
'messages': [],
'errors': [],
'data': {}
}
def _ide_log(self, message: str, level: str = 'info') -> None:
"""IDE 友好的日志输出"""
timestamp = time.strftime('%H:%M:%S')
formatted_message = f"[{timestamp}] {message}"
if self.config.verbose:
print(formatted_message)
self.ide_output['messages'].append({
'timestamp': timestamp,
'level': level,
'message': message
})
def _ide_error(self, message: str, exception: Exception = None) -> None:
"""IDE 友好的错误输出"""
self.ide_output['success'] = False
error_info = {'message': message}
if exception:
error_info['exception'] = str(exception)
error_info['type'] = type(exception).__name__
self.ide_output['errors'].append(error_info)
self._ide_log(f"错误: {message}", 'error')
async def connect(self) -> None:
"""建立连接并完成认证"""
try:
self._ide_log("建立WebSocket连接")
self.ws = await websockets.connect(
f"ws://{self.config.host}:{self.config.port}/"
)
await self._authenticate()
except Exception as e:
self._ide_error("连接失败", e)
raise
async def _authenticate(self) -> None:
"""挑战-响应认证（适配不同设备）"""
try:
self._ide_log("开始认证流程")
# 1. 接收挑战码（可能是字符串或字节）
self._ide_log("等待挑战码...")
challenge = await self.ws.recv()
self._ide_log(f"收到挑战码数据类型: {type(challenge)}")
# 统一转换为字节
if isinstance(challenge, str):
challenge_bytes = challenge.encode('utf-8')
challenge_str = challenge
self._ide_log(f"挑战码（字符串）: {repr(challenge_str)}")
else:
challenge_bytes = challenge
challenge_str = challenge.decode('utf-8', errors='ignore')
self._ide_log(f"挑战码（字节）长度: {len(challenge_bytes)}")
self._ide_log(f"挑战码（字符串）: {repr(challenge_str)}")
self._ide_log(f"收到挑战码: {len(challenge_bytes)}字节")
# 2. 根据挑战码内容选择认证策略
if 'Password:' in challenge_str:
# 简单密码认证（设备发送"Password:"提示）
self._ide_log("使用简单密码认证")
response = self.config.password + "\n"
self._ide_log(f"发送响应: {repr(response)}")
elif len(challenge_bytes) == 20:
# 标准挑战-响应认证
self._ide_log("使用标准挑战-响应认证")
response_bytes = hashlib.md5(
self.config.password.encode() + challenge_bytes
).digest()
response = response_bytes.decode('latin-1') # 转换为字符串发送
self._ide_log(f"发送响应长度: {len(response_bytes)}字节")
else:
# 其他长度的挑战码，尝试使用密码直接响应
self._ide_log(f"使用适配认证（挑战码长度: {len(challenge_bytes)}）")
response_bytes = hashlib.md5(
self.config.password.encode() + challenge_bytes
).digest()
response = response_bytes.decode('latin-1') # 转换为字符串发送
self._ide_log(f"发送响应长度: {len(response_bytes)}字节")
await self.ws.send(response)
self._ide_log("发送认证响应")
# 3. 验证结果
self._ide_log("等待认证结果...")
result = await self.ws.recv()
self._ide_log(f"收到认证结果数据类型: {type(result)}")
if isinstance(result, str):
self._ide_log(f"认证结果（字符串）: {repr(result)}")
else:
self._ide_log(f"认证结果（字节）长度: {len(result)}")
self._ide_log(f"认证结果（字符串）: {repr(result.decode('utf-8', errors='ignore'))}")
# 检查认证结果（可能的不同格式）
if isinstance(result, str):
if 'WebREPL connected' in result or '>>>' in result or 'Verified!' in result:
self._ide_log("认证成功")
else:
raise ProtocolError(f"认证失败: {result}")
else:
result_str = result.decode('utf-8', errors='ignore')
if 'WebREPL connected' in result_str or '>>>' in result_str or 'Verified!' in result_str:
self._ide_log("认证成功")
else:
raise ProtocolError(f"认证失败: {result_str}")
except Exception as e:
self._ide_error("认证过程异常", e)
raise
def _build_request(self, op: WebREPLOpCode, filename: str, filesize: int = 0) -> bytes:
"""构建78字节请求头（正确格式）"""
# 格式: <2sBB8sLH64s
name_bytes = filename.encode('utf-8')
name_padded = name_bytes.ljust(64, b'\0')
return struct.pack(
"<2sBB8sLH64s",
b"WA", # 固定头
op.value, # 操作码
0, # 保留
b'\0' * 8, # 8字节未知字段（必须全0）
filesize, # 文件大小
len(name_bytes), # 文件名长度
name_padded # 文件名（64字节）
)
def _parse_response(self, data: bytes) -> int:
"""解析4字节响应"""
if len(data) >= 4:
sig, status = struct.unpack("<2sH", data[:4])
if sig == b"WB":
return status
return -1
async def put_file(self, local_path: str, remote_path: str) -> bool:
"""上传文件（正确实现）"""
if not os.path.exists(local_path):
self._ide_error(f"本地文件不存在: {local_path}")
return False
try:
file_size = os.path.getsize(local_path)
self._ide_log(f"开始上传: {local_path} -> {remote_path} ({file_size} 字节)")
# 发送请求
header = self._build_request(WebREPLOpCode.PUT_FILE, remote_path, file_size)
await self.ws.send(header)
self._ide_log("发送PUT_FILE请求")
# 等待确认
resp = await self.ws.recv()
status = self._parse_response(resp)
if status != 0:
raise DeviceError(f"设备拒绝上传: status={status}")
self._ide_log("设备确认上传")
# 发送数据（1024字节块）
sent = 0
with open(local_path, 'rb') as f:
while True:
chunk = f.read(1024)
if not chunk:
break
await self.ws.send(chunk)
sent += len(chunk)
if sent % 1024 == 0:
self._ide_log(f"上传进度: {sent}/{file_size} 字节")
# 等待完成
final = await self.ws.recv()
status = self._parse_response(final)
if status != 0:
raise DeviceError(f"上传完成但状态异常: status={status}")
self._ide_log(f"上传成功: {file_size} 字节")
self.ide_output['data']['upload'] = {
'local_path': local_path,
'remote_path': remote_path,
'size': file_size,
'success': True
}
return True
except Exception as e:
self._ide_error("文件上传异常", e)
return False
async def get_file(self, remote_path: str, local_path: str) -> bool:
"""下载文件（使用直接的 WebSocket 方法）"""
try:
self._ide_log(f"开始下载: {remote_path} -> {local_path}")
# 构建GET_FILE请求
name_bytes = remote_path.encode('utf-8')
name_padded = name_bytes.ljust(64, b'\0')
# 构建请求头
header = struct.pack(
"<2sBB8sLH64s",
b"WA", # 固定头
2, # 操作码: GET_FILE
0, # 保留
b'\0' * 8, # 8字节未知字段
0, # 文件大小（GET_FILE时为0）
len(name_bytes), # 文件名长度
name_padded # 文件名
)
# 发送请求
await self.ws.send(header)
self._ide_log("发送GET_FILE请求")
# 等待确认
self._ide_log("等待设备确认...")
resp = await self.ws.recv()
self._ide_log(f"收到确认数据: {len(resp)} 字节")
# 解析响应状态
if len(resp) >= 4:
sig, status = struct.unpack("<2sH", resp[:4])
if status != 0:
raise DeviceError(f"设备拒绝下载: status={status}")
else:
raise DeviceError("无效的响应")
self._ide_log("设备确认下载")
# 开始接收
await self.ws.send(b'\0') # 首个引导字节
self._ide_log("发送首个引导字节")
file_data = bytearray()
buffer = b""
while True:
self._ide_log("等待接收数据...")
data = await self.ws.recv()
self._ide_log(f"收到数据: {len(data)} 字节, 类型: {type(data)}")
# 确保数据是字节类型
if isinstance(data, str):
data = data.encode('utf-8')
buffer += data
# 处理数据
while len(buffer) >= 2:
# 解析长度
length = buffer[0] | (buffer[1] << 8)
if length == 0:
# 文件结束
self._ide_log("收到EOF标记")
buffer = buffer[2:]
break
if len(buffer) >= 2 + length:
# 提取数据块
chunk = buffer[2:2+length]
file_data.extend(chunk)
buffer = buffer[2+length:]
# 发送下一个引导字节
await self.ws.send(b'\0')
self._ide_log(f"接收数据块: {len(chunk)} 字节")
else:
# 数据不完整
break
if length == 0:
break
# 等待最终状态
self._ide_log("等待最终状态...")
final = await self.ws.recv()
self._ide_log(f"收到最终状态: {len(final)} 字节")
# 解析最终状态
if len(final) >= 4:
sig, status = struct.unpack("<2sH", final[:4])
if status != 0:
raise DeviceError(f"下载完成但状态异常: status={status}")
else:
raise DeviceError("无效的最终响应")
# 保存文件
os.makedirs(os.path.dirname(local_path) or '.', exist_ok=True)
with open(local_path, 'wb') as f:
f.write(file_data)
self._ide_log(f"下载成功: {len(file_data)} 字节")
self.ide_output['data']['download'] = {
'remote_path': remote_path,
'local_path': local_path,
'size': len(file_data),
'success': True
}
return True
except Exception as e:
self._ide_error("文件下载异常", e)
import traceback
self._ide_log(f"异常详情: {traceback.format_exc()}")
return False
async def close(self) -> None:
"""关闭连接"""
if self.ws:
await self.ws.close()
self.ws = None
async def exec(self, command: str) -> str:
"""执行命令
参数:
command: 要执行的命令
返回:
str: 命令执行结果
"""
try:
self._ide_log(f"执行命令: {command}")
# 发送命令
await self.ws.send(command + "\r\n")
await asyncio.sleep(1.0) # 增加等待时间，确保命令执行完成
# 读取响应
response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 10.0: # 增加超时时间
try:
part = await asyncio.wait_for(self.ws.recv(), timeout=1.0)
self._ide_log(f"收到响应部分: {repr(part)}")
response += part
if '>>>' in response:
self._ide_log("收到提示符，停止读取")
break
except asyncio.TimeoutError:
self._ide_log("读取超时，停止读取")
break
self._ide_log(f"完整响应: {repr(response)}")
# 解析响应
# 过滤掉命令echo和提示符
lines = response.split('\n')
filtered = []
for line in lines:
line = line.strip()
self._ide_log(f"处理行: {repr(line)}")
if line and line != '>>>' and not line.startswith('>>>') and not line == command:
filtered.append(line)
result = '\n'.join(filtered) if filtered else response
self._ide_log(f"命令执行结果: {repr(result)}")
return result
except Exception as e:
self._ide_error("命令执行异常", e)
return ""
class WebSocketFrame:
"""WebSocket 帧解析器（用于原生 socket）"""
def __init__(self, sock, verbose=False):
self.s = sock
self.buf = b""
self.verbose = verbose
def recvexactly(self, sz):
"""精确读取指定字节数"""
res = b""
while sz:
try:
data = self.s.recv(sz)
if not data:
break
res += data
sz -= len(data)
except socket.timeout:
break
return res
def write(self, data):
"""发送二进制数据（WebSocket 帧）"""
l = len(data)
if l < 126:
hdr = struct.pack(">BB", 0x82, l)
else:
hdr = struct.pack(">BBH", 0x82, 126, l)
self.s.send(hdr)
self.s.send(data)
def writetext(self, data):
"""发送文本数据（WebSocket 帧）"""
l = len(data)
if l < 126:
hdr = struct.pack(">BB", 0x81, l)
else:
hdr = struct.pack(">BBH", 0x81, 126, l)
self.s.send(hdr)
self.s.send(data)
def read_text_frame(self):
"""读取一个文本帧"""
# 如果缓冲区中有数据，先检查是否是文本帧
if self.buf:
if self.verbose:
print(f"[WebSocketFrame] 缓冲区长度: {len(self.buf)}, 内容: {repr(self.buf[:100])}")
# 尝试解析缓冲区中的数据
if len(self.buf) >= 2:
fl = struct.unpack(">B", self.buf[0:1])[0]
sz = struct.unpack(">B", self.buf[1:2])[0]
if self.verbose:
print(f"[WebSocketFrame] 帧头: FIN={fl & 0x80}, opcode={fl & 0x0F}, 长度={sz}")
# 检查是否为文本帧
if fl == 0x81:
# 计算实际数据长度
data_start = 2
if sz == 126:
if len(self.buf) >= 4:
sz = struct.unpack(">H", self.buf[2:4])[0]
data_start = 4
else:
# 数据不完整，等待更多数据
if self.verbose:
print("[WebSocketFrame] 数据不完整（sz=126）")
pass
elif sz == 127:
if len(self.buf) >= 10:
sz = struct.unpack(">Q", self.buf[2:10])[0]
data_start = 10
else:
# 数据不完整，等待更多数据
if self.verbose:
print("[WebSocketFrame] 数据不完整（sz=127）")
pass
if self.verbose:
print(f"[WebSocketFrame] 数据长度: {sz}, 数据起始位置: {data_start}")
# 检查是否有足够的数据
if data_start + sz <= len(self.buf):
data = self.buf[data_start:data_start + sz]
self.buf = self.buf[data_start + sz:]
if self.verbose:
print(f"[WebSocketFrame] 解析成功: {repr(data[:100])}")
return data
else:
if self.verbose:
print(f"[WebSocketFrame] 数据不完整: 需要 {data_start + sz} 字节，实际有 {len(self.buf)} 字节")
else:
if self.verbose:
print(f"[WebSocketFrame] 不是文本帧 (opcode={fl & 0x0F})")
# 读取 WebSocket 帧头（第一个字节）
fl_data = self.s.recv(1)
if len(fl_data) < 1:
return b""
fl = struct.unpack(">B", fl_data)[0]
# 读取帧头（第二个字节）
sz_data = self.s.recv(1)
if len(sz_data) < 1:
return b""
sz = struct.unpack(">B", sz_data)[0]
# 检查是否为文本帧
if fl != 0x81:
# 不是文本帧，跳过
if sz == 126:
sz = struct.unpack(">H", self.recvexactly(2))[0]
elif sz == 127:
sz = struct.unpack(">Q", self.recvexactly(8))[0]
# 读取并丢弃数据
while sz:
chunk = self.s.recv(sz)
if not chunk:
break
sz -= len(chunk)
# 递归读取下一个帧
return self.read_text_frame()
# 读取文本帧数据
if sz == 126:
sz = struct.unpack(">H", self.recvexactly(2))[0]
elif sz == 127:
sz = struct.unpack(">Q", self.recvexactly(8))[0]
if sz == 0:
return b""
data = self.recvexactly(sz)
if len(data) < sz:
return b""
return data
def read(self, size, text_ok=False, size_match=True):
"""读取指定字节数"""
if not self.buf:
# 读取 WebSocket 帧头
hdr = self.recvexactly(2)
if len(hdr) < 2:
return b""
fl, sz = struct.unpack(">BB", hdr)
if sz == 126:
hdr = self.recvexactly(2)
if len(hdr) < 2:
return b""
(sz,) = struct.unpack(">H", hdr)
# 跳过非二进制帧（如果需要）
if fl == 0x82 or (text_ok and fl == 0x81):
data = self.recvexactly(sz)
if len(data) < sz:
return b""
self.buf = data
else:
# 跳过不期望的帧
while sz:
skip = self.s.recv(sz)
if not skip:
break
sz -= len(skip)
return self.read(size, text_ok, size_match)
d = self.buf[:size]
self.buf = self.buf[size:]
if size_match and len(d) != size:
return b""
return d
class WSRemote:
"""ESP32 WebSocket 远程调试工具类 - v3.2.1
改进特性：
- 自动检测 REPL 模式并设置标志位
- 使用 websockets 库作为默认执行方式
- 自动检测并显示固件版本信息
- 用户透明：明确提示固件版本和适配模式
API:
- connect() / disconnect() - 连接管理
- detect_repl_mode() - 检测 REPL 模式
- get_firmware_info() - 获取固件版本信息
- put() / get() - 文件传输（上传/下载）
- ls() - 列出文件
- exec() - 执行命令
"""
def __init__(self, connection_type='websocket', **kwargs):
self.connection_type = connection_type
self.verbose = kwargs.get('verbose', True)
# 保存连接参数
self.host = kwargs.get('host')
self.port = kwargs.get('port', 8266)
self.password = kwargs.get('password', '123456')
# REPL 模式标志
self.repl_mode = REPL_MODE_UNKNOWN
self.repl_mode_detected = False
# 固件信息
self.firmware_version = None
self.firmware_info = None
# 事件循环（延迟初始化，按需创建）
self._loop = None
# 持久的 REPL 连接（用于命令执行）
self._repl_connection = None
self._repl_connected = False
# 文件传输连接（独立连接，避免与 REPL 命令冲突）
self._file_connection = None
self._file_connected = False
# 现代化客户端支持
self._modern_client = None
self._use_modern_client = kwargs.get('use_modern', False)
if self.verbose:
print(f"[WSRemote] 初始化: host={self.host}, port={self.port}")
if self._use_modern_client:
print("[WSRemote] 现代化客户端模式已启用")
def use_modern_client(self, enable=True):
"""启用或禁用现代化客户端
Args:
enable: 是否启用现代化客户端
"""
self._use_modern_client = enable
if self.verbose:
if enable:
print("[WSRemote] 现代化客户端模式已启用")
else:
print("[WSRemote] 现代化客户端模式已禁用")
def _ensure_modern_client(self):
"""确保现代化客户端已初始化"""
if not self._modern_client:
config = WebREPLConfig(
host=self.host,
port=self.port,
password=self.password,
verbose=self.verbose
)
self._modern_client = ModernWebREPLClient(config)
def _run_modern_async(self, coro):
"""运行现代化客户端的异步协程"""
self._ensure_modern_client()
loop = self._get_event_loop()
return loop.run_until_complete(coro)
def _get_event_loop(self):
"""获取或创建事件循环
复用单个事件循环，避免每次执行都创建销毁
"""
if self._loop is None or self._loop.is_closed():
try:
# 尝试获取当前运行的事件循环（Python 3.7+）
self._loop = asyncio.get_running_loop()
except RuntimeError:
# 当前线程没有运行的事件循环，创建新的
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
return self._loop
def _run_async(self, coro):
"""运行异步协程
使用复用的事件循环执行异步操作
参数:
coro: 异步协程对象
返回:
协程的执行结果
"""
loop = self._get_event_loop()
return loop.run_until_complete(coro)
def _ensure_repl_connection(self):
"""确保 REPL 连接存在且有效
WebREPL 协议限制：不能同时维护多个连接
如果已有连接，先关闭再创建新连接
返回:
bool: 连接是否成功
"""
# WebREPL 协议限制：不能复用连接
# 如果已有连接，先关闭
if self._repl_connected and self._repl_connection:
if self.verbose:
print("[WSRemote] 关闭现有 REPL 连接...")
try:
self._run_async(self._repl_connection.close())
except:
pass
self._repl_connection = None
self._repl_connected = False
import time
time.sleep(0.5)
# 创建新连接
if self.verbose:
print("[WSRemote] 创建新的 REPL 连接...")
async def _connect_repl():
uri = f"ws://{self.host}:{self.port}/"
try:
websocket = await websockets.connect(
uri,
ping_interval=None,
ping_timeout=None,
close_timeout=1.0
)
except Exception as e:
if self.verbose:
print(f"[WSRemote] 连接失败: {e}")
return None
try:
# 登录
if not await self._webrepl_login_websockets(websocket):
await websocket.close()
return None
# 根据模式进入正确状态
if self.repl_mode == REPL_MODE_NORMAL:
await websocket.send("\x03") # Ctrl+C
else:
await websocket.send("\x01") # Ctrl+A 进入原始模式
await asyncio.sleep(1.0)
# 清空初始响应
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.5)
except asyncio.TimeoutError:
pass
return websocket
except Exception as e:
if self.verbose:
print(f"[WSRemote] 连接初始化失败: {e}")
await websocket.close()
return None
self._repl_connection = self._run_async(_connect_repl())
if self._repl_connection:
self._repl_connected = True
if self.verbose:
print("[WSRemote] REPL 连接已建立")
return True
else:
self._repl_connected = False
return False
def close(self):
"""关闭连接"""
# 关闭现代化客户端连接
if self._modern_client:
try:
self._run_modern_async(self._modern_client.close())
if self.verbose:
print("[WSRemote] 现代化客户端连接已关闭")
except:
pass
self._modern_client = None
# 关闭 REPL 连接
if self._repl_connection:
try:
self._run_async(self._repl_connection.close())
if self.verbose:
print("[WSRemote] REPL 连接已关闭")
except:
pass
self._repl_connection = None
self._repl_connected = False
# 关闭文件传输连接
if self._file_connection:
try:
self._file_connection.close()
if self.verbose:
print("[WSRemote] 文件传输连接已关闭")
except:
pass
self._file_connection = None
self._file_connected = False
# 关闭事件循环
if self._loop is not None and not self._loop.is_closed():
self._loop.close()
self._loop = None
if self.verbose:
print("[WSRemote] 事件循环已关闭")
def upload_file(self, local_path, remote_path, max_retries=3):
"""上传文件（带重试机制）
Args:
local_path: 本地文件路径
remote_path: 远程文件路径
max_retries: 最大重试次数
Returns:
bool: 上传是否成功
"""
for retry in range(max_retries):
if self.verbose:
print(f"[WSRemote] 上传文件尝试 {retry + 1}/{max_retries}")
try:
success = self.put(local_path, remote_path)
if success:
if self.verbose:
print(f"[WSRemote] 文件上传成功: {local_path} -> {remote_path}")
return True
else:
if self.verbose:
print(f"[WSRemote] 文件上传失败，重试...")
time.sleep(2)
continue
except Exception as e:
if self.verbose:
print(f"[WSRemote] 上传错误: {e}")
time.sleep(2)
continue
if self.verbose:
print(f"[WSRemote] 文件上传失败（已重试最大次数）")
return False
def upload_files(self, file_list, max_retries=3):
"""批量上传文件
Args:
file_list: 文件列表，格式为 [(local_path, remote_path), ...]
max_retries: 最大重试次数
Returns:
dict: 上传结果，键为本地文件路径，值为是否成功
"""
results = {}
for local_path, remote_path in file_list:
success = self.upload_file(local_path, remote_path, max_retries)
results[local_path] = success
# 短暂休息，避免连接问题
time.sleep(1)
return results
def restart_device(self):
"""重启设备
Returns:
bool: 重启命令是否发送成功
"""
try:
if self.verbose:
print("[WSRemote] 发送重启命令...")
self.exec("import machine; machine.reset()")
if self.verbose:
print("[WSRemote] 重启命令已发送")
return True
except Exception as e:
if self.verbose:
print(f"[WSRemote] 重启失败: {e}")
return False
def stop_running_program(self):
"""停止设备上运行的程序
Returns:
bool: 操作是否成功
"""
try:
if self.verbose:
print("[WSRemote] 停止运行程序...")
# 发送 Ctrl+C 停止当前运行的程序
self.exec("\x03")
time.sleep(1)
# 发送 Ctrl+D 确保进入REPL
self.exec("\x04")
time.sleep(1)
if self.verbose:
print("[WSRemote] 程序已停止")
return True
except Exception as e:
if self.verbose:
print(f"[WSRemote] 停止程序失败: {e}")
return False
def verify_file(self, remote_path):
"""验证远程文件是否存在
Args:
remote_path: 远程文件路径
Returns:
bool: 文件是否存在
"""
try:
filename = remote_path.split('/')[-1]
files = self.exec("import uos; print(uos.listdir('/'))")
return filename in str(files)
except Exception as e:
if self.verbose:
print(f"[WSRemote] 验证文件失败: {e}")
return False
def get_file_size(self, remote_path):
"""获取远程文件大小
Args:
remote_path: 远程文件路径
Returns:
int: 文件大小（字节），失败返回-1
"""
try:
size_output = self.exec(f"import uos; print(uos.stat('{remote_path}')[6])")
if size_output:
# 提取最后一行的数字
lines = size_output.split('\n')
for line in reversed(lines):
line = line.strip()
if line and line.isdigit():
return int(line)
return -1
except Exception as e:
if self.verbose:
print(f"[WSRemote] 获取文件大小失败: {e}")
return -1
def _create_file_connection(self):
"""创建独立的文件传输连接（原生 socket）
根据 webrepl.js 的完整 WebREPL 协议实现
注意：文件传输使用独立的连接，避免与 REPL 命令连接冲突
"""
if self.verbose:
print("[WSRemote] 创建文件传输连接...")
# 创建新的 socket 连接
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(15) # 增加超时时间
try:
sock.connect((self.host, self.port))
except Exception as e:
if self.verbose:
print(f"[WSRemote] 文件传输连接失败: {e}")
return None, None
# WebSocket 握手（根据 webrepl.js 的实现）
handshake = b"GET / HTTP/1.1\r\n"
handshake += b"Host: %s\r\n" % self.host.encode()
handshake += b"Connection: Upgrade\r\n"
handshake += b"Upgrade: websocket\r\n"
handshake += b"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" # 标准测试密钥
handshake += b"Sec-WebSocket-Version: 13\r\n"
handshake += b"\r\n"
sock.send(handshake)
# 读取 HTTP 握手响应
response = b""
start_time = time.time()
while time.time() - start_time < 5.0: # 5秒超时
try:
chunk = sock.recv(1024)
if not chunk:
break
response += chunk
# 检查是否已经收到完整的 HTTP 响应
if b"\r\n\r\n" in response:
# HTTP 响应已完成，检查是否握手成功
if b"101 Switching Protocols" in response:
if self.verbose:
print("[WSRemote] WebSocket 握手成功")
break
else:
if self.verbose:
print(f"[WSRemote] 握手失败: {repr(response[:200])}")
sock.close()
return None, None
except socket.timeout:
break
if self.verbose:
print(f"[WSRemote] 握手响应: {repr(response[:200])}")
# 创建 WebSocket 帧处理器
ws = WebSocketFrame(sock, verbose=self.verbose)
# 保存所有 HTTP 响应之后的数据到缓冲区
if b"\r\n\r\n" in response:
pos = response.find(b"\r\n\r\n") + 4
ws.buf = response[pos:]
if self.verbose and ws.buf:
print(f"[WSRemote] 缓冲区数据: {repr(ws.buf[:100])}")
return ws, sock
def _create_socket_connection(self):
"""创建原生 socket 连接并返回 WebSocketFrame（兼容旧代码）"""
return self._create_file_connection()
def _webrepl_login_native(self, ws, sock):
"""原生 socket WebREPL 登录（文件传输专用）
根据 webrepl.js 的实现，文件传输连接需要不同的认证方式
"""
if self.verbose:
print("[WSRemote] 文件传输连接认证...")
# 文件传输连接的特殊认证方式
# 根据 webrepl.js 的实现，文件传输连接可能不需要密码提示
# 直接发送密码并检查响应
# 首先检查缓冲区中是否有密码提示
password_prompt_received = False
start_time = time.time()
# 检查缓冲区中是否已有密码提示
if ws.buf:
if self.verbose:
print(f"[WSRemote] 缓冲区数据: {repr(ws.buf[:100])}")
try:
# 尝试解析缓冲区中的文本数据
data_str = ws.buf.decode('utf-8', errors='ignore')
if self.verbose:
print(f"[WSRemote] 缓冲区文本: {repr(data_str[:100])}")
if 'Password:' in data_str:
password_prompt_received = True
if self.verbose:
print("[WSRemote] 缓冲区中已找到密码提示")
except Exception as e:
if self.verbose:
print(f"[WSRemote] 解析缓冲区错误: {e}")
# 如果没有在缓冲区中找到，尝试直接发送密码
if not password_prompt_received:
if self.verbose:
print("[WSRemote] 尝试直接发送密码...")
# 直接发送密码（根据 webrepl.js 的实现）
ws.writetext(self.password.encode("utf-8") + b"\n")
time.sleep(1.0)
# 检查响应
try:
# 尝试读取响应
data = ws.read_text_frame()
if data:
data_str = data.decode('utf-8', errors='ignore')
if self.verbose:
print(f"[WSRemote] 密码响应: {repr(data_str[:100])}")
# 检查是否认证成功
if b">>>" in data or b"WebREPL connected" in data_str:
if self.verbose:
print("[WSRemote] 文件传输连接认证成功")
return True
else:
# 没有响应，可能连接已关闭
if self.verbose:
print("[WSRemote] 未收到密码响应")
except Exception as e:
if self.verbose:
print(f"[WSRemote] 密码认证错误: {e}")
# 传统方式：等待密码提示
if not password_prompt_received:
while time.time() - start_time < 10.0:
try:
# 尝试读取文本帧
data = ws.read_text_frame()
if data:
data_str = data.decode('utf-8', errors='ignore')
if self.verbose:
print(f"[WSRemote] 收到数据: {repr(data_str[:100])}")
if 'Password:' in data_str:
password_prompt_received = True
if self.verbose:
print("[WSRemote] 收到密码提示")
break
else:
# 没有读到数据，等待一下再试
time.sleep(0.5)
except socket.timeout:
continue
except Exception as e:
if self.verbose:
print(f"[WSRemote] 读取错误: {e}")
time.sleep(0.5)
continue
if password_prompt_received:
# 发送密码
if self.verbose:
print("[WSRemote] 发送密码...")
ws.writetext(self.password.encode("utf-8") + b"\n")
time.sleep(2.0)
# 读取登录响应
login_resp = b""
start_time = time.time()
while time.time() - start_time < 10.0:
try:
data = ws.read_text_frame()
if data:
login_resp += data
if self.verbose:
print(f"[WSRemote] 登录响应: {repr(data.decode('utf-8', errors='ignore')[:100])}")
if b">>>" in login_resp or b"WebREPL connected" in login_resp:
break
else:
break
except socket.timeout:
continue
except Exception as e:
if self.verbose:
print(f"[WSRemote] 读取登录响应错误: {e}")
break
if b"WebREPL connected" in login_resp or b">>>" in login_resp:
if self.verbose:
print("[WSRemote] WebREPL 登录成功")
return True
if self.verbose:
print("[WSRemote] 文件传输连接认证失败")
return False
async def _webrepl_login_websockets(self, websocket):
"""websockets 库 WebREPL 登录"""
try:
await asyncio.sleep(1.5)
# 接收密码提示
response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 8.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=3.0)
response += part
if self.verbose:
print(f"[WSRemote] 收到: {repr(part[:50])}")
if ':' in response:
break
except asyncio.TimeoutError:
if response:
break
continue
except Exception as e:
if self.verbose:
print(f"[WSRemote] 接收错误: {e}")
break
if ':' not in response:
if self.verbose:
print(f"[WSRemote] 未收到密码提示，响应: {repr(response[:100])}")
return False
# 发送密码
if self.verbose:
print("[WSRemote] 发送密码...")
await websocket.send(self.password + "\n")
await asyncio.sleep(1.5)
# 接收登录响应
login_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 8.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=3.0)
login_response += part
if self.verbose:
print(f"[WSRemote] 登录响应: {repr(part[:50])}")
if 'WebREPL connected' in login_response or '>>>' in login_response:
break
except asyncio.TimeoutError:
if login_response:
break
continue
except Exception as e:
if self.verbose:
print(f"[WSRemote] 接收登录响应错误: {e}")
break
if 'WebREPL connected' not in login_response and '>>>' not in login_response:
if self.verbose:
print(f"[WSRemote] 登录失败，响应: {repr(login_response[:100])}")
return False
if self.verbose:
print("[WSRemote] WebREPL 登录成功")
return True
except Exception as e:
if self.verbose:
print(f"[WSRemote] WebREPL 登录错误: {e}")
return False
def get_firmware_info(self):
"""获取固件版本信息
返回:
dict: 包含固件版本信息的字典
"""
if self.firmware_version:
return {
'version': self.firmware_version,
'info': self.firmware_info,
'repl_mode': self.repl_mode
}
# 使用 websockets 方式获取固件信息
result = self._exec_websockets("import sys; print(sys.version)")
if result:
self.firmware_version = result.strip()
result = self._exec_websockets("import sys; print(sys.implementation)")
if result:
self.firmware_info = result.strip()
return {
'version': self.firmware_version,
'info': self.firmware_info,
'repl_mode': self.repl_mode
}
def detect_repl_mode(self):
"""检测 REPL 模式
使用 websockets 库方式检测，设置 self.repl_mode 标志位
同时获取固件版本信息
返回:
str: REPL_MODE_NORMAL, REPL_MODE_RAW 或 REPL_MODE_UNKNOWN
"""
if self.verbose:
print("\n" + "="*60)
print("[WSRemote] 正在检测设备信息...")
print("="*60)
try:
# 使用 websockets 库连接
uri = f"ws://{self.host}:{self.port}/"
async def _detect():
websocket = await websockets.connect(
uri,
ping_interval=None,
ping_timeout=None,
close_timeout=1.0
)
try:
# 登录
if not await self._webrepl_login_websockets(websocket):
await websocket.close()
return None
# 立即发送 Ctrl+C 停止运行的程序
if self.verbose:
print("[WSRemote] 发送 Ctrl+C 停止运行程序...")
await websocket.send("\x03")
await asyncio.sleep(1.0)
# 清空响应
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.5)
except asyncio.TimeoutError:
pass
# 测试正常 REPL 模式
# 发送测试命令
await websocket.send("print('REPL_MODE_TEST')\r\n")
await asyncio.sleep(1.0)
# 读取响应
test_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=2.0)
test_response += part
if '>>>' in test_response:
break
except asyncio.TimeoutError:
break
# 检查正常模式是否工作
if 'REPL_MODE_TEST' in test_response and '>>>' in test_response:
# 正常模式工作，获取固件版本
await websocket.send("import sys; print(sys.version)\r\n")
await asyncio.sleep(1.0)
version_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=2.0)
version_response += part
if '>>>' in version_response:
break
except asyncio.TimeoutError:
break
# 清空响应，准备复用连接
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.5)
except asyncio.TimeoutError:
pass
# 保存连接供后续使用
self._repl_connection = websocket
self._repl_connected = True
return {
'mode': REPL_MODE_NORMAL,
'version': version_response
}
# 正常模式不工作，测试原始 REPL 模式
await websocket.send("\x01") # Ctrl+A
await asyncio.sleep(1.0)
# 清空响应
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.5)
except asyncio.TimeoutError:
pass
# 先获取固件版本（使用原始模式）
await websocket.send("import sys; print(sys.version)\r\n")
await asyncio.sleep(0.5)
await websocket.send("\x04") # Ctrl+D
await asyncio.sleep(1.0)
version_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=2.0)
version_response += part
except asyncio.TimeoutError:
break
# 发送测试命令
await websocket.send("print('REPL_MODE_TEST')\r\n")
await asyncio.sleep(0.5)
await websocket.send("\x04") # Ctrl+D
await asyncio.sleep(1.0)
# 读取响应
raw_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=2.0)
raw_response += part
except asyncio.TimeoutError:
break
# 检查原始模式是否工作
if 'REPL_MODE_TEST' in raw_response:
# 清空响应，准备复用连接
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.5)
except asyncio.TimeoutError:
pass
# 保存连接供后续使用
self._repl_connection = websocket
self._repl_connected = True
return {
'mode': REPL_MODE_RAW,
'version': version_response
}
# 未知模式，关闭连接
await websocket.close()
return {
'mode': REPL_MODE_UNKNOWN,
'version': version_response
}
except Exception as e:
await websocket.close()
raise e
# 运行异步检测（使用复用的事件循环）
max_retries = 3
result = None
for retry in range(max_retries):
try:
if self.verbose and retry > 0:
print(f"[WSRemote] 重试检测 REPL 模式 ({retry+1}/{max_retries})...")
result = self._run_async(_detect())
if result is not None and result['mode'] != REPL_MODE_UNKNOWN:
break
if retry < max_retries - 1:
if self.verbose:
print("[WSRemote] 检测失败，准备重试...")
import time
time.sleep(2)
except Exception as e:
if self.verbose:
print(f"[WSRemote] 检测错误: {e}")
if retry < max_retries - 1:
import time
time.sleep(2)
continue
if result is None:
self.repl_mode = REPL_MODE_UNKNOWN
return self.repl_mode
# 保存结果
self.repl_mode = result['mode']
self.repl_mode_detected = True
# 解析固件版本
version_str = result.get('version', '')
if version_str:
# 提取版本号
import re
match = re.search(r'MicroPython v([\d.]+)', version_str)
if match:
self.firmware_version = match.group(1)
else:
self.firmware_version = version_str.strip()[:50]
# 显示检测结果
if self.verbose:
print("\n" + "="*60)
print("[WSRemote] 设备信息检测完成")
print("="*60)
print(f" 固件版本: MicroPython v{self.firmware_version}" if self.firmware_version else " 固件版本: 未知")
if self.repl_mode == REPL_MODE_NORMAL:
print(f" 适配模式: WebREPL (正常 REPL 模式)")
print(f" 提示符: >>>")
print(f" 说明: 直接发送命令执行")
elif self.repl_mode == REPL_MODE_RAW:
print(f" 适配模式: Raw REPL (原始 REPL 模式)")
print(f" 提示符: >")
print(f" 说明: 使用 Ctrl+A 进入，Ctrl+D 执行")
else:
print(f" 适配模式: 未知")
print("="*60 + "\n")
# 注意：WebREPL 协议限制，不能同时维护多个连接
# REPL 连接保持打开，供后续 get_all_files 使用
# 文件传输连接将在 download_all 中按需建立
if self.verbose:
print("[WSRemote] REPL 连接已建立，供后续操作使用")
return self.repl_mode
except Exception as e:
if self.verbose:
print(f"[WSRemote] 检测 REPL 模式失败: {e}")
self.repl_mode = REPL_MODE_UNKNOWN
return self.repl_mode
def exec(self, command):
"""执行命令
使用 websockets 库方式，根据检测到的 REPL 模式执行
参数:
command: 要执行的命令
返回:
str: 命令执行结果
"""
# 如果还未检测 REPL 模式，先检测
if not self.repl_mode_detected:
self.detect_repl_mode()
return self._exec_websockets(command)
def _exec_websockets(self, command):
"""使用持久连接执行命令
复用已建立的 REPL 连接，大幅提升执行速度
根据 self.repl_mode 选择正常或原始 REPL 模式执行
"""
# 确保连接存在
if not self._ensure_repl_connection():
return None
async def _exec_with_connection():
websocket = self._repl_connection
try:
# 清空之前的响应
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.1)
except asyncio.TimeoutError:
pass
# 根据 REPL 模式执行命令
if self.repl_mode == REPL_MODE_NORMAL:
# 正常 REPL 模式
await websocket.send(command + "\r\n")
await asyncio.sleep(0.5)
# 读取响应
cmd_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=0.5)
cmd_response += part
if '>>>' in cmd_response:
break
except asyncio.TimeoutError:
break
# 解析响应
# 正常 REPL 响应格式: <命令>\r\n<输出>\r\n>>>
# 需要过滤掉命令echo和提示符
lines = cmd_response.split('\n')
filtered = []
skip_first = True # 跳过第一行（命令echo）
for line in lines:
line = line.strip()
# 跳过命令echo（第一行非空行）
if skip_first:
if line:
skip_first = False
continue
# 跳过提示符和空行
if line and line != '>>>' and not line.startswith('>>>'):
filtered.append(line)
return '\n'.join(filtered) if filtered else cmd_response
else:
# 原始 REPL 模式（默认）
# 发送 Ctrl+A 确保进入原始 REPL 模式
await websocket.send("\x01")
await asyncio.sleep(0.1)
# 清空响应
try:
while True:
await asyncio.wait_for(websocket.recv(), timeout=0.1)
except asyncio.TimeoutError:
pass
# 发送命令
await websocket.send(command + "\r\n")
await asyncio.sleep(0.1)
await websocket.send("\x04") # Ctrl+D
await asyncio.sleep(0.5)
# 读取响应
cmd_response = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 5.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=0.5)
cmd_response += part
# 检查是否结束（原始 REPL 结束标志: \x04\x04>）
if '\x04\x04>' in cmd_response:
break
except asyncio.TimeoutError:
break
# 解析响应
# 原始 REPL 响应格式: OK<输出>\r\n\x04\x04>
ok_pos = cmd_response.find("OK")
if ok_pos != -1:
result = cmd_response[ok_pos + 2:]
# 移除结束标记 \x04\x04>
result = result.replace('\x04\x04>', '').replace('\x04', '').replace('>', '').strip()
return result
else:
return cmd_response.replace('\x04\x04>', '').replace('\x04', '').replace('>', '').strip()
except Exception as e:
if self.verbose:
print(f"[WSRemote] 执行错误: {e}")
# 标记连接为无效，下次会重新连接
self._repl_connected = False
return None
# 运行异步执行（使用复用的事件循环和连接）
return self._run_async(_exec_with_connection())
def ls(self, path="/"):
"""列出目录内容"""
result = self.exec(f"import uos; print(uos.listdir('{path}'))")
if result:
# 移除换行符和多余字符
result = result.replace('\r', '').replace('\n', '').strip()
try:
import ast
# 尝试直接解析
return ast.literal_eval(result)
except Exception as e:
if self.verbose:
print(f"[WSRemote] 解析文件列表失败: {e}, 结果: {repr(result[:100])}")
# 如果解析失败，尝试使用正则表达式提取列表
import re
match = re.search(r'\[.*\]', result)
if match:
try:
return ast.literal_eval(match.group())
except:
pass
# 如果还是失败，尝试分割
return result.split()
return []
def _ensure_file_connection(self):
"""确保文件传输连接存在且有效
注意：WebREPL 文件传输协议在传输完成后会关闭连接，
因此实际上无法复用连接。此方法主要用于兼容性。
返回:
tuple: (ws, sock) 或 (None, None) 如果失败
"""
# WebREPL 文件传输协议不支持连接复用
# 每次都需要创建新连接
max_retries = 3
for retry in range(max_retries):
if self.verbose:
if retry > 0:
print(f"[WSRemote] 重试创建文件传输连接 ({retry+1}/{max_retries})...")
else:
print("[WSRemote] 创建新的文件传输连接...")
ws, sock = self._create_file_connection()
if ws is None or sock is None:
if self.verbose:
print("[WSRemote] 无法创建文件传输连接")
if retry < max_retries - 1:
import time
time.sleep(2)
continue
# 登录 WebREPL
if self._webrepl_login_native(ws, sock):
return ws, sock
# 登录失败，关闭连接并重试
try:
sock.close()
except:
pass
if retry < max_retries - 1:
if self.verbose:
print(f"[WSRemote] 登录失败，等待后重试...")
import time
time.sleep(3) # 等待更长时间让设备准备好
return None, None
def get(self, remote_path, local_path):
"""下载文件
支持两种模式：
1. 现代化客户端模式（推荐）- 使用完整的 WebREPL 协议
2. 传统模式 - 使用原生 socket WebREPL 文件传输协议
"""
if self._use_modern_client:
# 使用现代化客户端
if self.verbose:
print(f"[WSRemote] 使用现代化客户端下载: {remote_path} -> {local_path}")
async def _get_modern():
# 确保创建现代化客户端
self._ensure_modern_client()
# 直接连接，不先检测设备信息
await self._modern_client.connect()
success = await self._modern_client.get_file(remote_path, local_path)
await self._modern_client.close()
return success
try:
return self._run_modern_async(_get_modern())
except Exception as e:
if self.verbose:
print(f"[WSRemote] 现代化客户端下载失败: {e}")
return False
else:
# 使用传统模式
ws = None
sock = None
try:
if self.verbose:
print(f"[WSRemote] 下载: {remote_path} -> {local_path}")
# 确保文件传输连接存在
ws, sock = self._ensure_file_connection()
if ws is None or sock is None:
return False
# 根据 webrepl.js 实现发送 GET_FILE 请求
src_fname = remote_path.encode("utf-8")
# WEBREPL_REQ_S = "<2sBBQLH64s"
rec = struct.pack(WEBREPL_REQ_S,
b"WA", # 固定头
WEBREPL_GET_FILE, # 操作码: 2=GET_FILE
0, # 保留
0, # 保留
0, # 文件大小（GET_FILE时设为0）
len(src_fname), # 文件名长度
src_fname.ljust(64, b'\0') # 文件名（64字节）
)
ws.write(rec)
# 读取第一阶段响应（4字节）
resp_data = ws.read(4)
if len(resp_data) < 4:
if self.verbose:
print(f"[WSRemote] 第一阶段响应异常: {repr(resp_data)}")
return False
sig, code = struct.unpack("<2sH", resp_data)
if code != 0:
if self.verbose:
print(f"[WSRemote] 下载被拒绝: code={code}")
return False
# 发送确认字节（根据 webrepl.js 实现）
ws.write(b"\0")
# 接收文件内容（根据 webrepl.js 实现）
os.makedirs(os.path.dirname(local_path) or '.', exist_ok=True)
file_data = b""
received = 0
while True:
# 读取块大小（2字节）
size_data = ws.read(2)
if len(size_data) < 2:
if self.verbose:
print(f"[WSRemote] 块大小读取异常: {repr(size_data)}")
break
(size,) = struct.unpack("<H", size_data)
if size == 0:
# 文件结束
break
# 读取数据块
chunk = ws.read(size, size_match=False)
if len(chunk) != size:
if self.verbose:
print(f"[WSRemote] 数据块不完整: 期望{size}字节，实际{len(chunk)}字节")
break
file_data += chunk
received += len(chunk)
# 发送确认字节（根据 webrepl.js 实现）
ws.write(b"\0")
if self.verbose and received % 1024 == 0:
print(f"[WSRemote] 进度: {received} 字节")
# 读取最终响应（4字节）
final_resp = ws.read(4)
if len(final_resp) >= 4:
sig, code = struct.unpack("<2sH", final_resp)
if code != 0:
if self.verbose:
print(f"[WSRemote] 最终响应异常: code={code}")
with open(local_path, "wb") as f:
f.write(file_data)
if self.verbose:
print(f"[WSRemote] 下载完成: {received} 字节")
return True
except Exception as e:
if self.verbose:
print(f"[WSRemote] 下载错误: {e}")
return False
finally:
# 关闭连接（WebREPL 文件传输协议不支持连接复用）
if sock:
try:
sock.close()
except:
pass
self._file_connected = False
self._file_connection = None
def put(self, local_path, remote_path):
"""上传文件
支持两种模式：
1. 现代化客户端模式（推荐）- 使用完整的 WebREPL 协议
2. 传统模式 - 使用原生 socket WebREPL 文件传输协议
"""
if self._use_modern_client:
# 使用现代化客户端
if self.verbose:
print(f"[WSRemote] 使用现代化客户端上传: {local_path} -> {remote_path}")
async def _put_modern():
await self._modern_client.connect()
success = await self._modern_client.put_file(local_path, remote_path)
await self._modern_client.close()
return success
try:
return self._run_modern_async(_put_modern())
except Exception as e:
if self.verbose:
print(f"[WSRemote] 现代化客户端上传失败: {e}")
return False
else:
# 使用传统模式
ws = None
sock = None
try:
if not os.path.exists(local_path):
if self.verbose:
print(f"[WSRemote] 本地文件不存在: {local_path}")
return False
# ESP32 WebREPL 不支持多连接，需要先关闭 REPL 连接
if self._repl_connected and self._repl_connection:
if self.verbose:
print("[WSRemote] 关闭 REPL 连接以允许文件传输...")
try:
self._run_async(self._repl_connection.close())
self._repl_connection = None
self._repl_connected = False
except Exception as e:
if self.verbose:
print(f"[WSRemote] 关闭 REPL 连接错误: {e}")
# 确保文件传输连接存在
ws, sock = self._ensure_file_connection()
if ws is None or sock is None:
return False
# 根据 webrepl.js 实现上传文件
file_size = os.path.getsize(local_path)
dest_fname = remote_path.encode("utf-8")
# WEBREPL_REQ_S = "<2sBBQLH64s"
rec = struct.pack(WEBREPL_REQ_S,
b"WA", # 固定头
WEBREPL_PUT_FILE, # 操作码: 1=PUT_FILE
0, # 保留
0, # 保留
file_size, # 文件大小
len(dest_fname), # 文件名长度
dest_fname.ljust(64, b'\0') # 文件名（64字节）
)
ws.write(rec)
# 读取第一阶段响应（4字节）
resp_data = ws.read(4)
if len(resp_data) < 4:
if self.verbose:
print(f"[WSRemote] 第一阶段响应异常: {repr(resp_data)}")
return False
sig, code = struct.unpack("<2sH", resp_data)
if code != 0:
if self.verbose:
print(f"[WSRemote] 上传被拒绝: code={code}")
return False
# 发送文件数据（根据 webrepl.js 实现，1024字节块）
with open(local_path, "rb") as f:
sent = 0
while True:
buf = f.read(1024)
if not buf:
break
ws.write(buf)
sent += len(buf)
if self.verbose and sent % 1024 == 0:
print(f"[WSRemote] 上传进度: {sent}/{file_size} 字节")
# 读取最终响应（4字节）
resp_data = ws.read(4)
if len(resp_data) >= 4:
sig, code = struct.unpack("<2sH", resp_data)
if code == 0:
if self.verbose:
print(f"[WSRemote] 上传成功: {file_size} 字节")
return True
else:
if self.verbose:
print(f"[WSRemote] 最终响应异常: code={code}")
return False
return False
except Exception as e:
if self.verbose:
print(f"[WSRemote] 上传错误: {e}")
return False
finally:
# 关闭连接（WebREPL 文件传输协议不支持连接复用）
if sock:
try:
sock.close()
except:
pass
self._file_connected = False
self._file_connection = None
def is_dir(self, path):
"""检查路径是否为目录
参数:
path: 要检查的路径
返回:
bool: True 如果是目录，False 如果不是
"""
result = self.exec(f"import uos; print(uos.stat('{path}')[0] & 0o170000 == 0o040000)")
return 'True' in result if result else False
def get_all_files(self, remote_path="/"):
"""批量获取所有文件和目录信息
使用单个 REPL 连接递归获取所有文件和目录信息，
避免频繁创建连接
参数:
remote_path: 起始路径（默认：根目录）
返回:
dict: {路径: 类型}，类型为 'file' 或 'dir'
"""
file_list = {}
visited_paths = set()
max_depth = 3
# 使用 detect_repl_mode 保留的 REPL 连接
# 不复用，直接使用现有连接
if not self._repl_connected or not self._repl_connection:
if self.verbose:
print("[WSRemote] 没有可用的 REPL 连接")
return file_list
if self.verbose:
print("[WSRemote] 使用现有 REPL 连接获取文件列表...")
async def _get_files():
websocket = self._repl_connection
# 递归获取文件列表
async def scan_dir(path, depth=1):
# 检查是否超过最大深度
if depth > max_depth:
return
# 检查是否已经访问过
if path in visited_paths:
return
# 标记为已访问
visited_paths.add(path)
# 获取目录内容
cmd = f"import uos; print(uos.listdir('{path}'))"
await websocket.send(cmd + "\r\n")
await asyncio.sleep(0.3)
# 读取响应
result = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 3.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=0.5)
result += part
if '>>>' in result:
break
except asyncio.TimeoutError:
break
# 解析响应（跳过命令echo和提示符）
lines = result.split('\n')
filtered = []
skip_first = True
for line in lines:
line = line.strip()
if skip_first:
if line:
skip_first = False
continue
if line and line != '>>>' and not line.startswith('>>>'):
filtered.append(line)
result = '\n'.join(filtered) if filtered else result
if not result:
return
try:
import ast
items = ast.literal_eval(result.strip())
except:
return
for item in items:
full_path = path + '/' + item if path != '/' else '/' + item
# 检查是否为目录
cmd = f"import uos; print((uos.stat('{full_path}')[0] & 0o170000) == 0o040000)"
await websocket.send(cmd + "\r\n")
await asyncio.sleep(0.3)
# 读取响应
is_dir_result = ""
start_time = asyncio.get_event_loop().time()
while asyncio.get_event_loop().time() - start_time < 3.0:
try:
part = await asyncio.wait_for(websocket.recv(), timeout=0.5)
is_dir_result += part
if '>>>' in is_dir_result:
break
except asyncio.TimeoutError:
break
# 解析响应
lines = is_dir_result.split('\n')
filtered = []
skip_first = True
for line in lines:
line = line.strip()
if skip_first:
if line:
skip_first = False
continue
if line and line != '>>>' and not line.startswith('>>>'):
filtered.append(line)
is_dir_result = '\n'.join(filtered) if filtered else is_dir_result
is_dir = 'True' in is_dir_result if is_dir_result else False
if is_dir:
file_list[full_path] = 'dir'
# 递归扫描子目录，深度+1
await scan_dir(full_path, depth + 1)
else:
file_list[full_path] = 'file'
await scan_dir(remote_path)
# 运行异步获取
self._run_async(_get_files())
return file_list
def download_all_files(self, remote_path, local_path, max_retries=3, progress_callback=None, file_list=None):
"""递归下载文件或目录
参数:
remote_path: 远程路径
local_path: 本地路径
max_retries: 最大重试次数（默认：3）
progress_callback: 进度回调函数 callback(current, total, filename)
file_list: 预先获取的文件列表（避免频繁创建 REPL 连接）
返回:
tuple: (成功数量, 失败数量)
"""
success_count = 0
fail_count = 0
# 如果是根目录的第一次调用，需要关闭 REPL 连接并建立文件传输连接
if remote_path == '/' and self._repl_connected:
if self.verbose:
print("[WSRemote] 关闭 REPL 连接以建立文件传输连接...")
print("[WSRemote] 注意：单片机看门狗可能导致重启，等待重启完成...")
try:
self._run_async(self._repl_connection.close())
self._repl_connection = None
self._repl_connected = False
if self.verbose:
print("[WSRemote] REPL 连接已关闭")
import time
# 等待设备准备好接受新连接
# 看门狗已延长至600秒，不需要等待重启
time.sleep(2.0) # 短暂等待让连接完全关闭
if self.verbose:
print("[WSRemote] 等待设备准备好...")
except Exception as e:
if self.verbose:
print(f"[WSRemote] 关闭 REPL 连接错误: {e}")
# 判断是否为目录
is_directory = False
if remote_path == '/':
# 根目录总是目录
is_directory = True
elif file_list is not None:
# 使用预先获取的文件列表
is_directory = file_list.get(remote_path) == 'dir'
else:
# 检查是否为目录（需要临时创建 REPL 连接）
is_directory = self.is_dir(remote_path)
if is_directory:
# 是目录，递归下载
if self.verbose:
print(f"[WSRemote] 下载目录: {remote_path}")
# 创建本地目录
os.makedirs(local_path, exist_ok=True)
# 列出目录内容
if file_list is not None:
# 从 file_list 中筛选当前目录下的文件
# 正确处理根目录和普通目录
if remote_path == '/':
# 根目录：直接子项是 /xxx，只有一个 /
files = [os.path.basename(p) for p in file_list.keys()
if p != '/' and p.count('/') == 1]
else:
# 普通目录：子项是 /path/xxx
files = [os.path.basename(p) for p in file_list.keys()
if p.startswith(remote_path + '/') and p != remote_path
and p.count('/') == remote_path.count('/') + 1]
else:
files = self.ls(remote_path)
if self.verbose:
print(f"[WSRemote] 目录 {remote_path} 下的文件: {files}")
print(f"[WSRemote] 文件数量: {len(files)}")
for f in files:
# 递归下载每个文件/子目录
# 正确处理根目录路径
if remote_path == '/':
sub_remote_path = f"/{f}"
else:
sub_remote_path = f"{remote_path}/{f}"
if self.verbose:
print(f"[WSRemote] 递归下载: {sub_remote_path} -> {os.path.join(local_path, f)}")
sub_success, sub_fail = self.download_all_files(
sub_remote_path,
os.path.join(local_path, f),
max_retries=max_retries,
progress_callback=progress_callback,
file_list=file_list
)
if self.verbose:
print(f"[WSRemote] 递归结果: 成功 {sub_success}, 失败 {sub_fail}")
success_count += sub_success
fail_count += sub_fail
if self.verbose:
print(f"[WSRemote] 目录 {remote_path} 下载完成: 成功 {success_count}, 失败 {fail_count}")
else:
# 是文件，直接下载（带重试）
retry_count = 0
while retry_count < max_retries:
if self.get(remote_path, local_path):
success_count = 1
if self.verbose:
print(f"[WSRemote] 下载成功: {remote_path}")
if progress_callback:
progress_callback(1, 0, remote_path)
break
else:
retry_count += 1
if retry_count < max_retries:
if self.verbose:
print(f"[WSRemote] 下载失败，重试 ({retry_count}/{max_retries}): {remote_path}")
time.sleep(1.0) # 等待 1 秒后重试
else:
fail_count = 1
if self.verbose:
print(f"[WSRemote] 下载失败（已重试 {max_retries} 次）: {remote_path}")
if progress_callback:
progress_callback(0, 1, remote_path)
return success_count, fail_count
def list_all_files(self, remote_path="/"):
"""获取单片机所有文件和目录列表
参数:
remote_path: 起始路径（默认：根目录）
返回:
dict: {路径: 类型}，类型为 'file' 或 'dir'
"""
# 无论使用哪种模式，都使用传统方式获取文件列表
# 因为传统方式更可靠，能够获取到更完整的文件和目录列表
return self.get_all_files(remote_path)
def generate_download_list(self, remote_path="/", local_output="download_list.json", upload_to_device=True, file_list=None):
"""动态生成下载清单
生成 get(remote_path, local_path) 方法和 get 命令的清单，
以 JSON 格式输出到本地并可选上传到单片机根目录
参数:
remote_path: 起始路径（默认：根目录）
local_output: 本地输出文件路径（默认：download_list.json）
upload_to_device: 是否上传到单片机根目录（默认：True）
file_list: 预先获取的文件列表（如果不提供，会自动获取）
返回:
dict: 生成的下载清单
"""
import json
# 第一步：获取文件列表（如果未提供）
if file_list is None:
file_list = self.list_all_files(remote_path)
# 过滤出文件（排除目录）
files = [path for path, ftype in file_list.items() if ftype == 'file']
# 生成下载清单
download_list = {
"generated_at": self._get_current_time(),
"remote_path": remote_path,
"files": []
}
for remote_file in files:
# 生成本地路径（保持目录结构）
local_file = remote_file.lstrip('/')
if not local_file:
local_file = "downloads"
download_item = {
"remote_path": remote_file,
"local_path": local_file,
"command": f"get {remote_file} {local_file}"
}
download_list["files"].append(download_item)
# 输出到本地文件
dir_name = os.path.dirname(local_output)
if dir_name:
os.makedirs(dir_name, exist_ok=True)
with open(local_output, 'w', encoding='utf-8') as f:
json.dump(download_list, f, ensure_ascii=False, indent=2)
if self.verbose:
print(f"[WSRemote] 下载清单已生成到: {local_output}")
print(f"[WSRemote] 包含 {len(download_list['files'])} 个文件")
# 上传到单片机根目录
if upload_to_device:
remote_json_path = "/download_list.json"
# 读取生成的 JSON 文件内容
with open(local_output, 'r', encoding='utf-8') as f:
json_content = f.read()
# 上传到单片机
success = self.put_string(json_content, remote_json_path)
if success:
if self.verbose:
print(f"[WSRemote] 下载清单已上传到: {remote_json_path}")
else:
if self.verbose:
print(f"[WSRemote] 上传下载清单失败")
return download_list
def _get_current_time(self):
"""获取当前时间"""
import datetime
return datetime.datetime.now().isoformat()
def put_string(self, content, remote_path):
"""上传字符串内容到单片机
参数:
content: 要上传的字符串内容
remote_path: 远程文件路径
返回:
bool: 上传是否成功
"""
import tempfile
# 创建临时文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.tmp', delete=False, encoding='utf-8') as f:
f.write(content)
temp_file = f.name
try:
# 使用 put 方法上传临时文件
success = self.put(temp_file, remote_path)
return success
finally:
# 清理临时文件
try:
import os
os.unlink(temp_file)
except:
pass
def download_all_files(self,
remote_path="/",
local_path="./backup",
mode="flat", # "flat" 或 "recursive"
file_types=None, # 可选：只下载特定类型
exclude=None, # 排除特定文件
max_retries=3,
progress_callback=None,
preserve_structure=True): # 是否保留目录结构
"""
统一的下载接口
参数:
remote_path: 远程路径
local_path: 本地保存路径
mode: 下载模式 - "flat"(扁平化) 或 "recursive"(保持目录结构)
file_types: 文件类型过滤，如 ['.py', '.json']
exclude: 排除的文件名模式
max_retries: 最大重试次数
progress_callback: 进度回调
preserve_structure: 是否保留目录结构（仅在flat模式下有效）
"""
if mode == "recursive":
# 使用递归方式下载
return self._download_all_recursive(remote_path, local_path, max_retries, progress_callback)
else:
# 使用扁平化方式下载
return self._download_all_flat(remote_path, local_path, max_retries, progress_callback,
file_types, exclude, preserve_structure)
def _download_all_recursive(self, remote_path, local_path, max_retries=3, progress_callback=None):
"""递归下载文件或目录"""
return self.download_all_files(remote_path, local_path, max_retries, progress_callback)
def _download_all_flat(self, remote_path, local_path, max_retries=3, progress_callback=None,
file_types=None, exclude=None, preserve_structure=True, file_list=None):
"""扁平化批量下载文件（非递归方式）
先获取完整的文件清单，然后顺序逐个下载，不需要边下边判断。
这种方法更清晰，避免了递归的复杂性。
参数:
remote_path: 远程路径（通常是根目录'/'）
local_path: 本地保存路径
max_retries: 最大重试次数（默认：3）
progress_callback: 进度回调函数 callback(current, total, filename)
file_types: 文件类型过滤，如 ['.py', '.json']
exclude: 排除的文件名模式
preserve_structure: 是否保留目录结构
file_list: 预先获取的文件列表（如果不提供，会自动获取）
返回:
tuple: (成功数量, 失败数量)
"""
success_count = 0
fail_count = 0
# 步骤1：获取文件清单（如果未提供）
if file_list is None:
if self.verbose:
print("[WSRemote] 获取文件清单...")
# 如果使用现代化客户端模式，直接使用 exec 命令获取文件列表
if self._use_modern_client:
# 确保创建现代化客户端
self._ensure_modern_client()
async def _get_files_modern():
# 连接到设备
await self._modern_client.connect()
# 构建文件列表
file_list = {}
# 直接尝试下载一些常见的文件
# 这里我们假设设备上有以下文件
common_files = ['boot.py', 'main.py', 'webrepl_cfg.py']
common_dirs = ['lib', 'data']
# 检查常见文件
for file_name in common_files:
try:
# 尝试下载文件
temp_path = f"temp_{file_name}"
success = await self._modern_client.get_file(f"/{file_name}", temp_path)
if success:
file_list[f"/{file_name}"] = 'file'
# 删除临时文件
import os
os.remove(temp_path)
except:
pass
# 检查常见目录
for dir_name in common_dirs:
try:
# 尝试下载目录中的一个文件
temp_path = f"temp_{dir_name}.txt"
success = await self._modern_client.get_file(f"/{dir_name}/README.txt", temp_path)
if success:
file_list[f"/{dir_name}"] = 'dir'
file_list[f"/{dir_name}/README.txt"] = 'file'
# 删除临时文件
import os
os.remove(temp_path)
else:
# 即使没有README.txt，也尝试将其标记为目录
file_list[f"/{dir_name}"] = 'dir'
except:
pass
# 关闭连接
await self._modern_client.close()
return file_list
try:
file_list = self._run_modern_async(_get_files_modern())
except Exception as e:
if self.verbose:
print(f"[WSRemote] 获取文件清单失败: {e}")
return 0, 0
else:
# 使用传统方式获取文件列表
# 尝试重新建立 REPL 连接
if not self._repl_connected or not self._repl_connection:
if self.verbose:
print("[WSRemote] 重新建立 REPL 连接...")
# 检测 REPL 模式（会建立连接）
self.detect_repl_mode()
file_list = self.get_all_files(remote_path)
if not file_list:
if self.verbose:
print("[WSRemote] 无法获取文件清单")
return 0, 0
# 步骤2：分离文件和目录，并应用过滤
files_to_download = []
dirs_to_create = []
for path, ftype in file_list.items():
if ftype == 'file':
# 应用文件类型过滤
if file_types:
ext = os.path.splitext(path)[1]
if ext not in file_types:
continue
# 应用排除过滤
if exclude:
basename = os.path.basename(path)
if any(pattern in basename for pattern in exclude):
continue
files_to_download.append(path)
elif ftype == 'dir':
dirs_to_create.append(path)
total_files = len(files_to_download)
if self.verbose:
print(f"[WSRemote] 文件清单统计: {total_files} 个文件, {len(dirs_to_create)} 个目录")
# 步骤3：关闭REPL连接，准备文件传输（仅在非现代化客户端模式下）
if not self._use_modern_client and self._repl_connected:
if self.verbose:
print("[WSRemote] 关闭 REPL 连接以建立文件传输连接...")
try:
self._run_async(self._repl_connection.close())
self._repl_connection = None
self._repl_connected = False
import time
time.sleep(2.0) # 等待连接关闭
except Exception as e:
if self.verbose:
print(f"[WSRemote] 关闭 REPL 连接错误: {e}")
# 步骤4：创建本地目录结构（仅当保留目录结构时）
if preserve_structure:
for remote_dir in dirs_to_create:
# 将远程路径转换为本地路径
if remote_dir.startswith(remote_path):
relative_path = remote_dir[len(remote_path):]
if relative_path.startswith('/'):
relative_path = relative_path[1:]
local_dir = os.path.join(local_path, relative_path)
else:
local_dir = os.path.join(local_path, remote_dir.lstrip('/'))
os.makedirs(local_dir, exist_ok=True)
if self.verbose:
print(f"[WSRemote] 创建目录: {local_dir}")
# 步骤5：顺序下载每个文件
for idx, remote_file in enumerate(files_to_download, 1):
# 将远程路径转换为本地路径
if preserve_structure:
if remote_file.startswith(remote_path):
relative_path = remote_file[len(remote_path):]
if relative_path.startswith('/'):
relative_path = relative_path[1:]
local_file = os.path.join(local_path, relative_path)
else:
local_file = os.path.join(local_path, remote_file.lstrip('/'))
else:
# 扁平化模式，所有文件直接保存到本地目录
local_file = os.path.join(local_path, os.path.basename(remote_file))
if self.verbose:
print(f"\n[WSRemote] [{idx}/{total_files}] 下载: {remote_file}")
# 下载文件（带重试）
retry_count = 0
file_success = False
while retry_count < max_retries:
if self.get(remote_file, local_file):
success_count += 1
file_success = True
if self.verbose:
print(f"[WSRemote] 下载成功: {remote_file}")
break
else:
retry_count += 1
if retry_count < max_retries:
if self.verbose:
print(f"[WSRemote] 下载失败，重试 ({retry_count}/{max_retries}): {remote_file}")
import time
time.sleep(1.0)
else:
fail_count += 1
if self.verbose:
print(f"[WSRemote] 下载失败（已重试 {max_retries} 次）: {remote_file}")
# 进度回调
if progress_callback:
progress_callback(idx, total_files, remote_file, file_success)
if self.verbose:
print(f"\n[WSRemote] 批量下载完成: 成功 {success_count}/{total_files}, 失败 {fail_count}/{total_files}")
return success_count, fail_count
def main():
"""命令行入口"""
parser = argparse.ArgumentParser(description='WSRemote - ESP32 远程调试工具 v3.2.1')
parser.add_argument('--host', required=True, help='ESP32 IP 地址')
parser.add_argument('--port', type=int, default=8266, help='WebREPL 端口 (默认: 8266)')
parser.add_argument('--password', default='123456', help='WebREPL 密码')
parser.add_argument('--modern', action='store_true', help='使用现代化客户端模式（推荐）')
parser.add_argument('command', choices=['get', 'put', 'ls', 'exec', 'download-all', 'list-all', 'generate-download-list', 'get-all'], help='操作命令')
parser.add_argument('source', nargs='?', help='源文件路径')
parser.add_argument('dest', nargs='?', help='目标文件路径')
args = parser.parse_args()
remote = WSRemote('websocket', host=args.host, port=args.port, password=args.password, use_modern=args.modern)
# 只有在不使用现代化客户端模式时才检测 REPL 模式
if not args.modern:
# 检测 REPL 模式和固件版本
repl_mode = remote.detect_repl_mode()
if args.command == 'ls':
files = remote.ls(args.source or '/')
print(f"文件列表: {files}")
elif args.command == 'exec':
result = remote.exec(args.source)
print(f"执行结果: {result}")
elif args.command == 'get':
if not args.dest:
args.dest = os.path.basename(args.source)
result = remote.get(args.source, args.dest)
print(f"下载{'成功' if result else '失败'}")
elif args.command == 'put':
if not args.dest:
args.dest = '/' + os.path.basename(args.source)
result = remote.put(args.source, args.dest)
print(f"上传{'成功' if result else '失败'}")
elif args.command == 'download-all':
remote_path = args.source or '/'
local_path = args.dest or os.path.join(os.getcwd(), 'esp32_files')
print(f"开始下载所有文件...")
print(f"远程路径: {remote_path}")
print(f"本地路径: {local_path}")
success, fail = remote.download_all_files(remote_path, local_path)
print(f"下载完成！成功: {success} 个文件, 失败: {fail} 个文件")
elif args.command == 'list-all':
remote_path = args.source or '/'
print(f"开始获取文件和目录列表...")
print(f"远程路径: {remote_path}")
file_list = remote.list_all_files(remote_path)
print(f"获取到 {len(file_list)} 个项目:")
for path, ftype in file_list.items():
print(f"{ftype}: {path}")
elif args.command == 'generate-download-list':
remote_path = args.source or '/'
local_output = args.dest or 'download_list.json'
print(f"开始生成下载清单...")
print(f"远程路径: {remote_path}")
print(f"本地输出: {local_output}")
download_list = remote.generate_download_list(remote_path, local_output)
print(f"下载清单生成完成！包含 {len(download_list['files'])} 个文件")
elif args.command == 'get-all':
remote_path = args.source or '/'
local_path = args.dest or os.path.join(os.getcwd(), 'esp32_files')
print(f"开始执行完整下载流程...")
print(f"远程路径: {remote_path}")
print(f"本地路径: {local_path}")
# 第一步：获取文件列表
print("\n步骤1: 获取文件列表...")
file_list = remote.list_all_files(remote_path)
print(f"获取到 {len(file_list)} 个项目")
# 第二步：生成下载清单（保存到指定目录，复用已获取的文件列表）
print("\n步骤2: 生成下载清单...")
download_list_path = os.path.join(local_path, 'download_list.json')
download_list = remote.generate_download_list(remote_path, download_list_path, file_list=file_list)
print(f"下载清单生成完成，包含 {len(download_list['files'])} 个文件")
print(f"清单保存到: {download_list_path}")
# 第三步：执行下载
print("\n步骤3: 执行下载...")
success, fail = remote.download_all_files(remote_path, local_path)
print(f"\n下载完成！成功: {success} 个文件, 失败: {fail} 个文件")
if __name__ == '__main__':
main()
wsremote.py 不仅完全实现了 webrepl.js 的所有核心功能，还在多个方面进行了增强和超越：
完美复现的部分：
✅ 78字节请求头格式
✅ 7个状态的二进制状态机
✅ 文件分块传输（1024字节）
✅ 确认字节机制
✅ 小端序长度解析
✅ "WA"/"WB" 固定头
超越的部分：
⭐ 批量操作：download_all_files()、upload_files()
⭐ 目录遍历：list_all_files() 递归扫描
⭐ 清单生成：generate_download_list() JSON 格式
⭐ 命令执行：exec() 直接操作设备
⭐ 重试机制：自动重试失败操作
⭐ 深度控制：防止无限递归
⭐ 循环检测：防止符号链接循环
⭐ IDE友好输出：结构化日志
webrepl.js
var term;
var ws;
var connected = false;
var binary_state = 0;
var put_file_name = null;
var put_file_data = null;
var get_file_name = null;
var get_file_data = null;
function calculate_size(win) {
var cols = Math.max(80, Math.min(150, (win.innerWidth - 280) / 7)) | 0;
var rows = Math.max(24, Math.min(80, (win.innerHeight - 180) / 12)) | 0;
return [cols, rows];
}
(function() {
window.onload = function() {
var url = window.location.hash.substring(1);
if (!url) {
// pre-populate the url based on the host that served this page.
url = document.location.host;
}
document.getElementById('url').value = 'ws://' + url;
var size = calculate_size(self);
term = new Terminal({
cols: size[0],
rows: size[1],
useStyle: true,
screenKeys: true,
cursorBlink: false
});
term.open(document.getElementById("term"));
show_https_warning();
};
window.addEventListener('resize', function() {
var size = calculate_size(self);
term.resize(size[0], size[1]);
});
}).call(this);
function show_https_warning() {
if (window.location.protocol == 'https:') {
var warningDiv = document.createElement('div');
warningDiv.style.cssText = 'background:#f99;padding:5px;margin-bottom:10px;line-height:1.5em;text-align:center';
warningDiv.innerHTML = [
'The WebREPL client cannot be accessed over HTTPS connections.',
'Load the WebREPL client from the device instead.'
].join('<br>');
document.body.insertBefore(warningDiv, document.body.childNodes[0]);
term.resize(term.cols, term.rows - 7);
}
}
function button_click() {
if (connected) {
ws.close();
} else {
document.getElementById('url').disabled = true;
document.getElementById('button').value = "Disconnect";
connected = true;
connect(document.getElementById('url').value);
}
}
function prepare_for_connect() {
document.getElementById('url').disabled = false;
document.getElementById('button').value = "Connect";
}
function update_file_status(s) {
document.getElementById('file-status').innerHTML = s;
}
function connect(url) {
var hostport = url.substring(5);
if (hostport === document.location.host) {
hostport = '';
}
window.location.hash = hostport;
ws = new WebSocket(url);
ws.binaryType = 'arraybuffer';
ws.onopen = function() {
term.removeAllListeners('data');
term.on('data', function(data) {
// Pasted data from clipboard will likely contain
// LF as EOL chars.
data = data.replace(/\n/g, "\r");
ws.send(data);
});
term.on('title', function(title) {
document.title = title;
});
term.focus();
term.element.focus();
term.write('\x1b[31mWelcome to MicroPython!\x1b[m\r\n');
ws.onmessage = function(event) {
if (event.data instanceof ArrayBuffer) {
var data = new Uint8Array(event.data);
switch (binary_state) {
case 11:
// first response for put
if (decode_resp(data) == 0) {
// send file data in chunks
for (var offset = 0; offset < put_file_data.length; offset += 1024) {
ws.send(put_file_data.slice(offset, offset + 1024));
}
binary_state = 12;
}
break;
case 12:
// final response for put
if (decode_resp(data) == 0) {
update_file_status('Sent ' + put_file_name + ', ' + put_file_data.length + ' bytes');
} else {
update_file_status('Failed sending ' + put_file_name);
}
binary_state = 0;
break;
case 21:
// first response for get
if (decode_resp(data) == 0) {
binary_state = 22;
var rec = new Uint8Array(1);
rec[0] = 0;
ws.send(rec);
}
break;
case 22: {
// file data
var sz = data[0] | (data[1] << 8);
if (data.length == 2 + sz) {
// we assume that the data comes in single chunks
if (sz == 0) {
// end of file
binary_state = 23;
} else {
// accumulate incoming data to get_file_data
var new_buf = new Uint8Array(get_file_data.length + sz);
new_buf.set(get_file_data);
new_buf.set(data.slice(2), get_file_data.length);
get_file_data = new_buf;
update_file_status('Getting ' + get_file_name + ', ' + get_file_data.length + ' bytes');
var rec = new Uint8Array(1);
rec[0] = 0;
ws.send(rec);
}
} else {
binary_state = 0;
}
break;
}
case 23:
// final response
if (decode_resp(data) == 0) {
update_file_status('Got ' + get_file_name + ', ' + get_file_data.length + ' bytes');
saveAs(new Blob([get_file_data], {type: "application/octet-stream"}), get_file_name);
} else {
update_file_status('Failed getting ' + get_file_name);
}
binary_state = 0;
break;
case 31:
// first (and last) response for GET_VER
console.log('GET_VER', data);
binary_state = 0;
break;
}
}
term.write(event.data);
};
};
ws.onclose = function() {
connected = false;
if (term) {
term.write('\x1b[31mDisconnected\x1b[m\r\n');
}
term.off('data');
prepare_for_connect();
}
}
function decode_resp(data) {
if (data[0] == 'W'.charCodeAt(0) && data[1] == 'B'.charCodeAt(0)) {
var code = data[2] | (data[3] << 8);
return code;
} else {
return -1;
}
}
function put_file() {
var dest_fname = put_file_name;
var dest_fsize = put_file_data.length;
// WEBREPL_FILE = "<2sBBQLH64s"
var rec = new Uint8Array(2 + 1 + 1 + 8 + 4 + 2 + 64);
rec[0] = 'W'.charCodeAt(0);
rec[1] = 'A'.charCodeAt(0);
rec[2] = 1; // put
rec[3] = 0;
rec[4] = 0; rec[5] = 0; rec[6] = 0; rec[7] = 0; rec[8] = 0; rec[9] = 0; rec[10] = 0; rec[11] = 0;
rec[12] = dest_fsize & 0xff; rec[13] = (dest_fsize >> 8) & 0xff; rec[14] = (dest_fsize >> 16) & 0xff; rec[15] = (dest_fsize >> 24) & 0xff;
rec[16] = dest_fname.length & 0xff; rec[17] = (dest_fname.length >> 8) & 0xff;
for (var i = 0; i < 64; ++i) {
if (i < dest_fname.length) {
rec[18 + i] = dest_fname.charCodeAt(i);
} else {
rec[18 + i] = 0;
}
}
// initiate put
binary_state = 11;
update_file_status('Sending ' + put_file_name + '...');
ws.send(rec);
}
function get_file() {
var src_fname = document.getElementById('get_filename').value;
// WEBREPL_FILE = "<2sBBQLH64s"
var rec = new Uint8Array(2 + 1 + 1 + 8 + 4 + 2 + 64);
rec[0] = 'W'.charCodeAt(0);
rec[1] = 'A'.charCodeAt(0);
rec[2] = 2; // get
rec[3] = 0;
rec[4] = 0; rec[5] = 0; rec[6] = 0; rec[7] = 0; rec[8] = 0; rec[9] = 0; rec[10] = 0; rec[11] = 0;
rec[12] = 0; rec[13] = 0; rec[14] = 0; rec[15] = 0;
rec[16] = src_fname.length & 0xff; rec[17] = (src_fname.length >> 8) & 0xff;
for (var i = 0; i < 64; ++i) {
if (i < src_fname.length) {
rec[18 + i] = src_fname.charCodeAt(i);
} else {
rec[18 + i] = 0;
}
}
// initiate get
binary_state = 21;
get_file_name = src_fname;
get_file_data = new Uint8Array(0);
update_file_status('Getting ' + get_file_name + '...');
ws.send(rec);
}
function get_ver() {
// WEBREPL_REQ_S = "<2sBBQLH64s"
var rec = new Uint8Array(2 + 1 + 1 + 8 + 4 + 2 + 64);
rec[0] = 'W'.charCodeAt(0);
rec[1] = 'A'.charCodeAt(0);
rec[2] = 3; // GET_VER
// rest of "rec" is zero
// initiate GET_VER
binary_state = 31;
ws.send(rec);
}
function handle_put_file_select(evt) {
// The event holds a FileList object which is a list of File objects,
// but we only support single file selection at the moment.
var files = evt.target.files;
// Get the file info and load its data.
var f = files[0];
put_file_name = f.name;
var reader = new FileReader();
reader.onload = function(e) {
put_file_data = new Uint8Array(e.target.result);
document.getElementById('put-file-list').innerHTML = '' + escape(put_file_name) + ' - ' + put_file_data.length + ' bytes';
document.getElementById('put-file-button').disabled = false;
};
reader.readAsArrayBuffer(f);
}
document.getElementById('put-file-select').addEventListener('click', function(){
this.value = null;
}, false);
document.getElementById('put-file-select').addEventListener('change', handle_put_file_select, false);
document.getElementById('put-file-button').disabled = true;
智能硬件社区聚焦AI智能硬件技术生态，汇聚嵌入式AI、物联网硬件开发者，打造交流分享平台，同步全国赛事资讯、开展 OPC 核心人才招募，助力技术落地与开发者成长。
更多推荐
0
0- 0
已为社区贡献4条内容
所有评论(0)