206 lines
7.6 KiB
Python
206 lines
7.6 KiB
Python
import paho.mqtt.client as mqtt
|
||
import json
|
||
import logging
|
||
import taos
|
||
import binascii
|
||
from datetime import datetime
|
||
|
||
conn = taos.connect(host="123.6.102.119", user="readonly_user ", password="Aassword123", database="antsev.charge_jiuxing")
|
||
print("Connected successfully")
|
||
conn.close()
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
filename='parsed_data.log',
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
encoding='utf-8'
|
||
)
|
||
|
||
class ChargeDataParser:
|
||
def __init__(self, mqtt_host="123.6.102.119", mqtt_port=1883, mqtt_username="emqx_test", mqtt_password="emqx_test",
|
||
tdengine_host="localhost", tdengine_user="root", tdengine_password="taosdata"):
|
||
# MQTT 配置
|
||
self.mqtt_client = mqtt.Client(client_id="ChargeDataParser", protocol=mqtt.MQTTv311,
|
||
callback_api_version=mqtt.CallbackAPIVersion.VERSION1)
|
||
self.mqtt_client.username_pw_set(mqtt_username, mqtt_password)
|
||
self.mqtt_client.on_connect = self.on_connect
|
||
self.mqtt_client.on_message = self.on_message
|
||
self.mqtt_host = mqtt_host
|
||
self.mqtt_port = mqtt_port
|
||
self.connected = False
|
||
|
||
# TDengine 配置
|
||
self.tdengine_host = tdengine_host
|
||
self.tdengine_user = tdengine_user
|
||
self.tdengine_password = tdengine_password
|
||
self.tdengine_conn = None
|
||
self.tdengine_cursor = None
|
||
|
||
def connect(self):
|
||
"""连接 MQTT 和 TDengine"""
|
||
# 连接 MQTT
|
||
try:
|
||
self.mqtt_client.connect(self.mqtt_host, self.mqtt_port, 60)
|
||
self.mqtt_client.loop_start()
|
||
logging.info("Connected to MQTT broker")
|
||
except Exception as e:
|
||
logging.error(f"MQTT connection error: {str(e)}")
|
||
raise
|
||
|
||
# 连接 TDengine
|
||
try:
|
||
self.tdengine_conn = taos.connect(
|
||
host=self.tdengine_host,
|
||
user=self.tdengine_user,
|
||
password=self.tdengine_password,
|
||
database="tms_design"
|
||
)
|
||
self.tdengine_cursor = self.tdengine_conn.cursor()
|
||
logging.info("Connected to TDengine")
|
||
except Exception as e:
|
||
logging.error(f"TDengine connection error: {str(e)}")
|
||
raise
|
||
|
||
def on_connect(self, client, userdata, flags, rc):
|
||
if rc == 0:
|
||
self.connected = True
|
||
self.mqtt_client.subscribe("hejin/charging/log", qos=1)
|
||
logging.info("Subscribed to hejin/charging/log")
|
||
else:
|
||
logging.error(f"Failed to connect to MQTT broker with code: {rc}")
|
||
|
||
def on_message(self, client, userdata, msg):
|
||
"""处理接收到的 MQTT 消息"""
|
||
try:
|
||
payload = msg.payload.decode('utf-8')
|
||
data = json.loads(payload)
|
||
logging.info(f"Received message: {data}")
|
||
|
||
# 解析报文并存储
|
||
sql = self.parse_and_generate_sql(data)
|
||
if sql:
|
||
self.tdengine_cursor.execute(sql)
|
||
self.tdengine_conn.commit()
|
||
logging.info(f"Inserted into TDengine: {sql}")
|
||
except Exception as e:
|
||
logging.error(f"Error processing message: {str(e)}")
|
||
|
||
def hex_to_ascii(self, hex_str):
|
||
"""将 HEX 字符串转为 ASCII"""
|
||
try:
|
||
return bytes.fromhex(hex_str).decode('ascii').strip('\x00')
|
||
except:
|
||
return ""
|
||
|
||
def parse_bcd_time(self, bcd):
|
||
"""解析 BCD 码时间（格式：YYMMDDHHMMSS）"""
|
||
try:
|
||
year = f"20{bcd[0:2]}" # 假设 19 表示 2019
|
||
month = bcd[2:4]
|
||
day = bcd[4:6]
|
||
hour = bcd[6:8]
|
||
minute = bcd[8:10]
|
||
second = bcd[10:12]
|
||
return f"{year}-{month}-{day}T{hour}:{minute}:{second}+08:00"
|
||
except:
|
||
return None
|
||
|
||
def parse_and_generate_sql(self, data):
|
||
"""解析报文并生成插入 SQL"""
|
||
hex_data = data[3].replace(" ", "") # 移除空格
|
||
pile_id = data[4] # 桩号
|
||
time = data[2].replace(" ", "T") + "+08:00" # 时间
|
||
|
||
# 提取基本字段
|
||
if len(hex_data) < 30: # 最小长度：14字节（HEX表示为28字符）+校验码(2)
|
||
logging.warning(f"数据包长度不足: {hex_data}")
|
||
return None
|
||
|
||
company = hex_data[0:4] # 4A58
|
||
if company != "4A58":
|
||
logging.warning(f"无效帧起始: {company}")
|
||
return None
|
||
|
||
cmd = hex_data[4:6] # 命令码
|
||
length_str = hex_data[24:28] # 数据域长度
|
||
length = int(length_str, 16) * 2 # HEX字符数
|
||
data_domain = hex_data[28:28 + length] if length > 0 else ""
|
||
|
||
# 从桩号提取信息
|
||
operator_id = pile_id[0:4] # 运营商编号
|
||
station_id = pile_id[6:12] # 站点编号
|
||
cabinet_no = pile_id[12:16] # 站内桩地址（作为柜号）
|
||
|
||
# 初始化默认值
|
||
battery_id = "unknown"
|
||
battery_bun_id = "unknown"
|
||
battery_faults = ""
|
||
created_at = time
|
||
updated_at = time
|
||
soc = 0.0
|
||
|
||
# 根据命令码解析数据域
|
||
if cmd == "0B": # 平台心跳
|
||
if len(data_domain) >= 12: # 6字节时间 + 1字节超时次数
|
||
time_str = data_domain[0:12] # BCD码时间
|
||
parsed_time = self.parse_bcd_time(time_str)
|
||
if parsed_time:
|
||
created_at = parsed_time
|
||
updated_at = parsed_time
|
||
|
||
elif cmd == "25": # 充电信息（表 3.9.9）
|
||
if len(data_domain) >= 92: # 确保数据域足够长
|
||
# 充电用户编号（字节 0-15，ASCII）
|
||
battery_id = self.hex_to_ascii(data_domain[0:32])
|
||
# 充电电压（字节 16-17，HEX，单位 0.1V）
|
||
voltage = int(data_domain[32:36], 16) / 10.0
|
||
# 充电电流（字节 18-19，HEX，单位 0.1A）
|
||
current = int(data_domain[36:40], 16) / 10.0
|
||
# 充电电量（字节 20-21，HEX，单位 0.01kWh）
|
||
energy = int(data_domain[40:44], 16) / 100.0
|
||
# SOC（字节 22，HEX，单位 %）
|
||
soc = int(data_domain[44:46], 16)
|
||
# 故障状态（字节 23-26，HEX）
|
||
battery_faults = data_domain[46:54]
|
||
# 其他字段可以继续解析...
|
||
|
||
else:
|
||
logging.info(f"未处理的命令码: {cmd}")
|
||
return None # 跳过未处理的命令
|
||
|
||
# 构建插入 SQL
|
||
sql = (
|
||
"INSERT INTO battery_charge_jiuxing (battery_id, operator_id, station_id, battery_bun_id, cabinet_no, "
|
||
"battery_faults, created_at, updated_at, SOC) VALUES "
|
||
f"('{battery_id}', '{operator_id}', '{station_id}', '{battery_bun_id}', '{cabinet_no}', "
|
||
f"'{battery_faults}', '{created_at}', '{updated_at}', {soc})"
|
||
)
|
||
return sql
|
||
|
||
def run(self):
|
||
"""启动程序"""
|
||
self.connect()
|
||
try:
|
||
while True:
|
||
pass # 保持程序运行
|
||
except KeyboardInterrupt:
|
||
self.mqtt_client.loop_stop()
|
||
self.mqtt_client.disconnect()
|
||
if self.tdengine_cursor:
|
||
self.tdengine_cursor.close()
|
||
if self.tdengine_conn:
|
||
self.tdengine_conn.close()
|
||
logging.info("Program stopped")
|
||
|
||
if __name__ == "__main__":
|
||
parser = ChargeDataParser(
|
||
mqtt_host="123.6.102.119",
|
||
mqtt_port=1883,
|
||
mqtt_username="emqx_test",
|
||
mqtt_password="emqx_test",
|
||
tdengine_host="localhost", # 根据实际环境修改
|
||
tdengine_user="root",
|
||
tdengine_password="taosdata"
|
||
)
|
||
parser.run() |