54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
import time
|
||
import struct
|
||
|
||
|
||
class Command06:
|
||
def __init__(self):
|
||
self.pile_id = b"\x00\x27\x02\x12\x34\x56\x12\x34" # 示例桩号
|
||
|
||
def build_06h_response(self, pile_id):
|
||
"""构建06H下发对时帧"""
|
||
frame = bytearray([0x4A, 0x58]) # 帧头: "JX"
|
||
frame.append(0x06) # 命令码: 06H
|
||
frame.extend(pile_id) # 桩号 (8字节)
|
||
frame.append(0x01) # 数据加密方式: 0x01 (不加密)
|
||
frame.extend([0x00, 0x00]) # 数据域长度 (初始为0，后续更新)
|
||
|
||
# 数据域
|
||
data = bytearray()
|
||
current_time = time.localtime()
|
||
data.extend([
|
||
current_time.tm_year - 2000, current_time.tm_mon, current_time.tm_mday,
|
||
current_time.tm_hour, current_time.tm_min, current_time.tm_sec
|
||
]) # 时间标识 (6字节)
|
||
|
||
frame.extend(data)
|
||
struct.pack_into(">H", frame, 12, len(data)) # 更新数据域长度
|
||
|
||
# 校验码
|
||
checksum = 0
|
||
for b in frame[2:-1]:
|
||
checksum ^= b
|
||
frame.append(checksum)
|
||
|
||
return frame
|
||
|
||
def process_06h(self, data):
|
||
"""处理06H数据（由充电桩调用）"""
|
||
if data and len(data) >= 20 and data[2] == 0x06:
|
||
time_data = data[14:20]
|
||
year = 2000 + time_data[0]
|
||
month, day, hour, minute, second = time_data[1:6]
|
||
logging.info(f"收到对时: {year}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}:{second:02d}")
|
||
return True
|
||
logging.warning("无效的06H数据")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import logging
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
cmd = Command06()
|
||
response = cmd.build_06h_response(b"\x00\x27\x02\x12\x34\x56\x12\x34")
|
||
print(f"06H响应帧: {response.hex().upper()}") |