C语言结构体数据通过管道高效传输至Python：ctypes与二进制流处理教程-Python教程-PHP中文网

C语言结构体数据通过管道高效传输至Python：ctypes与二进制流处理教程

本教程详细介绍了如何通过标准输出管道将C语言结构体数组的二进制数据高效传输至Python，并利用c++types模块进行精确解析。文章从C端的数据准备、二进制写入，到Python端的进程调用、数据捕获与结构化解析，提供了完整的代码示例。特别强调了C语言中正确引入头文件（如stdio.h）的重要性，并建议使用编译器警告标志（如-Wall）来提升代码质量和调试效率。

引言：C与Python数据交互的需求

在科学计算、数据分析或系统编程等场景中，经常需要将高性能c/c++程序生成的数据传递给python进行后续处理（如绘图、机器学习等）。虽然文件i/o是常见的做法，但通过管道（pipe）直接传输二进制数据可以实现更高效、更自动化的数据流。本文将聚焦于如何将c语言中定义的结构体数组，以二进制形式通过标准输出管道传输到python，并使用ctypes模块在python中进行结构化解析。

C端数据准备与二进制输出

在C语言中，我们可以定义一个结构体来封装相关数据，并创建一个该结构体的数组。为了将这些数据传输到Python，我们将使用fwrite函数将结构体数组的内存内容直接写入标准输出（stdout）。

结构体定义

假设我们有一个用于追踪污染浓度的结构体ChemicalDemands：

// ChemicalDemands.h 或直接在 .c 文件中定义
struct ChemicalDemands {
    double DO;  // 溶解氧
    double BOD; // 生化需氧量
    int time;   // 时间戳
};

登录后复制

数据填充与二进制输出

在C程序中，首先创建并填充结构体数组。然后，使用fwrite函数将整个数组的内存块写入stdout。

关键修正：stdio.h的重要性

立即学习“Python免费学习笔记（深入）”；

在C语言中，fwrite和stdout是标准库函数和宏，它们在<stdio.h>头文件中声明。如果未包含此头文件，编译器可能不会报告错误（尤其是在某些旧的C标准或编译器默认设置下），但程序行为将是未定义的，导致输出数据混乱。这是导致原始问题中Python端接收到错误数据的根本原因。因此，务必在C源文件中包含stdio.h。

#include <stdio.h>  // 务必包含此头文件以正确声明fwrite和stdout
#include <stdlib.h> // 包含此头文件以使用exit等函数（如果需要）

struct ChemicalDemands {
    double DO;
    double BOD;
    int time;
};

int main() {
    // 创建一个包含3个ChemicalDemands结构体的数组
    struct ChemicalDemands chemicalDemandsArray[3];

    // 填充数据
    for (int i = 0; i < 3; i++) {
        chemicalDemandsArray[i].DO = 1.0 + i * 0.1; // 示例数据
        chemicalDemandsArray[i].BOD = 2.0 + i * 0.2; // 示例数据
        chemicalDemandsArray[i].time = i + 1;
    }

    // 将结构体数组的二进制内容写入标准输出
    // 参数：数据指针，单个元素大小，元素数量，文件指针（stdout）
    size_t written_elements = fwrite(chemicalDemandsArray, sizeof(struct ChemicalDemands), 3, stdout);

    if (written_elements != 3) {
        fprintf(stderr, "Error: Failed to write all elements to stdout.\n");
        return 1;
    }

    return 0;
}

登录后复制

编译C程序

使用GCC等C编译器将上述C代码编译为可执行文件。

gcc -o c_data_producer main.c

登录后复制

推荐实践：使用编译器警告标志

在编译C/C++代码时，强烈建议始终启用编译器警告标志，例如：

-Wall: 启用所有常见的警告。
-Wextra: 启用额外的一些有用的警告。
-Wpedantic: 严格遵循C标准，报告不符合标准的构造。

例如：gcc -Wall -Wextra -o c_data_producer main.c

这些警告能帮助你发现潜在的错误，包括像忘记包含stdio.h这样的问题，尽管在这种特定情况下，某些编译器可能不会将其视为错误。理解并解决编译器发出的警告是编写健壮代码的关键一步。

Starry.ai

AI艺术绘画生成器

查看详情

Python端数据接收与解析

在Python端，我们将使用subprocess模块执行C程序并捕获其标准输出，然后使用ctypes模块根据C结构体的定义来解析二进制数据。

定义Python ctypes结构体

ctypes模块允许Python代码与C库进行交互。我们需要在Python中定义一个与C语言中完全对应的结构体，确保字段类型和顺序一致。

import subprocess
import ctypes
import sys

# 定义与C语言中struct ChemicalDemands对应的ctypes结构体
class ChemicalDemands(ctypes.Structure):
    _fields_ = [("DO", ctypes.c_double),
                ("BOD", ctypes.c_double),
                ("time", ctypes.c_int)]

登录后复制

注意事项：字节对齐

ctypes.Structure默认会遵循C编译器的字节对齐规则。通常情况下，对于基本数据类型（如double和int），ctypes的默认对齐方式与主流C编译器（如GCC）是一致的。如果C结构体使用了特殊的#pragma pack指令进行字节对齐，那么在Python ctypes中也需要通过_pack_属性进行相应的设置。例如：_pack_ = 1表示1字节对齐。对于本例中的结构体，默认对齐通常是正确的。

执行C程序并捕获输出

使用subprocess.run函数执行编译好的C程序，并捕获其标准输出。

# 执行编译后的C程序，并捕获其标准输出
# 假设C程序名为 c_data_producer
try:
    result = subprocess.run(["./c_data_producer"], capture_output=True, check=True)
    # check=True 会在返回码非零时抛出CalledProcessError
except subprocess.CalledProcessError as e:
    print(f"Error executing C program: {e}")
    print(f"Stderr: {e.stderr.decode()}")
    sys.exit(1)
except FileNotFoundError:
    print("Error: C program 'c_data_producer' not found. Make sure it's compiled and in the correct path.")
    sys.exit(1)

buffer = result.stdout # 捕获到的二进制数据

登录后复制

二进制数据解析

获取到二进制数据后，我们可以根据结构体的大小和元素数量来解析它。

# 确定结构体数组的元素数量
# 假设我们知道C程序输出了3个元素
num_elements = 3 

# 或者，更通用地，根据缓冲区大小和单个结构体大小计算元素数量
# if len(buffer) % ctypes.sizeof(ChemicalDemands) != 0:
#     print("Error: Buffer size is not a multiple of the structure size.")
#     sys.exit(1)
# num_elements = len(buffer) // ctypes.sizeof(ChemicalDemands)


# 创建一个ctypes数组类型，并从捕获的二进制缓冲区中复制数据
# (ChemicalDemands * num_elements) 创建一个包含num_elements个ChemicalDemands的数组类型
# .from_buffer_copy(buffer) 从二进制数据中复制内容到这个数组实例
results_array = (ChemicalDemands * num_elements).from_buffer_copy(buffer)

# 遍历并打印解析出的数据
print("解析出的数据：")
for i, item in enumerate(results_array):
    print(f"元素 {i}: DO={item.DO:.2f}, BOD={item.BOD:.2f}, Time={item.time}")

# 验证第一个元素
print(f"\n第一个元素 (DO, BOD, Time): {results_array[0].DO:.2f}, {results_array[0].BOD:.2f}, {results_array[0].time}")

登录后复制

完整Python代码示例

import subprocess
import ctypes
import sys

# 定义与C语言中struct ChemicalDemands对应的ctypes结构体
class ChemicalDemands(ctypes.Structure):
    _fields_ = [("DO", ctypes.c_double),
                ("BOD", ctypes.c_double),
                ("time", ctypes.c_int)]

def main():
    # 执行编译后的C程序，并捕获其标准输出
    c_program_path = "./c_data_producer" # 确保路径正确
    try:
        result = subprocess.run([c_program_path], capture_output=True, check=True)
    except subprocess.CalledProcessError as e:
        print(f"Error executing C program (return code {e.returncode}):")
        print(f"Stderr: {e.stderr.decode()}")
        sys.exit(1)
    except FileNotFoundError:
        print(f"Error: C program '{c_program_path}' not found. "
              "Please ensure it is compiled and accessible.")
        sys.exit(1)

    buffer = result.stdout # 捕获到的二进制数据

    # 确定结构体数组的元素数量
    # 这里我们假设C程序输出了3个元素，但在实际应用中，你可能需要更动态的方式
    # 例如，C程序可以在输出数据前先输出元素数量，或者通过缓冲区大小计算
    num_elements = 3 

    # 验证缓冲区大小是否与预期相符
    expected_buffer_size = num_elements * ctypes.sizeof(ChemicalDemands)
    if len(buffer) != expected_buffer_size:
        print(f"Warning: Buffer size mismatch. Expected {expected_buffer_size} bytes, got {len(buffer)} bytes.")
        # 根据实际情况，你可以选择在这里退出或尝试继续解析
        # sys.exit(1) 

    # 创建一个ctypes数组类型，并从捕获的二进制缓冲区中复制数据
    try:
        results_array = (ChemicalDemands * num_elements).from_buffer_copy(buffer)
    except ValueError as e:
        print(f"Error creating ctypes array from buffer: {e}")
        print("This might indicate a mismatch in structure definition or buffer size.")
        sys.exit(1)

    # 遍历并打印解析出的数据
    print("--- 解析出的数据 ---")
    for i, item in enumerate(results_array):
        print(f"元素 {i+1}: DO={item.DO:.3f}, BOD={item.BOD:.3f}, Time={item.time}")

    # 再次验证第一个元素
    print(f"\n验证第一个元素 (DO, BOD, Time): {results_array[0].DO:.3f}, {results_array[0].BOD:.3f}, {results_array[0].time}")

if __name__ == "__main__":
    main()

登录后复制