637 lines
24 KiB
Python
637 lines
24 KiB
Python
import struct
|
|
import datetime
|
|
|
|
class msg_ccUFbMsg:
|
|
def __init__(self):
|
|
# 消息头
|
|
self.start1 = 0x20
|
|
self.start2 = 0x20
|
|
self.id = 0x0051 # CCU_STATE_FB_ID
|
|
self.length = 0
|
|
self.checkCode = 0x00
|
|
|
|
# CCU状态数据 - 根据ccuState结构体定义
|
|
self.fc_mode = 0x00
|
|
self.fc_status = 0x00
|
|
self.fc_fault_level = 0x00
|
|
|
|
self.output_power_limit = 0
|
|
self.total_generation_time = 0
|
|
self.total_generation_power = 0
|
|
self.fault_level_1 = 0
|
|
self.fault_level_2 = 0
|
|
self.fault_level_3 = 0
|
|
self.fault_level_4 = 0
|
|
|
|
self.hydrogen_capacity = 0
|
|
self.liquid_oxygen_capacity = 0
|
|
|
|
self.palladium_temp = 0
|
|
self.main_pipe_pressure = 0
|
|
self.aux_pipe_pressure = 0
|
|
|
|
self.cabin_pressure1 = 0
|
|
self.cabin_pressure2 = 0
|
|
self.cabin_temp1 = 0
|
|
self.cabin_temp2 = 0
|
|
self.cabin_humidity1 = 0
|
|
self.cabin_humidity2 = 0
|
|
|
|
self.flame_detector1 = 0x00
|
|
self.flame_detector2 = 0x00
|
|
|
|
self.h2_concentration1 = 0
|
|
self.h2_concentration2 = 0
|
|
self.h2_concentration3 = 0
|
|
self.o2_concentration1 = 0
|
|
self.o2_concentration2 = 0
|
|
self.ch3oh_concentration1 = 0
|
|
self.ch3oh_concentration2 = 0
|
|
|
|
self.reserved1 = [0, 0]
|
|
self.emergency_battery1_voltage = 0
|
|
self.emergency_battery1_current = 0
|
|
self.emergency_battery1_max_temp = 0
|
|
self.emergency_battery1_fault_word = 0
|
|
self.emergency_battery2_voltage = 0
|
|
self.emergency_battery2_current = 0
|
|
self.emergency_battery2_max_temp = 0
|
|
self.emergency_battery2_fault_word = 0
|
|
|
|
self.cabin_ox_concentration = 0
|
|
self.cabin_temperature = 0
|
|
self.cabin_humidity = 0
|
|
self.cabin_pressure = 0
|
|
|
|
self.power_cabin_ox_concentration = 0
|
|
self.power_cabin_temperature = 0
|
|
self.power_cabin_humidity = 0
|
|
self.power_cabin_pressure = 0
|
|
self.reserved2 = [0]
|
|
|
|
self.dyn_alarm_flag = [0] * 6
|
|
self.ins_alarm_flag = [0] * 6
|
|
|
|
self.ins_relay_status1 = 0x00
|
|
self.ins_relay_status2 = 0x00
|
|
self.dyn_relay_status1 = 0x00
|
|
self.dyn_relay_status2 = 0x00
|
|
|
|
self.ins_max_discharge_power = 0
|
|
self.dyn_max_discharge_power = 0
|
|
|
|
self.ins_soc = 0
|
|
self.dyn_soc = 0
|
|
|
|
self.ins_total_energy = 0
|
|
self.dyn_total_energy = 0
|
|
|
|
self.ins_power_input = 0
|
|
self.dyn_power_input = 0
|
|
|
|
self.ins_charge_status = 0x00
|
|
self.dyn_charge_status = 0x00
|
|
|
|
self.ins_voltage_link = 0
|
|
self.ins_voltage_pack = 0
|
|
self.ins_current = 0
|
|
self.ins_resistance_pos = 0
|
|
self.ins_resistance_neg = 0
|
|
|
|
self.dyn_voltage_link = 0
|
|
self.dyn_voltage_pack = 0
|
|
self.dyn_current = 0
|
|
self.dyn_resistance_pos = 0
|
|
self.dyn_resistance_neg = 0
|
|
|
|
self.ins_emergency_status = 0x00
|
|
self.dyn_emergency_status = 0x00
|
|
|
|
self.reserved3 = [0] * 2
|
|
|
|
self.ins_soc_threshold1 = 0
|
|
self.ins_soc_threshold2 = 0
|
|
self.ins_soc_threshold3 = 0
|
|
self.dyn_soc_threshold1 = 0
|
|
self.dyn_soc_threshold2 = 0
|
|
self.dyn_soc_threshold3 = 0
|
|
|
|
self.ins_power_limit1 = 0
|
|
self.ins_power_limit2 = 0
|
|
self.dyn_power_limit1 = 0
|
|
self.dyn_power_limit2 = 0
|
|
|
|
self.device_online_flag1 = 0
|
|
self.device_online_flag2 = 0
|
|
|
|
self.dyn_charge_level = 0x00
|
|
self.dyn_water_leak_status = 0x00
|
|
self.ins_water_leak_dcdc_status = 0x00
|
|
self.reserved4 = 0x00
|
|
self.heartbeat = 0
|
|
|
|
def pack(self):
|
|
"""打包消息为字节数据"""
|
|
# 重置校验码
|
|
self.checkCode = 0x00
|
|
|
|
# 打包消息头
|
|
data = struct.pack('<BBHH',
|
|
self.start1,
|
|
self.start2,
|
|
self.id,
|
|
self.length)
|
|
|
|
# 打包CCU状态数据
|
|
data += struct.pack('<BBB',
|
|
self.fc_mode,
|
|
self.fc_status,
|
|
self.fc_fault_level)
|
|
|
|
data += struct.pack('<HHHHHHH',
|
|
self.output_power_limit,
|
|
self.total_generation_time,
|
|
self.total_generation_power,
|
|
self.fault_level_1,
|
|
self.fault_level_2,
|
|
self.fault_level_3,
|
|
self.fault_level_4)
|
|
|
|
data += struct.pack('<BB',
|
|
self.hydrogen_capacity,
|
|
self.liquid_oxygen_capacity)
|
|
|
|
data += struct.pack('<HHH',
|
|
self.palladium_temp,
|
|
self.main_pipe_pressure,
|
|
self.aux_pipe_pressure)
|
|
|
|
data += struct.pack('<HHHHHH',
|
|
self.cabin_pressure1,
|
|
self.cabin_pressure2,
|
|
self.cabin_temp1,
|
|
self.cabin_temp2,
|
|
self.cabin_humidity1,
|
|
self.cabin_humidity2)
|
|
|
|
data += struct.pack('<BB',
|
|
self.flame_detector1,
|
|
self.flame_detector2)
|
|
|
|
data += struct.pack('<HHHHHHH',
|
|
self.h2_concentration1,
|
|
self.h2_concentration2,
|
|
self.h2_concentration3,
|
|
self.o2_concentration1,
|
|
self.o2_concentration2,
|
|
self.ch3oh_concentration1,
|
|
self.ch3oh_concentration2)
|
|
|
|
# 保留字段和应急电池数据
|
|
data += struct.pack('<HHHHHHHH',
|
|
self.reserved1[0], self.reserved1[1],
|
|
self.emergency_battery1_voltage,
|
|
self.emergency_battery1_current,
|
|
self.emergency_battery1_max_temp,
|
|
self.emergency_battery1_fault_word,
|
|
self.emergency_battery2_voltage,
|
|
self.emergency_battery2_current)
|
|
|
|
data += struct.pack('<HHHH',
|
|
self.emergency_battery2_max_temp,
|
|
self.emergency_battery2_fault_word,
|
|
self.cabin_ox_concentration,
|
|
self.cabin_temperature)
|
|
|
|
data += struct.pack('<HHHHHHHH',
|
|
self.cabin_humidity,
|
|
self.cabin_pressure,
|
|
self.power_cabin_ox_concentration,
|
|
self.power_cabin_temperature,
|
|
self.power_cabin_humidity,
|
|
self.power_cabin_pressure,
|
|
self.reserved2[0], 0) # 填充第二个reserved2字节
|
|
|
|
# 报警标志数组
|
|
for flag in self.dyn_alarm_flag:
|
|
data += struct.pack('<B', flag)
|
|
for flag in self.ins_alarm_flag:
|
|
data += struct.pack('<B', flag)
|
|
|
|
# 继电器状态
|
|
data += struct.pack('<BBBB',
|
|
self.ins_relay_status1,
|
|
self.ins_relay_status2,
|
|
self.dyn_relay_status1,
|
|
self.dyn_relay_status2)
|
|
|
|
# 功率和SOC数据
|
|
data += struct.pack('<HHBBHHHH',
|
|
self.ins_max_discharge_power,
|
|
self.dyn_max_discharge_power,
|
|
self.ins_soc,
|
|
self.dyn_soc,
|
|
self.ins_total_energy,
|
|
self.dyn_total_energy,
|
|
self.ins_power_input,
|
|
self.dyn_power_input)
|
|
|
|
# 充电状态
|
|
data += struct.pack('<BB',
|
|
self.ins_charge_status,
|
|
self.dyn_charge_status)
|
|
|
|
# 电压电流和绝缘电阻
|
|
data += struct.pack('<HHHHHHHHHH',
|
|
self.ins_voltage_link,
|
|
self.ins_voltage_pack,
|
|
self.ins_current,
|
|
self.ins_resistance_pos,
|
|
self.ins_resistance_neg,
|
|
self.dyn_voltage_link,
|
|
self.dyn_voltage_pack,
|
|
self.dyn_current,
|
|
self.dyn_resistance_pos,
|
|
self.dyn_resistance_neg)
|
|
|
|
# 紧急状态和保留字段
|
|
data += struct.pack('<BB',
|
|
self.ins_emergency_status,
|
|
self.dyn_emergency_status)
|
|
|
|
data += struct.pack('<BB',
|
|
self.reserved3[0], self.reserved3[1])
|
|
|
|
# SOC门限值
|
|
data += struct.pack('<BBBBBB',
|
|
self.ins_soc_threshold1,
|
|
self.ins_soc_threshold2,
|
|
self.ins_soc_threshold3,
|
|
self.dyn_soc_threshold1,
|
|
self.dyn_soc_threshold2,
|
|
self.dyn_soc_threshold3)
|
|
|
|
# 功率限制值
|
|
data += struct.pack('<BBBB',
|
|
self.ins_power_limit1,
|
|
self.ins_power_limit2,
|
|
self.dyn_power_limit1,
|
|
self.dyn_power_limit2)
|
|
|
|
# 设备在线状态和心跳
|
|
data += struct.pack('<II',
|
|
self.device_online_flag1,
|
|
self.device_online_flag2)
|
|
|
|
data += struct.pack('<BBBB',
|
|
self.dyn_charge_level,
|
|
self.dyn_water_leak_status,
|
|
self.ins_water_leak_dcdc_status,
|
|
self.reserved4)
|
|
|
|
data += struct.pack('<I', self.heartbeat)
|
|
|
|
# 计算校验码
|
|
self.checkCode = 0
|
|
for byte in data:
|
|
self.checkCode = (self.checkCode + byte) & 0xFF
|
|
|
|
# 添加校验码
|
|
data += struct.pack('<B', self.checkCode)
|
|
|
|
# 更新长度字段
|
|
self.length = len(data) - 6 # 减去头部6字节
|
|
|
|
return data
|
|
|
|
def unpack(self, data):
|
|
"""从字节数据解包消息"""
|
|
if len(data) < 10: # 最小消息长度检查
|
|
return False
|
|
|
|
# 解析消息头
|
|
self.start1 = data[0]
|
|
self.start2 = data[1]
|
|
self.id = struct.unpack_from('<H', data, 2)[0]
|
|
self.length = struct.unpack_from('<H', data, 4)[0]
|
|
|
|
offset = 6 # 头部结束位置
|
|
|
|
# 检查数据长度是否足够
|
|
if len(data) < offset + self.length + 1: # +1 for checksum
|
|
return False
|
|
|
|
# 解析CCU状态数据
|
|
(self.fc_mode, self.fc_status, self.fc_fault_level) = struct.unpack_from('<BBB', data, offset); offset += 3
|
|
|
|
(self.output_power_limit, self.total_generation_time, self.total_generation_power,
|
|
self.fault_level_1, self.fault_level_2, self.fault_level_3, self.fault_level_4) = struct.unpack_from('<HHHHHHH', data, offset); offset += 14
|
|
|
|
(self.hydrogen_capacity, self.liquid_oxygen_capacity) = struct.unpack_from('<BB', data, offset); offset += 2
|
|
|
|
(self.palladium_temp, self.main_pipe_pressure, self.aux_pipe_pressure) = struct.unpack_from('<HHH', data, offset); offset += 6
|
|
|
|
(self.cabin_pressure1, self.cabin_pressure2, self.cabin_temp1, self.cabin_temp2,
|
|
self.cabin_humidity1, self.cabin_humidity2) = struct.unpack_from('<HHHHHH', data, offset); offset += 12
|
|
|
|
(self.flame_detector1, self.flame_detector2) = struct.unpack_from('<BB', data, offset); offset += 2
|
|
|
|
(self.h2_concentration1, self.h2_concentration2, self.h2_concentration3,
|
|
self.o2_concentration1, self.o2_concentration2, self.ch3oh_concentration1,
|
|
self.ch3oh_concentration2) = struct.unpack_from('<HHHHHHH', data, offset); offset += 14
|
|
|
|
# 保留字段和应急电池数据
|
|
(self.reserved1[0], self.reserved1[1], self.emergency_battery1_voltage,
|
|
self.emergency_battery1_current, self.emergency_battery1_max_temp,
|
|
self.emergency_battery1_fault_word, self.emergency_battery2_voltage,
|
|
self.emergency_battery2_current) = struct.unpack_from('<HHHHHHHH', data, offset); offset += 16
|
|
|
|
(self.emergency_battery2_max_temp, self.emergency_battery2_fault_word,
|
|
self.cabin_ox_concentration, self.cabin_temperature) = struct.unpack_from('<HHHH', data, offset); offset += 8
|
|
|
|
(self.cabin_humidity, self.cabin_pressure, self.power_cabin_ox_concentration,
|
|
self.power_cabin_temperature, self.power_cabin_humidity, self.power_cabin_pressure,
|
|
reserved2_0, reserved2_1) = struct.unpack_from('<HHHHHHBB', data, offset); offset += 16
|
|
self.reserved2 = [reserved2_0, reserved2_1]
|
|
|
|
# 报警标志数组
|
|
for i in range(6):
|
|
self.dyn_alarm_flag[i] = data[offset]; offset += 1
|
|
for i in range(6):
|
|
self.ins_alarm_flag[i] = data[offset]; offset += 1
|
|
|
|
# 继电器状态
|
|
(self.ins_relay_status1, self.ins_relay_status2,
|
|
self.dyn_relay_status1, self.dyn_relay_status2) = struct.unpack_from('<BBBB', data, offset); offset += 4
|
|
|
|
# 功率和SOC数据
|
|
(self.ins_max_discharge_power, self.dyn_max_discharge_power,
|
|
self.ins_soc, self.dyn_soc, self.ins_total_energy, self.dyn_total_energy,
|
|
self.ins_power_input, self.dyn_power_input) = struct.unpack_from('<HHBBHHHH', data, offset); offset += 14
|
|
|
|
# 充电状态
|
|
(self.ins_charge_status, self.dyn_charge_status) = struct.unpack_from('<BB', data, offset); offset += 2
|
|
|
|
# 电压电流和绝缘电阻
|
|
(self.ins_voltage_link, self.ins_voltage_pack, self.ins_current,
|
|
self.ins_resistance_pos, self.ins_resistance_neg, self.dyn_voltage_link,
|
|
self.dyn_voltage_pack, self.dyn_current, self.dyn_resistance_pos,
|
|
self.dyn_resistance_neg) = struct.unpack_from('<HHHHHHHHHH', data, offset); offset += 20
|
|
|
|
# 紧急状态和保留字段
|
|
(self.ins_emergency_status, self.dyn_emergency_status) = struct.unpack_from('<BB', data, offset); offset += 2
|
|
|
|
(self.reserved3[0], self.reserved3[1]) = struct.unpack_from('<BB', data, offset); offset += 2
|
|
|
|
# SOC门限值
|
|
(self.ins_soc_threshold1, self.ins_soc_threshold2, self.ins_soc_threshold3,
|
|
self.dyn_soc_threshold1, self.dyn_soc_threshold2, self.dyn_soc_threshold3) = struct.unpack_from('<BBBBBB', data, offset); offset += 6
|
|
|
|
# 功率限制值
|
|
(self.ins_power_limit1, self.ins_power_limit2,
|
|
self.dyn_power_limit1, self.dyn_power_limit2) = struct.unpack_from('<BBBB', data, offset); offset += 4
|
|
|
|
# 设备在线状态和心跳
|
|
(self.device_online_flag1, self.device_online_flag2) = struct.unpack_from('<II', data, offset); offset += 8
|
|
|
|
(self.dyn_charge_level, self.dyn_water_leak_status,
|
|
self.ins_water_leak_dcdc_status, self.reserved4) = struct.unpack_from('<BBBB', data, offset); offset += 4
|
|
|
|
(self.heartbeat,) = struct.unpack_from('<I', data, offset); offset += 4
|
|
|
|
# 校验校验码
|
|
checkCode = 0
|
|
for byte in data[:-1]: # 不包括校验码本身
|
|
checkCode = (checkCode + byte) & 0xFF
|
|
|
|
return checkCode == data[-1]
|
|
|
|
def print_info(self):
|
|
"""打印消息信息"""
|
|
print(f"消息头: start1=0x{self.start1:02X}, start2=0x{self.start2:02X}, id=0x{self.id:04X}, length={self.length}")
|
|
print(f"燃料电池模式: 0x{self.fc_mode:02X}")
|
|
print(f"燃料电池状态: 0x{self.fc_status:02X}")
|
|
print(f"故障等级: 0x{self.fc_fault_level:02X}")
|
|
print(f"输出功率限制: {self.output_power_limit}W")
|
|
print(f"累计发电时间: {self.total_generation_time * 0.1}小时")
|
|
print(f"发电总功率: {self.total_generation_power}W")
|
|
print(f"储氢容量: {self.hydrogen_capacity}%")
|
|
print(f"液氧容量: {self.liquid_oxygen_capacity}%")
|
|
print(f"钯膜温度: {self.palladium_temp}℃")
|
|
print(f"主水路压力: {self.main_pipe_pressure * 0.01}kPa")
|
|
print(f"辅水路压力: {self.aux_pipe_pressure * 0.01}kPa")
|
|
print(f"仪表SOC: {self.ins_soc}%")
|
|
print(f"动力SOC: {self.dyn_soc}%")
|
|
print(f"心跳计数器: {self.heartbeat}")
|
|
print(f"校验码: 0x{self.checkCode:02X}")
|
|
|
|
|
|
class msg_ccUCmdMsg:
|
|
def __init__(self):
|
|
# 消息头
|
|
self.start1 = 0x20
|
|
self.start2 = 0x20
|
|
self.id = 0x0040 # CCU_CMD_ID
|
|
self.length = 0
|
|
self.checkCode = 0x00
|
|
|
|
# 时间戳
|
|
self.year = 0
|
|
self.month = 0
|
|
self.day = 0
|
|
self.hour = 0
|
|
self.minute = 0
|
|
self.second = 0
|
|
self.milliseconds = 0
|
|
|
|
# 复合管控器操作指令
|
|
self.fcmode = 0x00 # 燃料电池运行模式
|
|
self.fcuCmd = 0x00 # 燃料电池控制命令
|
|
self.fcuPowerEfficiency = 0x00 # 燃料电池功率效率
|
|
self.insChargeEnable = 0x00 # 仪表电池充电使能
|
|
self.insBatCmd = 0x00 # 仪表电池命令
|
|
self.powerBatCmd = 0x00 # 动力电池命令
|
|
self.powerBatEfficiency = 0 # 动力电池功率效率
|
|
self.heartbeat = 0 # 心跳
|
|
self.masterState = 0x00 # 主控状态
|
|
self.paceholder1 = 0 # 占位符1
|
|
self.paceholder2 = 0 # 占位符2
|
|
|
|
def pack(self):
|
|
"""打包消息为字节数据"""
|
|
# 重置校验码
|
|
self.checkCode = 0x00
|
|
|
|
# 打包消息头
|
|
data = struct.pack('<BBHH',
|
|
self.start1,
|
|
self.start2,
|
|
self.id,
|
|
self.length)
|
|
|
|
# 打包时间戳
|
|
data += struct.pack('<HBBBBBB',
|
|
self.year,
|
|
self.month,
|
|
self.day,
|
|
self.hour,
|
|
self.minute,
|
|
self.second,
|
|
self.milliseconds)
|
|
|
|
# 打包复合管控器操作指令
|
|
data += struct.pack('<BBBBBBHBBII',
|
|
self.fcmode,
|
|
self.fcuCmd,
|
|
self.fcuPowerEfficiency,
|
|
self.insChargeEnable,
|
|
self.insBatCmd,
|
|
self.powerBatCmd,
|
|
self.powerBatEfficiency,
|
|
self.heartbeat,
|
|
self.masterState,
|
|
self.paceholder1,
|
|
self.paceholder2)
|
|
|
|
# 计算校验码
|
|
self.checkCode = 0
|
|
for byte in data:
|
|
self.checkCode = (self.checkCode + byte) & 0xFF
|
|
|
|
# 添加校验码
|
|
data += struct.pack('<B', self.checkCode)
|
|
|
|
# 更新长度字段
|
|
self.length = len(data) - 6 # 减去头部6字节
|
|
|
|
return data
|
|
|
|
def unpack(self, data):
|
|
"""从字节数据解包消息"""
|
|
if len(data) < 10: # 最小消息长度检查
|
|
return False
|
|
|
|
# 解析消息头
|
|
self.start1 = data[0]
|
|
self.start2 = data[1]
|
|
self.id = struct.unpack_from('<H', data, 2)[0]
|
|
self.length = struct.unpack_from('<H', data, 4)[0]
|
|
|
|
offset = 6 # 头部结束位置
|
|
|
|
# 检查数据长度是否足够
|
|
if len(data) < offset + self.length + 1: # +1 for checksum
|
|
return False
|
|
|
|
# 解析时间戳
|
|
(self.year, self.month, self.day, self.hour,
|
|
self.minute, self.second, self.milliseconds) = struct.unpack_from('<HBBBBBB', data, offset); offset += 8
|
|
|
|
# 解析复合管控器操作指令
|
|
(self.fcmode, self.fcuCmd, self.fcuPowerEfficiency,
|
|
self.insChargeEnable, self.insBatCmd, self.powerBatCmd,
|
|
self.powerBatEfficiency, self.heartbeat, self.masterState,
|
|
self.paceholder1, self.paceholder2) = struct.unpack_from('<BBBBBBHBBII', data, offset); offset += 20
|
|
|
|
# 校验校验码
|
|
checkCode = 0
|
|
for byte in data[:-1]: # 不包括校验码本身
|
|
checkCode = (checkCode + byte) & 0xFF
|
|
|
|
return checkCode == data[-1]
|
|
|
|
def print_info(self):
|
|
"""打印消息信息"""
|
|
print(f"消息头: start1=0x{self.start1:02X}, start2=0x{self.start2:02X}, id=0x{self.id:04X}, length={self.length}")
|
|
print(f"时间戳: {self.year}-{self.month:02d}-{self.day:02d} {self.hour:02d}:{self.minute:02d}:{self.second:02d}.{self.milliseconds:03d}")
|
|
print(f"燃料电池模式: 0x{self.fcmode:02X}")
|
|
print(f"燃料电池控制命令: 0x{self.fcuCmd:02X}")
|
|
print(f"燃料电池功率效率: {self.fcuPowerEfficiency}%")
|
|
print(f"仪表电池充电使能: 0x{self.insChargeEnable:02X}")
|
|
print(f"仪表电池命令: 0x{self.insBatCmd:02X}")
|
|
print(f"动力电池命令: 0x{self.powerBatCmd:02X}")
|
|
print(f"动力电池功率效率: {self.powerBatEfficiency}%")
|
|
print(f"心跳: {self.heartbeat}")
|
|
print(f"主控状态: 0x{self.masterState:02X}")
|
|
print(f"占位符1: {self.paceholder1}")
|
|
print(f"占位符2: {self.paceholder2}")
|
|
print(f"校验码: 0x{self.checkCode:02X}")
|
|
|
|
|
|
# 测试函数
|
|
def test_msg_ccUFbMsg():
|
|
"""测试msg_ccUFbMsg类"""
|
|
msg = msg_ccUFbMsg()
|
|
|
|
# 设置一些测试数据
|
|
msg.fc_mode = 0x02 # 自动模式
|
|
msg.fc_status = 0x06 # 运行状态
|
|
msg.output_power_limit = 15000
|
|
msg.total_generation_time = 100 # 10小时
|
|
msg.hydrogen_capacity = 80
|
|
msg.liquid_oxygen_capacity = 75
|
|
msg.ins_soc = 85
|
|
msg.dyn_soc = 90
|
|
msg.heartbeat = 12345
|
|
|
|
# 打包消息
|
|
data = msg.pack()
|
|
print(f"打包后的数据长度: {len(data)} 字节")
|
|
|
|
# 解包消息
|
|
msg2 = msg_ccUFbMsg()
|
|
if msg2.unpack(data):
|
|
print("解包成功!")
|
|
msg2.print_info()
|
|
else:
|
|
print("解包失败!")
|
|
|
|
return msg2
|
|
|
|
|
|
def test_msg_ccUCmdMsg():
|
|
"""测试msg_ccUCmdMsg类"""
|
|
msg = msg_ccUCmdMsg()
|
|
|
|
# 设置一些测试数据
|
|
msg.year = 2025
|
|
msg.month = 9
|
|
msg.day = 11
|
|
msg.hour = 17
|
|
msg.minute = 5
|
|
msg.second = 30
|
|
msg.milliseconds = 123
|
|
|
|
msg.fcmode = 0x02 # 自动模式
|
|
msg.fcuCmd = 0x02 # 启动命令
|
|
msg.fcuPowerEfficiency = 85 # 85%效率
|
|
msg.insChargeEnable = 0x01 # 使能充电
|
|
msg.insBatCmd = 0x10 # 仪表电池启动
|
|
msg.powerBatCmd = 0x10 # 动力电池启动
|
|
msg.powerBatEfficiency = 90 # 90%效率
|
|
msg.heartbeat = 123
|
|
msg.masterState = 0x01 # 主控正常
|
|
msg.paceholder1 = 0
|
|
msg.paceholder2 = 0
|
|
|
|
# 打包消息
|
|
data = msg.pack()
|
|
print(f"CCU命令消息打包后的数据长度: {len(data)} 字节")
|
|
|
|
# 解包消息
|
|
msg2 = msg_ccUCmdMsg()
|
|
if msg2.unpack(data):
|
|
print("CCU命令消息解包成功!")
|
|
msg2.print_info()
|
|
else:
|
|
print("CCU命令消息解包失败!")
|
|
|
|
return msg2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=== 测试CCU状态反馈消息 ===")
|
|
test_msg_ccUFbMsg()
|
|
print("\n=== 测试CCU命令消息 ===")
|
|
test_msg_ccUCmdMsg()
|