# -*- coding: utf-8 -*-
"""
SPI 写操作测试模块
-连续写2MB递增数据，用于测试SPI传输完整性和数据丢失检测
+连续写2MB递增数据，仅当收到外部中断时发送数据块，用于测试SPI传输完整性和数据丢失检测
"""
import log
import utime
from machine import SPI
+from machine import Pin
+from machine import ExtInt
# SPI配置说明：
# spi - 通道选择[0,1]，int类型
SPI_MODE = 0 # Mode 0
SPI_CLK = 8 # 1.625MHz（可根据需要调整：8=1.625MHz, 9=3.25MHz, 12=26MHz）
+# 全局中断标志位：标记是否收到中断（仅中断回调可置位，主逻辑复位）
+interrupt_triggered = False
+# 中断等待超时时间（单位：秒），避免无限等待
+INTERRUPT_TIMEOUT = 60
+
+
+# 中断回调函数：收到中断时置位标志位
+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()))
+
+# 初始化外部中断（GPIO3，上升/下降沿触发，无上下拉）
+extint = ExtInt(ExtInt.GPIO3, ExtInt.IRQ_RISING_FALLING, ExtInt.PULL_DISABLE, ExtIntfun)
+extint.enable()
+clog.info("External interrupt initialized: GPIO3, RISING_FALLING edge")
+
+# 创建gpio对象（唤醒MCU的引脚，GPIO4，输出模式，初始高电平）
+gpioWakeBT = Pin(Pin.GPIO4, Pin.OUT, Pin.PULL_DISABLE, 1)
+# 设置输入输出模式
+gpioWakeBT.set_dir(Pin.OUT)
+
+# 设置引脚电平
+gpioWakeBT.write(1)
+
class SpiWriteTest:
"""
SPI写操作测试类
Returns:
bool: 发送是否成功
"""
+ gpioWakeBT.write(0)
+ # clog.info("gpioWakeBT.write(0)")
+ utime.sleep_ms(2)
+ gpioWakeBT.write(1)
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)
- 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)
+ 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
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)))
return False
def close(self):
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)
total_bytes: 发送的总字节数
data_pattern_info: 数据模式信息（用于MCU端验证）
"""
+ global interrupt_triggered
+
try:
# 初始化SPI
if not self.__init_spi():
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
+
+ # 首次发送不等待中断，后续每次发送前等待中断
+ 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))
+
# 计算本次发送的块大小（最后一块可能小于chunk_size）
remaining = self.data_size - total_sent
current_chunk_size = min(self.chunk_size, remaining)
{ HAL_PIO_FUNC_UART1, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_UP, HAL_PIO_IE_ENBALE }, // S_MGPIO20 uart1 rx
{ HAL_PIO_FUNC_GPIO, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_MAX, HAL_PIO_IE_DISABLE }, // S_MGPIO21 uart1 rts
{ HAL_PIO_FUNC_GPIO, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_UP, HAL_PIO_IE_ENBALE }, // S_MGPIO22 uart1 cts
- { HAL_PIO_FUNC_SPI2_M1, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_DOWN, HAL_PIO_IE_DISABLE }, // S_MGPIO23 NC
- { HAL_PIO_FUNC_SPI2_M1, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_DOWN, HAL_PIO_IE_DISABLE }, // S_MGPIO24 NC
+ { HAL_PIO_FUNC_GPIO, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_UP, HAL_PIO_IE_DISABLE }, // S_MGPIO23 NC
+ { HAL_PIO_FUNC_GPIO, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_MAX, HAL_PIO_IE_DISABLE }, // S_MGPIO24 NC
{ HAL_PIO_FUNC_SPI2_M1, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_DOWN, HAL_PIO_IE_DISABLE }, // S_MGPIO25 NC
{ HAL_PIO_FUNC_SPI2_M1, HAL_PIO_DRIVE_MAX, HAL_PIO_PULL_DOWN, HAL_PIO_IE_DISABLE }, // S_MGPIO26 NC
#if (defined QSPI_DISPLAY)
${CMAKE_CURRENT_SOURCE_DIR}/custom_resfile_check.c
${CMAKE_CURRENT_SOURCE_DIR}/custom_save_runtime.c
${CMAKE_CURRENT_SOURCE_DIR}/ln_compat_stubs.c
+ ${CMAKE_CURRENT_SOURCE_DIR}/custom_spi.c
)
--- /dev/null
+
+/**
+ * Copyright (c) @CompanyNameMagicTag 2023-2023. All rights reserved. \n
+ *
+ * Description: Button driver source file \n
+ * Author: @CompanyNameTag \n
+ * History: \n
+ * 2023-02-24, Create file. \n
+ */
+
+#include "pinctrl.h"
+#include "gpio.h"
+#include "pinctrl.h"
+#include "errcode.h"
+#include "los_base.h"
+#include "tcxo.h"
+#include <stdint.h>
+#include "spi.h"
+#include "custom_spi.h"
+
+static int total_cnt = 0;
+static uint8_t g_rx_buffer[1024] = {0};
+void custom_spi_read(void)
+{
+ spi_xfer_data_t spi_data = {0};
+ 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)
+ {
+ 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(ret == ERRCODE_SPI_DMA_TRANSFER_ERROR)
+ // {
+ // uint8_t data = 2;
+ // 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;
+ // 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);
+ // }
+
+
+ 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_init(void)
+{
+ spi_bus_t bus = SPI_BUS_1;
+ // uint8_t spi_buffer[102];
+
+
+ // Initialize SPI1 as Slave (Client) mode
+ spi_attr_t attr = {0};
+ spi_extra_attr_t extra_attr = {0};
+
+ attr.is_slave = true; // Slave模式（Client）
+ attr.slave_num = 0; // Slave编号
+ attr.bus_clk = 32000000; // 总线时钟32MHz
+ attr.freq_mhz = 0; // Slave模式下由Master控制（26MHz由Master设置）
+ attr.clk_polarity = SPI_CFG_CLK_CPOL_0; // CPOL=0
+ attr.clk_phase = SPI_CFG_CLK_CPHA_0; // CPHA=0 (Mode 0)
+ 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.sste = SPI_CFG_SSTE_DISABLE;
+ attr.ndf = 1;
+
+ // DMA模式配置
+ // 注意：在自动切换模式下，extra_attr.rx_use_dma设置为true只是告诉驱动支持DMA
+ // 实际是否使用DMA由uapi_spi_slave_read根据数据大小自动决定
+ extra_attr.tx_use_dma = false;
+ extra_attr.rx_use_dma = true; // 启用DMA接收模式支持（自动切换模式下会根据数据大小决定）
+
+
+ memset(&extra_attr.qspi_param, 0, sizeof(extra_attr.qspi_param));
+ memset(&extra_attr.sspi_param, 0, sizeof(extra_attr.sspi_param));
+
+ // Deinitialize SPI first (if already initialized)
+ uapi_spi_deinit(bus);
+
+ // Initialize SPI
+ errcode_t ret = uapi_spi_init(bus, &attr, &extra_attr);
+ if (ret != ERRCODE_SUCC) {
+ printf("SPI1 init failed: 0x%x\n", ret);
+ return;
+ }
+ uapi_pin_set_mode(S_MGPIO23,HAL_PIO_FUNC_GPIO);
+ uapi_gpio_set_dir(S_MGPIO23, GPIO_DIRECTION_INPUT);
+ uapi_pin_set_ie(S_MGPIO23,HAL_PIO_IE_ENBALE);
+ 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;
+ }
+}
+
+void custom_spi_deinit(void)
+{
+ 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);
+ if (ret != ERRCODE_SUCC) {
+ printf("uapi_gpio_unregister_isr_func failed: 0x%x\n", ret);
+ return;
+ }
+}
+
+
+void custom_spi_rx_callback(pin_t pin, uintptr_t param)
+{
+ // printf("custom_spi_rx_callback %d 0x%X\n",pin,param);
+ uapi_gpio_clear_interrupt(pin);
+
+ // do_custom_cmd_service_check_msg(CUSTOM_CMD_MSG_SPI_READ,NULL,0);
+}
--- /dev/null
+/*
+ * @Author: penglei 1624952103@qq.com
+ * @Date: 2025-12-04 16:26:22
+ * @LastEditors: penglei 1624952103@qq.com
+ * @LastEditTime: 2025-12-04 16:31:52
+ * @FilePath: /FD230/code/sdk/zhihui/utils/custom_spi.h
+ * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
+ */
+/**
+ * Copyright (c) @CompanyNameMagicTag 2023-2023. All rights reserved. \n
+ *
+ * Description: Button driver header file \n
+ * Author: @CompanyNameTag \n
+ * History: \n
+ * 2023-02-24, Create file. \n
+ */
+#ifndef CUSTOM_SPI_H
+#define CUSTOM_SPI_H
+#ifdef __cplusplus
+extern "C" {
+#endif /* __cplusplus */
+
+#include <stdint.h>
+#include <stdbool.h>
+#include "errcode.h"
+#include "platform_core.h"
+
+void custom_spi_init(void);
+void custom_spi_deinit(void);
+void custom_spi_read(void);
+void custom_spi_rx_callback(pin_t pin, uintptr_t param);
+
+#ifdef __cplusplus
+}
+#endif /* __cplusplus */
+
+#endif
\ No newline at end of file
}
-errcode_t custom_spi_init(uint8_t bus_index)
-{
- spi_bus_t bus;
- spi_attr_t attr;
- bus = bus_index;
- attr.is_slave = false;
- attr.slave_num = 1;
- attr.bus_clk = 32000000;
- attr.freq_mhz = 8;
- attr.clk_polarity = (uint32_t)SPI_CFG_CLK_CPOL_1;
- attr.clk_phase = (uint32_t)SPI_CFG_CLK_CPHA_1;
- attr.frame_format = SPI_CFG_FRAME_FORMAT_MOTOROLA_SPI;
- attr.spi_frame_format = (uint32_t)HAL_SPI_FRAME_FORMAT_STANDARD;
- attr.frame_size = (uint32_t)HAL_SPI_FRAME_SIZE_8;
- attr.tmod = HAL_SPI_TRANS_MODE_TXRX;
- attr.sste = SPI_CFG_SSTE_ENABLE;
- attr.ndf = 1;
-
- errcode_t ret = uapi_spi_init(bus,&attr,NULL);
- if (ret != ERRCODE_SUCC) {
- PRINT("spi init fail! ret=0x%x\n", ret);
- }
- return ret;
-}
+// errcode_t custom_spi_init(uint8_t bus_index)
+// {
+// spi_bus_t bus;
+// spi_attr_t attr;
+// bus = bus_index;
+// attr.is_slave = false;
+// attr.slave_num = 1;
+// attr.bus_clk = 32000000;
+// attr.freq_mhz = 8;
+// attr.clk_polarity = (uint32_t)SPI_CFG_CLK_CPOL_1;
+// attr.clk_phase = (uint32_t)SPI_CFG_CLK_CPHA_1;
+// attr.frame_format = SPI_CFG_FRAME_FORMAT_MOTOROLA_SPI;
+// attr.spi_frame_format = (uint32_t)HAL_SPI_FRAME_FORMAT_STANDARD;
+// attr.frame_size = (uint32_t)HAL_SPI_FRAME_SIZE_8;
+// attr.tmod = HAL_SPI_TRANS_MODE_TXRX;
+// attr.sste = SPI_CFG_SSTE_ENABLE;
+// attr.ndf = 1;
+
+// errcode_t ret = uapi_spi_init(bus,&attr,NULL);
+// if (ret != ERRCODE_SUCC) {
+// PRINT("spi init fail! ret=0x%x\n", ret);
+// }
+// return ret;
+// }
/*省电模式*/
void uapi_enter_power_saving_mode(void)
uint32_t custom_spi_suspend(void);
uint32_t custom_spi_resume(void);
-errcode_t custom_spi_init(uint8_t bus_index);
+// errcode_t custom_spi_init(uint8_t bus_index);
/**
* 资源文件校验码保存在文件 /user/fileschecksum.txt 中
projects / mirrors / FD230.git / commitdiff