import logging import os.path import os, struct import csv from pyxcp.dllif import getKey from pya2l import DB import pya2l.model as model DATATYPES = { "SBYTE": 1, "SWORD": 2, "SLONG": 4, "A_INT64": 8, "UBYTE": 1, "UWORD": 2, "ULONG": 4, "A_UINT64": 8, "FLOAT16_IEEE": 2, "FLOAT32_IEEE": 4, "FLOAT64_IEEE": 8 } DATATYPE_MAP = { "UBYTE": "U8", "SBYTE": "I8", "UWORD": "U16", "SWORD": "I16", "ULONG": "U32", "SLONG": "I32", "A_UINT64": "U64", "A_INT64": "I64", "FLOAT16_IEEE": "F16", "FLOAT32_IEEE": "F32", "FLOAT64_IEEE": "F64", } FLOAT_TYPES = {"FLOAT16_IEEE", "FLOAT32_IEEE", "FLOAT64_IEEE"} INT_TYPES = {"SBYTE", "SWORD", "SLONG", "A_INT64"} UINT_TYPES = {"UBYTE", "UWORD", "ULONG", "A_UINT64"} def SeedKeyXCP(seed: bytes): temp0 = seed[0] temp1 = seed[1] temp2 = seed[2] temp3 = seed[3] temp = (temp3 << 24) | (temp2 << 16) | (temp1 << 8) | temp0 temp = (temp >> 5) | (temp << 23) temp = (temp * 7) temp = temp ^ 0x26031961 key = bytearray(9) key[0] = ((temp >> 0) & 0x00FF) key[1] = ((temp >> 8) & 0x00FF) key[2] = ((temp >> 16) & 0x00FF) key[3] = ((temp >> 24) & 0x00FF) return bytes(key) def open_db_session(a2l_file_path): db = DB() db_file_path = a2l_file_path + "db" if os.path.exists(db_file_path): session = db.open_existing(db_file_path) else: session = db.import_a2l(a2l_file_path) return db, session def create_measurement_list(measurement_names, session): daq_measurements = [] measurement_dict_by_name = {m.name: m for m in session.query(model.Measurement).all()} for m_name in measurement_names: m = measurement_dict_by_name.get(m_name) datatype = DATATYPE_MAP.get(m.datatype) ecu_address = m.ecu_address.address ext = m.ecu_address_extension.extension daq_measurements.append((m_name, ecu_address, ext, datatype)) return daq_measurements def parse_csv(csv_file): data_dict = {} with open(csv_file, mode='r', encoding="utf-8") as file: csv_reader = csv.reader(file) headers = next(csv_reader)[2:] for header in headers: data_dict[header] = [] for row in csv_reader: for i, value in enumerate(row[2:]): data_dict[headers[i]].append(value) return data_dict class XCPclass: def __init__(self, master): self.master = master self.logger = logging.getLogger(__name__) self.logger.setLevel(logging.DEBUG) self.config = master.config def get_seed_and_unlock(self, master, resource, dll_path=None): try: key_parts = [] res = master.getSeed(0, resource) seed = res.seed seed = bytearray(seed) length = res.length while length - len(seed): res = master.getSeed(1, resource) seed += res.seed if dll_path is None: key = SeedKeyXCP(seed) else: _, key = getKey(self.logger, self.config.custom_dll_loader, dll_path, 1, seed, False) for i in range(0, len(key), 6): key_parts.append(key[i:i + 6]) master.unlock(len(key), key_parts[0]) except Exception as e: print(e) print("Failed to get seed or unlock resource") exit() def build_checksum_for_slave(self, master, session): try: mod_par = session.query(model.ModPar).all() master.setMta(mod_par[0].memory_segment[0].address) master.buildChecksum(mod_par[0].memory_segment[0].size) master.setMta(mod_par[0].memory_segment[0].address) master.buildChecksum(128) master.setMta(mod_par[0].memory_segment[0].address + 0x80) master.buildChecksum(128) session.close() except Exception as e: print(e) print("Failed to build checksum for slave") exit() def set_mta_and_build_checksum(self, master, session): try: mod_par = session.query(model.ModPar).all() addr = mod_par[0].memory_segment[0].address size = mod_par[0].memory_segment[0].size while size > 0: if size < 128 and size != 0: master.setMta(addr) master.buildChecksum(size) size = 0 else: master.setMta(addr) master.buildChecksum(128) size = size - 128 addr = addr + 0x80 session.close() except Exception as e: print(e) print("Failed to set_mta_and_build_checksum for slave") exit() def set_characteristic_value(self, master, value, name, session): try: char_dict = session.query(model.Characteristic).all() rec_layout = session.query(model.RecordLayout).all() wanted_dict = {} for i in range(len(char_dict)): if char_dict[i].name == name: wanted_dict = char_dict[i] for i in range(len(rec_layout)): if wanted_dict.deposit == rec_layout[i].name: wanted_dict.datatype = rec_layout[i].fnc_values.datatype break break min = wanted_dict.lowerLimit max = wanted_dict.upperLimit address = wanted_dict.address addr_ext = wanted_dict.ecu_address_extension.extension data_type = wanted_dict.datatype if min <= value <= max: master.setMta(address, addr_ext) byte_order = "little" if wanted_dict.byte_order.byteOrder == "MSB_LAST" else "big" if data_type in FLOAT_TYPES: float_value = struct.pack('f', value) master.download(data=float_value) elif data_type in UINT_TYPES: data_size = DATATYPES[data_type] pwm_level = value.to_bytes(data_size, byte_order, signed=False) master.download(data=pwm_level) elif data_type in INT_TYPES: data_size = DATATYPES[data_type] pwm_level = value.to_bytes(data_size, byte_order, signed=True) master.download(data=pwm_level) else: print("Value type not valid") else: print("Value not in range") session.close() except Exception as e: print(e) print("Failed to set characteristic value") exit() def get_characteristic_value(self, master, name, session): try: char_dict = session.query(model.Characteristic).all() rec_layout = session.query(model.RecordLayout).all() wanted_dict = {} for i in range(len(char_dict)): if char_dict[i].name == name: wanted_dict = char_dict[i] for i in range(len(rec_layout)): if wanted_dict.deposit == rec_layout[i].name: wanted_dict.datatype = rec_layout[i].fnc_values.datatype break break data_type = wanted_dict.datatype data_size = DATATYPES[data_type] address = wanted_dict.address addr_ext = wanted_dict.ecu_address_extension.extension byte_order = "little" if wanted_dict.byte_order.byteOrder == "MSB_LAST" else "big" master.setMta(address, addr_ext) result = master.upload(length=data_size) if data_type in FLOAT_TYPES: result = struct.unpack('f', result) elif data_type in UINT_TYPES: result = int.from_bytes(result, byte_order) elif data_type in INT_TYPES: result = int.from_bytes(result, byte_order) session.close() return result except Exception as e: print(e) print("Failed to get characteristic value") exit() def get_measurement(self, master, name, session): try: measurement_dict = session.query(model.Measurement).all() wanted_dict = {} for i in range(len(measurement_dict)): if measurement_dict[i].name == name: wanted_dict = measurement_dict[i] break data_type = wanted_dict.datatype address = wanted_dict.ecu_address.address addr_ext = wanted_dict.ecu_address_extension.extension byte_order = "little" if wanted_dict.byte_order.byteOrder == "MSB_LAST" else "big" data_size = DATATYPES[data_type] master.setMta(address, addr_ext) result = master.upload(length=data_size) # result = master.shortUpload(address=address, length=data_size, addressExt=addr_ext) if data_type in FLOAT_TYPES: result = struct.unpack('f', result) elif data_type in UINT_TYPES: result = int.from_bytes(result, byte_order) elif data_type in INT_TYPES: result = int.from_bytes(result, byte_order) session.close() return result except Exception as e: print(e) print("Failed to get characteristic dict") exit()