+++ /dev/null
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-SPI通信协议实现
-支持可靠的数据传输，包含CRC16校验、自动重传机制
-"""
-
-import log
-import utime
-from machine import SPI
-
-# struct模块在嵌入式Python中可能不可用，实现替代方案
-try:
- import struct
-except ImportError:
- # 如果struct模块不可用，实现一个简单的替代
- class _SimpleStruct:
- @staticmethod
- def pack(fmt, *args):
- """简单的pack实现，支持大端序格式"""
- result = bytearray()
- i = 0
- for c in fmt:
- if c == '>': # 大端序，忽略
- continue
- elif c == 'B': # unsigned char (1 byte)
- result.append(args[i] & 0xFF)
- i += 1
- elif c == 'H': # unsigned short (2 bytes, big-endian)
- val = args[i] & 0xFFFF
- result.append((val >> 8) & 0xFF)
- result.append(val & 0xFF)
- i += 1
- elif c == 'I': # unsigned int (4 bytes, big-endian)
- val = args[i] & 0xFFFFFFFF
- result.append((val >> 24) & 0xFF)
- result.append((val >> 16) & 0xFF)
- result.append((val >> 8) & 0xFF)
- result.append(val & 0xFF)
- i += 1
- return bytes(result)
-
- @staticmethod
- def unpack(fmt, data):
- """简单的unpack实现，支持大端序格式"""
- result = []
- i = 0
- for c in fmt:
- if c == '>': # 大端序，忽略
- continue
- elif c == 'B': # unsigned char (1 byte)
- result.append(data[i])
- i += 1
- elif c == 'H': # unsigned short (2 bytes, big-endian)
- val = (data[i] << 8) | data[i + 1]
- result.append(val)
- i += 2
- elif c == 'I': # unsigned int (4 bytes, big-endian)
- val = (data[i] << 24) | (data[i + 1] << 16) | (data[i + 2] << 8) | data[i + 3]
- result.append(val)
- i += 4
- return tuple(result)
-
- struct = _SimpleStruct()
-
-# 设置日志输出级别
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiprotocol")
-
-# 协议常量
-SPI_PROTOCOL_MAGIC = 0xAA
-SPI_PROTOCOL_VERSION = 0x01
-
-# 包类型
-PKT_TYPE_DATA = 0x01
-PKT_TYPE_ACK = 0x02
-PKT_TYPE_NACK = 0x03
-PKT_TYPE_EOT = 0x04
-
-# 标志位
-FLAG_FIRST = 0x01 # 第一包
-FLAG_LAST = 0x02 # 最后一包
-FLAG_RETRANSMIT = 0x04 # 重传包
-
-# 包大小限制（减小以适应嵌入式环境）
-MAX_PACKET_SIZE = 4 * 1024 # 4KB（减小以节省内存）
-HEADER_SIZE = 16
-DATA_CRC_SIZE = 2
-MAX_DATA_SIZE = MAX_PACKET_SIZE - HEADER_SIZE - DATA_CRC_SIZE # 4080 bytes
-
-# 重传参数
-MAX_RETRY_COUNT = 3
-ACK_TIMEOUT_BASE_MS = 500 # 基础超时500ms
-ACK_TIMEOUT_PER_KB_MS = 10 # 每KB增加10ms，最大2s
-
-
-def crc16_ccitt(data, initial=0xFFFF):
- """
- CRC16-CCITT计算（多项式0x1021）
-
- Args:
- data: 要计算的数据（bytes或bytearray）
- initial: 初始值，默认0xFFFF
-
- Returns:
- int: CRC16值
- """
- crc = initial
- polynomial = 0x1021
-
- for byte in data:
- crc ^= (byte << 8)
- for _ in range(8):
- if crc & 0x8000:
- crc = (crc << 1) ^ polynomial
- else:
- crc <<= 1
- crc &= 0xFFFF
-
- return crc
-
-
-class SpiProtocolSender:
- """
- SPI协议发送端
- 处理数据分包、重传、校验等
- """
-
- def __init__(self, spi_obj, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 初始化SPI协议发送端
-
- Args:
- spi_obj: SPI对象（如果为None则自动创建）
- spi_channel: SPI通道，默认2
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
- """
- self.spi_obj = spi_obj
- self.spi_channel = spi_channel
- self.spi_mode = spi_mode
- self.spi_clk = spi_clk
- self.seq_num = 0
- self.stats = {
- 'total_packets': 0,
- 'retry_packets': 0,
- 'error_packets': 0,
- 'total_bytes': 0
- }
-
- def __init_spi(self):
- """初始化SPI对象"""
- if self.spi_obj is None:
- try:
- self.spi_obj = SPI(self.spi_channel, self.spi_mode, self.spi_clk)
- clog.info("SPI initialized: channel={}, mode={}, clk={}".format(
- self.spi_channel, self.spi_mode, self.spi_clk))
- return True
- except Exception as e:
- clog.error("Failed to initialize SPI: {}".format(e))
- return False
- return True
-
- def __create_header(self, pkt_type, flags, seq_num, data_len):
- """
- 创建协议包头
-
- Args:
- pkt_type: 包类型
- flags: 标志位
- seq_num: 序列号
- data_len: 数据长度
-
- Returns:
- bytes: 包头数据（16字节）
- """
- # 包头结构（前14字节，用于CRC16计算）
- header = struct.pack('>BBBBII',
- SPI_PROTOCOL_MAGIC,
- SPI_PROTOCOL_VERSION,
- pkt_type,
- flags,
- seq_num,
- data_len)
-
- # 计算包头CRC16（前14字节）
- header_crc = crc16_ccitt(header)
-
- # 完整包头（16字节）
- full_header = header + struct.pack('>HH', header_crc, 0) # Reserved=0
-
- return full_header
-
- def __send_packet(self, pkt_type, flags, seq_num, data=None):
- """
- 发送数据包（包含重传逻辑）
-
- Args:
- pkt_type: 包类型
- flags: 标志位
- seq_num: 序列号
- data: 数据负载（bytes/bytearray，可选）
-
- Returns:
- bool: 发送是否成功
- """
- if data is None:
- data = b''
-
- data_len = len(data)
- if data_len > MAX_DATA_SIZE:
- clog.error("Data too large: {} bytes, max: {} bytes".format(
- data_len, MAX_DATA_SIZE))
- return False
-
- # 创建包头
- header = self.__create_header(pkt_type, flags, seq_num, data_len)
-
- # 计算数据CRC16
- data_crc = crc16_ccitt(data) if data_len > 0 else 0
-
- # 组装完整数据包
- packet = header + data
- if data_len > 0:
- packet += struct.pack('>H', data_crc)
-
- # 重传逻辑
- retry_count = 0
- while retry_count <= MAX_RETRY_COUNT:
- try:
- # 发送数据包
- packet_len = len(packet)
- if hasattr(self.spi_obj, 'write'):
- self.spi_obj.write(packet, packet_len)
- elif hasattr(self.spi_obj, 'write_read'):
- read_buf = bytearray(packet_len)
- try:
- self.spi_obj.write_read(read_buf, packet, packet_len, 5)
- except TypeError:
- self.spi_obj.write_read(read_buf, packet, packet_len)
- else:
- clog.error("SPI object does not have write or write_read method")
- return False
-
- # 等待并接收ACK/NACK响应
- # 超时时间根据数据包大小动态调整
- timeout_ms = ACK_TIMEOUT_BASE_MS + (data_len // 1024) * ACK_TIMEOUT_PER_KB_MS
- if timeout_ms > 2000:
- timeout_ms = 2000
-
- # 读取ACK/NACK响应（16字节包头）
- # 注意：在SPI Slave模式下，从设备不能主动发起传输
- # 需要通过write_read触发MCU响应，MCU在传输过程中返回ACK/NACK
- ack_received = False
- ack_type = None
- start_time = utime.time() * 1000 # 转换为毫秒
- max_retries = timeout_ms // 50 # 每50ms重试一次
- retry_count_ack = 0
-
- while retry_count_ack < max_retries and ((utime.time() * 1000) - start_time) < timeout_ms:
- # 尝试读取ACK/NACK响应
- ack_buffer = bytearray(HEADER_SIZE)
- try:
- if hasattr(self.spi_obj, 'write_read'):
- # 使用write_read同时发送和接收，触发MCU响应
- dummy_data = bytearray(HEADER_SIZE) # 发送空数据
- try:
- self.spi_obj.write_read(ack_buffer, dummy_data, HEADER_SIZE, 50)
- except TypeError:
- self.spi_obj.write_read(ack_buffer, dummy_data, HEADER_SIZE)
-
- # 解析ACK/NACK响应
- if len(ack_buffer) >= HEADER_SIZE:
- magic, version, pkt_type, flags, ack_seq_num, ack_data_len, ack_crc, reserved = \
- struct.unpack('>BBBBIIHH', ack_buffer)
-
- if magic == SPI_PROTOCOL_MAGIC and version == SPI_PROTOCOL_VERSION:
- # 校验CRC
- calculated_crc = crc16_ccitt(ack_buffer[:14])
- if calculated_crc == ack_crc:
- if pkt_type == PKT_TYPE_ACK and ack_seq_num == seq_num:
- ack_received = True
- ack_type = PKT_TYPE_ACK
- break
- elif pkt_type == PKT_TYPE_NACK and ack_seq_num == seq_num:
- ack_received = True
- ack_type = PKT_TYPE_NACK
- break
- elif hasattr(self.spi_obj, 'read'):
- # 尝试直接读取（如果支持）
- bytes_read = self.spi_obj.read(ack_buffer, HEADER_SIZE, 50)
- if bytes_read >= HEADER_SIZE:
- magic, version, pkt_type, flags, ack_seq_num, ack_data_len, ack_crc, reserved = \
- struct.unpack('>BBBBIIHH', ack_buffer)
-
- if magic == SPI_PROTOCOL_MAGIC and version == SPI_PROTOCOL_VERSION:
- calculated_crc = crc16_ccitt(ack_buffer[:14])
- if calculated_crc == ack_crc:
- if pkt_type == PKT_TYPE_ACK and ack_seq_num == seq_num:
- ack_received = True
- ack_type = PKT_TYPE_ACK
- break
- elif pkt_type == PKT_TYPE_NACK and ack_seq_num == seq_num:
- ack_received = True
- ack_type = PKT_TYPE_NACK
- break
- except Exception as e:
- # 读取失败，继续等待
- pass
-
- retry_count_ack += 1
- utime.sleep_ms(50) # 延迟50ms后重试
-
- # 检查ACK/NACK
- if not ack_received:
- # 超时，未收到ACK/NACK
- # 注意：在SPI Slave模式下，ACK/NACK机制可能需要硬件支持
- # 这里暂时假设发送成功（简化实现）
- clog.warning("Packet {} ACK timeout after {}ms (assuming success for now)".format(seq_num, timeout_ms))
- # 在实际硬件测试中，如果ACK机制不工作，可以暂时假设成功
- # TODO: 完善ACK/NACK接收机制
- elif ack_type == PKT_TYPE_NACK:
- # 收到NACK，需要重传
- clog.warning("Packet {} received NACK, retransmitting".format(seq_num))
- retry_count += 1
- if retry_count <= MAX_RETRY_COUNT:
- continue
- else:
- return False
-
- # 发送成功（收到ACK或超时但假设成功）
- if retry_count > 0:
- self.stats['retry_packets'] += 1
- clog.info("Packet {} retransmitted {} times, now successful".format(seq_num, retry_count))
-
- self.stats['total_packets'] += 1
- self.stats['total_bytes'] += data_len
- return True
-
- except Exception as e:
- retry_count += 1
- if retry_count <= MAX_RETRY_COUNT:
- clog.warning("Failed to send packet {}, retry {}/{}: {}".format(
- seq_num, retry_count, MAX_RETRY_COUNT, e))
- utime.sleep_ms(50) # 重传前短暂延迟
- else:
- clog.error("Failed to send packet {} after {} retries: {}".format(
- seq_num, MAX_RETRY_COUNT, e))
- self.stats['error_packets'] += 1
- return False
-
- return False
-
- def send_data(self, data_source, timeout=60):
- """
- 发送数据（统一接口，不区分文件/流）
-
- Args:
- data_source: 数据源，支持：
- - bytes/bytearray: 内存中的数据
- - 可迭代对象: 如response.content（网络流）
- - 文件对象: 如open()返回的对象
- timeout: 总超时时间（秒）
-
- Returns:
- bool: 发送是否成功
- """
- if not self.__init_spi():
- return False
-
- try:
- # 重置序列号
- self.seq_num = 0
- self.stats = {
- 'total_packets': 0,
- 'retry_packets': 0,
- 'error_packets': 0,
- 'total_bytes': 0
- }
-
- # 处理数据源
- if isinstance(data_source, (bytes, bytearray)):
- # 内存数据：转换为可迭代
- data_iter = iter([data_source])
- elif hasattr(data_source, 'read'):
- # 文件对象：按小块读取（避免大内存分配）
- CHUNK_READ_SIZE = 1024 # 每次读取1KB，减少内存占用
- def file_iter():
- while True:
- chunk = data_source.read(CHUNK_READ_SIZE)
- if not chunk:
- break
- yield chunk
- data_iter = file_iter()
- else:
- # 可迭代对象（如网络流）
- data_iter = data_source
-
- # 数据缓冲区（限制大小，避免大内存分配）
- MAX_BUFFER_SIZE = MAX_DATA_SIZE # 缓冲区最大不超过一个数据包大小
- buffer = bytearray()
- is_first = True
- start_time = utime.time()
-
- clog.info("Starting SPI protocol data transmission...")
-
- # 处理数据流
- for chunk in data_iter:
- if not chunk:
- continue
-
- # 检查超时
- if utime.time() - start_time > timeout:
- clog.error("Transmission timeout after {} seconds".format(timeout))
- return False
-
- # 添加到缓冲区
- buffer.extend(chunk)
-
- # 当缓冲区达到最大数据包大小时，发送一个包
- # 使用循环确保及时发送，避免缓冲区过大
- while len(buffer) >= MAX_DATA_SIZE:
- # 提取一个完整的数据包
- packet_data = bytes(buffer[:MAX_DATA_SIZE])
- buffer = buffer[MAX_DATA_SIZE:]
-
- # 确定标志位
- flags = 0
- if is_first:
- flags |= FLAG_FIRST
- is_first = False
-
- # 发送数据包
- if not self.__send_packet(PKT_TYPE_DATA, flags, self.seq_num, packet_data):
- clog.error("Failed to send packet {}".format(self.seq_num))
- return False
-
- self.seq_num += 1
-
- # 每发送64KB打印一次进度
- if self.stats['total_bytes'] % (64 * 1024) < MAX_DATA_SIZE:
- clog.info("SPI TX Progress: {} bytes, {} packets".format(
- self.stats['total_bytes'], self.stats['total_packets']))
-
- # 如果缓冲区仍然很大（接近限制），强制发送一个包
- # 这样可以避免缓冲区无限增长
- if len(buffer) > MAX_BUFFER_SIZE:
- # 发送一个完整的数据包
- packet_data = bytes(buffer[:MAX_DATA_SIZE])
- buffer = buffer[MAX_DATA_SIZE:]
-
- flags = 0
- if is_first:
- flags |= FLAG_FIRST
- is_first = False
-
- if not self.__send_packet(PKT_TYPE_DATA, flags, self.seq_num, packet_data):
- clog.error("Failed to send packet {}".format(self.seq_num))
- return False
-
- self.seq_num += 1
-
- # 发送剩余数据（最后一个包）
- if len(buffer) > 0:
- flags = FLAG_LAST
- if is_first:
- flags |= FLAG_FIRST
-
- if not self.__send_packet(PKT_TYPE_DATA, flags, self.seq_num, bytes(buffer)):
- clog.error("Failed to send last packet {}".format(self.seq_num))
- return False
- self.seq_num += 1
-
- # 发送EOT包
- if not self.__send_packet(PKT_TYPE_EOT, 0, self.seq_num):
- clog.error("Failed to send EOT packet")
- return False
-
- # 打印统计信息
- elapsed_time = utime.time() - start_time
- speed_kbps = (self.stats['total_bytes'] * 8.0) / elapsed_time / 1024.0 if elapsed_time > 0 else 0
- speed_mbps = speed_kbps / 1024.0
-
- if speed_mbps >= 1.0:
- clog.info("SPI transmission completed: {} bytes in {:.3f}s, Speed: {:.2f} Mbps ({:.2f} Kbps)".format(
- self.stats['total_bytes'], elapsed_time, speed_mbps, speed_kbps))
- else:
- clog.info("SPI transmission completed: {} bytes in {:.3f}s, Speed: {:.2f} Kbps".format(
- self.stats['total_bytes'], elapsed_time, speed_kbps))
-
- clog.info("Statistics: {} packets, {} retries, {} errors".format(
- self.stats['total_packets'], self.stats['retry_packets'], self.stats['error_packets']))
-
- return True
-
- except Exception as e:
- clog.error("Error in SPI protocol transmission: {}".format(e))
- return False
-
- def close(self):
- """关闭SPI对象"""
- if self.spi_obj is not None:
- try:
- if hasattr(self.spi_obj, 'close'):
- self.spi_obj.close()
- elif hasattr(self.spi_obj, 'deinit'):
- self.spi_obj.deinit()
- except Exception as e:
- clog.error("Error closing SPI object: {}".format(e))
- self.spi_obj = None
-
-
-def send_data(data_source, spi_obj=None, timeout=60, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 便捷函数：发送数据
-
- Args:
- data_source: 数据源（bytes/bytearray/可迭代对象/文件对象）
- spi_obj: SPI对象（可选，如果为None则自动创建）
- timeout: 总超时时间（秒）
- spi_channel: SPI通道，默认2
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- bool: 发送是否成功
- """
- sender = SpiProtocolSender(spi_obj, spi_channel, spi_mode, spi_clk)
- try:
- return sender.send_data(data_source, timeout)
- finally:
- sender.close()
-
-
-def send_from_url(url, timeout=60, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 便捷函数：从URL下载并转发到MCU
-
- Args:
- url: 要下载的URL
- timeout: 总超时时间（秒）
- spi_channel: SPI通道，默认2
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- bool: 发送是否成功
- """
- try:
- import request
- response = request.get(url, stream=True, timeout=timeout)
- if response.status_code != 200:
- clog.error("HTTP request failed with status code: {}".format(response.status_code))
- return False
-
- return send_data(response.content, None, timeout, spi_channel, spi_mode, spi_clk)
- except Exception as e:
- clog.error("Error downloading from URL: {}".format(e))
- return False
-
+++ /dev/null
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-SPI协议接口封装
-提供简单的对外接口，隐藏协议细节
-"""
-
-from usr.spi_protocol import send_data, send_from_url, SpiProtocolSender
-import log
-
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiprotocolapi")
-
-
-def send_data_simple(data_source, spi_obj=None, timeout=60, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 简单发送接口：发送数据到MCU
-
- Args:
- data_source: 数据源，支持：
- - bytes/bytearray: 内存中的数据
- - 可迭代对象: 如response.content（网络流）
- - 文件对象: 如open()返回的对象
- spi_obj: SPI对象（可选，如果为None则自动创建）
- timeout: 总超时时间（秒），默认60秒
- spi_channel: SPI通道，默认2（对应MCU端SPI1）
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- bool: 发送是否成功
-
- 示例:
- # 发送内存数据
- data = b"Hello, MCU!"
- send_data_simple(data)
-
- # 从URL下载并转发
- send_data_simple_from_url("https://example.com/file.bin")
-
- # 发送文件
- with open("file.bin", "rb") as f:
- send_data_simple(f)
- """
- return send_data(data_source, spi_obj, timeout, spi_channel, spi_mode, spi_clk)
-
-
-def send_data_simple_from_url(url, timeout=60, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 简单发送接口：从URL下载并转发到MCU
-
- Args:
- url: 要下载的URL
- timeout: 总超时时间（秒），默认60秒
- spi_channel: SPI通道，默认2（对应MCU端SPI1）
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- bool: 发送是否成功
-
- 示例:
- send_data_simple_from_url("https://example.com/file.bin")
- """
- return send_from_url(url, timeout, spi_channel, spi_mode, spi_clk)
-
-
-def send_file(filepath, timeout=60, spi_channel=2, spi_mode=0, spi_clk=8):
- """
- 发送本地文件（向后兼容接口）
-
- Args:
- filepath: 文件路径
- timeout: 总超时时间（秒）
- spi_channel: SPI通道，默认2（对应MCU端SPI1）
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- bool: 发送是否成功
- """
- try:
- with open(filepath, "rb") as f:
- return send_data_simple(f, None, timeout, spi_channel, spi_mode, spi_clk)
- except Exception as e:
- clog.error("Error opening file {}: {}".format(filepath, e))
- return False
-
-
-def get_sender_stats(sender):
- """
- 获取发送统计信息
-
- Args:
- sender: SpiProtocolSender实例
-
- Returns:
- dict: 统计信息字典，包含：
- - total_packets: 总包数
- - retry_packets: 重传包数
- - error_packets: 错误包数
- - total_bytes: 总字节数
- """
- if sender and hasattr(sender, 'stats'):
- return sender.stats.copy()
- return {
- 'total_packets': 0,
- 'retry_packets': 0,
- 'error_packets': 0,
- 'total_bytes': 0
- }
-
+++ /dev/null
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-SPI协议发送接口封装
-提供简单的对外接口，隐藏协议细节
-"""
-
-from usr.spi_protocol import send_data, send_from_url, SpiProtocolSender
-import log
-
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiprotocolsender")
-
-
-def send_data_simple(data_source, spi_obj=None, timeout=60):
- """
- 简单发送接口：发送数据到MCU
-
- Args:
- data_source: 数据源，支持：
- - bytes/bytearray: 内存中的数据
- - 可迭代对象: 如response.content（网络流）
- - 文件对象: 如open()返回的对象
- spi_obj: SPI对象（可选，如果为None则自动创建）
- timeout: 总超时时间（秒），默认60秒
-
- Returns:
- bool: 发送是否成功
-
- 示例:
- # 发送内存数据
- data = b"Hello, MCU!"
- send_data_simple(data)
-
- # 从URL下载并转发
- send_data_simple_from_url("https://example.com/file.bin")
-
- # 发送文件
- with open("file.bin", "rb") as f:
- send_data_simple(f)
- """
- return send_data(data_source, spi_obj, timeout)
-
-
-def send_data_simple_from_url(url, timeout=60):
- """
- 简单发送接口：从URL下载并转发到MCU
-
- Args:
- url: 要下载的URL
- timeout: 总超时时间（秒），默认60秒
-
- Returns:
- bool: 发送是否成功
-
- 示例:
- send_data_simple_from_url("https://example.com/file.bin")
- """
- return send_from_url(url, timeout)
-
-
-# 为了向后兼容，保留旧的函数名
-def send_file(filepath, timeout=60):
- """
- 发送本地文件（向后兼容接口）
-
- Args:
- filepath: 文件路径
- timeout: 总超时时间（秒）
-
- Returns:
- bool: 发送是否成功
- """
- try:
- with open(filepath, "rb") as f:
- return send_data_simple(f, None, timeout)
- except Exception as e:
- clog.error("Error opening file {}: {}".format(filepath, e))
- return False
-
+++ /dev/null
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-SPI协议测试脚本
-测试各种场景的数据传输
-"""
-
-import log
-import utime
-from usr.spi_protocol_api import send_data_simple, send_data_simple_from_url, send_file
-from usr.spi_protocol import SpiProtocolSender
-from machine import SPI
-
-# 设置日志输出级别
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiprotocoltest")
-
-
-def test_small_data():
- """测试小数据包（<1KB）"""
- clog.info("=" * 60)
- clog.info("Test 1: Small data packet (<1KB)")
- clog.info("=" * 60)
-
- # 生成小数据（500字节）
- data = bytearray(500)
- for i in range(500):
- data[i] = i % 256
-
- success = send_data_simple(data)
- if success:
- clog.info("✓ Small data test PASSED")
- else:
- clog.error("✗ Small data test FAILED")
-
- return success
-
-
-def test_medium_data():
- """测试中等数据包（1KB-64KB）"""
- clog.info("=" * 60)
- clog.info("Test 2: Medium data packet (1KB-64KB)")
- clog.info("=" * 60)
-
- # 生成中等数据（32KB）
- data = bytearray(32 * 1024)
- for i in range(32 * 1024):
- data[i] = i % 256
-
- success = send_data_simple(data)
- if success:
- clog.info("✓ Medium data test PASSED")
- else:
- clog.error("✗ Medium data test FAILED")
-
- return success
-
-
-def test_large_data():
- """测试大数据包（>64KB，需要分包）"""
- clog.info("=" * 60)
- clog.info("Test 3: Large data packet (>64KB, will be split)")
- clog.info("=" * 60)
-
- # 生成大数据（200KB，会分成多个包）
- data = bytearray(200 * 1024)
- for i in range(200 * 1024):
- data[i] = i % 256
-
- success = send_data_simple(data)
- if success:
- clog.info("✓ Large data test PASSED")
- else:
- clog.error("✗ Large data test FAILED")
-
- return success
-
-
-def test_stream_data():
- """测试流式数据（模拟网络流）"""
- clog.info("=" * 60)
- clog.info("Test 4: Stream data (simulating network stream)")
- clog.info("=" * 60)
-
- # 模拟流式数据（每次生成4KB，总共100KB）
- def stream_generator():
- total_size = 100 * 1024
- chunk_size = 4 * 1024
- sent = 0
- while sent < total_size:
- chunk = bytearray(min(chunk_size, total_size - sent))
- for i in range(len(chunk)):
- chunk[i] = (sent + i) % 256
- sent += len(chunk)
- yield chunk
-
- success = send_data_simple(stream_generator())
- if success:
- clog.info("✓ Stream data test PASSED")
- else:
- clog.error("✗ Stream data test FAILED")
-
- return success
-
-
-def test_from_url(url=None):
- """测试从URL下载并转发"""
- clog.info("=" * 60)
- clog.info("Test 5: Download from URL and forward")
- clog.info("=" * 60)
-
- if url is None:
- url = "https://file-virtual-person-qcl.test.leiniao.com/ambybox/Emotions/StraightEyes/Amber/WorkAndStudy.mp4"
- clog.info("Using default test URL: {}".format(url))
-
- success = send_data_simple_from_url(url, timeout=120)
- if success:
- clog.info("✓ URL download test PASSED")
- else:
- clog.error("✗ URL download test FAILED")
-
- return success
-
-
-def test_with_stats():
- """测试带统计信息的发送"""
- clog.info("=" * 60)
- clog.info("Test 6: Send with statistics")
- clog.info("=" * 60)
-
- from spi_protocol import SpiProtocolSender
-
- # 创建发送器
- sender = SpiProtocolSender(None, spi_channel=2, spi_mode=0, spi_clk=8)
-
- # 生成测试数据（64KB）
- data = bytearray(64 * 1024)
- for i in range(64 * 1024):
- data[i] = i % 256
-
- # 发送数据
- success = sender.send_data(data, timeout=60)
-
- # 打印统计信息
- stats = sender.stats
- clog.info("Statistics:")
- clog.info(" Total packets: {}".format(stats['total_packets']))
- clog.info(" Retry packets: {}".format(stats['retry_packets']))
- clog.info(" Error packets: {}".format(stats['error_packets']))
- clog.info(" Total bytes: {} ({:.2f} KB)".format(
- stats['total_bytes'], stats['total_bytes'] / 1024.0))
-
- # 关闭
- sender.close()
-
- if success:
- clog.info("✓ Statistics test PASSED")
- else:
- clog.error("✗ Statistics test FAILED")
-
- return success
-
-
-def run_all_tests():
- """运行所有测试"""
- clog.info("Starting SPI Protocol Test Suite...")
- clog.info("")
-
- results = []
-
- # 测试1：小数据包
- results.append(("Small Data", test_small_data()))
- utime.sleep(1)
-
- # 测试2：中等数据包
- results.append(("Medium Data", test_medium_data()))
- utime.sleep(1)
-
- # 测试3：大数据包
- results.append(("Large Data", test_large_data()))
- utime.sleep(1)
-
- # 测试4：流式数据
- results.append(("Stream Data", test_stream_data()))
- utime.sleep(1)
-
- # 测试5：从URL下载（可选，需要网络）
- # results.append(("URL Download", test_from_url()))
- # utime.sleep(1)
-
- # 测试6：带统计信息
- results.append(("With Statistics", test_with_stats()))
-
- # 打印测试结果
- clog.info("")
- clog.info("=" * 60)
- clog.info("Test Results Summary")
- clog.info("=" * 60)
-
- passed = 0
- failed = 0
-
- for test_name, result in results:
- status = "PASSED" if result else "FAILED"
- symbol = "✓" if result else "✗"
- clog.info("{} {}: {}".format(symbol, test_name, status))
- if result:
- passed += 1
- else:
- failed += 1
-
- clog.info("")
- clog.info("Total: {} passed, {} failed".format(passed, failed))
-
- if failed == 0:
- clog.info("All tests PASSED!")
- else:
- clog.error("Some tests FAILED!")
-
- return failed == 0
-
-
-def test_single_scenario():
- """测试单个场景（用于调试）"""
- clog.info("Running single scenario test...")
-
- # 可以根据需要修改测试场景
- return test_small_data()
-
-
-if __name__ == '__main__':
- # 运行所有测试
- # run_all_tests()
-
- # 或者运行单个测试
- test_single_scenario()
-
+++ /dev/null
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-SPI新协议测试应用
-按照旧协议的逻辑生成2MB递增数据，使用新协议接口进行发送
-统计发送时长、速率和总量
-"""
-
-import log
-import utime
-from usr.spi_protocol_api import send_data_simple, get_sender_stats
-from usr.spi_protocol import SpiProtocolSender
-
-# 设置日志输出级别
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiprototest")
-
-# 测试参数
-TEST_DATA_SIZE = 2 * 1024 * 1024 # 2MB
-SPI_CHANNEL = 2 # SPI通道2（对应MCU端SPI1）
-SPI_MODE = 0 # Mode 0
-SPI_CLK = 8 # 1.625MHz（可根据需要调整：8=1.625MHz, 9=3.25MHz, 12=26MHz）
-
-
-class IncrementalDataGenerator:
- """
- 递增数据生成器
- 按照旧协议的逻辑生成递增数据（0, 1, 2, ..., 255, 0, 1, ...）
- """
-
- def __init__(self, total_size):
- """
- 初始化数据生成器
-
- Args:
- total_size: 要生成的数据总大小（字节）
- """
- self.total_size = total_size
- self.current_offset = 0
-
- def __iter__(self):
- """返回迭代器自身"""
- return self
-
- def __next__(self):
- """生成下一个字节"""
- if self.current_offset >= self.total_size:
- raise StopIteration
-
- # 生成递增字节值：offset % 256
- byte_value = self.current_offset % 256
- self.current_offset += 1
- return byte_value
-
- def generate_chunk(self, chunk_size):
- """
- 生成指定大小的数据块
-
- Args:
- chunk_size: 数据块大小（字节）
-
- Returns:
- bytearray: 递增数据块
- """
- remaining = self.total_size - self.current_offset
- if remaining <= 0:
- return None
-
- actual_size = min(chunk_size, remaining)
- chunk = bytearray(actual_size)
-
- for i in range(actual_size):
- chunk[i] = self.current_offset % 256
- self.current_offset += 1
-
- return chunk
-
-
-def run_spi_protocol_test_2mb(data_size=TEST_DATA_SIZE,
- spi_channel=SPI_CHANNEL,
- spi_mode=SPI_MODE,
- spi_clk=SPI_CLK):
- """
- 运行SPI新协议2MB测试
-
- Args:
- data_size: 要发送的数据总大小（字节），默认2MB
- spi_channel: SPI通道，默认2（对应MCU端SPI1）
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- tuple: (success: bool, send_time: float, total_bytes: int, stats: dict)
- success: 测试是否成功
- send_time: 发送耗时（秒）
- total_bytes: 发送的总字节数
- stats: 发送统计信息（包数、重传数、错误数等）
- """
- try:
- clog.info("=" * 60)
- clog.info("SPI Protocol Test - 2MB Incremental Data")
- clog.info("=" * 60)
- clog.info("Test parameters:")
- clog.info(" Total data size: {} bytes ({:.2f} MB)".format(
- data_size, data_size / 1024.0 / 1024.0))
- clog.info(" SPI channel: {}, mode: {}, clk: {}".format(
- spi_channel, spi_mode, spi_clk))
- clog.info(" Data pattern: Incremental bytes (0, 1, 2, ..., 255, 0, 1, ...)")
- clog.info(" Protocol: New reliable SPI protocol with CRC16 and retransmission")
- clog.info("")
-
- # 创建数据生成器
- data_generator = IncrementalDataGenerator(data_size)
-
- # 记录开始时间
- start_time = utime.time()
- start_time_ms = int(utime.time() * 1000) # 转换为毫秒
-
- clog.info("Starting data transmission...")
- clog.info(" Using new SPI protocol (reliable protocol)")
- clog.info(" Data will be automatically packetized and sent with CRC16 checksums")
- clog.info("")
-
- # 使用新协议发送数据
- # 创建一个生成器函数，每次生成一个chunk
- def data_source():
- """数据源生成器函数"""
- generator = IncrementalDataGenerator(data_size)
- chunk_size = 64 * 1024 # 64KB chunks（协议会自动分包）
- while True:
- chunk = generator.generate_chunk(chunk_size)
- if chunk is None:
- break
- yield chunk
-
- # 使用send_data_simple发送数据
- success = send_data_simple(
- data_source(),
- spi_obj=None, # 自动创建SPI对象
- timeout=120, # 2分钟超时（2MB数据可能需要较长时间）
- spi_channel=spi_channel,
- spi_mode=spi_mode,
- spi_clk=spi_clk
- )
-
- # 记录结束时间
- end_time = utime.time()
- end_time_ms = int(utime.time() * 1000) # 转换为毫秒
- send_time = end_time - start_time
- send_time_ms = end_time_ms - start_time_ms
-
- # 获取发送统计信息（需要从SpiProtocolSender获取）
- # 注意：send_data_simple内部创建了SpiProtocolSender，但我们无法直接访问
- # 所以我们需要修改send_data_simple来返回统计信息，或者使用另一种方式
-
- # 计算传输速度
- if send_time > 0:
- speed_bps = (data_size * 8.0) / send_time # bits per second
- speed_kbps = speed_bps / 1024.0 # Kbps
- speed_mbps = speed_kbps / 1024.0 # Mbps
- else:
- speed_bps = 0
- speed_kbps = 0
- speed_mbps = 0
-
- # 打印结果
- clog.info("")
- clog.info("=" * 60)
- clog.info("Test Results")
- clog.info("=" * 60)
- clog.info(" Success: {}".format("YES" if success else "NO"))
- clog.info(" Total bytes sent: {} ({:.2f} MB, {:.2f} KB)".format(
- data_size, data_size / 1024.0 / 1024.0, data_size / 1024.0))
- clog.info(" Send time: {:.3f} seconds ({:.0f} ms)".format(
- send_time, send_time_ms))
-
- if speed_mbps >= 1.0:
- clog.info(" Transfer speed: {:.2f} Mbps ({:.2f} Kbps, {:.0f} bps)".format(
- speed_mbps, speed_kbps, speed_bps))
- else:
- clog.info(" Transfer speed: {:.2f} Kbps ({:.0f} bps)".format(
- speed_kbps, speed_bps))
-
- clog.info("")
- clog.info("Data pattern info:")
- clog.info(" Pattern: Incremental bytes (0, 1, 2, ..., 255, 0, 1, ...)")
- clog.info(" Start value: 0")
- clog.info(" Cycle: Every 256 bytes")
- clog.info(" Expected byte at offset N: (N % 256)")
- clog.info("")
-
- # 统计信息（暂时无法获取，需要修改send_data_simple）
- stats = {
- 'total_bytes': data_size,
- 'send_time': send_time,
- 'send_time_ms': send_time_ms,
- 'speed_bps': speed_bps,
- 'speed_kbps': speed_kbps,
- 'speed_mbps': speed_mbps,
- 'success': success
- }
-
- if success:
- clog.info("✓ Test PASSED")
- else:
- clog.error("✗ Test FAILED")
-
- clog.info("=" * 60)
-
- return success, send_time, data_size, stats
-
- except Exception as e:
- clog.error("Error in SPI protocol test: {}".format(e))
- # 在嵌入式环境中，traceback可能不可用，只打印错误信息
- try:
- import traceback
- traceback.print_exc()
- except ImportError:
- # traceback模块不可用，只打印错误类型和消息
- clog.error("Exception type: {}, message: {}".format(type(e).__name__, str(e)))
- return False, 0.0, 0, {}
-
-
-def run_spi_protocol_test_with_stats(data_size=TEST_DATA_SIZE,
- spi_channel=SPI_CHANNEL,
- spi_mode=SPI_MODE,
- spi_clk=SPI_CLK):
- """
- 运行SPI新协议测试（带详细统计信息）
- 直接使用SpiProtocolSender以获取统计信息
-
- Args:
- data_size: 要发送的数据总大小（字节），默认2MB
- spi_channel: SPI通道，默认2（对应MCU端SPI1）
- spi_mode: SPI模式，默认0
- spi_clk: SPI时钟频率，默认8（1.625MHz）
-
- Returns:
- tuple: (success: bool, send_time: float, total_bytes: int, stats: dict)
- """
- try:
- clog.info("=" * 60)
- clog.info("SPI Protocol Test - 2MB Incremental Data (with stats)")
- clog.info("=" * 60)
- clog.info("Test parameters:")
- clog.info(" Total data size: {} bytes ({:.2f} MB)".format(
- data_size, data_size / 1024.0 / 1024.0))
- clog.info(" SPI channel: {}, mode: {}, clk: {}".format(
- spi_channel, spi_mode, spi_clk))
- clog.info(" Data pattern: Incremental bytes (0, 1, 2, ..., 255, 0, 1, ...)")
- clog.info("")
-
- # 创建SPI协议发送器
- sender = SpiProtocolSender(None, spi_channel, spi_mode, spi_clk)
-
- # 初始化SPI
- if not sender.__init_spi():
- clog.error("Failed to initialize SPI")
- return False, 0.0, 0, {}
-
- # 记录开始时间
- start_time = utime.time()
- start_time_ms = int(utime.time() * 1000) # 转换为毫秒
-
- clog.info("Starting data transmission...")
- clog.info(" Using new SPI protocol (reliable protocol)")
- clog.info("")
-
- # 创建数据生成器
- data_generator = IncrementalDataGenerator(data_size)
-
- # 发送数据（使用send_data方法，它会自动处理分包和标志位）
- # 创建一个生成器函数作为数据源（使用小块生成，减少内存占用）
- def data_source():
- """数据源生成器函数"""
- generator = IncrementalDataGenerator(data_size)
- chunk_size = 1024 # 1KB chunks（协议会自动分包，使用小块减少内存占用）
- while True:
- chunk = generator.generate_chunk(chunk_size)
- if chunk is None:
- break
- yield chunk
-
- # 使用send_data方法发送数据（会自动处理分包、标志位等）
- success = sender.send_data(data_source(), timeout=120)
-
- if not success:
- clog.error("Failed to send data")
- sender.close()
- return False, 0.0, 0, {}
-
- # 记录结束时间
- end_time = utime.time()
- end_time_ms = int(utime.time() * 1000) # 转换为毫秒
- send_time = end_time - start_time
- send_time_ms = end_time_ms - start_time_ms
-
- # 获取统计信息
- stats = get_sender_stats(sender)
- total_sent = stats.get('total_bytes', data_size) # 使用统计信息中的总字节数
- stats['send_time'] = send_time
- stats['send_time_ms'] = send_time_ms
-
- # 计算传输速度
- if send_time > 0:
- speed_bps = (total_sent * 8.0) / send_time
- speed_kbps = speed_bps / 1024.0
- speed_mbps = speed_kbps / 1024.0
- else:
- speed_bps = 0
- speed_kbps = 0
- speed_mbps = 0
-
- stats['speed_bps'] = speed_bps
- stats['speed_kbps'] = speed_kbps
- stats['speed_mbps'] = speed_mbps
-
- # 关闭SPI
- sender.close()
-
- # 打印结果
- clog.info("")
- clog.info("=" * 60)
- clog.info("Test Results")
- clog.info("=" * 60)
- clog.info(" Success: YES")
- clog.info(" Total bytes sent: {} ({:.2f} MB, {:.2f} KB)".format(
- total_sent, total_sent / 1024.0 / 1024.0, total_sent / 1024.0))
- clog.info(" Send time: {:.3f} seconds ({:.0f} ms)".format(
- send_time, send_time_ms))
-
- if speed_mbps >= 1.0:
- clog.info(" Transfer speed: {:.2f} Mbps ({:.2f} Kbps)".format(
- speed_mbps, speed_kbps))
- else:
- clog.info(" Transfer speed: {:.2f} Kbps".format(speed_kbps))
-
- clog.info("")
- clog.info("Protocol Statistics:")
- clog.info(" Total packets: {}".format(stats.get('total_packets', 0)))
- clog.info(" Retry packets: {}".format(stats.get('retry_packets', 0)))
- clog.info(" Error packets: {}".format(stats.get('error_packets', 0)))
- clog.info(" Total bytes: {} ({:.2f} MB)".format(
- stats.get('total_bytes', 0),
- stats.get('total_bytes', 0) / 1024.0 / 1024.0))
- clog.info("")
-
- clog.info("✓ Test PASSED")
- clog.info("=" * 60)
-
- return True, send_time, total_sent, stats
-
- except Exception as e:
- clog.error("Error in SPI protocol test: {}".format(e))
- # 在嵌入式环境中，traceback可能不可用，只打印错误信息
- try:
- import traceback
- traceback.print_exc()
- except ImportError:
- # traceback模块不可用，只打印错误类型和消息
- clog.error("Exception type: {}, message: {}".format(type(e).__name__, str(e)))
- return False, 0.0, 0, {}
-
-
-# 主函数
-if __name__ == '__main__':
- # 运行2MB递增数据测试（带详细统计）
- success, send_time, total_bytes, stats = run_spi_protocol_test_with_stats(
- data_size=2 * 1024 * 1024, # 2MB
- spi_channel=2, # SPI通道2（对应MCU端SPI1）
- spi_mode=0, # Mode 0
- spi_clk=8 # 1.625MHz（可根据需要调整）
- )
-
- if success:
- clog.info("")
- clog.info("Final Summary:")
- clog.info(" Total bytes: {} ({:.2f} MB)".format(
- total_bytes, total_bytes / 1024.0 / 1024.0))
- clog.info(" Send time: {:.3f} seconds".format(send_time))
- if stats.get('speed_mbps', 0) >= 1.0:
- clog.info(" Speed: {:.2f} Mbps".format(stats.get('speed_mbps', 0)))
- else:
- clog.info(" Speed: {:.2f} Kbps".format(stats.get('speed_kbps', 0)))
- else:
- clog.error("Test failed!")
-
projects / mirrors / FD230.git / commitdiff