基于 Art_DAQ、InfluxDB 和 PyQt 的传感器数据采集、存储与可视化

发布于:2025-09-13 ⋅ 阅读:(23) ⋅ 点赞:(0)

📚 目标

该项目旨在构建一个完整的传感器数据采集系统,集成以下三大核心技术:

  • Art_DAQ:用于高精度模拟信号采集(兼容 USB8814 等国产 DAQ 设备)
  • InfluxDB:高性能时序数据库,用于高效存储与查询时间序列数据
  • PyQt5 + pyqtgraph:构建跨平台图形用户界面(GUI)并实现实时数据可视化

最终成果:一个可独立运行的 .exe 软件,具备硬件连接检测、实时波形显示、数据保存为 CSV 等功能。

源码:https://gitee.com/phm720/signal-monitor


🧰 一、环境准备

1. 安装 Python

下载地址:https://www.python.org/downloads/

⚠️ 推荐使用 Python 3.9,避免某些库的兼容性问题。


2. 创建独立虚拟环境(避免依赖冲突)

# 创建虚拟环境
python -m venv .venv

# 激活环境
# Windows:
.venv\Scripts\activate

3. 安装核心依赖

pip install pyqt5
pip install pyqtgraph
pip install numpy
pip install pandas
pip install influxdb-client
pip install pyinstaller

✅ 或者执行此命令安装依赖 pip install -r requirements.txt


4. 安装 InfluxDB

5. 安装 DAQ 驱动

  • USB8814 用户:作者使用的数据采集卡型号是USB8814,该型号数据采集卡的参数如下所示:

在这里插入图片描述

在这里插入图片描述

  • 安装方法:双击 Setup.exe,安装 ArtDAQ 1.8.31

在这里插入图片描述

  • 提取码可在评论区联系作者获取(发送自己的邮箱,5个工作日之内有回复)。

✅ 如果设备管理器或 Device Management & Configuration Tool - ART Technology 软件中,所连接的采集卡左侧有黄色三角警告图标,很有可能是数字签名问题,建议查阅驱动安装包目录下 ARTDAQ 使用方法-V6.01.pdf 的最后几页,以禁用驱动程序数字签名。

设备管理器界面:

在这里插入图片描述

ARTDAQ 软件界面:

在这里插入图片描述

6.硬件连接

整体连接:

在这里插入图片描述

这里用的是三轴加速度传感器,可读取 XYZ 三个方向上的加速度,对应 ai0ai1ai2 三个通道。

三轴加速度传感器:

在这里插入图片描述

USB8814 数据采集卡:

在这里插入图片描述


🛠️ 二、项目结构

sensor_monitor/
├── main.py                 # 主程序
├── mainwindow.ui           # Qt Designer 设计的界面
├── icon.ico                # 软件图标
├── requirements.txt        # 依赖列表
├── main.spec.bac           # 打包配置文件备份
├── README.md               # 项目说明
├── LICENSE                 # 软件许可
├── .gitignore              # git 项目文件
├── .venv/                  # 虚拟环境
├── assets/                 # README.md 依赖的资源
└── artdaq/                 # ArtDAQ 驱动包

🎨 三、设计用户界面(.ui 文件)

使用 Qt Designer 创建 mainwindow.ui,包含以下控件:

  • QPushButton:开始采集、停止采集、保存数据
  • QLabel:状态显示(如“采集中”、“空闲”)
  • PlotWidget:实时波形显示区域(pyqtgraph)
  • QLineEdit:采样率、通道数等参数输入

💡 保存后,用 uic.loadUi('mainwindow.ui', self) 在代码中加载。


💻 四、核心代码实现

1. 导入必要模块

import sys
import numpy as np
import pandas as pd
from datetime import datetime, timezone
from influxdb_client import InfluxDBClient
from PyQt5 import QtWidgets, uic
from PyQt5.QtWidgets import QMessageBox, QFileDialog
import pyqtgraph as pg
import artdaq  # 项目目录下要包含 artdaq 文件

2. 资源路径处理函数(关键!用于打包)

def resource_path(relative_path):
    try:
        base_path = sys._MEIPASS
    except Exception:
        base_path = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(base_path, relative_path)

3. 主窗口类定义

class MainWindow(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        # 动态加载 UI
        ui_path = resource_path('mainwindow.ui')
        uic.loadUi(ui_path, self)

        # 初始化 InfluxDB 客户端
        self.influx_client = InfluxDBClient(
            url="http://localhost:8086",
            token="your-token",
            org="your-org"
        )
        self.write_api = self.influx_client.write_api()

        # 硬件检测
        self.hardware_connected = False
        self.test_hardware_connection()

        if not self.hardware_connected:
            QMessageBox.critical(self, "硬件错误", "无法连接到采集卡,请检查设备。")
            self.startButton.setEnabled(False)
        else:
            QMessageBox.information(self, "连接成功", "已检测到采集卡。")

        # 初始化绘图
        self.plot_widget = self.findChild(pg.PlotWidget, 'plotWidget')
        self.curve = self.plot_widget.plot(pen='y')

        # 连接按钮信号
        self.startButton.clicked.connect(self.start_acquisition)
        self.stopButton.clicked.connect(self.stop_acquisition)
        self.saveButton.clicked.connect(self.save_data)

        self.is_running = False
        self.session_start = None
        self.session_end = None

4. 硬件连接检测

def test_hardware_connection(self):
    try:
        task = artdaq.Task()
        task.ai_channels.add_ai_voltage_chan("Dev1/ai0")  # 测试通道
        task.close()
        self.hardware_connected = True
        return True
    except Exception as e:
        print(f"硬件检测失败: {e}")
        self.hardware_connected = False
        return False

5. 数据采集(多线程)

from PyQt5.QtCore import QThread, pyqtSignal

class DataAcquisitionThread(QThread):
    data_ready = pyqtSignal(np.ndarray)

    def __init__(self):
        super().__init__()
        self.running = False

    def run(self):
        task = artdaq.Task()
        task.ai_channels.add_ai_voltage_chan("Dev1/ai0:7")  # 8 通道
        task.timing.cfg_samp_clk_timing(1000)  # 1kHz 采样率

        self.running = True
        while self.running:
            data = task.read(number_of_samples_per_channel=100)
            self.data_ready.emit(np.array(data))
        task.close()

MainWindow 中启动线程:

def start_acquisition(self):
    self.acq_thread = DataAcquisitionThread()
    self.acq_thread.data_ready.connect(self.update_plot)
    self.acq_thread.start()
    self.is_running = True
    self.session_start = datetime.now().timestamp()

6. 实时绘图更新

def update_plot(self, data):
    if data.ndim > 1:
        data = data[0]  # 取第一通道示例
    self.curve.setData(data)

7. 数据保存为 CSV

def save_data(self):
    if not self.session_start or not self.session_end:
        QMessageBox.warning(self, "保存失败", "无采集数据。")
        return

    try:
        query = f'''
        from(bucket: "sensor_data")
            |> range(start: {self.session_start}s, stop: {self.session_end}s)
            |> filter(fn: (r) => r._measurement == "voltage")
            |> pivot(rowKey:["_time"], columnKey: ["channel"], valueColumn: "_value")
        '''
        result = self.influx_client.query_api().query(query)
        # ... 转为 DataFrame 并保存为 CSV
        filename, _ = QFileDialog.getSaveFileName(self, "保存数据", "", "CSV Files (*.csv)")
        if filename:
            df.to_csv(filename, index=False)
            QMessageBox.information(self, "保存成功", f"数据已保存至:\n{filename}")
    except Exception as e:
        QMessageBox.critical(self, "保存失败", str(e))

8. 程序执行

程序执行前务必在电脑上安装 ArtDAQ 驱动,并且连接好数据采集卡、传感器等硬件,同时启动 InfluxDB 服务,配置好 InfluxDB 各项参数,否则程序无法正常运行。
执行以下命令,执行程序:

python main.py

📦 五、打包为独立 .exe 文件

1. 生成 .spec 文件

pyinstaller --onefile --windowed --icon=icon.ico main.py

2. 编辑 main.spec(关键配置,可参照 main.spec.bac)

a = Analysis(
    ['main.py'],
    datas=[
        ('mainwindow.ui', '.'),      # 包含 UI 文件
        ('artdaq', 'artdaq'),      # 包含驱动文件
    ],
    hiddenimports=[
        'influxdb_client',
        'numpy',
    ],
    excludes=[
        'tkinter', 'matplotlib', 'PyQt5.Qt3D*', 'PyQt5.QtWeb*'
    ],
    ...
)

exe = EXE(
    ...,
    console=False,          # 无控制台
    icon='icon.ico',        # 图标
)

3. 打包

pyinstaller main.spec

输出文件位于 dist/Signaller.exe


🧪 六、部署与运行

用户需准备:

  1. 安装 ArtDAQ 驱动
  2. 启动 InfluxDB 服务
  3. 运行 Signaller.exe

软件功能:

  • 启动时自动检测硬件连接
  • 实时显示多通道波形
  • 支持开始/停止采集
  • 可将历史数据导出为 CSV

截图:

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

在这里插入图片描述

在这里插入图片描述
在这里插入图片描述

完整的演示视频:

https://v.youku.com/v_show/id_XNjQ5MzAwMDIxNg==.html


🔚 七、总结

技术 作用
Art_DAQ 高精度、低延迟数据采集
InfluxDB 高效存储与查询时间序列数据
PyQt + pyqtgraph 构建专业级 GUI 与实时可视化
PyInstaller 打包为可分发的 .exe 软件

✅ 本系统适用于实验室监测、工业传感器采集、环境监控等场景,具备良好的扩展性与实用性。


📎 附录:常见问题

Q1:打包后 UI 加载失败?

→ 确保使用 resource_path() 获取路径。

Q2:图标不显示?

→ 安装 pillow,或使用专业工具生成标准 .ico 文件。

Q3:体积太大?

→ 使用干净虚拟环境 + excludes 排除无用模块。