import struct import logging import binascii class Command30: def __init__(self): self.command = 0x30 # BMS状态信息命令 def parse_30h_bms_status(self, data): """ 解析30H BMS状态需求报文 :param data: 完整的30H命令报文 :return: 解析后的字典或None """ try: # 验证基本帧格式 if len(data) < 14 or data[0:2] != b'JX' or data[2] != 0x30: logging.warning(f"30H命令帧格式不正确，原始报文: {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 = 20 # 解析电压需求 voltage_request = struct.unpack("> 2) & 0x03), "充电过流": ((warning_bytes >> 4) & 0x03), "动力蓄电池温度过高": ((warning_bytes >> 6) & 0x03) } current_index += 1 # 打印解析结果 print("\n30H BMS状态需求报文解析结果:") print(f"桩号: {pile_id_bytes.hex()}") print(f"时间标识: {timestamp}") print(f"电压需求: {voltage_request}V") print(f"电流需求: {current_request}A") print(f"充电模式: {self.get_charging_mode_text(charging_mode)}") print(f"最高单体电压: {max_cell_voltage}V") print(f"最高单体电压所在电池组号: {max_cell_group}") print(f"SOC: {soc}%") print(f"剩余充电时间: {remaining_charging_time}分钟") print(f"最高单体电压电池组编号: {max_voltage_group_number}") print(f"最高动力蓄电池温度: {max_battery_temperature}°C") print(f"最高温度检测点编号: {max_temperature_point}") print(f"最低动力蓄电池温度: {min_battery_temperature}°C") print(f"最低温度监测点编号: {min_temperature_point}") print("告警信息:") for warning, level in warnings.items(): print(f" {warning}: {self.get_warning_level_text(level)}") return { "pile_id": pile_id_bytes.hex(), "timestamp": timestamp, "voltage_request": voltage_request, "current_request": current_request, "charging_mode": self.get_charging_mode_text(charging_mode), "max_cell_voltage": max_cell_voltage, "max_cell_group": max_cell_group, "soc": soc, "remaining_charging_time": remaining_charging_time, "max_voltage_group_number": max_voltage_group_number, "max_battery_temperature": max_battery_temperature, "max_temperature_point": max_temperature_point, "min_battery_temperature": min_battery_temperature, "min_temperature_point": min_temperature_point, "warnings": {k: self.get_warning_level_text(v) for k, v in warnings.items()} } except Exception as e: logging.error(f"解析30H命令失败: {str(e)}") logging.error(f"原始报文: {binascii.hexlify(data)}") return None def get_charging_mode_text(self, mode): """ 解析充电模式 :param mode: 充电模式字节 :return: 充电模式文本描述 """ mode_map = { 0x01: "恒流充电", 0x02: "恒压充电", 0x03: "涓流充电", 0x04: "充电完成", 0x05: "充电终止" } return mode_map.get(mode, f"未知模式 (0x{mode:02X})") def get_warning_level_text(self, level): """ 解析告警级别 :param level: 告警级别 :return: 告警级别文本描述 """ level_map = { 0x00: "正常", 0x01: "预警", 0x02: "严重告警", 0x03: "故障" } return level_map.get(level, f"未知级别 (0x{level:02X})") def process_30h_bms_status(self, data): """ 处理30H BMS状态信息命令 :param data: 完整的30H命令报文 :return: 是否成功处理 """ try: parsed_data = self.parse_30h_bms_status(data) if parsed_data is None: logging.warning("30H命令解析失败") return False # 记录BMS状态信息日志 logging.info( f"收到桩号 {parsed_data['pile_id']} 的BMS状态信息: " f"SOC {parsed_data['soc']}%, " f"电压需求 {parsed_data['voltage_request']}V, " f"电流需求 {parsed_data['current_request']}A" ) return True except Exception as e: logging.error(f"处理30H命令出错: {str(e)}") return False # 测试用例 if __name__ == "__main__": # 30H命令测试报文 test_30_data = bytes.fromhex( "4A 58 30 03 17 66 56 11 36 06 37 01 1C 00 19 01 09 0B 25 15 01 60 1B 8D 07 02 DA 07 A0 0F 4A C1 49 78 00 01 40 03 3C 05 00 10 64") parser = Command30() # 测试解析30H命令 parser.process_30h_bms_status(test_30_data)