自动化通讯协议列表：从原理到实战的全方位指南

import paho.mqtt.client as mqtt
import time
 
# 配置参数
BROKER = "mqtt.example.com"
PORT = 8883  # TLS端口
TOPIC = "industrial/sensor/temperature"
QOS = 1
 
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("与MQTT Broker连接成功")
    else:
        print(f"连接失败，错误码: {rc}")
 
# 初始化客户端并启用TLS
client = mqtt.Client()
client.on_connect = on_connect
client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
client.username_pw_set("device001", "secure_password")
client.connect(BROKER, PORT, 60)
 
# 模拟传感器数据发布
client.loop_start()
try:
    while True:
        temperature = 25.5 + (time.time() % 10 -5)*0.5  # 模拟温度波动
        payload = f'{{"temperature": {temperature:.1f}, "timestamp": {int(time.time())}}}'
        result = client.publish(TOPIC, payload, qos=QOS)
        result.wait_for_publish()
        print(f"发布数据: {payload}")
        time.sleep(5)
except KeyboardInterrupt:
    print("停止发布")
finally:
    client.loop_stop()
    client.disconnect()

from opcua import Client
 
# 连接到OPC UA服务器
url = "opc.tcp://opcua.example.com:4840/freeopcua/server/"
client = Client(url)
client.set_user("admin")
client.set_password("opcua_pass")
client.connect()
 
# 获取根节点与Objects节点
root = client.get_root_node()
print(f"根节点: {root}")
 
# 读取自定义变量
temp_node = client.get_node("ns=2;s=TemperatureSensor.Value")
temp_value = temp_node.get_value()
print(f"当前温度: {temp_value} °C")
 
client.disconnect()

from pymodbus.client import ModbusTcpClient
 
# 连接到Modbus TCP服务器
client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()
 
# 读取保持寄存器（地址0，数量2）
result = client.read_holding_registers(address=0, count=2, unit=1)
if not result.isError():
    print(f"保持寄存器值: {result.registers}")
else:
    print(f"读取失败: {result}")
 
# 写入保持寄存器（地址0，值100）
result = client.write_register(address=0, value=100, unit=1)
if not result.isError():
    print("写入成功")
else:
    print(f"写入失败: {result}")
 
client.close()

import requests
import json
 
# 从REST API获取传感器数据
url = "https://api.example.com/v1/sensors/1"
headers = {"Authorization": "Bearer YOUR_TOKEN"}
response = requests.get(url, headers=headers)
 
if response.status_code == 200:
    data = response.json()
    print(f"传感器数据: {data}")
else:
    print(f"请求失败: {response.status_code}")
 
# 发送控制指令到设备
payload = {"command": "start", "speed": 50}
response = requests.post("https://api.example.com/v1/devices/1/control", 
                         headers=headers, json=payload)
print(f"指令发送结果: {response.json()}")

import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient
import json
 
# MQTT配置
BROKER = "mqtt.example.com"
PORT = 8883
TOPIC = "industrial/line1/+/status"
USERNAME = "monitor_user"
PASSWORD = "monitor_pass"
 
# InfluxDB配置
INFLUXDB_HOST = "localhost"
INFLUXDB_PORT = 8086
INFLUXDB_DB = "industrial_monitor"
 
# 初始化InfluxDB客户端
influx_client = InfluxDBClient(INFLUXDB_HOST, INFLUXDB_PORT, database=INFLUXDB_DB)
influx_client.create_database(INFLUXDB_DB)
 
def on_connect(client, userdata, flags, rc):
    print(f"连接成功，错误码: {rc}")
    client.subscribe(TOPIC, qos=1)
 
def on_message(client, userdata, msg):
    topic_parts = msg.topic.split('/')
    machine_id = topic_parts[2]
    data = json.loads(msg.payload.decode())
    
    # 构造InfluxDB数据点
    points = [
        {
            "measurement": "temperature",
            "tags": {"machine_id": machine_id},
            "fields": {"value": data["temperature"]},
            "time": data["timestamp"]
        },
        {
            "measurement": "vibration",
            "tags": {"machine_id": machine_id},
            "fields": {"value": data["vibration"]},
            "time": data["timestamp"]
        }
    ]
    
    influx_client.write_points(points)
    print(f"存储数据: {machine_id} - {data}")
 
# 初始化MQTT客户端
client = mqtt.Client()
client.username_pw_set(USERNAME, PASSWORD)
client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, 60)
 
client.loop_forever()