#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
-SPI 写操作测试模块
-连续写2MB递增数据，仅当收到外部中断时发送数据块，用于测试SPI传输完整性和数据丢失检测
+SPI 写操作测试模块（保留原始时序版）
+核心：严格保留原有延时/时序 + 适配datasize参数 + 解决内存/赋值报错
"""
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
-# 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
+# 设置日志输出级别
+log.basicConfig(level=log.INFO)
+clog = log.getLogger("spiwritetest")
-# 测试参数
-TEST_DATA_SIZE = 2 * 1024 * 1024 # 2MB
-CHUNK_SIZE = 1024 # 每次发送1KB
-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）
+# 测试参数（完全保留原始值）
+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
-# å\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）
+# å\85¨å±\80ä¸æ\96æ \87å¿\97ä½\8dï¼\88ä¿\9dç\95\99å\8e\9få§\8b）
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°ï¼\9aæ\94¶å\88°ä¸æ\96æ\97¶ç½®ä½\8dæ \87å¿\97ä½\8d
+# ä¸æ\96å\9b\9eè°\83å\87½æ\95°ï¼\88å®\8cå\85¨ä¿\9dç\95\99å\8e\9få§\8bé\80»è¾\91ï¼\89
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")
- # 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()))
+ clog.info("interrupt_triggered = True (need read slave data)")
-# 初始化外部中断（GPIO3，上升/下降沿触发，无上下拉）
+# 初始化外部中断（保留原始时序/配置）
extint = ExtInt(ExtInt.GPIO3, ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_DISABLE, ExtIntfun)
extint.enable()
clog.info("External interrupt initialized: GPIO3, RISING_FALLING edge")
-# å\88\9b建gpio对象ï¼\88å\94¤é\86\92MCUç\9a\84å¼\95è\84\9aï¼\8cGPIO4ï¼\8cè¾\93å\87ºæ¨¡å¼\8fï¼\8cå\88\9då§\8bé«\98ç\94µå¹³）
+# å\94¤é\86\92MCUå¼\95è\84\9aï¼\88å®\8cå\85¨ä¿\9dç\95\99å\8e\9få§\8bé\85\8dç½®/æ\97¶åº\8f）
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, data_size=TEST_DATA_SIZE, chunk_size=CHUNK_SIZE,
+ def __init__(self, total_data_size=TEST_DATA_SIZE, packet_size=PACKET_TOTAL_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）
- """
- self.data_size = data_size
- self.chunk_size = chunk_size
+ self.total_data_size = total_data_size
+ self.packet_size = packet_size
+ self.header_size = HEADER_SIZE
+ self.data_body_size = DATA_BODY_SIZE
self.spi_channel = spi_channel
self.spi_mode = spi_mode
self.spi_clk = spi_clk
self.spi_obj = None
- self.current_value = 0 # 当前数据值（从0开始递增）
-
+
+ # 包缓存区：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
+
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_chunk(self, start_value, chunk_size):
- """
- 生成递增数据块
-
- Args:
- start_value: 起始值（字节值，0-255）
- chunk_size: 数据块大小（字节）
-
- Returns:
- bytearray: 递增数据块
- """
- chunk = bytearray(chunk_size)
+ def __generate_data_body(self, start_value):
+ """生成1023字节数据体（保留原始逻辑，解决赋值报错）"""
+ # 改为直接构建bytes，避免bytearray赋值
+ data_body_list = []
value = start_value
- for i in range(chunk_size):
- chunk[i] = value & 0xFF # 只取低8位
- value = (value + 1) % 256 # 递增，超过255后回到0
- return chunk
+ for i in range(self.data_body_size):
+ data_body_list.append(value & 0xFF)
+ value = (value + 1) % 256
+ return bytes(data_body_list)
- 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)
+ 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
+
+ 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)
+
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("Error sending chunk via SPI: {}".format(e))
- import sys
- clog.error("Exception type: {}, details: {}".format(type(e).__name__, str(e)))
+ clog.error("send seq {} fail: {}".format(seq, 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("Error closing SPI object: {}".format(e))
+ clog.error("Close SPI failed: {}".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("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, ...)")
+ # 保留原始日志
+ 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))
- # 记录开始时间
start_time = utime.time()
- # 发送数据
- 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
+ # 保留原始发送逻辑
+ while self.total_sent_bytes < self.total_data_size:
+ clog.info("Sending packet {} (seq {})".format(self.packet_count, self.current_seq))
- # 首次发送不等待中断，后续每次发送前等待中断
- 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))
+ # 构建当前包（保留原始逻辑）
+ packet = self.__build_packet(self.current_seq, self.current_data_start)
+ # 缓存上一包（保留原始逻辑）
+ self.packet_cache[self.current_seq] = packet
- # 计算本次发送的块大小（最后一块可能小于chunk_size）
- remaining = self.data_size - total_sent
- current_chunk_size = min(self.chunk_size, remaining)
+ # 发送包（保留原始时序/逻辑）
+ send_ok = self.__send_packet(packet, self.current_seq)
- # 生成递增数据块
- chunk = self.__generate_chunk(current_byte_value, current_chunk_size)
+ # 保留原始中断处理逻辑
+ if interrupt_triggered:
+ clog.info("Interrupt triggered, read slave feedback...")
+ self.__read_slave_feedback() # 读取从机，为主机提供CLK
+ interrupt_triggered = False # 重置中断标志
- # 更新当前字节值（为下一块准备）
- current_byte_value = (current_byte_value + current_chunk_size) % 256
-
- # 通过SPI发送数据块
- if not self.__send_chunk(chunk):
- clog.error("Failed to send chunk {} at offset {}".format(
- chunk_count, total_sent))
+ if not send_ok:
+ clog.error("Send packet {} (seq {}) failed".format(self.packet_count, self.current_seq))
self.close()
- return False, 0.0, total_sent, {}
+ return False, 0.0, self.total_sent_bytes, {}
- total_sent += current_chunk_size
- chunk_count += 1
+ # 保留原始统计更新逻辑
+ 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
- # 每次发送后延迟5ms，减少SPI压力，避免FIFO溢出
- utime.sleep_ms(5)
+ # 保留原始进度打印逻辑
+ 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))
- # 每发送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))
+ # 保留原始延时
+ utime.sleep_ms(5)
- # 记录结束时间并计算耗时
+ # 保留原始测试完成逻辑
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
- # 计算传输速度
- speed_kbps = (total_sent * 8.0) / send_time / 1024.0 # Kbps
- speed_mbps = speed_kbps / 1024.0 # Mbps
-
- # 数据模式信息（用于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'
- }
+ 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))
- 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
-
+ 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
+ }
+
except Exception as e:
- clog.error("Error in SPI write test: {}".format(e))
+ clog.error("Test failed: {}".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(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（可根据需要调整）
- )
-
+# 便捷运行函数（保留原始）
+def run_spi_write_test():
+ test = SpiWriteTest()
+ success, send_time, total_bytes, info = test.run_test()
if success:
- 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))
+ clog.info("Test PASSED | Total bytes sent: {}".format(total_bytes))
else:
- clog.error("Test failed!")
-
-# 如果作为模块导入，不自动运行
-# 可以通过以下方式使用：
-# from spi_write_test import run_spi_write_test
-# success, send_time, total_bytes = run_spi_write_test()
+ clog.error("Test FAILED")
+ return success
+if __name__ == '__main__':
+ run_spi_write_test()
\ No newline at end of file
#include <stdint.h>
#include "spi.h"
#include "custom_spi.h"
+#include "custom_utils.h"
+
+osSemaphoreId_t g_semaphore_spi_id = NULL;
static int 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;
+
void custom_spi_read(void)
{
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;
- errcode_t ret = uapi_spi_slave_read(SPI_BUS_1, &spi_data, 50);
- // osal_irq_restore(irq_sts);
- printf("uapi_spi_slave_read ret : %x \n",ret);
- printf("spi_data.rx_bytes : %d\n",spi_data.rx_bytes);
- if(ret == 0)
+ ret = uapi_spi_slave_read(SPI_BUS_1, &spi_data, 50);
+
+ 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]);
+
+ // 新增：包计数自增
+ 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);
+ }
+ }
+ }
+
+ // 更新上一次接收的序号（用于下次判定）
+ last_expect_seq = current_seq;
+
+ total_cnt += spi_data.rx_bytes;
+ printf("total_cnt : %d, 当前接收包数：%d\n", total_cnt, pkg_recv_cnt);
+ }
+ else
{
- 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);
+ // 新增：包计数自增（失败/长度不符的包也计数，若仅统计成功包可移到上面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
+ // {
+ // 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);
+
+ // 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);
+ // }
+
+
+ // 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);
// int ret = fwrite(spi_data.rx_buff,1,spi_data.rx_bytes,spi_fp);
// printf("fwrite ret:%d\n",ret);
// }
- }
- // else if(ret == ERRCODE_SPI_DMA_TRANSFER_ERROR)
+ // }
+ // else
// {
- // uint8_t data = 2;
+ // 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);
- // data = 0;
- // spi_data.tx_buff = &data;
+ // 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);
- // printf("uapi_spi_slave_write ret : %x \n",ret);
+ // 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_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);
+ // 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)
// {
// }
}
+void custom_spi_service_task(void *arg)
+{
+ printf("SPI处理任务启动，开始等待中断...\r\n");
+
+ // 任务循环：永久运行，处理每次中断
+ while (1) {
+ // 3. 等待信号量（阻塞，直到中断释放/超时）
+ errcode_t ret = osSemaphoreAcquire(g_semaphore_spi_id, 60 * 1000);
+
+ if (ret == 0) {
+ // 4. 信号量获取成功：处理SPI数据
+ // printf("获取信号量成功，开始读取SPI数据...\r\n");
+
+ custom_spi_read();
+
+ // 5. 数据处理完成后，信号量自动回到「锁住」状态（无需手动操作）
+ // 下一次循环会再次调用osSemaphoreAcquire，等待下一次中断释放
+ // printf("SPI数据处理完成，重新等待中断...\r\n");
+
+ } else if (ret == osErrorTimeout) {
+ // 超时处理：打印日志，继续循环等待
+ printf("SPI中断等待超时（%d秒），继续等待...\r\n", 60);
+ } else {
+ // 信号量错误：退出循环，任务结束（可选重试逻辑）
+ 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)
{
spi_bus_t bus = SPI_BUS_1;
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_RX; // 只接收模式
+ attr.tmod = HAL_SPI_TRANS_MODE_TXRX; // 只接收模式
attr.sste = SPI_CFG_SSTE_DISABLE;
attr.ndf = 1;
printf("uapi_gpio_register_isr_func failed: 0x%x\n", ret);
return;
}
+
+ g_semaphore_spi_id = osSemaphoreNew(1, 0, NULL);
+ if(g_semaphore_spi_id == NULL)
+ {
+ printf("osSemaphoreNew failed\n");
+ return;
+ }
+
+ 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.stack_size = 0x2000;
+ taskcfg.func = custom_spi_service_task;
+ taskcfg.use_mem = USE_EXTEND_MEM;
+
+ ret=uapi_create_osThread_custom(&taskcfg);
+ if(ret != ERRCODE_SUCC)
+ {
+ printf("create custom_spi_service_task failed %d!\r\n",ret);
+ return;
+ }
+
+
+ // ret = osSemaphoreAcquire(g_semaphore_spi_id, 60 * 1000);
+ // if (ret != 0)
+ // {
+ // printf("SPI中断等待超时/失败（状态：%d）\r\n", ret);
+ // // 超时处理：销毁任务+信号量
+ // // uapi_delete_osThread_custom(taskcfg.attr.name); // 假设存在任务销毁接口
+ // osSemaphoreDelete(g_semaphore_spi_id);
+ // g_semaphore_spi_id = NULL;
+ // return;
+ // }
+
+ // printf("SPI中断触发，初始化完成\r\n");
+
}
void custom_spi_deinit(void)
{
+ discontinuous_cnt = 0;
+ last_expect_seq = 1;
+ total_cnt = 0;
+ memset_s(g_rx_buffer,sizeof(g_rx_buffer),0,sizeof(g_rx_buffer));
uapi_spi_deinit(SPI_BUS_1);
uapi_pin_set_ie(S_MGPIO23,HAL_PIO_IE_DISABLE);
errcode_t ret = uapi_gpio_unregister_isr_func(S_MGPIO23);
printf("uapi_gpio_unregister_isr_func failed: 0x%x\n", ret);
return;
}
+
+ if (g_semaphore_spi_id != NULL)
+ {
+ osSemaphoreDelete(g_semaphore_spi_id);
+ g_semaphore_spi_id = NULL;
+ }
}
void custom_spi_rx_callback(pin_t pin, uintptr_t param)
{
- // printf("custom_spi_rx_callback %d 0x%X\n",pin,param);
+ // printf("custom_spi_rx_callback \n");
+ // 1. 清除中断标志（避免重复触发）
uapi_gpio_clear_interrupt(pin);
-
- // do_custom_cmd_service_check_msg(CUSTOM_CMD_MSG_SPI_READ,NULL,0);
+
+ // 2. 安全释放信号量（空指针检查）
+ if (g_semaphore_spi_id != NULL) {
+ osSemaphoreRelease(g_semaphore_spi_id);
+ // printf("SPI中断触发，释放信号量\r\n");
+ }
}
+
+void custom_spi_start(void)
+{
+ 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);
+}
\ No newline at end of file
projects / mirrors / FD230.git / commitdiff