import _thread
from machine import SPI
from machine import Pin
+import uctypes
# spi - 通道选择[0,1]，int类型。
# 设置引脚电平
gpioWakeBT.write(1)
+# 提前创建1024字节SPI发送缓冲区（全局初始化，避免重复创建）
+SPI_TX_BUFFER = bytearray(1024)
+
+# 兼容型字节拷贝函数（替代切片赋值，适配所有MicroPython版本）
+def copy_bytes_to_buffer(buffer, data, offset=0):
+ """
+ 高速内存拷贝（替代逐字节循环）
+ :param buffer: 目标缓冲区（bytearray）
+ :param data: 源数据（bytes/bytearray）
+ :param offset: 缓冲区起始偏移
+ """
+ data_len = len(data)
+ if data_len == 0 or offset + data_len > len(buffer):
+ return
+ try:
+ # 优先使用内存块拷贝（最快）
+ uctypes.memmove(uctypes.addressof(buffer) + offset,
+ uctypes.addressof(data),
+ data_len
+ )
+ except:
+ # 降级到逐字节拷贝（仅兼容）
+ for i in range(data_len):
+ buffer[offset + i] = data[i]
+
class HttpFileDownloader:
"""
HTTP 文件下载类
def __download_and_send(self, conn, url):
"""
- 下载文件并通过 SPI 发送数据
+ 下载文件并通过 SPI 发送数据（提前补0优化版，兼容所有MicroPython）
Args:
conn: 客户端连接对象（用于接收 URL 请求，数据通过 SPI 发送）
remaining_data = b''
# 累计补0字节数
total_padded_zeros = 0
+ # 全局SPI缓冲区引用
+ tx_buffer = SPI_TX_BUFFER
# response.content 是 bytes 类型生成器
for chunk in response.content:
combined_data = remaining_data + chunk
remaining_data = b''
- # 拆分数据为1024字节的块
+ # 拆分数据为1024字节的块（提前填充缓冲区）
while len(combined_data) >= 1024:
- # 取前1024字节
- send_chunk = combined_data[:1024]
+ # 1. 提前填充缓冲区（兼容型逐字节拷贝，替代切片赋值）
+ copy_bytes_to_buffer(tx_buffer, combined_data[:1024], 0)
# 剩余数据保留
combined_data = combined_data[1024:]
+ # 2. GPIO电平切换（原有时序）
gpioWakeBT.write(0)
utime.sleep_ms(2)
gpioWakeBT.write(1)
utime.sleep_ms(2)
- # 通过 SPI 发送1024字节数据块
- chunk_len = len(send_chunk) # 固定为1024
+ # 3. 发送缓冲区数据（已提前填充，无拼接耗时）
if hasattr(self.spi_obj, 'write'):
- self.spi_obj.write(send_chunk, chunk_len)
- utime.sleep_ms(5)
+ self.spi_obj.write(tx_buffer, 1024)
elif hasattr(self.spi_obj, 'write_read'):
- read_buf = bytearray(chunk_len)
+ read_buf = bytearray(1024)
try:
- self.spi_obj.write_read(read_buf, send_chunk, chunk_len, 5)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024, 5)
except TypeError:
- self.spi_obj.write_read(read_buf, send_chunk, chunk_len)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024)
else:
clog.error("SPI object does not have write or write_read method")
return False, 0.0, -1
- downloaded_size += len(send_chunk)
+ downloaded_size += 1024
# 显示进度（每发送 64KB 打印一次进度）
if total_size > 0:
clog.error("Error sending data via SPI: {}".format(e))
return False, 0.0, -1
- # å¤\84ç\90\86æ\9c\80å\90\8eä¸\8d足1024å\97è\8a\82ç\9a\84å\89©ä½\99æ\95°æ\8d®ï¼\88æ\8b¼æ\8e¥ç©ºå\97è\8a\82ï¼\8cå®\8cå\85¨é\81¿å¼\80bytearrayä¸\8bæ \87èµ\8bå\80¼）
+ # å¤\84ç\90\86æ\9c\80å\90\8eä¸\8d足1024å\97è\8a\82ç\9a\84å\89©ä½\99æ\95°æ\8d®ï¼\88æ\8f\90å\89\8då¡«å\85\85ç¼\93å\86²å\8cºè¡¥0ï¼\8cå\85¼å®¹ç\89\88）
if len(remaining_data) > 0:
try:
- # 计算需要补的空字节数
- pad_len = 1024 - len(remaining_data)
- total_padded_zeros += pad_len # 累计补0字节数
- # 拼接空字节（直接生成pad_len个0字节的bytes）
- send_chunk = remaining_data + b'\x00' * pad_len
+ # 1. 计算补0数（提前计算）
+ data_len = len(remaining_data)
+ pad_len = 1024 - data_len
+ total_padded_zeros += pad_len
+
+ # 2. 提前填充缓冲区（兼容型逐字节拷贝，无切片赋值）
+ # 清空缓冲区（可选，bytearray初始为0，这里显式清空确保兼容性）
+ for i in range(1024):
+ tx_buffer[i] = 0
+ # 填充有效数据（逐字节拷贝，避免下标赋值报错）
+ copy_bytes_to_buffer(tx_buffer, remaining_data, 0)
+ clog.info("=== Pre-filled SPI Buffer Info ===")
+ clog.info("Buffer filled with {} bytes valid data (pre-send)".format(data_len))
+ clog.info("Buffer padded with {} bytes zeros (pre-send)".format(pad_len))
+ clog.info("Buffer filled via byte-by-byte copy (compatible mode)")
+ clog.info("====================================")
+
+ # 3. GPIO电平切换（原有时序）
gpioWakeBT.write(0)
utime.sleep_ms(2)
gpioWakeBT.write(1)
utime.sleep_ms(2)
+ # 4. 发送已补0的缓冲区（无任何拼接操作）
if hasattr(self.spi_obj, 'write'):
- self.spi_obj.write(send_chunk, 1024)
- utime.sleep_ms(5)
+ self.spi_obj.write(tx_buffer, 1024)
elif hasattr(self.spi_obj, 'write_read'):
read_buf = bytearray(1024)
try:
- self.spi_obj.write_read(read_buf, send_chunk, 1024, 5)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024, 5)
except TypeError:
- self.spi_obj.write_read(read_buf, send_chunk, 1024)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024)
- downloaded_size += len(remaining_data) # 只统计实际数据长度
- clog.info("=== Final SPI Chunk Info ===")
- clog.info("Actual data bytes: {}".format(len(remaining_data)))
- clog.info("Padded zero bytes: {}".format(pad_len))
- clog.info("Total padded zeros in whole transmission: {}".format(total_padded_zeros))
- clog.info("============================")
+ downloaded_size += data_len # 只统计实际数据长度
+ clog.info("=== Final SPI Chunk Detail ===")
+ clog.info("Last chunk actual data: {} bytes".format(data_len))
+ clog.info("Last chunk padded zeros: {} bytes (pre-send)".format(pad_len))
+ clog.info("Total padded zeros in transmission: {} bytes".format(total_padded_zeros))
+ clog.info("Buffer fill mode: Byte-by-byte copy (compatible with all MicroPython)")
+ clog.info("==============================")
# 最后一块进度显示
if total_size > 0:
# 打印整体传输统计
clog.info("=== SPI Transmission Summary ===")
clog.info("Total actual data sent: {} bytes".format(downloaded_size))
- clog.info("Total padded zero bytes: {} bytes".format(total_padded_zeros))
+ clog.info("Total padded zero bytes: {} bytes (all pre-send)".format(total_padded_zeros))
clog.info("Total SPI packets sent: {} (each 1024 bytes)".format((downloaded_size + total_padded_zeros) // 1024))
clog.info("Transmission time: {:.3f} seconds".format(download_time))
+ clog.info("Buffer fill mode: Byte-by-byte copy (compatible mode)")
clog.info("================================")
return True, download_time, downloaded_size
def start_download_and_send(self):
"""
- 开始下载文件并通过SPI发送（用于测试SPI传输速度）
+ 开始下载文件并通过SPI发送（提前补0优化版，兼容所有MicroPython）
Returns:
tuple: (success: bool, download_time: float, file_size: int)
remaining_data = b''
# 累计补0字节数
total_padded_zeros = 0
+ # 全局SPI缓冲区引用
+ tx_buffer = SPI_TX_BUFFER
# response.content是bytes类型生成器
for chunk in response.content:
combined_data = remaining_data + chunk
remaining_data = b''
- # 拆分数据为1024字节的块
+ # 拆分数据为1024字节的块（提前填充缓冲区，兼容版）
while len(combined_data) >= 1024:
- # 取前1024字节
- send_chunk = combined_data[:1024]
+ # 1. 提前填充缓冲区（逐字节拷贝，替代切片赋值）
+ copy_bytes_to_buffer(tx_buffer, combined_data[:1024], 0)
# 剩余数据保留
combined_data = combined_data[1024:]
+ # 2. GPIO电平切换（原有时序）
gpioWakeBT.write(0)
utime.sleep_ms(2)
gpioWakeBT.write(1)
utime.sleep_ms(2)
- # 通过SPI发送1024字节数据块
- chunk_len = len(send_chunk) # 固定为1024
+ # 3. 发送缓冲区数据（无拼接耗时）
+ chunk_len = 1024
if hasattr(self.spi_obj, 'write'):
- self.spi_obj.write(send_chunk, chunk_len)
- utime.sleep_ms(5)
+ self.spi_obj.write(tx_buffer, chunk_len)
elif hasattr(self.spi_obj, 'write_read'):
read_buf = bytearray(chunk_len)
try:
- self.spi_obj.write_read(read_buf, send_chunk, chunk_len, 5)
+ self.spi_obj.write_read(read_buf, tx_buffer, chunk_len, 5)
except TypeError:
- self.spi_obj.write_read(read_buf, send_chunk, chunk_len)
+ self.spi_obj.write_read(read_buf, tx_buffer, chunk_len)
else:
clog.error("SPI object does not have write or write_read method")
self.close() # 关闭SPI
return False, 0.0, -1
- downloaded_size += len(send_chunk)
+ downloaded_size += 1024
# 显示进度（每发送64KB打印一次，减少打印数量）
if total_size > 0:
self.close() # 关闭SPI
return False, 0.0, -1
- # å¤\84ç\90\86æ\9c\80å\90\8eä¸\8d足1024å\97è\8a\82ç\9a\84å\89©ä½\99æ\95°æ\8d®ï¼\88æ\8b¼æ\8e¥ç©ºå\97è\8a\82ï¼\8cå®\8cå\85¨é\81¿å¼\80bytearrayä¸\8bæ \87èµ\8bå\80¼）
+ # å¤\84ç\90\86æ\9c\80å\90\8eä¸\8d足1024å\97è\8a\82ç\9a\84å\89©ä½\99æ\95°æ\8d®ï¼\88æ\8f\90å\89\8då¡«å\85\85ç¼\93å\86²å\8cºè¡¥0ï¼\8cå\85¼å®¹ç\89\88）
if len(remaining_data) > 0:
try:
- # 计算需要补的空字节数
- pad_len = 1024 - len(remaining_data)
- total_padded_zeros += pad_len # 累计补0字节数
- # 直接拼接空字节（b'\x00' * pad_len），无需bytearray
- send_chunk = remaining_data + (b'\x00' * pad_len)
+ # 1. 提前计算补0数和填充缓冲区
+ data_len = len(remaining_data)
+ pad_len = 1024 - data_len
+ total_padded_zeros += pad_len
+
+ # 2. 提前填充缓冲区（逐字节拷贝，无切片赋值）
+ # 清空缓冲区（显式清空，确保兼容性）
+ for i in range(1024):
+ tx_buffer[i] = 0
+ # 填充有效数据（逐字节拷贝）
+ copy_bytes_to_buffer(tx_buffer, remaining_data, 0)
+
+ clog.info("=== Pre-filled SPI Buffer Info ===")
+ clog.info("Buffer filled with {} bytes valid data (pre-send)".format(data_len))
+ clog.info("Buffer padded with {} bytes zeros (pre-send)".format(pad_len))
+ clog.info("Buffer fill mode: Byte-by-byte copy (compatible with all MicroPython)")
+ clog.info("No slice assignment used (avoid 'bytearray item assignment' error)")
+ clog.info("====================================")
+ # 3. GPIO电平切换（原有时序）
gpioWakeBT.write(0)
utime.sleep_ms(2)
gpioWakeBT.write(1)
utime.sleep_ms(2)
+ # 4. 发送已补0的缓冲区（无任何拼接/计算）
if hasattr(self.spi_obj, 'write'):
- self.spi_obj.write(send_chunk, 1024)
- utime.sleep_ms(5)
+ self.spi_obj.write(tx_buffer, 1024)
elif hasattr(self.spi_obj, 'write_read'):
read_buf = bytearray(1024)
try:
- self.spi_obj.write_read(read_buf, send_chunk, 1024, 5)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024, 5)
except TypeError:
- self.spi_obj.write_read(read_buf, send_chunk, 1024)
+ self.spi_obj.write_read(read_buf, tx_buffer, 1024)
- downloaded_size += len(remaining_data) # 只统计实际数据长度
+ downloaded_size += data_len # 只统计实际数据长度
# 打印最后一块补0详情
clog.info("=== Final SPI Chunk Detail ===")
- clog.info("Last chunk actual data: {} bytes".format(len(remaining_data)))
- clog.info("Last chunk padded zeros: {} bytes".format(pad_len))
+ clog.info("Last chunk actual data: {} bytes".format(data_len))
+ clog.info("Last chunk padded zeros: {} bytes (all pre-send)".format(pad_len))
clog.info("Total padded zeros in transmission: {} bytes".format(total_padded_zeros))
+ clog.info("Buffer fill mode: Byte-by-byte copy (compatible)")
clog.info("==============================")
# 最后一块进度显示
# 打印整体传输统计信息
clog.info("=== SPI Transmission Complete Summary ===")
clog.info("Total actual data transmitted: {} bytes ({:.2f} MB)".format(downloaded_size, downloaded_size/1024/1024))
- clog.info("Total padded zero bytes: {} bytes ({:.2f} KB)".format(total_padded_zeros, total_padded_zeros/1024))
+ clog.info("Total padded zero bytes: {} bytes ({:.2f} KB) (all pre-send)".format(total_padded_zeros, total_padded_zeros/1024))
clog.info("Total SPI packets sent: {} (each 1024 bytes)".format((downloaded_size + total_padded_zeros) // 1024))
clog.info("Total transmission time: {:.3f} seconds".format(download_time))
if speed_mbps >= 1.0:
clog.info("Effective transmission speed: {:.2f} Mbps ({:.2f} Kbps)".format(speed_mbps, speed_kbps))
else:
clog.info("Effective transmission speed: {:.2f} Kbps".format(speed_kbps))
+ clog.info("Padding strategy: All zeros padded before SPI transmission (no runtime overhead)")
+ clog.info("Compatibility mode: Byte-by-byte buffer copy (no slice assignment)")
clog.info("==========================================")
# 传输完成后关闭SPI
projects / mirrors / FD230.git / commitdiff