292 lines
10 KiB
Python
292 lines
10 KiB
Python
import socket
|
|
import struct
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
import binascii
|
|
|
|
|
|
|
|
class Command02:
|
|
def __init__(self):
|
|
self.command = 0x02 # 回复命令码02H
|
|
self.qr_fixed = "https://platform.enneagon.cn/ScanCharging?connectorCode="
|
|
|
|
def parse_pile_id(self, pile_id_bytes):
|
|
"""解析桩号"""
|
|
try:
|
|
vendor_id = struct.unpack("<H", pile_id_bytes[0:2])[0] # 运营商编号
|
|
gun_info = pile_id_bytes[2] # 枪数量信息
|
|
|
|
if gun_info <= 31:
|
|
gun_type = "交流"
|
|
gun_count = gun_info
|
|
elif 51 <= gun_info <= 81:
|
|
gun_type = "直流"
|
|
gun_count = gun_info - 50
|
|
else:
|
|
gun_type = "未知"
|
|
gun_count = gun_info
|
|
|
|
site_id = int.from_bytes(pile_id_bytes[3:6], 'little') # 站点编号
|
|
addr_in_site = struct.unpack("<H", pile_id_bytes[6:8])[0] # 站内桩地址
|
|
|
|
return {
|
|
"vendor_id": f"{vendor_id:04d}",
|
|
"gun_type": gun_type,
|
|
"gun_count": gun_count,
|
|
"site_id": f"{site_id:06d}",
|
|
"addr_in_site": f"{addr_in_site:04d}"
|
|
}
|
|
except Exception as e:
|
|
logging.error(f"Parse pile ID failed: {str(e)}")
|
|
print(f"解析桩号失败: {str(e)}")
|
|
return None
|
|
|
|
def validate_frame(self, data):
|
|
"""验证帧格式"""
|
|
try:
|
|
print("\n开始验证帧格式:")
|
|
print(f"数据内容: {data.hex().upper()}")
|
|
print(f"数据长度: {len(data)}字节")
|
|
|
|
# 1. 基本格式检查
|
|
if len(data) < 14:
|
|
print("数据长度不足14字节")
|
|
return False
|
|
|
|
if data[0:2] != b'JX':
|
|
print("帧起始标志不是'JX'")
|
|
return False
|
|
|
|
# 2. 获取数据域长度
|
|
data_len = struct.unpack("<H", data[12:14])[0]
|
|
expected_total_len = 14 + data_len + 1 # 头部(14) + 数据域 + 校验码(1)
|
|
|
|
# 如果数据长度不匹配，尝试修正数据
|
|
if len(data) != expected_total_len:
|
|
print(f"数据长度不匹配，尝试修正...")
|
|
if len(data) > expected_total_len:
|
|
print(f"截断多余数据")
|
|
data = data[:expected_total_len]
|
|
else:
|
|
print("数据不完整")
|
|
return False
|
|
|
|
# 3. 显示数据结构
|
|
print(f"\n数据结构分析:")
|
|
print(f"起始标识: {data[0:2].hex().upper()}")
|
|
print(f"命令字: {data[2]:02X}")
|
|
print(f"桩号: {data[3:11].hex().upper()}")
|
|
print(f"加密方式: {data[11]:02X}")
|
|
print(f"数据长度: {data_len}")
|
|
print(f"数据域: {data[14:14 + data_len].hex().upper()}")
|
|
print(f"校验码: {data[-1]:02X}")
|
|
|
|
# 4. 校验码验证
|
|
check_data = data[2:-1]
|
|
calculated_check = 0
|
|
for b in check_data:
|
|
calculated_check ^= b
|
|
|
|
if calculated_check != data[-1]:
|
|
print(f"校验码不匹配: 计算值={calculated_check:02X}, 接收值={data[-1]:02X}")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"验证帧格式异常: {str(e)}")
|
|
return False
|
|
|
|
def debug_print_fields(self, data):
|
|
"""打印数据包的详细字段"""
|
|
print("\n===== 数据包解析 =====")
|
|
print("1. 固定头部:")
|
|
print(f" 起始标识 (2字节): {data[0:2].hex().upper()}")
|
|
print(f" 命令码 (1字节): {data[2]:02X}")
|
|
print(f" 桩号 (8字节): {data[3:11].hex().upper()}")
|
|
print(f" 加密方式 (1字节): {data[11]:02X}")
|
|
print(f" 数据域长度 (2字节): {struct.unpack('<H', data[12:14])[0]}")
|
|
|
|
data_len = struct.unpack("<H", data[12:14])[0]
|
|
print("\n2. 数据域:")
|
|
print(f" 时间标识 (6字节): {data[14:20].hex().upper()}")
|
|
print(f" 密钥版本 (2字节): {data[20:22].hex().upper()}")
|
|
print(f" 校验密文 (8字节): {data[22:30].hex().upper()}")
|
|
|
|
print(f"\n3. 校验码 (1字节): {data[-1]:02X}")
|
|
|
|
# 检查是否有多余字节
|
|
expected_len = 14 + data_len + 1
|
|
if len(data) > expected_len:
|
|
print(f"\n警告: 发现多余字节:")
|
|
print(f"多余字节内容: {data[expected_len:].hex().upper()}")
|
|
|
|
def parse_01h(self, data):
|
|
"""解析01H命令数据"""
|
|
try:
|
|
print("\n开始解析01H命令...")
|
|
if not self.validate_frame(data):
|
|
raise ValueError("帧格式验证失败")
|
|
|
|
# 提取基本信息
|
|
command = data[2]
|
|
pile_id = data[3:11]
|
|
encrypt_mode = data[11]
|
|
data_field = data[14:-1] # 从数据域开始到校验码之前
|
|
|
|
# 解析数据域内容
|
|
time_bytes = data_field[0:6]
|
|
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}"
|
|
|
|
key_version = struct.unpack("<H", data_field[6:8])[0]
|
|
check_text = data_field[8:16]
|
|
|
|
result = {
|
|
"command": command,
|
|
"pile_id": pile_id,
|
|
"encrypt_mode": encrypt_mode,
|
|
"timestamp": timestamp,
|
|
"key_version": key_version,
|
|
"check_text": check_text
|
|
}
|
|
|
|
return result
|
|
except Exception as e:
|
|
print(f"解析失败: {str(e)}")
|
|
return None
|
|
|
|
def build_02h_response(self, pile_id, allow=True, reject_reason=0):
|
|
"""构建02H响应命令"""
|
|
try:
|
|
frame = bytearray()
|
|
frame.extend(b'JX') # 帧起始标志
|
|
frame.append(self.command) # 命令
|
|
frame.extend(pile_id) # 桩号
|
|
frame.append(0x01) # 数据加密方式
|
|
|
|
data = bytearray()
|
|
|
|
# 添加时间标识 (6字节)
|
|
now = datetime.now()
|
|
data.extend(struct.pack("<BBBBBB",
|
|
now.year - 2000, now.month, now.day,
|
|
now.hour, now.minute, now.second))
|
|
|
|
# 添加请求结果和拒绝原因
|
|
data.append(0x01 if allow else 0x02)
|
|
data.append(reject_reason)
|
|
|
|
if allow:
|
|
# 添加二维码固定段 (100字节)
|
|
fixed_part = self.qr_fixed.encode()
|
|
data.extend(fixed_part.ljust(100, b'\x00'))
|
|
|
|
# 添加二维码枪号段数量
|
|
data.append(0x02) # 2个充电枪
|
|
|
|
# 添加枪号段 (每个20字节)
|
|
pile_id_str = ''.join([f"{b:02X}" for b in pile_id])
|
|
gun1 = f"{pile_id_str}001"
|
|
gun2 = f"{pile_id_str}002"
|
|
data.extend(gun1.encode().ljust(20, b'\x00'))
|
|
data.extend(gun2.encode().ljust(20, b'\x00'))
|
|
|
|
# 添加数据域长度
|
|
frame.extend(struct.pack("<H", len(data)))
|
|
|
|
# 添加数据域
|
|
frame.extend(data)
|
|
|
|
# 计算并添加校验码
|
|
check = 0
|
|
for b in frame[2:]: # 从命令字节开始异或
|
|
check ^= b
|
|
frame.append(check)
|
|
|
|
return bytes(frame)
|
|
|
|
except Exception as e:
|
|
logging.error(f"构建02H响应失败: {str(e)}")
|
|
return None
|
|
|
|
def process_and_respond(self, received_data, sock):
|
|
"""处理收到的01H命令并回复02H"""
|
|
try:
|
|
# 基础验证
|
|
if not self.validate_frame(received_data):
|
|
logging.error("01H命令帧格式验证失败")
|
|
return False
|
|
|
|
# 提取必要信息
|
|
pile_id = received_data[3:11] # 桩号
|
|
time_bytes = received_data[14:20] # 时间标识
|
|
key_version = struct.unpack("<H", received_data[20:22])[0] # 密钥版本
|
|
|
|
# 检查时间是否在合理范围内
|
|
cmd_time = datetime(2000 + time_bytes[0], time_bytes[1], time_bytes[2],
|
|
time_bytes[3], time_bytes[4], time_bytes[5])
|
|
time_diff = abs((datetime.now() - cmd_time).total_seconds())
|
|
|
|
if time_diff > 600: # 超过10分钟
|
|
logging.warning(f"时间差异过大: {time_diff}秒")
|
|
response = self.build_02h_response(pile_id, False, 6) # 拒绝原因6-时差过大
|
|
else:
|
|
# 这里可以添加更多的验证逻辑
|
|
response = self.build_02h_response(pile_id, True, 0)
|
|
|
|
if response and hasattr(sock, 'send'):
|
|
sock.send(response)
|
|
logging.info(f"成功发送02H响应, 长度: {len(response)}字节")
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
logging.error(f"处理01H命令失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def test_command():
|
|
"""测试函数"""
|
|
print("开始测试01H/02H命令处理...")
|
|
|
|
# 创建响应处理器
|
|
handler = Command02()
|
|
|
|
# 原始测试数据
|
|
hex_string = "4A5801031767631136065701100019010909371501000000000000000000004D"
|
|
|
|
# 解析数据长度字段以确定正确的数据包长度
|
|
expected_len = 14 + 16 + 1 # 头部(14字节) + 数据域(16字节) + 校验码(1字节)
|
|
|
|
# 确保只取需要的字节
|
|
test_data = bytes.fromhex(hex_string[:expected_len * 2]) # *2是因为hex字符串中一个字节用两个字符表示
|
|
|
|
print("\n测试数据:")
|
|
print(f"原始数据: {hex_string}")
|
|
print(f"处理后数据: {test_data.hex().upper()}")
|
|
print(f"数据长度: {len(test_data)}字节")
|
|
|
|
# 打印详细字段解析
|
|
handler.debug_print_fields(test_data)
|
|
|
|
# 创建模拟socket
|
|
class MockSocket:
|
|
def send(self, data):
|
|
print(f"\n模拟发送响应数据:")
|
|
print(f"数据内容: {data.hex().upper()}")
|
|
print(f"数据长度: {len(data)}字节")
|
|
|
|
mock_sock = MockSocket()
|
|
|
|
# 测试完整处理流程
|
|
result = handler.process_and_respond(test_data, mock_sock)
|
|
print(f"\n最终处理结果: {'成功' if result else '失败'}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_command() |