从零到精用树莓派Python玩转工业传感器数据采集压力/温度/光电在工业自动化和物联网技术蓬勃发展的今天如何低成本、高效率地实现工业现场数据采集成为了许多技术人员关注的焦点。传统工业数据采集系统往往价格昂贵、配置复杂对于中小型企业、创客团队或个人开发者来说门槛较高。而树莓派这款信用卡大小的微型计算机配合Python这一简单易用的编程语言为我们提供了一条快速搭建工业数据采集系统的捷径。本文将带领读者从零开始逐步构建一个完整的工业传感器数据采集系统。我们将重点介绍三种常见工业传感器——压力传感器、温度传感器和光电传感器的连接方法、数据采集原理以及Python编程实现。通过实际案例演示读者将掌握如何将这些物理量转换为数字信号并进行可视化展示和数据存储。无论你是工业自动化领域的新手还是希望扩展技能的经验丰富的工程师这套基于开源硬件的解决方案都将为你打开工业物联网应用开发的大门。1. 硬件准备与系统搭建1.1 树莓派选型与系统配置树莓派有多种型号可供选择对于工业数据采集应用我们推荐使用树莓派4B或更新型号主要考虑其更强的处理能力、更大的内存和更丰富的接口。以下是各型号树莓派的对比型号CPU内存USB接口网络功耗价格3B1.4GHz四核1GB4×USB2.0千兆以太网5W中等4B1.5GHz四核2/4/8GB2×USB3.02×USB2.0千兆以太网7.5W较高Zero 2W1GHz四核512MB1×micro USB无线2.5W低系统方面我们推荐使用Raspberry Pi OS Lite版本无桌面环境这样可以节省系统资源。安装完成后需要进行以下基本配置# 更新系统 sudo apt update sudo apt upgrade -y # 安装必要工具 sudo apt install -y python3-pip git vim # 设置Python环境 pip3 install --upgrade pip1.2 传感器选型与接口工业传感器种类繁多接口标准也不尽相同。我们主要关注以下几种常见类型压力传感器如MPX5700系列模拟电压输出0-5V量程0-700kPa温度传感器如DS18B20数字接口1-Wire测量范围-55°C至125°C光电传感器如E3Z光电开关数字输出NPN/PNP对于模拟输出的传感器树莓派需要通过ADC模数转换器模块进行连接因为树莓派本身没有模拟输入引脚。推荐使用MCP3008这款8通道10位ADC芯片价格低廉且性能可靠。1.3 电路连接与安全注意事项工业现场环境复杂在连接传感器时需要注意以下事项电气隔离工业传感器工作电压可能较高如24V务必使用光耦或隔离模块保护树莓派信号调理对于模拟信号可能需要使用运算放大器进行信号调理接地处理避免形成地环路导致信号干扰防反接保护在电源线上串联二极管防止电源反接以下是MCP3008与树莓派的连接方式MCP3008 树莓派 VDD 3.3V VREF 3.3V AGND GND CLK GPIO11 (SCLK) DOUT GPIO9 (MISO) DIN GPIO10 (MOSI) CS/SHDN GPIO8 (CE0) DGND GND2. Python数据采集基础2.1 GPIO控制与SPI通信树莓派通过GPIO引脚与外部设备通信。Python中可以使用RPi.GPIO或gpiozero库来操作GPIO。对于SPI设备如MCP3008我们需要先启用SPI接口sudo raspi-config # 选择Interfacing Options - SPI - Yes安装必要的Python库pip3 install RPi.GPIO gpiozero spidev下面是一个读取MCP3008通道0的示例代码import spidev import time # 创建SPI对象 spi spidev.SpiDev() spi.open(0, 0) # 使用SPI0片选0 spi.max_speed_hz 1000000 # 设置SPI时钟频率 def read_adc(channel): # 确保通道号在0-7之间 if channel 0 or channel 7: return -1 # MCP3008的SPI通信协议 # 起始位 单端/差分选择 通道号高位 通道号低位 cmd 0b11000000 | (channel 3) resp spi.xfer2([cmd, 0x00, 0x00]) # 解析返回的10位数据 value ((resp[0] 0x01) 9) | (resp[1] 1) | (resp[2] 7) return value while True: value read_adc(0) # 读取通道0 voltage (value * 3.3) / 1023 # 转换为电压值 print(fADC值: {value}, 电压: {voltage:.2f}V) time.sleep(1)2.2 传感器数据处理与校准从传感器获取的原始数据通常需要经过处理和校准才能得到有意义的物理量。以MPX5700压力传感器为例其输出电压与压力呈线性关系压力(kPa) (Vout - 0.2V) × (700kPa / (4.7V - 0.2V))在Python中可以实现如下def pressure_from_voltage(voltage): # MPX5700压力传感器转换公式 return (voltage - 0.2) * (700 / 4.5) if voltage 0.2 else 0.0 # 使用示例 adc_value read_adc(0) voltage (adc_value * 3.3) / 1023 pressure pressure_from_voltage(voltage) print(f当前压力: {pressure:.2f} kPa)对于数字传感器如DS18B20温度传感器我们需要先启用1-Wire接口sudo raspi-config # 选择Interfacing Options - 1-Wire - Yes然后可以通过文件系统直接读取温度数据def read_ds18b20(): base_dir /sys/bus/w1/devices/ device_folder glob.glob(base_dir 28*)[0] device_file device_folder /w1_slave with open(device_file, r) as f: lines f.readlines() if lines[0].strip()[-3:] ! YES: return None equals_pos lines[1].find(t) if equals_pos -1: return None temp_str lines[1][equals_pos2:] temp_c float(temp_str) / 1000.0 return temp_c while True: temp read_ds18b20() if temp is not None: print(f当前温度: {temp:.2f}°C) time.sleep(1)2.3 多线程数据采集在实际应用中我们通常需要同时采集多个传感器的数据。这时可以使用Python的threading模块实现多线程采集import threading from queue import Queue class SensorReader(threading.Thread): def __init__(self, sensor_type, interval1, queueNone): super().__init__() self.sensor_type sensor_type self.interval interval self.queue queue self._stop_event threading.Event() def stop(self): self._stop_event.set() def run(self): while not self._stop_event.is_set(): if self.sensor_type pressure: value read_pressure() elif self.sensor_type temperature: value read_ds18b20() elif self.sensor_type photo: value read_photo_sensor() if self.queue: self.queue.put((self.sensor_type, value, time.time())) time.sleep(self.interval) # 使用示例 data_queue Queue() sensors [ SensorReader(pressure, 0.5, data_queue), SensorReader(temperature, 1, data_queue), SensorReader(photo, 0.2, data_queue) ] for sensor in sensors: sensor.start() try: while True: sensor_type, value, timestamp data_queue.get() print(f[{timestamp}] {sensor_type}: {value}) except KeyboardInterrupt: for sensor in sensors: sensor.stop() for sensor in sensors: sensor.join()3. 数据存储与可视化3.1 数据库存储方案对于工业数据采集系统可靠的数据存储至关重要。我们推荐使用轻量级的SQLite数据库或时序数据库InfluxDB。SQLite方案import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(sensor_data.db) c conn.cursor() c.execute(CREATE TABLE IF NOT EXISTS sensor_data (id INTEGER PRIMARY KEY AUTOINCREMENT, sensor_type TEXT, value REAL, timestamp DATETIME)) conn.commit() conn.close() def save_to_db(sensor_type, value): conn sqlite3.connect(sensor_data.db) c conn.cursor() c.execute(INSERT INTO sensor_data (sensor_type, value, timestamp) VALUES (?, ?, ?), (sensor_type, value, datetime.now())) conn.commit() conn.close() # 初始化数据库 init_db() # 在数据采集线程中调用 save_to_db(pressure, 123.45)InfluxDB方案from influxdb import InfluxDBClient client InfluxDBClient(hostlocalhost, port8086) client.switch_database(sensor_data) def save_to_influx(sensor_type, value): json_body [ { measurement: sensor_type, fields: { value: value } } ] client.write_points(json_body) # 使用示例 save_to_influx(temperature, 25.6)3.2 实时数据可视化Python生态系统提供了丰富的数据可视化工具如Matplotlib、Plotly和PyQtGraph。对于实时数据显示PyQtGraph性能最佳import pyqtgraph as pg from pyqtgraph.Qt import QtGui import numpy as np app QtGui.QApplication([]) win pg.GraphicsWindow(title实时传感器数据) # 创建三个图表 pressure_plot win.addPlot(title压力 (kPa)) temp_plot win.addPlot(title温度 (°C)) photo_plot win.addPlot(title光电传感器) # 初始化曲线 pressure_curve pressure_plot.plot(penr) temp_curve temp_plot.plot(penb) photo_curve photo_plot.plot(peng) # 数据缓冲区 data_len 100 pressure_data np.zeros(data_len) temp_data np.zeros(data_len) photo_data np.zeros(data_len) ptr 0 def update(): global ptr, pressure_data, temp_data, photo_data # 模拟数据更新 pressure_data[ptr] read_pressure() temp_data[ptr] read_ds18b20() photo_data[ptr] read_photo_sensor() ptr 1 if ptr data_len: ptr 0 # 更新曲线 pressure_curve.setData(pressure_data) temp_curve.setData(temp_data) photo_curve.setData(photo_data) # 自动调整Y轴范围 pressure_plot.enableAutoRange(y, True) temp_plot.enableAutoRange(y, True) photo_plot.enableAutoRange(y, True) timer pg.QtCore.QTimer() timer.timeout.connect(update) timer.start(50) # 更新间隔(ms) QtGui.QApplication.instance().exec_()3.3 Web界面与远程监控对于需要远程监控的场景可以搭建一个简单的Web界面。使用Flask框架可以快速实现from flask import Flask, render_template, jsonify from threading import Thread import time app Flask(__name__) # 模拟传感器数据 current_data { pressure: 0.0, temperature: 0.0, photo: 0.0 } def sensor_loop(): while True: current_data[pressure] read_pressure() current_data[temperature] read_ds18b20() current_data[photo] read_photo_sensor() time.sleep(1) app.route(/) def index(): return render_template(index.html) app.route(/data) def get_data(): return jsonify(current_data) if __name__ __main__: # 启动传感器线程 Thread(targetsensor_loop, daemonTrue).start() app.run(host0.0.0.0, port8080)对应的HTML模板templates/index.html!DOCTYPE html html head title传感器监控面板/title script srchttps://cdn.plot.ly/plotly-latest.min.js/script style .gauge { width: 300px; height: 200px; display: inline-block; } .chart { width: 100%; height: 300px; } /style /head body h1工业传感器监控系统/h1 div idpressureGauge classgauge/div div idtempGauge classgauge/div div idphotoGauge classgauge/div div idhistoryChart classchart/div script // 初始化仪表盘 function createGauge(elementId, title, min, max) { return { domain: { row: 0, column: 0 }, value: 0, title: { text: title }, type: indicator, mode: gaugenumber, gauge: { axis: { range: [min, max] }, bar: { color: blue } } }; } const pressureGauge createGauge(pressureGauge, 压力 (kPa), 0, 700); const tempGauge createGauge(tempGauge, 温度 (°C), -20, 100); const photoGauge createGauge(photoGauge, 光电传感器, 0, 1023); Plotly.newPlot(pressureGauge, [pressureGauge]); Plotly.newPlot(tempGauge, [tempGauge]); Plotly.newPlot(photoGauge, [photoGauge]); // 历史数据图表 const historyChart { y: [], type: scatter, name: 压力历史 }; Plotly.newPlot(historyChart, [historyChart]); // 定时更新数据 function updateData() { fetch(/data) .then(response response.json()) .then(data { // 更新仪表盘 Plotly.update(pressureGauge, { value: data.pressure }); Plotly.update(tempGauge, { value: data.temperature }); Plotly.update(photoGauge, { value: data.photo }); // 更新历史图表 historyChart.y.push(data.pressure); if (historyChart.y.length 50) { historyChart.y.shift(); } Plotly.update(historyChart, { y: [historyChart.y] }); }); } setInterval(updateData, 1000); /script /body /html4. 高级应用与系统集成4.1 工业通信协议集成在工业环境中Modbus和OPC UA是两种最常见的通信协议。我们可以使用Python库来实现对这些协议的支持。Modbus RTU/TCP集成from pymodbus.client import ModbusSerialClient as ModbusClient # 初始化Modbus RTU客户端 modbus_rtu ModbusClient( methodrtu, port/dev/ttyUSB0, baudrate9600, timeout1 ) # 连接设备 if modbus_rtu.connect(): # 读取保持寄存器 result modbus_rtu.read_holding_registers(address0, count10, slave1) if not result.isError(): print(Modbus读取结果:, result.registers) else: print(Modbus读取错误:, result) modbus_rtu.close()OPC UA集成from opcua import Client opcua_url opc.tcp://localhost:4840 client Client(opcua_url) try: client.connect() # 获取节点 root client.get_root_node() objects client.get_objects_node() # 读取变量 temp_node client.get_node(ns2;i2) temperature temp_node.get_value() print(OPC UA温度值:, temperature) finally: client.disconnect()4.2 异常检测与预警系统工业数据采集系统的一个重要功能是异常检测和预警。我们可以使用简单的统计方法或机器学习算法来实现import numpy as np from sklearn.ensemble import IsolationForest # 历史数据 pressure_history np.loadtxt(pressure_history.csv) # 训练异常检测模型 clf IsolationForest(contamination0.05) clf.fit(pressure_history.reshape(-1, 1)) def check_anomaly(value): # 预测是否为异常-1表示异常 return clf.predict([[value]])[0] -1 # 使用示例 current_pressure read_pressure() if check_anomaly(current_pressure): print(f警告检测到异常压力值: {current_pressure}) # 触发警报如发送邮件或短信4.3 与工业云平台集成将本地采集的数据上传到工业云平台可以实现更高级的分析和远程监控。以阿里云IoT平台为例from aliyunIoT import Device # 设备认证信息 product_key your_product_key device_name your_device_name device_secret your_device_secret # 初始化设备 device Device(product_key, device_name, device_secret) # 属性上报回调 def on_props(request, response): print(属性上报结果:, response) # 连接云端 device.connect() # 上报传感器数据 def report_to_cloud(pressure, temp, photo): props { Pressure: pressure, Temperature: temp, PhotoSensor: photo } device.postProps(props, on_props) # 在数据采集循环中调用 while True: pressure read_pressure() temp read_ds18b20() photo read_photo_sensor() report_to_cloud(pressure, temp, photo) time.sleep(60) # 每分钟上报一次4.4 系统优化与性能调优随着采集点增多系统性能可能成为瓶颈。以下是一些优化建议数据采集优化使用DMA直接内存访问方式读取SPI/I2C数据合理设置采样频率避免不必要的高频采样对传感器数据进行滤波处理减少噪声数据处理优化使用NumPy/Pandas进行向量化运算对于复杂计算考虑使用Cython或Numba加速实现数据批处理减少数据库写入次数系统架构优化将数据采集、处理和存储分离到不同进程/机器使用消息队列如RabbitMQ解耦系统组件对于大规模部署考虑使用Kubernetes进行容器编排# 使用Numba加速数据处理示例 from numba import jit import numpy as np jit(nopythonTrue) def process_sensor_data(data): # 实现高效的数据处理逻辑 mean np.mean(data) std np.std(data) return mean, std # 使用示例 data np.random.normal(0, 1, 1000000) # 模拟100万个数据点 mean, std process_sensor_data(data) print(f均值: {mean:.4f}, 标准差: {std:.4f})通过树莓派和Python的组合我们实现了一个完整的工业传感器数据采集系统涵盖了从硬件连接到数据处理、存储、可视化和系统集成的全流程。这套方案不仅成本低廉而且灵活可扩展能够满足从简单实验室监控到小型工业自动化系统的各种需求。随着技术的不断进步树莓派的性能也在持续提升使得基于开源硬件的工业解决方案变得越来越可行。
从零到精：用树莓派+Python玩转工业传感器数据采集（压力/温度/光电）
发布时间：2026/5/15 13:44:24
从零到精用树莓派Python玩转工业传感器数据采集压力/温度/光电在工业自动化和物联网技术蓬勃发展的今天如何低成本、高效率地实现工业现场数据采集成为了许多技术人员关注的焦点。传统工业数据采集系统往往价格昂贵、配置复杂对于中小型企业、创客团队或个人开发者来说门槛较高。而树莓派这款信用卡大小的微型计算机配合Python这一简单易用的编程语言为我们提供了一条快速搭建工业数据采集系统的捷径。本文将带领读者从零开始逐步构建一个完整的工业传感器数据采集系统。我们将重点介绍三种常见工业传感器——压力传感器、温度传感器和光电传感器的连接方法、数据采集原理以及Python编程实现。通过实际案例演示读者将掌握如何将这些物理量转换为数字信号并进行可视化展示和数据存储。无论你是工业自动化领域的新手还是希望扩展技能的经验丰富的工程师这套基于开源硬件的解决方案都将为你打开工业物联网应用开发的大门。1. 硬件准备与系统搭建1.1 树莓派选型与系统配置树莓派有多种型号可供选择对于工业数据采集应用我们推荐使用树莓派4B或更新型号主要考虑其更强的处理能力、更大的内存和更丰富的接口。以下是各型号树莓派的对比型号CPU内存USB接口网络功耗价格3B1.4GHz四核1GB4×USB2.0千兆以太网5W中等4B1.5GHz四核2/4/8GB2×USB3.02×USB2.0千兆以太网7.5W较高Zero 2W1GHz四核512MB1×micro USB无线2.5W低系统方面我们推荐使用Raspberry Pi OS Lite版本无桌面环境这样可以节省系统资源。安装完成后需要进行以下基本配置# 更新系统 sudo apt update sudo apt upgrade -y # 安装必要工具 sudo apt install -y python3-pip git vim # 设置Python环境 pip3 install --upgrade pip1.2 传感器选型与接口工业传感器种类繁多接口标准也不尽相同。我们主要关注以下几种常见类型压力传感器如MPX5700系列模拟电压输出0-5V量程0-700kPa温度传感器如DS18B20数字接口1-Wire测量范围-55°C至125°C光电传感器如E3Z光电开关数字输出NPN/PNP对于模拟输出的传感器树莓派需要通过ADC模数转换器模块进行连接因为树莓派本身没有模拟输入引脚。推荐使用MCP3008这款8通道10位ADC芯片价格低廉且性能可靠。1.3 电路连接与安全注意事项工业现场环境复杂在连接传感器时需要注意以下事项电气隔离工业传感器工作电压可能较高如24V务必使用光耦或隔离模块保护树莓派信号调理对于模拟信号可能需要使用运算放大器进行信号调理接地处理避免形成地环路导致信号干扰防反接保护在电源线上串联二极管防止电源反接以下是MCP3008与树莓派的连接方式MCP3008 树莓派 VDD 3.3V VREF 3.3V AGND GND CLK GPIO11 (SCLK) DOUT GPIO9 (MISO) DIN GPIO10 (MOSI) CS/SHDN GPIO8 (CE0) DGND GND2. Python数据采集基础2.1 GPIO控制与SPI通信树莓派通过GPIO引脚与外部设备通信。Python中可以使用RPi.GPIO或gpiozero库来操作GPIO。对于SPI设备如MCP3008我们需要先启用SPI接口sudo raspi-config # 选择Interfacing Options - SPI - Yes安装必要的Python库pip3 install RPi.GPIO gpiozero spidev下面是一个读取MCP3008通道0的示例代码import spidev import time # 创建SPI对象 spi spidev.SpiDev() spi.open(0, 0) # 使用SPI0片选0 spi.max_speed_hz 1000000 # 设置SPI时钟频率 def read_adc(channel): # 确保通道号在0-7之间 if channel 0 or channel 7: return -1 # MCP3008的SPI通信协议 # 起始位 单端/差分选择 通道号高位 通道号低位 cmd 0b11000000 | (channel 3) resp spi.xfer2([cmd, 0x00, 0x00]) # 解析返回的10位数据 value ((resp[0] 0x01) 9) | (resp[1] 1) | (resp[2] 7) return value while True: value read_adc(0) # 读取通道0 voltage (value * 3.3) / 1023 # 转换为电压值 print(fADC值: {value}, 电压: {voltage:.2f}V) time.sleep(1)2.2 传感器数据处理与校准从传感器获取的原始数据通常需要经过处理和校准才能得到有意义的物理量。以MPX5700压力传感器为例其输出电压与压力呈线性关系压力(kPa) (Vout - 0.2V) × (700kPa / (4.7V - 0.2V))在Python中可以实现如下def pressure_from_voltage(voltage): # MPX5700压力传感器转换公式 return (voltage - 0.2) * (700 / 4.5) if voltage 0.2 else 0.0 # 使用示例 adc_value read_adc(0) voltage (adc_value * 3.3) / 1023 pressure pressure_from_voltage(voltage) print(f当前压力: {pressure:.2f} kPa)对于数字传感器如DS18B20温度传感器我们需要先启用1-Wire接口sudo raspi-config # 选择Interfacing Options - 1-Wire - Yes然后可以通过文件系统直接读取温度数据def read_ds18b20(): base_dir /sys/bus/w1/devices/ device_folder glob.glob(base_dir 28*)[0] device_file device_folder /w1_slave with open(device_file, r) as f: lines f.readlines() if lines[0].strip()[-3:] ! YES: return None equals_pos lines[1].find(t) if equals_pos -1: return None temp_str lines[1][equals_pos2:] temp_c float(temp_str) / 1000.0 return temp_c while True: temp read_ds18b20() if temp is not None: print(f当前温度: {temp:.2f}°C) time.sleep(1)2.3 多线程数据采集在实际应用中我们通常需要同时采集多个传感器的数据。这时可以使用Python的threading模块实现多线程采集import threading from queue import Queue class SensorReader(threading.Thread): def __init__(self, sensor_type, interval1, queueNone): super().__init__() self.sensor_type sensor_type self.interval interval self.queue queue self._stop_event threading.Event() def stop(self): self._stop_event.set() def run(self): while not self._stop_event.is_set(): if self.sensor_type pressure: value read_pressure() elif self.sensor_type temperature: value read_ds18b20() elif self.sensor_type photo: value read_photo_sensor() if self.queue: self.queue.put((self.sensor_type, value, time.time())) time.sleep(self.interval) # 使用示例 data_queue Queue() sensors [ SensorReader(pressure, 0.5, data_queue), SensorReader(temperature, 1, data_queue), SensorReader(photo, 0.2, data_queue) ] for sensor in sensors: sensor.start() try: while True: sensor_type, value, timestamp data_queue.get() print(f[{timestamp}] {sensor_type}: {value}) except KeyboardInterrupt: for sensor in sensors: sensor.stop() for sensor in sensors: sensor.join()3. 数据存储与可视化3.1 数据库存储方案对于工业数据采集系统可靠的数据存储至关重要。我们推荐使用轻量级的SQLite数据库或时序数据库InfluxDB。SQLite方案import sqlite3 from datetime import datetime def init_db(): conn sqlite3.connect(sensor_data.db) c conn.cursor() c.execute(CREATE TABLE IF NOT EXISTS sensor_data (id INTEGER PRIMARY KEY AUTOINCREMENT, sensor_type TEXT, value REAL, timestamp DATETIME)) conn.commit() conn.close() def save_to_db(sensor_type, value): conn sqlite3.connect(sensor_data.db) c conn.cursor() c.execute(INSERT INTO sensor_data (sensor_type, value, timestamp) VALUES (?, ?, ?), (sensor_type, value, datetime.now())) conn.commit() conn.close() # 初始化数据库 init_db() # 在数据采集线程中调用 save_to_db(pressure, 123.45)InfluxDB方案from influxdb import InfluxDBClient client InfluxDBClient(hostlocalhost, port8086) client.switch_database(sensor_data) def save_to_influx(sensor_type, value): json_body [ { measurement: sensor_type, fields: { value: value } } ] client.write_points(json_body) # 使用示例 save_to_influx(temperature, 25.6)3.2 实时数据可视化Python生态系统提供了丰富的数据可视化工具如Matplotlib、Plotly和PyQtGraph。对于实时数据显示PyQtGraph性能最佳import pyqtgraph as pg from pyqtgraph.Qt import QtGui import numpy as np app QtGui.QApplication([]) win pg.GraphicsWindow(title实时传感器数据) # 创建三个图表 pressure_plot win.addPlot(title压力 (kPa)) temp_plot win.addPlot(title温度 (°C)) photo_plot win.addPlot(title光电传感器) # 初始化曲线 pressure_curve pressure_plot.plot(penr) temp_curve temp_plot.plot(penb) photo_curve photo_plot.plot(peng) # 数据缓冲区 data_len 100 pressure_data np.zeros(data_len) temp_data np.zeros(data_len) photo_data np.zeros(data_len) ptr 0 def update(): global ptr, pressure_data, temp_data, photo_data # 模拟数据更新 pressure_data[ptr] read_pressure() temp_data[ptr] read_ds18b20() photo_data[ptr] read_photo_sensor() ptr 1 if ptr data_len: ptr 0 # 更新曲线 pressure_curve.setData(pressure_data) temp_curve.setData(temp_data) photo_curve.setData(photo_data) # 自动调整Y轴范围 pressure_plot.enableAutoRange(y, True) temp_plot.enableAutoRange(y, True) photo_plot.enableAutoRange(y, True) timer pg.QtCore.QTimer() timer.timeout.connect(update) timer.start(50) # 更新间隔(ms) QtGui.QApplication.instance().exec_()3.3 Web界面与远程监控对于需要远程监控的场景可以搭建一个简单的Web界面。使用Flask框架可以快速实现from flask import Flask, render_template, jsonify from threading import Thread import time app Flask(__name__) # 模拟传感器数据 current_data { pressure: 0.0, temperature: 0.0, photo: 0.0 } def sensor_loop(): while True: current_data[pressure] read_pressure() current_data[temperature] read_ds18b20() current_data[photo] read_photo_sensor() time.sleep(1) app.route(/) def index(): return render_template(index.html) app.route(/data) def get_data(): return jsonify(current_data) if __name__ __main__: # 启动传感器线程 Thread(targetsensor_loop, daemonTrue).start() app.run(host0.0.0.0, port8080)对应的HTML模板templates/index.html!DOCTYPE html html head title传感器监控面板/title script srchttps://cdn.plot.ly/plotly-latest.min.js/script style .gauge { width: 300px; height: 200px; display: inline-block; } .chart { width: 100%; height: 300px; } /style /head body h1工业传感器监控系统/h1 div idpressureGauge classgauge/div div idtempGauge classgauge/div div idphotoGauge classgauge/div div idhistoryChart classchart/div script // 初始化仪表盘 function createGauge(elementId, title, min, max) { return { domain: { row: 0, column: 0 }, value: 0, title: { text: title }, type: indicator, mode: gaugenumber, gauge: { axis: { range: [min, max] }, bar: { color: blue } } }; } const pressureGauge createGauge(pressureGauge, 压力 (kPa), 0, 700); const tempGauge createGauge(tempGauge, 温度 (°C), -20, 100); const photoGauge createGauge(photoGauge, 光电传感器, 0, 1023); Plotly.newPlot(pressureGauge, [pressureGauge]); Plotly.newPlot(tempGauge, [tempGauge]); Plotly.newPlot(photoGauge, [photoGauge]); // 历史数据图表 const historyChart { y: [], type: scatter, name: 压力历史 }; Plotly.newPlot(historyChart, [historyChart]); // 定时更新数据 function updateData() { fetch(/data) .then(response response.json()) .then(data { // 更新仪表盘 Plotly.update(pressureGauge, { value: data.pressure }); Plotly.update(tempGauge, { value: data.temperature }); Plotly.update(photoGauge, { value: data.photo }); // 更新历史图表 historyChart.y.push(data.pressure); if (historyChart.y.length 50) { historyChart.y.shift(); } Plotly.update(historyChart, { y: [historyChart.y] }); }); } setInterval(updateData, 1000); /script /body /html4. 高级应用与系统集成4.1 工业通信协议集成在工业环境中Modbus和OPC UA是两种最常见的通信协议。我们可以使用Python库来实现对这些协议的支持。Modbus RTU/TCP集成from pymodbus.client import ModbusSerialClient as ModbusClient # 初始化Modbus RTU客户端 modbus_rtu ModbusClient( methodrtu, port/dev/ttyUSB0, baudrate9600, timeout1 ) # 连接设备 if modbus_rtu.connect(): # 读取保持寄存器 result modbus_rtu.read_holding_registers(address0, count10, slave1) if not result.isError(): print(Modbus读取结果:, result.registers) else: print(Modbus读取错误:, result) modbus_rtu.close()OPC UA集成from opcua import Client opcua_url opc.tcp://localhost:4840 client Client(opcua_url) try: client.connect() # 获取节点 root client.get_root_node() objects client.get_objects_node() # 读取变量 temp_node client.get_node(ns2;i2) temperature temp_node.get_value() print(OPC UA温度值:, temperature) finally: client.disconnect()4.2 异常检测与预警系统工业数据采集系统的一个重要功能是异常检测和预警。我们可以使用简单的统计方法或机器学习算法来实现import numpy as np from sklearn.ensemble import IsolationForest # 历史数据 pressure_history np.loadtxt(pressure_history.csv) # 训练异常检测模型 clf IsolationForest(contamination0.05) clf.fit(pressure_history.reshape(-1, 1)) def check_anomaly(value): # 预测是否为异常-1表示异常 return clf.predict([[value]])[0] -1 # 使用示例 current_pressure read_pressure() if check_anomaly(current_pressure): print(f警告检测到异常压力值: {current_pressure}) # 触发警报如发送邮件或短信4.3 与工业云平台集成将本地采集的数据上传到工业云平台可以实现更高级的分析和远程监控。以阿里云IoT平台为例from aliyunIoT import Device # 设备认证信息 product_key your_product_key device_name your_device_name device_secret your_device_secret # 初始化设备 device Device(product_key, device_name, device_secret) # 属性上报回调 def on_props(request, response): print(属性上报结果:, response) # 连接云端 device.connect() # 上报传感器数据 def report_to_cloud(pressure, temp, photo): props { Pressure: pressure, Temperature: temp, PhotoSensor: photo } device.postProps(props, on_props) # 在数据采集循环中调用 while True: pressure read_pressure() temp read_ds18b20() photo read_photo_sensor() report_to_cloud(pressure, temp, photo) time.sleep(60) # 每分钟上报一次4.4 系统优化与性能调优随着采集点增多系统性能可能成为瓶颈。以下是一些优化建议数据采集优化使用DMA直接内存访问方式读取SPI/I2C数据合理设置采样频率避免不必要的高频采样对传感器数据进行滤波处理减少噪声数据处理优化使用NumPy/Pandas进行向量化运算对于复杂计算考虑使用Cython或Numba加速实现数据批处理减少数据库写入次数系统架构优化将数据采集、处理和存储分离到不同进程/机器使用消息队列如RabbitMQ解耦系统组件对于大规模部署考虑使用Kubernetes进行容器编排# 使用Numba加速数据处理示例 from numba import jit import numpy as np jit(nopythonTrue) def process_sensor_data(data): # 实现高效的数据处理逻辑 mean np.mean(data) std np.std(data) return mean, std # 使用示例 data np.random.normal(0, 1, 1000000) # 模拟100万个数据点 mean, std process_sensor_data(data) print(f均值: {mean:.4f}, 标准差: {std:.4f})通过树莓派和Python的组合我们实现了一个完整的工业传感器数据采集系统涵盖了从硬件连接到数据处理、存储、可视化和系统集成的全流程。这套方案不仅成本低廉而且灵活可扩展能够满足从简单实验室监控到小型工业自动化系统的各种需求。随着技术的不断进步树莓派的性能也在持续提升使得基于开源硬件的工业解决方案变得越来越可行。
