Linux下FIFO二进制传感器数据Python读取解析问题

1. 修复ctypes结构体定义与FIFO读取阻塞问题 #

程序冻结大概率是因为FIFO无数据时的阻塞读取，或是结构体字节对齐/读取长度不匹配导致解析失败。先搞定结构体定义和安全读取逻辑：

匹配C代码的ctypes结构体定义 #

C中AXIS_DATA是5个double字段，Python端要确保字节对齐和类型完全对应：

import ctypes

class AXIS_DATA(ctypes.Structure):
    _fields_ = [
        ("axis1", ctypes.c_double),
        ("axis2", ctypes.c_double),
        ("axis3", ctypes.c_double),
        ("axis4", ctypes.c_double),
        ("axis5", ctypes.c_double)
    ]
    _pack_ = 8  # 显式指定8字节对齐，和Linux下C结构体默认对齐规则一致

安全读取FIFO的代码 #

按结构体固定大小（5*8=40字节）读取，避免不完整数据包；如果需要非阻塞模式，可添加错误处理：

import os
import time
import errno

FIFO_PATH = "/path/to/your/fifo"

# 确保FIFO存在，不存在则创建
if not os.path.exists(FIFO_PATH):
    os.mkfifo(FIFO_PATH)

# 阻塞模式读取（传感器持续输出时推荐）
fd = os.open(FIFO_PATH, os.O_RDONLY)

try:
    while True:
        # 读取刚好一个结构体的字节数
        data = os.read(fd, ctypes.sizeof(AXIS_DATA))
        if len(data) != ctypes.sizeof(AXIS_DATA):
            print(f"Received incomplete data: {len(data)} bytes (expected 40)")
            time.sleep(0.1)
            continue
        # 解析二进制数据为结构体
        axis_data = AXIS_DATA.from_buffer_copy(data)
        # 转成字典方便后续处理
        data_dict = {
            "axis1": axis_data.axis1,
            "axis2": axis_data.axis2,
            "axis3": axis_data.axis3,
            "axis4": axis_data.axis4,
            "axis5": axis_data.axis5
        }
        print(data_dict)
        # 后续转JSON、导入InfluxDB的逻辑放在这里
finally:
    os.close(fd)

如果需要非阻塞模式（避免无数据时卡死），修改打开方式并处理EAGAIN错误：

fd = os.open(FIFO_PATH, os.O_RDONLY | os.O_NONBLOCK)

try:
    while True:
        try:
            data = os.read(fd, ctypes.sizeof(AXIS_DATA))
            if len(data) == 0:
                print("FIFO write end closed")
                break
            if len(data) != 40:
                print(f"Incomplete data: {len(data)} bytes")
                continue
            axis_data = AXIS_DATA.from_buffer_copy(data)
            data_dict = {f"axis{i+1}": getattr(axis_data, f"axis{i+1}") for i in range(5)}
            print(data_dict)
        except OSError as e:
            if e.errno == errno.EAGAIN:
                # 暂时无数据，休眠后重试
                time.sleep(0.1)
            else:
                raise
finally:
    os.close(fd)

2. 转JSON并导入InfluxDB #

用官方influxdb-client库实现写入，先安装：pip install influxdb-client

完整整合代码 #

import ctypes
import os
import time
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

# 定义C结构体
class AXIS_DATA(ctypes.Structure):
    _fields_ = [
        ("axis1", ctypes.c_double),
        ("axis2", ctypes.c_double),
        ("axis3", ctypes.c_double),
        ("axis4", ctypes.c_double),
        ("axis5", ctypes.c_double)
    ]
    _pack_ = 8

# InfluxDB配置
INFLUXDB_URL = "http://your-influxdb-host:8086"
INFLUXDB_TOKEN = "your-auth-token"
INFLUXDB_ORG = "your-org-name"
INFLUXDB_BUCKET = "your-bucket-name"

FIFO_PATH = "/path/to/your/fifo"

def main():
    # 初始化InfluxDB客户端
    client = InfluxDBClient(url=INFLUXDB_URL, token=INFLUXDB_TOKEN, org=INFLUXDB_ORG)
    write_api = client.write_api(write_options=SYNCHRONOUS)

    # 打开FIFO
    if not os.path.exists(FIFO_PATH):
        os.mkfifo(FIFO_PATH)
    fd = os.open(FIFO_PATH, os.O_RDONLY)

    try:
        while True:
            data = os.read(fd, 40)
            if len(data) != 40:
                print(f"Invalid data length: {len(data)}")
                time.sleep(0.1)
                continue
            # 解析结构体
            axis_data = AXIS_DATA.from_buffer_copy(data)
            # 构造InfluxDB数据点
            point = Point("sensor_axis_data") \
                .field("axis1", axis_data.axis1) \
                .field("axis2", axis_data.axis2) \
                .field("axis3", axis_data.axis3) \
                .field("axis4", axis_data.axis4) \
                .field("axis5", axis_data.axis5)
            # 写入InfluxDB
            write_api.write(bucket=INFLUXDB_BUCKET, org=INFLUXDB_ORG, record=point)
            print(f"Written data: axis1={axis_data.axis1:.2f}, axis2={axis_data.axis2:.2f}, axis3={axis_data.axis3:.2f}, axis4={axis_data.axis4:.2f}, axis5={axis_data.axis5:.2f}")
    except KeyboardInterrupt:
        print("Stopping data collection...")
    finally:
        os.close(fd)
        client.close()

if __name__ == "__main__":
    main()

3. 常见问题排查 #

• 程序冻结：检查传感器是否正常输出数据到FIFO，或改用非阻塞读取模式；如果FIFO写端关闭，读端会返回空数据，需处理退出逻辑。
• 结构体解析错误：用ctypes.sizeof(AXIS_DATA)确认大小为40字节；检查C结构体是否有额外填充字节，调整_pack_参数匹配C代码的对齐规则。
• InfluxDB写入失败：验证URL、token、组织和桶名是否正确，确保InfluxDB服务正常运行，网络连通。

内容的提问来源于stack exchange，提问作者Codejoy