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

手把手教你用Python解析MBUS水表数据（CJ/T 188协议实战）

news 2026/5/13 11:01:46

手把手教你用Python解析MBUS水表数据（CJ/T 188协议实战）

当你第一次拿到支持MBUS总线的智能水表时，可能会被那些十六进制数据帧搞得一头雾水。别担心，本文将带你从零开始，用Python一步步实现水表数据的读取和解析。无论你是物联网开发者还是嵌入式工程师，都能通过这个实战项目掌握MBUS通信的核心技能。

MBUS（Meter-Bus）是欧洲广泛使用的仪表总线标准，而CJ/T 188则是国内水表通信的行业标准协议。通过RS-485接口与这些智能水表通信，我们可以获取用水量、设备地址等关键信息。下面让我们从硬件连接到软件实现，构建一个完整的解决方案。

1. 硬件准备与基础配置

1.1 硬件连接指南

你需要准备以下硬件组件：

支持MBUS协议的智能水表
USB转RS-485转换器（如FT232芯片的转换模块）
双绞线（建议使用屏蔽双绞线以减少干扰）

连接步骤：

将USB转RS-485模块连接到电脑
确认水表的MBUS接口定义（通常为A/B两线制）
连接转换器的A/B线到水表对应接口
确保共地连接（如有必要）

注意：MBUS总线通常采用主从架构，一个主设备可以连接多个从设备。在接线时要注意总线终端是否需要匹配电阻。

1.2 串口参数设置

MBUS通信的标准参数如下：

参数	值
波特率	2400
数据位	8
校验位	偶校验
停止位	1
流控	无

在Python中，我们可以使用pyserial库来配置这些参数：

import serial ser = serial.Serial( port='/dev/ttyUSB0', # 根据实际设备修改 baudrate=2400, bytesize=serial.EIGHTBITS, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, timeout=1 )

2. MBUS协议帧结构解析

2.1 基本帧结构

MBUS协议的数据帧遵循特定格式，主要包含以下部分：

起始符：通常为0x68
地址域：标识从设备地址
控制码：定义操作类型（读/写）
数据长度：后续数据域的字节数
数据域：实际传输的数据
校验和：用于错误检测
结束符：通常为0x16

一个典型的查询帧结构如下：

FE FE FE 68 AA AA AA AA AA AA AA AA 03 03 81 0A 00 49 16

2.2 控制码详解

控制码（CTR）决定了操作的类型和方向：

控制码	含义	方向
0x01	读数据	主站到从站
0x03	读地址	主站到从站
0x81	从站响应读数据	从站到主站
0x83	从站响应读地址	从站到主站

控制码的最后一位表示传输方向：0表示主站发出，1表示从站发出。

3. Python实现MBUS通信

3.1 构建查询帧

让我们从最简单的地址查询开始。以下是构建查询地址帧的函数：

def build_address_query_frame(): # 基本地址查询帧 frame = [ 0xFE, 0xFE, 0xFE, 0x68, # 前导码和起始符 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, # 广播地址 0x03, # 数据长度 0x03, # 控制码（读地址） 0x81, 0x0A, 0x00, # 数据标识 ] # 计算校验和（从起始符到数据标识的累加和） checksum = sum(frame[3:]) frame.append(checksum & 0xFF) frame.append(0x16) # 结束符 return bytes(frame)

3.2 发送和接收数据

有了查询帧，我们可以实现数据的发送和接收：

def query_mbus_address(ser): query_frame = build_address_query_frame() ser.write(query_frame) # 等待响应（根据实际情况调整超时） response = ser.read(100) # 读取最多100字节 if not response: raise TimeoutError("设备未响应") return response

3.3 解析响应帧

收到响应后，我们需要解析其中的有用信息。以下是一个响应帧解析示例：

def parse_address_response(response): # 检查基本帧结构 if len(response) < 20 or response[0] != 0xFE or response[-1] != 0x16: raise ValueError("无效的响应帧") # 提取地址信息 address_bytes = response[4:11] manufacturer_code = address_bytes[-2:] device_address = address_bytes[:-2] # 将地址转换为可读格式 address_str = ''.join(f"{b:02X}" for b in reversed(device_address)) return { 'manufacturer_code': manufacturer_code, 'device_address': device_address, 'address_string': address_str }

4. 读取和解析用水量数据

4.1 构建用水量查询帧

要读取水表的累计流量，我们需要构建特定的查询帧：

def build_water_usage_query_frame(address): # 使用设备地址构建查询帧 frame = [ 0xFE, 0xFE, 0xFE, 0x68, # 前导码和起始符 *address, # 设备地址 0x01, # 控制码（读数据） 0x03, # 数据长度 0x90, 0x1F, 0x00, # 数据标识（1F90表示累计流量） ] # 计算校验和 checksum = sum(frame[3:]) frame.append(checksum & 0xFF) frame.append(0x16) # 结束符 return bytes(frame)

4.2 解析用水量数据

水表返回的用水量数据通常采用BCD编码或二进制格式。以下是解析示例：

def parse_water_usage(response): # 检查基本帧结构 if len(response) < 20 or response[0] != 0xFE or response[-1] != 0x16: raise ValueError("无效的响应帧") # 定位数据域（示例帧中的累计流量数据） # FE FE FE 68 10 18 02 12 20 20 00 00 81 16 90 1F 00 00 02 00 00 2C 00 02 00 00 2C 00 00 00 00 00 00 00 00 FF 85 16 # 累计流量数据位于特定位置（根据协议可能变化） usage_data = response[15:19] # 00 02 00 00 # 将字节转换为数值（小端序） value = (usage_data[3] << 24) | (usage_data[2] << 16) | (usage_data[1] << 8) | usage_data[0] # 转换为吨（假设最后两位是小数位） tons = value / 100 return tons

5. 完整示例与常见问题

5.1 完整Python脚本

结合以上部分，我们得到一个完整的MBUS水表数据读取脚本：

import serial import time class MBUSWaterMeterReader: def __init__(self, port='/dev/ttyUSB0'): self.ser = serial.Serial( port=port, baudrate=2400, bytesize=serial.EIGHTBITS, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE, timeout=1 ) def build_frame(self, address, ctrl, data_id): frame = [ 0xFE, 0xFE, 0xFE, 0x68, *address, ctrl, len(data_id), *data_id ] checksum = sum(frame[3:]) frame.append(checksum & 0xFF) frame.append(0x16) return bytes(frame) def query_device(self, frame): self.ser.write(frame) time.sleep(0.5) # 等待响应 return self.ser.read(100) def get_address(self): broadcast_address = [0xAA]*7 frame = self.build_frame(broadcast_address, 0x03, [0x81, 0x0A, 0x00]) response = self.query_device(frame) return self.parse_address(response) def get_water_usage(self, address): frame = self.build_frame(address, 0x01, [0x90, 0x1F, 0x00]) response = self.query_device(frame) return self.parse_water_usage(response) def parse_address(self, response): if len(response) < 20 or response[0] != 0xFE or response[-1] != 0x16: raise ValueError("Invalid response frame") return response[4:11] def parse_water_usage(self, response): if len(response) < 20 or response[0] != 0xFE or response[-1] != 0x16: raise ValueError("Invalid response frame") usage_data = response[15:19] value = (usage_data[3] << 24) | (usage_data[2] << 16) | (usage_data[1] << 8) | usage_data[0] return value / 100 def close(self): self.ser.close() # 使用示例 if __name__ == "__main__": reader = MBUSWaterMeterReader() try: address = reader.get_address() print(f"Device address: {address.hex()}") usage = reader.get_water_usage(address) print(f"Water usage: {usage:.2f} tons") finally: reader.close()