效果图

方案价值

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

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

flowchart LR
    meter[电表\nModbus-RTU] --> |RS485| converter[USB转RS485\n连接器]
    converter --> |USB| computer[电脑\nPython程序]
    computer --> |MQTT\n10s间隔| cloud[ThingsPanel\n物联网平台]
    
    subgraph 现场设备
    meter
    end
    
    subgraph 数据采集
    converter
    computer
    end
    
    subgraph 云端服务
    cloud --> |数据存储| db[(数据库)]
    cloud --> |可视化| dashboard[数据看板]
    end

    classDef device fill:#e1f5fe,stroke:#01579b,stroke-width:2px
    classDef computer fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px
    classDef cloud fill:#fff3e0,stroke:#ef6c00,stroke-width:2px
    
    class meter,converter device
    class computer computer
    class cloud,db,dashboard cloud

所需设备

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()

6. 运行程序

打开终端或命令提示符
进入程序所在目录
运行命令：

python meter_collector.py

7. 查看数据

登录 ThingsPanel 平台
进入设备详情页面
可以看到实时上报的电表数据

常见问题

连接不上设备？

检查接线是否正确
确认端口号是否正确
验证电表波特率是否为9600

数据显示异常？

确认电表地址是否为1
检查接线是否松动

无法连接平台？

检查网络连接
验证MQTT配置信息是否正确

扩展功能

可以通过修改代码中的 time.sleep(10) 来调整数据上报频率
支持添加多个电表，只需修改 slave_id 即可

现在，您的电表数据就可以实时推送到云平台了！

效果图