initial commit

serial_module.c

@@ -0,0 +1,262 @@
+#define PY_SSIZE_T_CLEAN
+#include <Python.h>
+#include <stdint.h>
+#include <string.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <zlib.h> // 引入 zlib 库
+#include <fec.h>  // 引入 libfec 库
+#include "hx_serial.h"
+#include "hx_ringbuffer.h"
+#include <pthread.h> // 引入 pthread 库
+
+// 宏定义
+#define USE_FEC
+#ifdef USE_FEC
+#define FEC_SIZE 32 // 前向纠错冗余数据大小
+#else
+#define FEC_SIZE 0 // 前向纠错冗余数据大小
+#endif
+#define FRAME_HEADER 0xAA55                                             // 帧头
+#define FRAME_SIZE (240)                                                // 每帧大小
+#define HEADER_SIZE (4)                                                 // 帧头 + 帧序号 + 数据长度
+#define CHECKSUM_SIZE (4)                                               // CRC32 校验和大小（4 字节）
+#define DATA_SIZE (FRAME_SIZE - HEADER_SIZE - CHECKSUM_SIZE - FEC_SIZE) // 数据段大小
+#define RING_BUFFER_SIZE (1024 * 10)                                    // 环形缓冲区大小 - default 10KB
+#define QUEUE_MAX_SIZE 1024                                             // 队列最大容量
+
+// 全局变量
+static by_serial_t serial_port;
+static by_ringbuf_t ring_buffer;
+static uint8_t send_buffer[FRAME_SIZE];
+static uint8_t frame_counter = 0;
+static uint8_t data_len = 0;
+
+// 定义队列结构体
+typedef struct {
+    uint8_t *data;
+    int length;
+} FrameData;
+
+// 队列相关定义
+static FrameData frame_queue[QUEUE_MAX_SIZE]; // 存储接收到的数据帧
+static int queue_head = 0;                   // 队列头指针
+static int queue_tail = 0;                   // 队列尾指针
+static pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER; // 保护队列的互斥锁
+static bool stop_receiving = false;          // 控制接收线程停止的标志
+
+// 前向声明
+void *receive_thread_func(void *arg);
+
+// 计算 CRC32 校验和
+uint32_t calculate_crc32(const uint8_t *data, size_t length) {
+    return crc32(0, data, length);
+}
+
+// 解析一帧数据
+int parse_frame(by_ringbuf_t *ringbuf, uint8_t *output_data, int *output_len) {
+    uint8_t frame[FRAME_SIZE];
+    int available_data = by_ringbuf_available_data(ringbuf);
+
+    // 检查是否有足够的数据解析一帧
+    if (available_data < HEADER_SIZE + CHECKSUM_SIZE + FEC_SIZE) {
+        return -1; // 数据不足，无法解析
+    }
+
+    // 查找帧头
+    uint16_t header = FRAME_HEADER;
+    int header_pos = by_ringbuf_find(ringbuf, (uint8_t *)&header, 2);
+    if (header_pos < 0) {
+        return -1; // 没有找到帧头
+    }
+
+    // 弹出帧头之前的数据
+    by_ringbuf_pop(ringbuf, frame, header_pos);
+
+    // 检查是否有足够的数据解析一帧
+    if (by_ringbuf_available_data(ringbuf) < FRAME_SIZE) {
+        return -1; // 数据不足，无法解析
+    }
+
+    // 读取帧数据
+    by_ringbuf_pop(ringbuf, frame, FRAME_SIZE);
+
+    // 解析帧序号、有效数据长度、数据段和 CRC32
+    uint8_t seq = frame[2];
+    uint8_t valid_data_len = frame[2 + 1];
+    uint8_t *data_segment = &frame[2 + 1 + 1];
+    uint32_t received_crc = *(uint32_t *)&frame[2 + 1 + 1 + DATA_SIZE];
+
+    // 计算 CRC32 校验
+    uint32_t calculated_crc = calculate_crc32(frame, HEADER_SIZE + DATA_SIZE + FEC_SIZE);
+    if (received_crc != calculated_crc) {
+        printf("CRC mismatch! Expected: %08X, Received: %08X\n", calculated_crc, received_crc);
+        return -1; // CRC 校验失败，丢弃该帧
+    }
+
+    // 将有效数据拼接到输出缓冲区
+    memcpy(&output_data[*output_len], data_segment, valid_data_len);
+    *output_len += valid_data_len;
+
+    // 判断是否为最后一帧
+    if (valid_data_len < DATA_SIZE) {
+        printf("Received last frame!\n");
+        return 1; // 最后一帧，解析完成
+    }
+    return 0; // 成功解析一帧，但可能还有更多帧
+}
+
+// 接收线程函数
+void *receive_thread_func(void *arg) {
+    while (!stop_receiving) {
+        uint8_t output_data[8192];
+        int output_len = 0;
+
+        // 从串口读取数据到环形缓冲区
+        uint8_t buffer[FRAME_SIZE];
+        int ret = by_serial_read(&serial_port, buffer, FRAME_SIZE);
+        if (ret > 0) {
+            by_ringbuf_append(&ring_buffer, buffer, ret);
+        }
+
+        // 尝试解析环形缓冲区中的数据
+        while (1) {
+            int parse_result = parse_frame(&ring_buffer, output_data, &output_len);
+            if (parse_result == 1) {
+                // 将解析后的数据放入队列
+                pthread_mutex_lock(&queue_mutex);
+                if ((queue_tail + 1) % QUEUE_MAX_SIZE != queue_head) { // 队列未满
+                    frame_queue[queue_tail].data = malloc(output_len);
+                    memcpy(frame_queue[queue_tail].data, output_data, output_len);
+                    frame_queue[queue_tail].length = output_len;
+                    queue_tail = (queue_tail + 1) % QUEUE_MAX_SIZE;
+                } else {
+                    printf("Queue is full, dropping frame!\n");
+                }
+                pthread_mutex_unlock(&queue_mutex);
+                break;
+            } else if (parse_result == -1) {
+                break; // 数据不足或解析失败，退出循环
+            }
+        }
+
+        usleep(1000); // 避免占用过多 CPU
+    }
+    return NULL;
+}
+
+// 初始化串口
+static PyObject *serial_init(PyObject *self, PyObject *args) {
+    const char *dev_name;
+    if (!PyArg_ParseTuple(args, "s", &dev_name)) {
+        return NULL;
+    }
+
+    if (by_serial_init(&serial_port, dev_name) != 0) {
+        PyErr_SetString(PyExc_IOError, "Failed to initialize serial port");
+        return NULL;
+    }
+
+    // 初始化环形缓冲区
+    if (by_ringbuf_init(&ring_buffer, RING_BUFFER_SIZE) != 0) {
+        PyErr_SetString(PyExc_IOError, "Failed to initialize ring buffer");
+        return NULL;
+    }
+
+    // 启动接收线程
+    pthread_t receive_thread;
+    stop_receiving = false;
+    if (pthread_create(&receive_thread, NULL, receive_thread_func, NULL) != 0) {
+        PyErr_SetString(PyExc_RuntimeError, "Failed to start receive thread");
+        return NULL;
+    }
+
+    Py_RETURN_NONE;
+}
+
+// 发送数据
+static PyObject *serial_send(PyObject *self, PyObject *args) {
+    const char *data;
+    Py_ssize_t length;
+    if (!PyArg_ParseTuple(args, "s#", &data, &length)) {
+        return NULL;
+    }
+
+    size_t offset = 0;
+    while (offset < length) {
+        memset(send_buffer, 0, FRAME_SIZE);
+
+        // 构造帧头
+        uint16_t header = FRAME_HEADER;
+        memcpy(send_buffer, &header, 2);
+
+        // 构造帧序号和数据长度
+        memcpy(send_buffer + 2, &frame_counter, 1);
+        if (length - offset > DATA_SIZE) {
+            data_len = DATA_SIZE;
+        } else {
+            data_len = length - offset;
+        }
+        memcpy(send_buffer + 3, &data_len, 1);
+
+        // 拷贝该帧对应数据段
+        memcpy(send_buffer + HEADER_SIZE, data + offset, data_len);
+
+        // 计算 CRC32 校验和
+        uint32_t crc = calculate_crc32(send_buffer, HEADER_SIZE + DATA_SIZE + FEC_SIZE);
+        memcpy(send_buffer + HEADER_SIZE + DATA_SIZE + FEC_SIZE, &crc, CHECKSUM_SIZE);
+
+        // 发送帧
+        if (by_serial_write(&serial_port, (const char *)send_buffer, FRAME_SIZE) != 0) {
+            PyErr_SetString(PyExc_IOError, "Failed to send data over serial port");
+            return NULL;
+        }
+
+        offset += DATA_SIZE;
+        frame_counter++;
+        usleep(80000);
+    }
+
+    frame_counter = 0;
+    Py_RETURN_NONE;
+}
+
+// 接收数据
+static PyObject *serial_receive(PyObject *self, PyObject *args) {
+    pthread_mutex_lock(&queue_mutex);
+    if (queue_head == queue_tail) {
+        pthread_mutex_unlock(&queue_mutex);
+        PyErr_SetString(PyExc_IOError, "No data available in the queue");
+        return NULL;
+    }
+
+    // 获取队列中最早的数据包
+    FrameData frame = frame_queue[queue_head];
+    queue_head = (queue_head + 1) % QUEUE_MAX_SIZE;
+    pthread_mutex_unlock(&queue_mutex);
+
+    // 返回解析后的数据
+    PyObject *result = Py_BuildValue("y#", frame.data, frame.length);
+    free(frame.data); // 释放内存
+    return result;
+}
+
+// 模块方法表
+static PyMethodDef SerialMethods[] = {
+    {"init", serial_init, METH_VARARGS, "Initialize serial port"},
+    {"send", serial_send, METH_VARARGS, "Send data over serial port"},
+    {"receive", serial_receive, METH_VARARGS, "Receive data from serial port"},
+    {NULL, NULL, 0, NULL}};
+
+// 模块定义
+static struct PyModuleDef serialmodule = {
+    PyModuleDef_HEAD_INIT,
+    "serial_module",
+    NULL,
+    -1,
+    SerialMethods};
+
+// 模块初始化函数
+PyMODINIT_FUNC PyInit_serial_module(void) {
+    return PyModule_Create(&serialmodule);
+}