当前位置：首页 > news >正文

硬件成本5元-USB串口采集电表数据完整方案-ThingsPanel快速入门

news 2025/7/4 4:09:29

ThingsPanel开源物联网平台支持广泛的协议，灵活自由，本文介绍ThingsPanel通过串口来采集电表数据，简单易行，成本低廉，适合入门者学习试验，也适合一些特定的应用场景做数据采集。

适用场景：

降低Modbus设备数据采集成本
快速测试Modbus电表数据采集
物联网入门学习Modbus相关知识

效果图

方案价值

这套方案最大的价值在于：

最低代码改动，复制修改配置即可运行
硬件成本低，5块钱，一个USB转RS485连接器即可
能快速实现电表数据远程监控，适用于学习、以及正式用途等各种场景
支持数据可视化和历史记录查询
无需布置复杂的网络，只需要一台电脑和网络即可
接入结构图

所需设备

USB转RS485连接器（淘宝搜索"USB转485"，5元左右）
支持Modbus协议的电表
一台电脑（Windows/Mac都可以）

操作步骤

1. 硬件连接

将USB转RS485连接器插入电脑
将连接器的A+和B-接线端子与电表对应端子连接
A+ 接电表的 A+
B- 接电表的 B-

2. 确认设备端口

Windows用户：

打开设备管理器
查看"端口(COM和LPT)"下的COM端口号

Mac用户：

打开终端
输入命令：ls /dev/tty.*
找到类似 /dev/tty.usbserial-110 的设备名

3. 安装Python环境

访问 https://www.python.org/downloads/
下载并安装Python 3.x版本

4. 安装必要的软件包

打开终端或命令提示符，运行：

pip install pyserial pip install pymodbus==3.7.2 pip install paho-mqtt

5. 创建并运行采集程序

创建一个新文件 meter_collector.py
将以下代码复制到文件中（注意修改端口号）：

===================代码区域开始=====================

import time
import json
from pymodbus.client.serial import ModbusSerialClient
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder
import paho.mqtt.client as mqtt

# MQTT配置-信息来自ThingsPanel平台，创建设备后在设备连接信息中。
MQTT_CONFIG = {
"broker": "47.115.210.16",
"port": 1883,
"username": "fe917b16-d786-3297-8f1",
"password": "f129542",
"client_id": "mqtt_3243936a-c23",
"telemetry_topic": "devices/telemetry",
"control_topic": "devices/telemetry/control/3243936a-c237-8cc5-7ea9-ba586098c2b7"
}

# Modbus配置（来自电表说明书，port需要自行在本机查询）
METER_CONFIG = {
"port": "/dev/tty.usbserial-110",
"baudrate": 9600,
"parity": 'E',
"stopbits": 1,
"bytesize": 8,
"byte_order": Endian.BIG
}
# 如下配置来自于电表设备说明书
DATA_ITEMS = [
{"name": "Voltage", "address": 0, "data_type": "int16-1", "factor": 0.1, "registers": 1},
{"name": "Current", "address": 3, "data_type": "int16-1", "factor": 0.01, "registers": 2},
{"name": "TotalActivePower", "address": 7, "data_type": "int16-1", "factor": 1, "registers": 1},
{"name": "TotalReactivePower", "address": 11, "data_type": "int16-1", "factor": 1, "registers": 1},
{"name": "TotalApparentPower", "address": 15, "data_type": "int16-1", "factor": 1, "registers": 1},
{"name": "TotalPowerFactor", "address": 19, "data_type": "int16-1", "factor": 0.001, "registers": 1},
{"name": "VoltageFrequency", "address": 26, "data_type": "int16-1", "factor": 0.01, "registers": 1}
]

# MQTT回调函数
def on_connect(client, userdata, flags, rc):
print(f"Connected to MQTT broker with result code {rc}")
# 订阅控制主题
client.subscribe(MQTT_CONFIG['control_topic'])

def on_message(client, userdata, msg):
try:
print(f"Received message on topic {msg.topic}: {msg.payload.decode()}")
# 这里可以添加处理接收到的控制命令的逻辑
data = json.loads(msg.payload.decode())
if 'switch' in data:
print(f"Received switch command: {data['switch']}")
except Exception as e:
print(f"Error processing message: {e}")

def create_mqtt_client():
client = mqtt.Client(client_id=MQTT_CONFIG['client_id'])
client.username_pw_set(MQTT_CONFIG['username'], MQTT_CONFIG['password'])
client.on_connect = on_connect
client.on_message = on_message

try:
client.connect(MQTT_CONFIG['broker'], MQTT_CONFIG['port'], 60)
client.loop_start()
return client
except Exception as e:
print(f"MQTT连接失败: {e}")
return None

# Modbus函数（与之前相同）
def create_modbus_client():
return ModbusSerialClient(
port=METER_CONFIG['port'],
baudrate=METER_CONFIG['baudrate'],
parity=METER_CONFIG['parity'],
stopbits=METER_CONFIG['stopbits'],
bytesize=METER_CONFIG['bytesize'],
timeout=1,
retries=3
)

def read_meter_data(client, slave_id):
data = {}
for item in DATA_ITEMS:
try:
result = client.read_holding_registers(
address=item['address'],
count=item['registers'],
slave=slave_id
)
if result and not result.isError():
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers,
byteorder=METER_CONFIG['byte_order'],
wordorder=Endian.BIG
)
if item['data_type'] == 'int16-1':
value = decoder.decode_16bit_int()
else:
value = decoder.decode_16bit_uint()
data[item['name']] = value * item['factor']
else:
data[item['name']] = None
except Exception as e:
print(f"读取 {item['name']} 时出错: {str(e)}")
data[item['name']] = None
return data

def main():
print("正在连接Modbus设备...")
modbus_client = create_modbus_client()
if not modbus_client.connect():
print("无法连接到Modbus设备")
return

print("正在连接MQTT服务器...")
mqtt_client = create_mqtt_client()
if not mqtt_client:
print("无法连接到MQTT服务器")
modbus_client.close()
return

print("开始数据采集和上报循环...")
try:
while True:
data = read_meter_data(modbus_client, slave_id=1)

# 过滤掉读取失败的数据
valid_data = {k: v for k, v in data.items() if v is not None}

if valid_data:
# 发布数据到MQTT
try:
mqtt_client.publish(
MQTT_CONFIG['telemetry_topic'],
json.dumps(valid_data)
)
print(f"数据已上报: {valid_data}")
except Exception as e:
print(f"MQTT发布失败: {e}")

# 等待10秒
time.sleep(10)

except KeyboardInterrupt:
print("\n程序终止...")
finally:
print("正在关闭连接...")
modbus_client.close()
mqtt_client.loop_stop()
mqtt_client.disconnect()
print("连接已关闭")

if __name__ == "__main__":
main()