#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
-SPI 写操作测试模块（保留原始时序版）
-核心：严格保留原有延时/时序 + 适配datasize参数 + 解决内存/赋值报错
+SPI 写操作测试模块
+连续写2MB递增数据，仅当收到外部中断时发送数据块，用于测试SPI传输完整性和数据丢失检测
"""
import log
+
+# 设置日志输出级别
+log.basicConfig(level=log.INFO)
+clog = log.getLogger("spiwritetest")
+
import utime
from machine import SPI
from machine import Pin
from machine import ExtInt
-# 设置日志输出级别
-log.basicConfig(level=log.INFO)
-clog = log.getLogger("spiwritetest")
+# SPI配置说明：
+# spi - 通道选择[0,1]，int类型
+# mode - SPI 的工作模式，int类型：
+# 0: CPOL=0, CPHA=0
+# 1: CPOL=0, CPHA=1
+# 2: CPOL=1, CPHA=0
+# 3: CPOL=1, CPHA=1
+# clk - 时钟频率：
+# 0: 100kHz
+# 1: 200kHz
+# 8: 1.625MHz
+# 9: 3.25MHz
+# 12: 26MHz
-# 测试参数（完全保留原始值）
-TEST_DATA_SIZE = 2 * 1024 * 1024 # 2MB总数据量
-PACKET_TOTAL_SIZE = 1024 # 每包总长度严格1024字节
-HEADER_SIZE = 1 # 1字节序号头
-DATA_BODY_SIZE = PACKET_TOTAL_SIZE - HEADER_SIZE # 1023字节数据体
-SPI_CHANNEL = 2 # SPI通道2
-SPI_MODE = 0 # Mode 0
-SPI_CLK = 8 # 1.625MHz
+# 测试参数
+TEST_DATA_SIZE = 18 * 1024 * 1024 # 2MB
+CHUNK_SIZE = 1024 # 每次发送1KB
+SPI_CHANNEL = 2 # SPI通道2（对应MCU端SPI1）
+SPI_MODE = 0 # Mode 0
+SPI_CLK = 12 # 1.625MHz（可根据需要调整：8=1.625MHz, 9=3.25MHz, 12=26MHz）
-# å\85¨å±\80ä¸æ\96æ \87å¿\97ä½\8dï¼\88ä¿\9dç\95\99å\8e\9få§\8b）
+# å\85¨å±\80ä¸æ\96æ \87å¿\97ä½\8dï¼\9aæ \87è®°æ\98¯å\90¦æ\94¶å\88°ä¸æ\96ï¼\88ä»\85ä¸æ\96å\9b\9eè°\83å\8f¯ç½®ä½\8dï¼\8c主é\80»è¾\91å¤\8dä½\8d）
interrupt_triggered = False
-INTERRUPT_TIMEOUT = 60 # 中断等待超时时间（秒）
+# 中断等待超时时间（单位：秒），避免无限等待
+INTERRUPT_TIMEOUT = 60
-# SPI接收缓存（保留原始）
-recv_buf = bytearray(1) # 通用接收缓存
-slave_feedback_buf = bytearray(1) # 从机反馈缓存（仅1字节：需要重发的序号）
-# ä¸æ\96å\9b\9eè°\83å\87½æ\95°ï¼\88å®\8cå\85¨ä¿\9dç\95\99å\8e\9få§\8bé\80»è¾\91ï¼\89
+# ä¸æ\96å\9b\9eè°\83å\87½æ\95°ï¼\9aæ\94¶å\88°ä¸æ\96æ\97¶ç½®ä½\8dæ \87å¿\97ä½\8d
def ExtIntfun(args):
global interrupt_triggered
+ # clog.info('### interrupt {} ###'.format(args)) # args[0]:gpio号 args[1]:上升沿或下降沿
if args[1] == 0:
interrupt_triggered = True
- clog.info("interrupt_triggered = True (need read slave data)")
+ clog.info("interrupt_triggered = True")
+ # pm.autosleep(0)
+ # pm.wakelock_lock(lpm_fd)
+ # elif args[1] == 1:
+ # clog.info("enable autosleep")
+ # pm.wakelock_unlock(lpm_fd)
+ # pm.autosleep(1)
+ # clog.info("recvDataPin={}".format(recvDataPin.read()))
-# 初始化外部中断（保留原始时序/配置）
+# 初始化外部中断（GPIO3，上升/下降沿触发，无上下拉）
extint = ExtInt(ExtInt.GPIO3, ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_DISABLE, ExtIntfun)
extint.enable()
clog.info("External interrupt initialized: GPIO3, RISING_FALLING edge")
-# å\94¤é\86\92MCUå¼\95è\84\9aï¼\88å®\8cå\85¨ä¿\9dç\95\99å\8e\9få§\8bé\85\8dç½®/æ\97¶åº\8f）
+# å\88\9b建gpio对象ï¼\88å\94¤é\86\92MCUç\9a\84å¼\95è\84\9aï¼\8cGPIO4ï¼\8cè¾\93å\87ºæ¨¡å¼\8fï¼\8cå\88\9då§\8bé«\98ç\94µå¹³）
gpioWakeBT = Pin(Pin.GPIO4, Pin.OUT, Pin.PULL_DISABLE, 1)
+# 设置输入输出模式
gpioWakeBT.set_dir(Pin.OUT)
+
+# 设置引脚电平
gpioWakeBT.write(1)
class SpiWriteTest:
- """SPI写操作测试类（严格保留原始时序）"""
+ """
+ SPI写操作测试类
+ 生成递增数据并通过SPI发送，用于测试数据完整性
+ """
- def __init__(self, total_data_size=TEST_DATA_SIZE, packet_size=PACKET_TOTAL_SIZE,
+ def __init__(self, data_size=TEST_DATA_SIZE, chunk_size=CHUNK_SIZE,
spi_channel=SPI_CHANNEL, spi_mode=SPI_MODE, spi_clk=SPI_CLK):
- self.total_data_size = total_data_size
- self.packet_size = packet_size
- self.header_size = HEADER_SIZE
- self.data_body_size = DATA_BODY_SIZE
+ """
+ 初始化SPI写测试
+
+ Args:
+ data_size: 要发送的数据总大小（字节），默认2MB
+ chunk_size: 每次发送的数据块大小（字节），默认1KB
+ spi_channel: SPI通道，默认2（对应MCU端SPI1）
+ spi_mode: SPI模式，默认0
+ spi_clk: SPI时钟频率，默认8（1.625MHz）
+ """
+ self.data_size = TEST_DATA_SIZE
+ self.chunk_size = chunk_size
self.spi_channel = spi_channel
self.spi_mode = spi_mode
- self.spi_clk = spi_clk
+ self.spi_clk = SPI_CLK
self.spi_obj = None
-
- # 包缓存区：key=序号, value=完整包数据（1024字节）
- self.packet_cache = {}
- self.retrans_count = 0 # 总重传次数
-
- # 发送控制（保留原始）
- self.current_seq = 0
- self.current_data_start = 0
- self.packet_count = 0
- self.total_sent_bytes = 0
-
+ self.current_value = 0 # 当前数据值（从0开始递增）
+
def __init_spi(self):
- """初始化SPI对象（保留原始逻辑）"""
+ """初始化SPI对象"""
try:
self.spi_obj = SPI(self.spi_channel, self.spi_mode, self.spi_clk)
- if hasattr(self.spi_obj, 'set_full_duplex'):
- self.spi_obj.set_full_duplex(True)
clog.info("SPI initialized: channel={}, mode={}, clk={}".format(
self.spi_channel, self.spi_mode, self.spi_clk))
return True
clog.error("Failed to initialize SPI: {}".format(e))
return False
- def __generate_data_body(self, start_value):
- """生成1023字节数据体（保留原始逻辑，解决赋值报错）"""
- # 改为直接构建bytes，避免bytearray赋值
- data_body_list = []
+ def __generate_chunk(self, start_value, chunk_size):
+ """
+ 生成递增数据块
+
+ Args:
+ start_value: 起始值（字节值，0-255）
+ chunk_size: 数据块大小（字节）
+
+ Returns:
+ bytearray: 递增数据块
+ """
+ chunk = bytearray(chunk_size)
value = start_value
- for i in range(self.data_body_size):
- data_body_list.append(value & 0xFF)
- value = (value + 1) % 256
- return bytes(data_body_list)
-
- def __build_packet(self, seq, start_value):
- """构建1024字节完整包并缓存（保留原始逻辑）"""
- # 1字节序号头
- header = bytes([seq & 0xFF])
- # 1023字节数据体
- data_body = self.__generate_data_body(start_value)
- # 拼接成1024字节总包
- packet = header + data_body
- # 缓存包（保留原始缓存逻辑）
- self.packet_cache[seq] = packet
- return packet
+ for i in range(chunk_size):
+ chunk[i] = value & 0xFF # 只取低8位
+ value = (value + 1) % 256 # 递增，超过255后回到0
+ return chunk
- def __send_packet(self, packet, seq):
- """发送指定包（仅修复datasize参数，严格保留原始时序）"""
- try:
- # 严格保留原始时序：write(0)→2ms→write(1)→10ms
- gpioWakeBT.write(0)
- utime.sleep_ms(2)
- gpioWakeBT.write(1)
- utime.sleep_ms(10)
+ def __send_chunk(self, chunk):
+ """
+ 通过SPI发送数据块
+
+ Args:
+ chunk: 要发送的数据块（bytearray或bytes）
+ Returns:
+ bool: 发送是否成功
+ """
+ gpioWakeBT.write(0)
+ # clog.info("gpioWakeBT.write(0)")
+ utime.sleep_ms(2)
+ gpioWakeBT.write(1)
+ utime.sleep_ms(2)
+ try:
if self.spi_obj is None:
+ clog.error("SPI object is None, cannot send chunk")
+ return False
+
+ chunk_len = len(chunk)
+ if hasattr(self.spi_obj, 'write'):
+
+ # write(data, datasize) - datasize 参数必需
+ clog.debug("Calling SPI.write() with {} bytes".format(chunk_len))
+ self.spi_obj.write(chunk, chunk_len)
+ clog.debug("SPI.write() completed successfully")
+ # elif hasattr(self.spi_obj, 'write_read'):
+ # # write_read(read_buf, write_data, datasize, timeout)
+ # # 或者 write_read(read_buf, write_data, datasize)
+ # read_buf = bytearray(chunk_len)
+ # try:
+ # # 尝试带timeout的调用
+ # self.spi_obj.write_read(read_buf, chunk, chunk_len, 5)
+ # except TypeError:
+ # # 尝试不带timeout的调用
+ # self.spi_obj.write_read(read_buf, chunk, chunk_len)
+ else:
+ clog.error("SPI object does not have write or write_read method")
return False
-
- # 核心修复：显式传入datasize参数，保留原始发送逻辑
- self.spi_obj.write(packet, len(packet))
- clog.info("reply seq {} (1024 bytes total)".format(seq))
return True
except Exception as e:
- clog.error("send seq {} fail: {}".format(seq, e))
+ clog.error("Error sending chunk via SPI: {}".format(e))
+ import sys
+ clog.error("Exception type: {}, details: {}".format(type(e).__name__, str(e)))
return False
- def __read_slave_feedback(self):
- """中断触发时读取从机反馈（保留原始时序/逻辑）"""
- global slave_feedback_buf, interrupt_triggered
- try:
- if self.spi_obj is None:
- return
-
- # 保留原始读取逻辑，补充datasize参数
- self.spi_obj.read(slave_feedback_buf, len(slave_feedback_buf))
- retrans_seq = slave_feedback_buf[0]
-
- # 保留原始重发逻辑
- if retrans_seq in self.packet_cache:
- self.retrans_count += 1
- clog.warning("Slave send NACK, need retransmit seq {}".format(retrans_seq))
- # 严格保留原始重发时序/逻辑
- self.__send_packet(self.packet_cache[retrans_seq], retrans_seq)
- else:
- clog.warning("seq {} not in cache, skip retrans".format(retrans_seq))
-
- interrupt_triggered = False
- except Exception as e:
- clog.error("Read slave feedback failed: {}".format(e))
- interrupt_triggered = False
-
def close(self):
- """关闭SPI（保留原始逻辑）"""
+ """关闭SPI对象，释放资源"""
if self.spi_obj is not None:
try:
if hasattr(self.spi_obj, 'close'):
self.spi_obj.close()
+ clog.info("SPI object closed via close() method")
elif hasattr(self.spi_obj, 'deinit'):
self.spi_obj.deinit()
+ clog.info("SPI object closed via deinit() method")
+ else:
+ clog.warning("SPI object does not have close() or deinit() method")
except Exception as e:
- clog.error("Close SPI failed: {}".format(e))
+ clog.error("Error closing SPI object: {}".format(e))
finally:
self.spi_obj = None
-
+
+ def __wait_for_interrupt(self, timeout_seconds=INTERRUPT_TIMEOUT):
+ """
+ 等待中断信号
+
+ Args:
+ timeout_seconds: 超时时间（秒），默认使用全局 INTERRUPT_TIMEOUT
+
+ Returns:
+ bool: 是否收到中断（True表示收到，False表示超时）
+ """
+ global interrupt_triggered
+
+ # 先检查是否已经有中断信号（可能在调用此函数之前就已经触发了）
+ if interrupt_triggered:
+ clog.info("Interrupt signal already present, consuming it immediately")
+ interrupt_triggered = False # 消费掉这个中断信号
+ return True
+
+ # 重置中断标志，准备等待新的中断
+ interrupt_triggered = False
+
+ # 等待中断，带超时机制
+ start_wait_time = utime.time()
+ check_count = 0
+ while not interrupt_triggered:
+ elapsed = utime.time() - start_wait_time
+ if elapsed >= timeout_seconds:
+ clog.warning("Interrupt wait timeout after {:.2f} seconds (checked {} times)".format(
+ elapsed, check_count))
+ return False
+ try:
+ utime.sleep_ms(5) # 每10ms检查一次，减少CPU占用
+ except Exception as e:
+ clog.error("Error in utime.sleep_ms: {}".format(e))
+ return False
+ check_count += 1
+ # 每1秒打印一次等待状态
+ if check_count % 100 == 0:
+ clog.info("Still waiting for interrupt... (elapsed: {:.2f}s)".format(elapsed))
+
+ clog.info("Interrupt received after {:.2f} seconds (checked {} times)".format(
+ utime.time() - start_wait_time, check_count))
+ # 重置标志，为下次等待做准备
+ interrupt_triggered = False
+ return True
+
def run_test(self):
- """运行测试（严格保留原始时序/逻辑，仅修复报错）"""
+ """
+ 运行SPI写测试
+ 首次发送不等待中断，后续每次发送前等待中断信号
+
+ Returns:
+ tuple: (success: bool, send_time: float, total_bytes: int, data_pattern_info: dict)
+ success: 测试是否成功
+ send_time: 发送耗时（秒）
+ total_bytes: 发送的总字节数
+ data_pattern_info: 数据模式信息（用于MCU端验证）
+ """
global interrupt_triggered
+
try:
+ # 初始化SPI
if not self.__init_spi():
return False, 0.0, 0, {}
- # 保留原始日志
- clog.info("=== SPI Write Test Start (1024B/packet: 1B seq + 1023B data) ===")
- clog.info("Total data size: {} bytes ({:.2f} MB)".format(
- self.total_data_size, self.total_data_size / 1024 / 1024))
- clog.info("Packet size: {} bytes (1B seq header + 1023B data body)".format(self.packet_size))
+ clog.info("Starting SPI write test")
+ clog.info("Test parameters:")
+ clog.info(" Total data size: {} bytes ({:.2f} MB)".format(
+ self.data_size, self.data_size / 1024.0 / 1024.0))
+ clog.info(" Chunk size: {} bytes".format(self.chunk_size))
+ clog.info(" SPI channel: {}, mode: {}, clk: {}".format(
+ self.spi_channel, self.spi_mode, self.spi_clk))
+ clog.info(" Delay between chunks: 5ms (to reduce SPI pressure)")
+ clog.info(" Data pattern: Incremental bytes (0, 1, 2, ..., 255, 0, 1, ...)")
+ # 记录开始时间
start_time = utime.time()
- # 保留原始发送逻辑
- while self.total_sent_bytes < self.total_data_size:
- clog.info("Sending packet {} (seq {})".format(self.packet_count, self.current_seq))
+ # 发送数据
+ total_sent = 0
+ chunk_count = 0
+ current_byte_value = 0 # 当前字节值（0-255循环）
+
+ # 计算需要发送的块数
+ total_chunks = (self.data_size + self.chunk_size - 1) // self.chunk_size
+
+ clog.info("Starting to send {} chunks...".format(total_chunks))
+
+ while total_sent < self.data_size:
+ # 计算剩余数据量
+ remaining_bytes = self.data_size - total_sent
+ remaining_chunks = (remaining_bytes + self.chunk_size - 1) // self.chunk_size
- # 构建当前包（保留原始逻辑）
- packet = self.__build_packet(self.current_seq, self.current_data_start)
- # 缓存上一包（保留原始逻辑）
- self.packet_cache[self.current_seq] = packet
+ # 首次发送不等待中断，后续每次发送前等待中断
+ # if chunk_count == 0:
+ # # 首次发送，不等待中断
+ # clog.info("Sending first chunk (chunk_count=0) without waiting for interrupt")
+ # else:
+ # # 后续发送，等待中断信号
+ # clog.info("Waiting for interrupt before sending chunk {} (remaining: {} bytes, {} chunks)...".format(
+ # chunk_count, remaining_bytes, remaining_chunks))
+ # if not self.__wait_for_interrupt():
+ # clog.error("Interrupt wait timeout at chunk {} (sent: {} bytes, remaining: {} bytes)".format(
+ # chunk_count, total_sent, remaining_bytes))
+ # self.close()
+ # return False, 0.0, total_sent, {}
+ # clog.info("Interrupt received, sending chunk {}".format(chunk_count))
- # 发送包（保留原始时序/逻辑）
- send_ok = self.__send_packet(packet, self.current_seq)
+ # 计算本次发送的块大小（最后一块可能小于chunk_size）
+ remaining = self.data_size - total_sent
+ current_chunk_size = min(self.chunk_size, remaining)
- # 保留原始中断处理逻辑
- if interrupt_triggered:
- clog.info("Interrupt triggered, read slave feedback...")
- self.__read_slave_feedback() # 读取从机，为主机提供CLK
- interrupt_triggered = False # 重置中断标志
+ # 生成递增数据块
+ chunk = self.__generate_chunk(current_byte_value, current_chunk_size)
- if not send_ok:
- clog.error("Send packet {} (seq {}) failed".format(self.packet_count, self.current_seq))
- self.close()
- return False, 0.0, self.total_sent_bytes, {}
+ # 更新当前字节值（为下一块准备）
+ current_byte_value = (current_byte_value + current_chunk_size) % 256
- # 保留原始统计更新逻辑
- self.total_sent_bytes += self.packet_size # 按1024字节/包统计
- self.current_data_start = (self.current_data_start + self.data_body_size) % 256 # 数据体起始值递增
- self.current_seq = (self.current_seq + 1) % 256 # 序号循环
- self.packet_count += 1
+ # 通过SPI发送数据块
+ if not self.__send_chunk(chunk):
+ clog.error("Failed to send chunk {} at offset {}".format(
+ chunk_count, total_sent))
+ self.close()
+ return False, 0.0, total_sent, {}
- # 保留原始进度打印逻辑
- if self.packet_count % 64 == 0:
- progress = (self.total_sent_bytes / self.total_data_size) * 100
- clog.info("Progress: {:.2f}% | Packets sent: {} | Total bytes: {} | Retrans: {} times".format(
- progress, self.packet_count, self.total_sent_bytes, self.retrans_count))
+ total_sent += current_chunk_size
+ chunk_count += 1
- # 保留原始延时
+ # 每次发送后延迟5ms，减少SPI压力，避免FIFO溢出
utime.sleep_ms(5)
+
+ # 每发送64KB打印一次进度
+ if total_sent % (64 * 1024) < self.chunk_size:
+ progress = (total_sent / self.data_size) * 100
+ clog.info("SPI TX Progress: {:.2f}% ({}/{}) bytes, chunks: {}".format(
+ progress, total_sent, self.data_size, chunk_count))
- # 保留原始测试完成逻辑
+ # 记录结束时间并计算耗时
end_time = utime.time()
send_time = end_time - start_time
- speed_kbps = (self.total_sent_bytes * 8) / send_time / 1024 if send_time > 0 else 0
- clog.info("=== Test Completed ===")
- clog.info("Total packets sent: {} ({} bytes total)".format(
- self.packet_count, self.total_sent_bytes))
- clog.info("Send time: {:.3f}s | Speed: {:.2f} Kbps".format(send_time, speed_kbps))
- clog.info("Total retrans times: {}".format(self.retrans_count))
+ # 计算传输速度
+ speed_kbps = (total_sent * 8.0) / send_time / 1024.0 # Kbps
+ speed_mbps = speed_kbps / 1024.0 # Mbps
- self.close()
- return True, send_time, self.total_sent_bytes, {
- "packet_size": self.packet_size,
- "header_size": self.header_size,
- "data_body_size": self.data_body_size,
- "retrans_count": self.retrans_count
+ # 数据模式信息（用于MCU端验证）
+ data_pattern_info = {
+ 'total_bytes': total_sent,
+ 'chunk_size': self.chunk_size,
+ 'pattern': 'incremental', # 递增模式
+ 'start_value': 0, # 起始值为0
+ 'cycle': 256, # 每256字节循环一次（0-255）
+ 'description': 'Data values increment from 0 to 255, then cycle back to 0'
}
-
+
+ if speed_mbps >= 1.0:
+ clog.info("SPI write test completed: {} bytes in {:.3f}s, Speed: {:.2f} Mbps ({:.2f} Kbps)".format(
+ total_sent, send_time, speed_mbps, speed_kbps))
+ else:
+ clog.info("SPI write test completed: {} bytes in {:.3f}s, Speed: {:.2f} Kbps".format(
+ total_sent, send_time, speed_kbps))
+
+ 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(" Total chunks: {}".format(chunk_count))
+ clog.info(" Expected byte at offset N: (N % 256)")
+
+ # 关闭SPI
+ self.close()
+
+ return True, send_time, total_sent, data_pattern_info
+
except Exception as e:
- clog.error("Test failed: {}".format(e))
+ clog.error("Error in SPI write test: {}".format(e))
self.close()
return False, 0.0, 0, {}
+
+ def get_expected_byte_at_offset(self, offset):
+ """
+ 获取指定偏移位置的期望字节值（用于MCU端验证）
+
+ Args:
+ offset: 字节偏移位置（0开始）
+
+ Returns:
+ int: 期望的字节值（0-255）
+ """
+ return offset % 256
-# 便捷运行函数（保留原始）
-def run_spi_write_test():
- test = SpiWriteTest()
- success, send_time, total_bytes, info = test.run_test()
+
+def run_spi_write_test(data_size=TEST_DATA_SIZE, chunk_size=CHUNK_SIZE,
+ spi_channel=SPI_CHANNEL, spi_mode=SPI_MODE, spi_clk=SPI_CLK):
+ """
+ 运行SPI写测试的便捷函数
+
+ Args:
+ data_size: 要发送的数据总大小（字节），默认2MB
+ chunk_size: 每次发送的数据块大小（字节），默认1KB
+ 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)
+ """
+ test = SpiWriteTest(data_size, chunk_size, spi_channel, spi_mode, spi_clk)
+ try:
+ success, send_time, total_bytes, pattern_info = test.run_test()
+ if success:
+ clog.info("SPI write test PASSED")
+ else:
+ clog.error("SPI write test FAILED")
+ return success, send_time, total_bytes
+ finally:
+ test.close()
+
+
+# 自动运行测试（如果直接执行此脚本）
+if __name__ == '__main__':
+ # 运行2MB递增数据测试
+ success, send_time, total_bytes = run_spi_write_test(
+ data_size=2 * 1024 * 1024, # 2MB
+ chunk_size=1024, # 1KB chunks
+ spi_channel=2, # SPI通道2（对应MCU端SPI1）
+ spi_mode=0, # Mode 0
+ spi_clk=8 # 1.625MHz（可根据需要调整）
+ )
+
if success:
- clog.info("Test PASSED | Total bytes sent: {}".format(total_bytes))
+ clog.info("Test completed successfully!")
+ clog.info("Total bytes sent: {} ({:.2f} MB)".format(
+ total_bytes, total_bytes / 1024.0 / 1024.0))
+ clog.info("Send time: {:.3f} seconds".format(send_time))
else:
- clog.error("Test FAILED")
- return success
+ clog.error("Test failed!")
+
+# 如果作为模块导入，不自动运行
+# 可以通过以下方式使用：
+# from spi_write_test import run_spi_write_test
+# success, send_time, total_bytes = run_spi_write_test()
-if __name__ == '__main__':
- run_spi_write_test()
\ No newline at end of file
#include "spi.h"
#include "custom_spi.h"
#include "custom_utils.h"
+#include "pm.h"
-osSemaphoreId_t g_semaphore_spi_id = NULL;
-static int total_cnt = 0;
+
+static osSemaphoreId_t g_semaphore_spi_id = NULL;
+static app_task_definition_t taskcfg={0};
+static uint32_t total_cnt = 0;
static uint8_t g_rx_buffer[1024] = {0};
-static uint8_t last_expect_seq = 1; // 初始期望递增序号（根据主设备实际起始值调整）
-static const uint8_t retrans_cmd = 0x01;
-#define PACKET_HEADER_OFFSET 0
-#define PACKET_TOTAL_SIZE 1024
-static uint8_t first_flag = 0;
-static int discontinuous_cnt = 0;
static uint32_t pkg_recv_cnt = 0; // 新增：接收包计数
-static int current_seq = 0;
-static int expected_seq = 0;
+static uint32_t total_byte_error_cnt = 0;
+
void custom_spi_read(void)
{
if(ret == 0 && spi_data.rx_bytes == 1024)
{
- uint8_t current_seq = spi_data.rx_buff[0];
- printf("spi_data.rx_bytes : %d, spi_data.rx_data[0]:0x%02X,spi_data.rx_data[1023]:0x%02X\n", spi_data.rx_bytes, current_seq,spi_data.rx_buff[1023]);
+ total_cnt += spi_data.rx_bytes;
+ printf("total_cnt:%d\n",total_cnt);
+ // printf("spi_data.rx_buff[0]:0x%x,total_cnt:%d\n",spi_data.rx_buff[0],total_cnt);
+ // printf("spi_data.rx_bytes : %d, spi_data.rx_data[0]:0x%02X,spi_data.rx_data[1023]:0x%02X\n",
+ // spi_data.rx_bytes, current_seq, spi_data.rx_buff[1023]);
- // 新增：包计数自增
+ // 包计数自增（从第一个包开始计数）
pkg_recv_cnt++;
-
- // 核心修改：仅在接收包数超过10个后，才执行连续性判定
- if (pkg_recv_cnt > 10) {
- if (last_expect_seq != 0xFF) { // 非首次接收才判定
- uint8_t expected_seq = (last_expect_seq == 0x00) ? 0xFF : (last_expect_seq - 1);
- if (current_seq != expected_seq) {
- discontinuous_cnt++;
- printf("不连续（跳变）：当前0x%02X, 期望0x%02X | 累计不连续数：%d\n",
- current_seq, expected_seq, discontinuous_cnt);
- uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
- uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- uapi_tcxo_delay_ms(2);
- uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
- uapi_tcxo_delay_ms(5);
- uint8_t data = expected_seq;
- spi_data.tx_buff = &data;
- spi_data.tx_bytes = 1;
- ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- printf("uapi_spi_slave_write ret : 0x%02X \n", ret);
- }
+
+ // --------------- 逐字节严格递增判断（无跳过，首个包就校验）---------------
+ uint8_t byte_error_cnt = 0; // 单包内字节错误计数
+ uint8_t prev_byte = spi_data.rx_buff[0]; // 初始化为第一个字节
+ // printf("head\n");
+ // 遍历1024字节（从第2个字节开始校验）
+ for (int i = 1; i < 1024; i++)
+ {
+ // if(i < 20)
+ // printf("(0x%02X)",spi_data.rx_buff[i]);
+ // if(i == 20)
+ // printf("\n");
+ uint8_t curr_byte = spi_data.rx_buff[i];
+ // 8位字节递增规则：0xFF后应为0x00，其余为前一字节+1
+ uint8_t expected_byte = (prev_byte == 0xFF) ? 0x00 : (prev_byte + 1);
+
+ // 校验当前字节是否等于期望的递增字节（无跳过）
+ if (curr_byte != expected_byte)
+ {
+ byte_error_cnt++;
+ // 打印错误详情（调试用，正式环境可注释）
+ printf("字节[%d]不连续：当前0x%02X, 期望0x%02X（前一字节0x%02X）\n",
+ i, curr_byte, expected_byte, prev_byte);
}
+
+ // 更新前一字节，用于下一轮校验
+ prev_byte = curr_byte;
+ }
+ // printf("该包无错误\n");
+ // 单包字节校验结果汇总（首个包就统计）
+ if (byte_error_cnt > 0)
+ {
+ printf("【包%d】字节递增错误数：%d | 累计字节错误数：%d\n",
+ pkg_recv_cnt, byte_error_cnt, total_byte_error_cnt += byte_error_cnt);
}
-
- // 更新上一次接收的序号（用于下次判定）
- last_expect_seq = current_seq;
-
- total_cnt += spi_data.rx_bytes;
- printf("total_cnt : %d, 当前接收包数：%d\n", total_cnt, pkg_recv_cnt);
- }
- else
- {
- // 新增：包计数自增（失败/长度不符的包也计数，若仅统计成功包可移到上面if内）
- pkg_recv_cnt++;
-
- uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
- uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- uapi_tcxo_delay_ms(2);
- uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
- uapi_tcxo_delay_ms(5);
- uint8_t data = expected_seq;
- spi_data.tx_buff = &data;
- spi_data.tx_bytes = 1;
- ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- printf("uapi_spi_slave_write ret : 0x%02X \n", ret);
- // printf("接收失败/长度不符，当前接收包数：%d\n", pkg_recv_cnt);
}
- // spi_xfer_data_t spi_data = {0};
- // errcode_t ret = ERRCODE_FAIL;
-
- // memset(g_rx_buffer, 0, sizeof(g_rx_buffer));
- // spi_data.rx_bytes = sizeof(g_rx_buffer);
- // spi_data.rx_buff = g_rx_buffer;
-
- // ret = uapi_spi_slave_read(SPI_BUS_1, &spi_data, 50);
-
- // if(ret == 0 && spi_data.rx_bytes == 1024)
- // {
- // printf("spi_data.rx_bytes : %d,spi_data.rx_data[0]:0x%x\n",spi_data.rx_bytes,spi_data.rx_buff[0]);
- // if(spi_data.rx_bytes == sizeof(g_rx_buffer))
- // total_cnt += spi_data.rx_bytes;
- // printf("total_cnt : %d\n",total_cnt);
-
-
- // // uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
-
- // // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- // // uapi_tcxo_delay_ms(10); /* delay 10 ms */
- // // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
- // // if(spi_fp == NULL)
- // // {
- // // spi_fp = fopen("/user/data/spi_text.txt","w+");
- // // }
-
- // // if(spi_fp)
- // // {
- // // int ret = fwrite(spi_data.rx_buff,1,spi_data.rx_bytes,spi_fp);
- // // printf("fwrite ret:%d\n",ret);
- // // }
- // }
// else
// {
+ // // 新增：包计数自增（失败/长度不符的包也计数，若仅统计成功包可移到上面if内）
+ // pkg_recv_cnt++;
+
// uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
-
// uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- // uapi_tcxo_delay_ms(2); /* delay 10 ms */
+ // uapi_tcxo_delay_ms(2);
// uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
-
- // uint8_t data = last_expect_seq;
- // last_expect_seq++;
+ // uapi_tcxo_delay_ms(5);
+ // uint8_t data = expected_seq;
// spi_data.tx_buff = &data;
// spi_data.tx_bytes = 1;
// ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // printf("uapi_spi_slave_write ret : %x \n",ret);
+ // printf("uapi_spi_slave_write ret : 0x%02X \n", ret);
+ // // printf("接收失败/长度不符，当前接收包数：%d\n", pkg_recv_cnt);
// }
- // if(ret == 0 && spi_data.rx_bytes == 1024)
- // {
- // if(first_flag == 0)
- // {
- // first_flag = 1;
- // last_expect_seq = spi_data.rx_buff[0];
- // }
- // else if(last_expect_seq != spi_data.rx_buff[0] - 1)
- // {
- // printf("data error,last_expect_seq:0x%x,spi_data.rx_data[0]:0x%x\n",last_expect_seq,spi_data.rx_buff[0]);
- // uint8_t data = last_expect_seq;
- // last_expect_seq++;
- // spi_data.tx_buff = &data;
- // spi_data.tx_bytes = 1;
- // ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // printf("uapi_spi_slave_write ret : %x \n",ret);
- // }
- // else if(last_expect_seq == 0 && spi_data.rx_buff[0] == 0xff)
- // {
- // last_expect_seq = spi_data.rx_buff[0];
- // printf("spi_data.rx_bytes : %d,spi_data.rx_data[0]:0x%x\n",spi_data.rx_bytes,spi_data.rx_buff[0]);
- // if(spi_data.rx_bytes == sizeof(g_rx_buffer))
- // total_cnt += spi_data.rx_bytes;
- // printf("total_cnt : %d\n",total_cnt);
- // }
- // else
- // {
- // last_expect_seq = spi_data.rx_buff[0];
- // printf("spi_data.rx_bytes : %d,spi_data.rx_data[0]:0x%x\n",spi_data.rx_bytes,spi_data.rx_buff[0]);
- // if(spi_data.rx_bytes == sizeof(g_rx_buffer))
- // total_cnt += spi_data.rx_bytes;
- // printf("total_cnt : %d\n",total_cnt);
- // }
-
-
-
- // uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
-
- // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- // uapi_tcxo_delay_ms(10); /* delay 10 ms */
- // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
- // if(spi_fp == NULL)
- // {
- // spi_fp = fopen("/user/data/spi_text.txt","w+");
- // }
-
- // if(spi_fp)
- // {
- // int ret = fwrite(spi_data.rx_buff,1,spi_data.rx_bytes,spi_fp);
- // printf("fwrite ret:%d\n",ret);
- // }
- // }
- // else
- // {
- // uint8_t data = last_expect_seq;
- // last_expect_seq++;
- // spi_data.tx_buff = &data;
- // spi_data.tx_bytes = 1;
- // ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // printf("uapi_spi_slave_write ret : %x \n",ret);
- // }
- // printf("\n=== SPI Read ===\n");
- // printf("uapi_spi_slave_read ret : 0x%x, rx bytes: %d\n", ret, spi_data.rx_bytes);
-
- // if (ret == 0 && spi_data.rx_bytes == PACKET_TOTAL_SIZE)
- // {
- // uint8_t current_seq = g_rx_buffer[PACKET_HEADER_OFFSET];
- // printf("Header seq: 0x%02X, Expect seq: 0x%02X\n", current_seq, last_expect_seq);
-
- // // 核心：判定序号是否为“递增连续”（仅允许递增，禁止递减）
- // uint8_t is_continuous = 0;
- // // 情况1：正常递增（如 0x7→0x8、0xFF→0x0）
- // if (current_seq == last_expect_seq)
- // {
- // is_continuous = 1;
- // }
- // // 情况2：255→0 溢出递增（合法）
- // else if (last_expect_seq == 0xFF && current_seq == 0x00)
- // {
- // is_continuous = 1;
- // printf("Seq overflow (255→0) → continuous\n");
- // }
-
- // if (is_continuous)
- // {
- // // 序号合法：累加计数 + 更新下一个期望序号（递增）
- // total_cnt += spi_data.rx_bytes;
- // last_expect_seq++; // uint8_t自动溢出（255→0）
- // printf("Seq OK! Total cnt: %d, Next expect: 0x%02X\n", total_cnt, last_expect_seq);
- // }
- // else
- // {
- // // 序号递减/跳变：触发重传，不更新计数
- // printf("Seq ERROR! Current:0x%02X, Expect:0x%02X → Send retrans cmd (0x01)\n",
- // current_seq, last_expect_seq);
-
- // spi_data.tx_buff = (uint8_t*)&retrans_cmd;
- // spi_data.tx_bytes = 1;
- // errcode_t write_ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // printf("Retrans cmd send ret: 0x%x\n", write_ret);
- // return; // 不更新计数，等待主设备重发正确序号
- // }
- // }
- // else if (ret == 0)
- // {
- // printf("Rx packet incomplete! Expect:1024, Actual:%d\n", spi_data.rx_bytes);
- // }
- // else
- // {
- // // 读取失败：回传当前期望序号，便于主设备定位
- // printf("SPI Read FAIL (0x%x) → Write back expect seq:0x%02X\n", ret, last_expect_seq);
- // spi_data.tx_buff = &last_expect_seq;
- // spi_data.tx_bytes = 1;
- // ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // if (ret != 0) ret = uapi_spi_slave_write(SPI_BUS_1, &spi_data, 50);
- // printf("Write back ret:0x%x, data:0x%02X\n", ret, last_expect_seq);
- // last_expect_seq++;
- // }
-
-
- // uapi_gpio_set_dir(S_AGPIO_L13, GPIO_DIRECTION_OUTPUT);
-
- // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_HIGH);
- // uapi_tcxo_delay_ms(2); /* delay 10 ms */
- // uapi_gpio_set_val(S_AGPIO_L13, GPIO_LEVEL_LOW);
-
- // if(ret == ERRCODE_SUCC)
- // {
- // printf("\nhead\n");
- // for(int i = 0; i < spi_data.rx_bytes ;i ++ )
- // {
- // printf("byte[%d]:(%x),",i,spi_data.rx_buff[i]);
- // }
- // printf("\nend\n");
- // }
}
void custom_spi_service_task(void *arg)
{
- printf("SPI处理任务启动，开始等待中断...\r\n");
+ printf("SPI处理任务启动,开始等待中断...\r\n");
// 任务循环：永久运行，处理每次中断
while (1) {
} else if (ret == osErrorTimeout) {
// 超时处理：打印日志，继续循环等待
- printf("SPI中断等待超时（%d秒），继续等待...\r\n", 60);
+ printf("SPI中断等待超时%d秒,继续等待...\r\n", 60);
} else {
// 信号量错误：退出循环，任务结束（可选重试逻辑）
- printf("信号量获取失败（状态：%d），任务退出\r\n", ret);
+ printf("信号量获取失败 状态：%d,任务退出\r\n", ret);
// break;
}
}
-
- // 任务退出前清理资源
- if (g_semaphore_spi_id != NULL) {
- osSemaphoreDelete(g_semaphore_spi_id);
- g_semaphore_spi_id = NULL;
- }
- printf("SPI处理任务退出\r\n");
- osThreadExit();
}
void custom_spi_init(void)
attr.frame_format = SPI_CFG_FRAME_FORMAT_MOTOROLA_SPI;
attr.spi_frame_format = (uint32_t)HAL_SPI_FRAME_FORMAT_STANDARD; // 标准SPI格式（非QUAD）
attr.frame_size = (uint32_t)HAL_SPI_FRAME_SIZE_8; // 8位数据（与Python端匹配）
- attr.tmod = HAL_SPI_TRANS_MODE_TXRX; // 只接收模式
+ attr.tmod = HAL_SPI_TRANS_MODE_RX; // 只接收模式
attr.sste = SPI_CFG_SSTE_DISABLE;
attr.ndf = 1;
ret = uapi_gpio_register_isr_func(S_MGPIO23, GPIO_INTERRUPT_FALLING_EDGE, custom_spi_rx_callback );
if (ret != ERRCODE_SUCC) {
printf("uapi_gpio_register_isr_func failed: 0x%x\n", ret);
- return;
+ goto FAILED;
}
g_semaphore_spi_id = osSemaphoreNew(1, 0, NULL);
if(g_semaphore_spi_id == NULL)
{
printf("osSemaphoreNew failed\n");
- return;
+ goto FAILED;
}
- app_task_definition_t taskcfg={0};
// taskcfg.create_queue_handle = create_ppg_queue;
taskcfg.attr.name="custom_spi_service";
- taskcfg.attr.priority = osPriorityNormal4;
+ taskcfg.attr.priority = osPriorityHigh;
taskcfg.attr.stack_size = 0x2000;
taskcfg.func = custom_spi_service_task;
if(ret != ERRCODE_SUCC)
{
printf("create custom_spi_service_task failed %d!\r\n",ret);
- return;
+ goto FAILED;
}
+ uint8_t cnt = pm_veto_get_id_cnt(PM_ID_SPI);
+ if (cnt == 0)
+ {
+ uapi_pm_add_sleep_veto(PM_ID_SPI);
+ }
+
+ ret = uapi_pm_request_mode(PM_ID_SPI, PM_MODE_PERFORMACE);
+ printf("uapi_pm_request_mode ret : %X\n",ret);
+ return;
// ret = osSemaphoreAcquire(g_semaphore_spi_id, 60 * 1000);
// if (ret != 0)
// {
// }
// printf("SPI中断触发，初始化完成\r\n");
-
+FAILED:
+ custom_spi_deinit();
}
void custom_spi_deinit(void)
{
- discontinuous_cnt = 0;
- last_expect_seq = 1;
+ pkg_recv_cnt = 0;
+ total_byte_error_cnt = 0;
total_cnt = 0;
- memset_s(g_rx_buffer,sizeof(g_rx_buffer),0,sizeof(g_rx_buffer));
- uapi_spi_deinit(SPI_BUS_1);
+ errcode_t ret = 0;
+
+ if(taskcfg.task_handle)
+ {
+ ret = uapi_terminate_thread(taskcfg.task_handle);
+ if (ret != ERRCODE_SUCC) {
+ printf("uapi_terminate_thread failed: 0x%x\n", ret);
+ }
+ taskcfg.task_handle = NULL;
+ }
+
+
uapi_pin_set_ie(S_MGPIO23,HAL_PIO_IE_DISABLE);
- errcode_t ret = uapi_gpio_unregister_isr_func(S_MGPIO23);
+
+ ret = uapi_gpio_unregister_isr_func(S_MGPIO23);
if (ret != ERRCODE_SUCC) {
printf("uapi_gpio_unregister_isr_func failed: 0x%x\n", ret);
- return;
+ }
+
+ ret = uapi_spi_deinit(SPI_BUS_1);
+ if (ret != ERRCODE_SUCC) {
+ printf("uapi_spi_deinit failed: 0x%x\n", ret);
}
if (g_semaphore_spi_id != NULL)
osSemaphoreDelete(g_semaphore_spi_id);
g_semaphore_spi_id = NULL;
}
+
+ uapi_pm_request_mode(PM_ID_SPI, PM_MODE_NORMAL);
+
+ uint8_t cnt = pm_veto_get_id_cnt(PM_ID_SPI);
+ if (cnt == 1)
+ {
+ uapi_pm_remove_sleep_veto(PM_ID_SPI);
+ }
}
projects / mirrors / FD230.git / commitdiff