好的，针对您提出的“Python采集三菱PLC数据并实现数据可视化与数据库存储的完整案例推荐”这一需求，我将基于提供的【参考资料】进行解构和推演，为您提供一个集成了数据采集、数据库存储和可视化的综合案例方案。 ### **案例概述：Python + 三菱PLC + InfluxDB + Grafana 实时监控系统** 本案例将展示如何使用Python与三菱PLC（以MELSEC Q系列为例，使用MC协议）进行通信，将采集到的数据存入时序数据库InfluxDB，并通过可视化工具Grafana创建实时监控看板。 ### **一、 系统架构与技术栈** 该方案的架构清晰，各组件职责明确，具体如下表所示： | 组件 | 技术选型 | 角色与作用 | | :--- | :--- | :--- | | **数据源** | 三菱MELSEC Q系列 PLC | 生产设备或产线控制器，提供原始运行数据（如温度、速度、运行状态）[ref_4]。 | | **数据采集层** | Python (库：`pymcprotocol`) | 通过以太网，使用三菱MC协议（Ethernet）与PLC通信，执行数据读取和写入[ref_3][ref_4]。 | | **数据传输/缓冲** | （可选）MQTT | 在复杂网络或多PLC场景下，可使用MQTT实现数据的异步发布/订阅，提高系统解耦度[ref_5][ref_6]。 | | **数据存储层** | InfluxDB | 专门为时序数据（如随时间变化的设备参数）设计的高性能数据库，支持高吞吐量写入和高效的时间范围查询[ref_4]。 | | **数据可视化层** | Grafana | 功能强大的开源数据可视化平台，能轻松连接InfluxDB等多种数据源，创建丰富的实时图表和看板[ref_4]。 | | **应用/监控服务** | Python Flask/Django | （可选）用于构建简单的Web监控页面或提供RESTful API，实现更复杂的业务逻辑[ref_1][ref_5]。 | ### **二、 核心环节实现案例** #### **1. Python 采集三菱PLC数据** 三菱PLC通信常用MC协议（MELSEC Communication Protocol）。在Python中，可以使用 `pymcprotocol` 库。 * **安装依赖库**: ```bash pip install pymcprotocol influxdb-client ``` * **Python 数据采集脚本示例 (`plc_collector.py`)**: 此脚本负责周期性地从PLC读取数据，并写入InfluxDB。 ```python import time import struct from pymcprotocol import Type3E # 针对Q系列3E帧 from influxdb_client import InfluxDBClient, Point from influxdb_client.client.write_api import SYNCHRONOUS # --- 1. 初始化PLC连接 --- # 创建Type3E实例（对应三菱Q系列MC协议3E帧） plc = Type3E() # 设置PLC的IP地址和端口（默认为4999） plc.connect("192.168.1.10", 5002) # 请替换为您的PLC实际IP[ref_4] # --- 2. 初始化InfluxDB连接 --- # 配置InfluxDB 2.x连接信息 influx_url = "http://localhost:8086" influx_token = "your_admin_token" influx_org = "your_org" influx_bucket = "plc_data_bucket" client = InfluxDBClient(url=influx_url, token=influx_token, org=influx_org) write_api = client.write_api(write_options=SYNCHRONOUS) # --- 3. 定义数据点标签和采集地址 --- device_id = "CNC_Machine_01" # 设备标识 # 假设从PLC的D100开始读取4个寄存器：D100（温度）， D101（压力）， D102（速度）， D103（状态） start_address = "D100" num_registers = 4 try: while True: # --- 4. 从PLC批量读取数据 --- # batchread_wordunits可以一次性读取多个字寄存器 values = plc.batchread_wordunits(headdevice=start_address, readsize=num_registers) # 解析读取到的值 # 注意：返回值可能是整数列表，具体含义需要根据PLC程序确定 temperature = values[0] / 10.0 # 假设D100存储温度，实际值为整数，需除以10 pressure = values[1] # D101存储压力值 speed = values[2] # D102存储速度值 status_code = values[3] # D103存储状态码 # --- 5. 构建InfluxDB数据点并写入 --- # 使用Point构造数据点，包含measurement（类似表名）、tag（索引字段）、field（数值字段）和时间戳 point = Point("machine_metrics") \ .tag("device_id", device_id) \ .field("temperature", temperature) \ .field("pressure", pressure) \ .field("speed", speed) \ .field("status", status_code) \ .time(time.time_ns()) # 使用纳秒时间戳 write_api.write(bucket=influx_bucket, record=point) print(f"数据已写入: Temp={temperature}, Pressure={pressure}, Speed={speed}, Status={status_code}") # --- 6. 等待下一个采集周期 --- time.sleep(1) # 每秒采集一次，可根据实际需求调整 except KeyboardInterrupt: print("程序被用户中断。") except Exception as e: print(f"发生错误: {e}") finally: # --- 7. 清理连接 --- plc.close() client.close() print("连接已关闭。") ``` * **关键说明**： 1. **协议与库选择**：三菱PLC型号众多，通信协议可能为MC、SLMP等，`pymcprotocol`是常用选择[ref_4]。务必根据PLC型号和设置选择正确的帧类型（如Type3E）。 2. **地址映射**：`D100`是软元件地址，需要与PLC程序员的地址规划完全一致[ref_2]。 3. **数据解析**：从PLC读取的通常是16位或32位整数，需要根据实际工程单位（如温度有小数）进行标度转换。 4. **错误处理**：工业现场网络不稳定，必须包含完善的异常处理（如重连机制），本例仅为演示[ref_1][ref_3]。 #### **2. 数据持久化到数据库** 本案例选用 **InfluxDB**，因为它专为工业物联网场景下的海量时序数据设计，写入和查询性能远超传统关系型数据库（如MySQL）[ref_4]。 * **InfluxDB 安装与配置 (Docker方式)**: ```bash # 拉取并运行InfluxDB 2.x docker run -d -p 8086:8086 \ --name influxdb \ -v $PWD/influxdb2:/var/lib/influxdb2 \ influxdb:latest ``` 启动后，访问 `http://localhost:8086` 完成首次初始化，创建`组织(Org)`、`桶(Bucket)`并生成访问令牌(Token)，这些信息需要填入上述Python脚本中。 * **数据模型优势**： InfluxDB的 `Measurement` (指标，如 `machine_metrics`)、`Tag` (标签，如 `device_id`，用于高效过滤和分组)、`Field` (字段，如 `temperature`，存储实际值) 和 `Timestamp` (时间戳) 模型，完美契合设备数据“带时间戳的键值对”特性[ref_4]。 #### **3. 数据可视化展示** 使用 **Grafana** 连接 InfluxDB，创建动态监控看板。 * **Grafana 安装 (Docker方式)**: ```bash docker run -d -p 3000:3000 --name grafana grafana/grafana-enterprise ``` 启动后，访问 `http://localhost:3000`，默认账号密码为 `admin/admin`。 * **配置数据源与创建看板**： 1. **添加数据源**：在Grafana界面，添加 `InfluxDB` 数据源，选择 `Flux` 查询语言，正确填写URL、Token、Org和Bucket信息。 2. **创建仪表板(Dashboard)**：新建一个仪表板。 3. **添加面板(Panel)**：在面板中，通过编写Flux查询语句来获取数据。例如，查询最近1小时某台设备的温度曲线： ```flux from(bucket: "plc_data_bucket") |> range(start: -1h) |> filter(fn: (r) => r["_measurement"] == "machine_metrics") |> filter(fn: (r) => r["device_id"] == "CNC_Machine_01") |> filter(fn: (r) => r["_field"] == "temperature") |> aggregateWindow(every: 10s, fn: mean, createEmpty: false) ``` 4. **选择可视化类型**：为查询结果选择合适的图表，如“时间序列图(Time series)”用于趋势展示，“仪表盘(Gauge)”用于当前值显示，“状态图(Stat)”用于显示状态码对应的文本（可通过值映射实现）[ref_5]。 5. **设置刷新频率**：将整个仪表板的刷新频率设置为 `5s` 或 `10s`，即可实现**准实时**监控。 ### **三、 案例扩展与进阶建议** 1. **多PLC与高并发采集**：对于需要连接数十上百台PLC的场景，上述简单循环脚本无法满足性能要求。应使用 **异步IO（asyncio）** 或多线程/多进程架构，并考虑引入消息队列（如 **RabbitMQ** 或 **Kafka**）作为数据总线，实现采集、处理和存储的解耦，提升系统稳定性和扩展性[ref_1][ref_3]。 2. **协议扩展**：若三菱PLC支持 **OPC UA**（新一代工业标准通信协议），强烈建议使用OPC UA。Python的 `opcua` 库提供了更标准化、更安全（内置加密）的访问方式，能获取更丰富的数据类型和结构信息，是未来集成的方向[ref_1][ref_2][ref_5]。 3. **边缘计算**：可在靠近PLC的工控机或网关设备上部署上述Python程序，先进行本地数据清洗、滤波和边缘告警，再将结果数据上报给中心的InfluxDB，这符合工业4.0的边云协同架构[ref_1][ref_4]。 4. **系统集成**：本案例生成的数据流，可以很容易地被上层的 **MES** (制造执行系统) 或 **SCADA** (监控与数据采集系统) 通过InfluxDB的API或直接查询来消费，从而实现生产数据与管理系统的融合[ref_2]。 ### **总结** 通过组合 `Python` (数据采集)、`InfluxDB` (时序存储) 和 `Grafana` (可视化) 这套开源技术栈，您可以快速构建一个专业、高效且低成本的三菱PLC数据监控系统。该案例涵盖了从底层协议通信到上层数据展示的全链路，具有良好的可扩展性，可以根据实际项目需求进行裁剪和增强[ref_1][ref_3][ref_4][ref_5]。