293 lines
11 KiB
Python
293 lines
11 KiB
Python
import struct
|
||
import logging
|
||
import binascii
|
||
from datetime import datetime
|
||
|
||
|
||
class Command2122:
|
||
def __init__(self):
|
||
self.command_21 = 0x21 # 充电启动结果命令
|
||
self.command_22 = 0x22 # 平台回复启动充电结果命令
|
||
|
||
def parse_21h_charging_start_result(self, data):
|
||
"""
|
||
解析21H充电启动结果命令
|
||
|
||
:param data: 完整的21H命令报文
|
||
:return: 解析后的字典或None
|
||
"""
|
||
try:
|
||
# 验证基本帧格式
|
||
if len(data) < 14 or data[0:2] != b'JX' or data[2] != 0x21:
|
||
logging.warning(f"21H命令帧格式不正确，原始报文: {binascii.hexlify(data)}")
|
||
return None
|
||
|
||
# 打印完整的原始报文以便调试
|
||
print(f"完整原始报文: {binascii.hexlify(data)}")
|
||
|
||
# 提取桩号
|
||
pile_id_bytes = data[3:11]
|
||
|
||
# 提取时间标识
|
||
time_bytes = data[14:20]
|
||
year = time_bytes[0] + 2000
|
||
month, day, hour, minute, second = time_bytes[1:6]
|
||
timestamp = f"{year:04d}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}:{second:02d}"
|
||
|
||
current_index = 22
|
||
|
||
# 解析充电订单号
|
||
charging_order_number = data[current_index:current_index + 32].decode('ascii').rstrip('\x00')
|
||
current_index += 32
|
||
|
||
# 解析用户ID
|
||
user_id = data[current_index:current_index + 32].decode('ascii').rstrip('\x00')
|
||
current_index += 32
|
||
|
||
# 解析用户类型
|
||
user_type = struct.unpack("<H", data[current_index:current_index + 2])[0]
|
||
current_index += 2
|
||
|
||
# 解析车牌号
|
||
vehicle_number = data[current_index:current_index + 9].decode('ascii').rstrip('\x00')
|
||
current_index += 9
|
||
|
||
# 解析控制方式
|
||
control_mode = data[current_index]
|
||
current_index += 1
|
||
|
||
# 解析控制参数
|
||
control_param = struct.unpack("<I", data[current_index:current_index + 4])[0]
|
||
current_index += 4
|
||
|
||
# 解析充电模式
|
||
charging_mode = data[current_index]
|
||
current_index += 1
|
||
|
||
# 解析充电桩类型
|
||
pile_type = data[current_index]
|
||
current_index += 1
|
||
|
||
# 解析启动结果
|
||
start_result = data[current_index]
|
||
current_index += 1
|
||
|
||
# 解析启动失败原因
|
||
start_failure_reason = struct.unpack("<H", data[current_index:current_index + 2])[0]
|
||
current_index += 2
|
||
|
||
# 解析充电起始时间
|
||
start_charging_time_bytes = data[current_index:current_index + 6]
|
||
start_charging_time = datetime(
|
||
start_charging_time_bytes[0] + 2000,
|
||
start_charging_time_bytes[1],
|
||
start_charging_time_bytes[2],
|
||
start_charging_time_bytes[3],
|
||
start_charging_time_bytes[4],
|
||
start_charging_time_bytes[5]
|
||
)
|
||
current_index += 6
|
||
|
||
# 解析充电起始电量
|
||
start_charging_amount = struct.unpack("<I", data[current_index:current_index + 4])[0] / 100 # 0.01kWh
|
||
current_index += 4
|
||
|
||
# 解析绝缘检测电压
|
||
insulation_voltage = struct.unpack("<H", data[current_index:current_index + 2])[0] / 10 # 0.1V
|
||
current_index += 2
|
||
|
||
# 打印解析结果
|
||
print("\n21H充电启动结果命令解析结果:")
|
||
print(f"桩号: {pile_id_bytes.hex()}")
|
||
print(f"时间标识: {timestamp}")
|
||
print(f"充电订单号: {charging_order_number}")
|
||
print(f"用户ID: {user_id}")
|
||
print(f"用户类型: {self.get_user_type_text(user_type)}")
|
||
print(f"车牌号: {vehicle_number}")
|
||
print(f"控制方式: {self.get_control_mode_text(control_mode)}")
|
||
print(f"控制参数: {control_param}")
|
||
print(f"充电模式: {self.get_charging_mode_text(charging_mode)}")
|
||
print(f"充电桩类型: {self.get_pile_type_text(pile_type)}")
|
||
print(f"启动结果: {self.get_start_result_text(start_result)}")
|
||
print(f"启动失败原因: {self.get_start_failure_reason_text(start_failure_reason)}")
|
||
print(f"充电起始时间: {start_charging_time}")
|
||
print(f"充电起始电量: {start_charging_amount}kWh")
|
||
print(f"绝缘检测电压: {insulation_voltage}V")
|
||
|
||
return {
|
||
"pile_id": pile_id_bytes.hex(),
|
||
"timestamp": timestamp,
|
||
"charging_order_number": charging_order_number,
|
||
"user_id": user_id,
|
||
"user_type": self.get_user_type_text(user_type),
|
||
"vehicle_number": vehicle_number,
|
||
"control_mode": self.get_control_mode_text(control_mode),
|
||
"control_param": control_param,
|
||
"charging_mode": self.get_charging_mode_text(charging_mode),
|
||
"pile_type": self.get_pile_type_text(pile_type),
|
||
"start_result": self.get_start_result_text(start_result),
|
||
"start_failure_reason": self.get_start_failure_reason_text(start_failure_reason),
|
||
"start_charging_time": start_charging_time,
|
||
"start_charging_amount": start_charging_amount,
|
||
"insulation_voltage": insulation_voltage
|
||
}
|
||
|
||
except Exception as e:
|
||
logging.error(f"解析21H命令失败: {str(e)}")
|
||
logging.error(f"原始报文: {binascii.hexlify(data)}")
|
||
return None
|
||
|
||
def generate_22h_charging_start_response(self, pile_id_bytes):
|
||
"""
|
||
生成22H平台回复启动充电结果命令
|
||
|
||
:param pile_id_bytes: 充电桩桩号字节
|
||
:return: 22H响应报文
|
||
"""
|
||
try:
|
||
# 构建帧
|
||
frame = bytearray()
|
||
frame.extend(b'JX') # 帧起始标志
|
||
frame.append(self.command_22) # 命令码
|
||
frame.extend(pile_id_bytes) # 桩号
|
||
frame.append(0x01) # 数据加密方式
|
||
|
||
# 构建数据域
|
||
data = bytearray()
|
||
|
||
# 时间标识（当前时间）
|
||
now = datetime.now()
|
||
data.extend(struct.pack("<BBBBBB",
|
||
now.year - 2000, now.month, now.day,
|
||
now.hour, now.minute, now.second))
|
||
|
||
# 数据域长度
|
||
frame.extend(struct.pack("<H", len(data)))
|
||
|
||
# 加入数据域
|
||
frame.extend(data)
|
||
|
||
# 计算校验码
|
||
check = 0
|
||
for b in frame[2:]:
|
||
check ^= b
|
||
frame.append(check)
|
||
|
||
print("22H充电启动结果响应数据构建成功:")
|
||
print(f"数据内容: {frame.hex()}")
|
||
print(f"数据长度: {len(frame)}字节")
|
||
|
||
return bytes(frame)
|
||
|
||
except Exception as e:
|
||
logging.error(f"生成22H充电启动结果响应出错: {str(e)}")
|
||
return None
|
||
|
||
def process_21h_charging_start_result(self, data):
|
||
"""
|
||
处理21H充电启动结果命令
|
||
|
||
:param data: 完整的21H命令报文
|
||
:return: 是否成功处理
|
||
"""
|
||
try:
|
||
parsed_data = self.parse_21h_charging_start_result(data)
|
||
|
||
if parsed_data is None:
|
||
logging.warning("21H命令解析失败")
|
||
return False
|
||
|
||
# 记录充电启动结果信息日志
|
||
logging.info(
|
||
f"收到桩号 {parsed_data['pile_id']} 的充电启动结果: "
|
||
f"订单号 {parsed_data['charging_order_number']}, "
|
||
f"启动结果 {parsed_data['start_result']}"
|
||
)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(f"处理21H命令出错: {str(e)}")
|
||
return False
|
||
|
||
def get_user_type_text(self, user_type):
|
||
"""解析用户类型"""
|
||
type_map = {
|
||
1: "超级卡",
|
||
2: "在线卡",
|
||
3: "离线卡",
|
||
5: "本地管理员",
|
||
6: "VIN鉴权"
|
||
}
|
||
return type_map.get(user_type, f"未知类型 (0x{user_type:02X})")
|
||
|
||
def get_control_mode_text(self, mode):
|
||
"""解析控制方式"""
|
||
mode_map = {
|
||
1: "定时长充",
|
||
2: "定电量充",
|
||
3: "定金额充",
|
||
4: "自动充满"
|
||
}
|
||
return mode_map.get(mode, f"未知方式 (0x{mode:02X})")
|
||
|
||
def get_charging_mode_text(self, mode):
|
||
"""解析充电模式"""
|
||
mode_map = {
|
||
1: "普通充电",
|
||
2: "轮充",
|
||
3: "大功率",
|
||
4: "超级充",
|
||
5: "电池维护",
|
||
6: "柔性充"
|
||
}
|
||
return mode_map.get(mode, f"未知模式 (0x{mode:02X})")
|
||
|
||
def get_pile_type_text(self, pile_type):
|
||
"""解析充电桩类型"""
|
||
type_map = {
|
||
1: "交流",
|
||
2: "直流"
|
||
}
|
||
return type_map.get(pile_type, f"未知类型 (0x{pile_type:02X})")
|
||
|
||
def get_start_result_text(self, result):
|
||
"""解析启动结果"""
|
||
result_map = {
|
||
1: "成功",
|
||
2: "失败"
|
||
}
|
||
return result_map.get(result, f"未知结果 (0x{result:02X})")
|
||
|
||
def get_start_failure_reason_text(self, reason):
|
||
"""解析启动失败原因"""
|
||
reason_map = {
|
||
1: "设备故障",
|
||
2: "充电枪使用中",
|
||
3: "充电设备已被预约",
|
||
4: "不允许充电",
|
||
5: "参数不支持",
|
||
6: "其他原因"
|
||
}
|
||
return reason_map.get(reason, f"未知原因 (0x{reason:04X})")
|
||
|
||
|
||
# 测试用例
|
||
if __name__ == "__main__":
|
||
# 21H命令测试报文
|
||
test_21_data = bytes.fromhex(
|
||
"4A 58 21 03 17 66 56 11 36 06 37 01 B3 00 19 01 09 0B 28 1D 01 31 38 37 37 31 39 38 35 39 35 39 37 30 35 39 36 38 36 00 00 00 00 00 00 00 00 00 00 00 00 00 38 34 30 34 33 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 16 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 03 F4 01 00 00 01 02 01 00 00 19 01 09 0B 27 37 D4 6C 68 01 00 00 00 00 00 00 00 00 00 03 D0 11 26 18 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 4C 5A 47 4A 4C 4D 34 34 35 50 58 31 31 34 35 33 37 01 01 00 00 00 00 00 00 77 01 DC 05 03 0B 60 1B 73 DA 02 E8 18 E8")
|
||
|
||
# 22H命令测试报文
|
||
test_22_data = bytes.fromhex("4A 58 22 03 17 66 56 11 36 06 37 01 07 00 19 01 09 0B 28 20 01 05")
|
||
|
||
parser = Command2122()
|
||
|
||
# 测试解析21H命令
|
||
parser.process_21h_charging_start_result(test_21_data)
|
||
|
||
# 测试生成22H响应
|
||
pile_id_bytes = bytes.fromhex("0317665611360637")
|
||
response = parser.generate_22h_charging_start_response(pile_id_bytes)
|
||
print("\n22H充电启动结果响应:")
|
||
print(response.hex()) |