import struct from typing import MutableMapping, Union from . import tools def parse(response: bytes) -> MutableMapping[str, Union[int, bytes, str, float]]: # print(response, 1111111111111) out: MutableMapping[str, Union[int, bytes, str, float]] = dict() out["full_response"] = response if response[0] & 0x80: # long address out["address"] = int.from_bytes(response[1:6], "big") response = response[6:] else: # short address out["address"] = response[1] response = response[2:] command, bytecount, response_code, device_status = struct.unpack_from(">BBBB", response) out["device_status"] = device_status out["response_code"] = response_code if response_code == 0: out["status"] = "success" else: out["status"] = "error" out["error"] = f"Device responded with error code: {response_code}" data = response[4 : 4 + bytecount] out["command"] = command out["command_name"] = f"hart_command_{command}" out["bytecount"] = bytecount out["data"] = data # handle error return if bytecount == 2: return out # universal commands if command in [0, 11]: out["command_name"] = "read_unique_identifier" out["manufacturer_id"] = data[1] out["manufacturer_device_type"] = data[2] out["number_response_preamble_characters"] = data[3] out["universal_command_revision_level"] = data[4] out["transmitter_specific_command_revision_level"] = data[5] out["software_revision_level"] = data[6] out["hardware_revision_level"] = data[7] out["device_id"] = int.from_bytes(data[9:12], "big") elif command in [1]: out["command_name"] = "read_primary_variable" units, variable = struct.unpack_from(">Bf", data) out["primary_variable_units"] = units out["primary_variable"] = variable elif command in [2]: out["command_name"] = "read_loop_current_and_percent" analog_signal, primary_variable = struct.unpack_from(">ff", data) out["analog_signal"] = analog_signal out["primary_variable"] = primary_variable elif command in [3]: out["command_name"] = "read_dynamic_variables_and_loop_current" if len(data) >= 4: analog_signal = struct.unpack_from(">f", data, 0)[0] out["analog_signal"] = analog_signal # 计算剩余数据长度并确定支持的变量数量 remaining_length = len(data) - 4 variable_count = remaining_length // 5 # 每个变量占5字节 (1字节单位 + 4字节值) # 动态解析变量数据 offset = 4 # 从第4字节开始是变量数据 for i in range(variable_count): if offset + 5 > len(data): # 确保有足够数据 break # 解析单位枚举值 (1字节) unit_enum = struct.unpack_from(">B", data, offset)[0] offset += 1 # 解析变量值 (4字节) variable_value = struct.unpack_from(">f", data, offset)[0] offset += 4 # 根据变量索引设置字段名 if i == 0: out["primary_variable_units"] = unit_enum out["primary_variable"] = variable_value elif i == 1: out["secondary_variable_units"] = unit_enum out["secondary_variable"] = variable_value elif i == 2: out["tertiary_variable_units"] = unit_enum out["tertiary_variable"] = variable_value elif i == 3: out["quaternary_variable_units"] = unit_enum out["quaternary_variable"] = variable_value elif command in [6]: out["command_name"] = "write_polling_address" polling_address = struct.unpack_from(">B", data)[0] out["polling_address"] = polling_address elif command in [12]: # print(data) out["command_name"] = "read_message" out["message"] = data[0:24] # print(out, 111111111) elif command in [13]: out["command_name"] = "read_tag_descriptor_date" # Tag is 8 chars packed into 6 bytes out["device_tag_name"] = tools.unpack_packed_ascii(data[0:6], 8) # Descriptor is 16 chars packed into 12 bytes out["device_descriptor"] = tools.unpack_packed_ascii(data[6:18], 16) # Date is 3 bytes: day, month, year (year is offset from 1900) day, month, year_offset = struct.unpack_from(">BBB", data, 18) out["date"] = {"day": day, "month": month, "year": year_offset + 1900} elif command in [14]: out["command_name"] = "read_primary_variable_information" out["serial_no"] = data[0:3] sensor_limits_code, upper_limit, lower_limit, min_span = struct.unpack_from( ">xxxBfff", data ) out["sensor_limits_code"] = sensor_limits_code out["upper_limit"] = upper_limit out["lower_limit"] = lower_limit out["min_span"] = min_span elif command in [15]: out["command_name"] = "read_output_information" ( alarm_code, transfer_fn_code, primary_variable_range_code, upper_range_value, lower_range_value, damping_value, write_protect, private_label, ) = struct.unpack_from(">BBBfffBB", data) out["alarm_code"] = alarm_code out["transfer_fn_code"] = transfer_fn_code out["primary_variable_range_code"] = primary_variable_range_code out["upper_range_value"] = upper_range_value out["lower_range_value"] = lower_range_value out["damping_value"] = damping_value out["write_protect"] = write_protect out["private_label"] = private_label elif command in [16]: out["command_name"] = "read_final_assembly_number" # print(data) out["final_assembly_no"] = int.from_bytes(data[0:3], "big") elif command in [17]: out["command_name"] = "write_message" out["message"] = data[0:24] elif command in [18]: out["command_name"] = "write_tag_descriptor_date" out["device_tag_name"] = data[0:6] out["device_descriptor"] = data[6:18] out["date"] = data[18:21] elif command in [19]: out["command_name"] = "write_final_assembly_number" out["final_assembly_no"] = int.from_bytes(data[0:2], "big") elif command in [34]: out["command_name"] = "write_primary_variable_damping" out["damping_time"] = struct.unpack_from(">f", data)[0] elif command in [35]: out["command_name"] = "write_primary_variable_range" out["upper_range"] = struct.unpack_from(">f", data)[0] out["lower_range"] = struct.unpack_from(">f", data[4:])[0] elif command in [36]: out["command_name"] = "calibrate_20ma" # 20mA校准命令无返回数据 elif command in [37]: out["command_name"] = "calibrate_4ma" # 4mA校准命令无返回数据 elif command in [40]: out["command_name"] = "set_fixed_current_output" if len(data) > 0: out["fixed_output_enabled"] = data[0] == 1 if len(data) > 1: out["fixed_current_value"] = struct.unpack_from(">f", data[1:])[0] elif command in [43]: out["command_name"] = "calibrate_zero_point" # 零点校准命令无返回数据 elif command in [44]: out["command_name"] = "write_primary_variable_units" out["units_code"] = data[0] elif command in [45]: out["command_name"] = "trim_loop_current_4ma" out["measured_current"] = struct.unpack_from(">f", data)[0] elif command in [46]: out["command_name"] = "trim_loop_current_20ma" out["measured_current"] = struct.unpack_from(">f", data)[0] elif command in [47]: out["command_name"] = "write_primary_variable_output_function" out["transfer_function_code"] = data[0] # COMMON COMMANDS # elif command in [37]: # out["command_name"] = "set_primary_variable_lower_range_value" # out[""] = # request data bytes = NONE, response data bytes = NONE # elif command in [38]: # out["command_name"] = "reset_configuration_changed_flag" # out[""] = # request data bytes = NONE, response data bytes = NONE # elif command in [42]: # out["command_name"] = "perform_master_reset" # out[""] = # request data bytes = NONE, response data bytes = NONE # elif command in [48]: # out["command_name"] = "read_additional_transmitter_status" # out[""] = # request data bytes = NONE, response data bytes = NONE elif command in [50]: out["command_name"] = "read_dynamic_variable_assignments" ( primary_transmitter_variable, secondary_transmitter_variable, tertiary_transmitter_variable, quaternary_transmitter_variable, ) = struct.unpack_from(">BBBB", data) out["primary_transmitter_variable"] = primary_transmitter_variable out["secondary_transmitter_variable"] = secondary_transmitter_variable out["tertiary_transmitter_variable"] = tertiary_transmitter_variable # NOT USED out["quaternary_transmitter_variable"] = quaternary_transmitter_variable # NOT USED elif command in [59]: out["command_name"] = "write_number_of_response_preambles" n_response_preambles = struct.unpack_from(">B", data)[0] out["n_response_preambles"] = n_response_preambles elif command in [66]: out["command_name"] = "toggle_analog_output_mode" ( analog_output_selection, analog_output_units_code, fixed_analog_output, ) = struct.unpack_from(">BBf", data) out["analog_output_selection"] = analog_output_selection out["analog_output_units_code"] = analog_output_units_code out["fixed_analog_output"] = fixed_analog_output elif command in [67]: out["command_name"] = "trim_analog_output_zero" analog_output_code, analog_output_units_code, measured_analog_output = struct.unpack_from( ">BBf", data ) out["analog_output_code"] = analog_output_code out["analog_output_units_code"] = analog_output_units_code out["measured_analog_output"] = measured_analog_output elif command in [68]: out["command_name"] = "trim_analog_output_span" analog_output_code, analog_output_units_code, measured_analog_output = struct.unpack_from( ">BBf", data ) out["analog_output_code"] = analog_output_code out["analog_output_units_code"] = analog_output_units_code out["measured_analog_output"] = measured_analog_output elif command in [123]: out["command_name"] = "select_baud_rate" out["baud_rate"] = int.from_bytes(data, "big") return out