#!/usr/bin/env python import os import os.path import struct import sys from pya2l import DB, model from pyxcp.dllif import getKey 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, } FLOAT_TYPES = {"FLOAT16_IEEE", "FLOAT32_IEEE", "FLOAT64_IEEE"} INT_TYPES = {"SBYTE", "SWORD", "SLONG", "A_INT64"} UINT_TYPES = {"UBYTE", "UWORD", "ULONG", "A_UINT64"} def get_byte_order(byte_order): if byte_order: byte_order = "little" if byte_order.byteOrder == "MSB_LAST" else "big" else: byte_order = "little" return byte_order def set_and_start_daq(master, mode, daq, event_channel, prescalar, priority): try: master.setDaqListMode(mode, daq, event_channel, prescalar, priority) master.startStopDaqList(0x02, daq) except Exception as e: print(e) print("Failed to set and start DAQ") sys.exit() def alloc_daq(master, daq_num): try: master.freeDaq() master.allocDaq(daq_num) except Exception as e: print(e) print("Failed to alloc DAQ") sys.exit() def alloc_odt_to_daq(master, daq, odt_count): try: master.allocOdt(daq, odt_count) except Exception as e: print(e) print("Failed to alloc ODT to DAQ") sys.exit() def alloc_odt_entry(master, daq, odt_number, odt_entries_count): try: master.allocOdtEntry(daq, odt_number, odt_entries_count) except Exception as e: print(e) print("Failed to alloc ODT entry") sys.exit() class XCPclass: def __init__(self, master): self.master = master self.common_dict = [] self.one_big_dict = {} self.tmp_values = [] def my_callback(self, msg_type, response, counter, length, timestamp): data = response odt = int.from_bytes(data[0:1], "big") daq = int.from_bytes(data[1:2], "big") for measurement in self.common_dict: datatype = measurement["datatype"] if measurement["has_two_odts"]: if measurement["ODT_0"] == odt and measurement["DAQ"] == daq: size = measurement["entry_size_0"] position = measurement["ODT_ENTRY_pos_0"] if datatype in FLOAT_TYPES or datatype in INT_TYPES: value = data[2 + position : 2 + position + size] self.tmp_values.insert(0, value) else: value = data[2 + position : 2 + position + size] self.tmp_values.insert(0, value) elif measurement["ODT_1"] == odt and measurement["DAQ"] == daq: size = measurement["entry_size_1"] position = measurement["ODT_ENTRY_pos_1"] name = measurement["name"] byte_order = measurement["byte_order"] if datatype in FLOAT_TYPES: result = data[2 + position : 2 + position + size] value = struct.unpack("d", self.tmp_values[0] + result) self.one_big_dict[f"{name}"].append(value) elif datatype in INT_TYPES: result = data[2 + position : 2 + position + size] value = int.from_bytes( self.tmp_values[0] + result, byte_order, signed=True ) self.one_big_dict[f"{name}"].append(hex(value)) else: result = data[2 + position : 2 + position + size] value = int.from_bytes( self.tmp_values[0] + result, byte_order, signed=False ) self.one_big_dict[f"{name}"].append(hex(value)) else: if measurement["ODT"] == odt and measurement["DAQ"] == daq: size = measurement["entry_size"] position = measurement["ODT_ENTRY_pos"] data_type = measurement["datatype"] name = measurement["name"] byte_order = measurement["byte_order"] if data_type in FLOAT_TYPES: result = data[2 + position : 2 + position + size] value = struct.unpack("f", result) self.one_big_dict[f"{name}"].append(value) elif data_type in INT_TYPES: value = int.from_bytes( data[2 + position : 2 + position + size], byte_order, signed=True, ) self.one_big_dict[f"{name}"].append(hex(value)) else: value = int.from_bytes( data[2 + position : 2 + position + size], byte_order, signed=False, ) self.one_big_dict[f"{name}"].append(hex(value)) def get_seed_and_unlock(self, master, resource, dll_path): try: key_parts = [] res = master.getSeed(0, resource) seed = res.seed length = res.length while length - len(seed): res = master.getSeed(1, resource) seed += res.seed result_code, key = getKey(dll_path, 1, seed, False) if result_code == 0: for i in range(0, len(key), 6): key_parts.append(key[i : i + 6]) master.unlock(len(key), key_parts[0]) for i in range(1, len(key_parts)): master.unlock(len(key_parts[i]), key_parts[i]) except Exception as e: print(e) print("Failed to get seed or unlock resource") sys.exit() def build_checksum_for_slave(self, master, a2l_file_path): try: 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) 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") sys.exit() def set_mta_and_build_checksum(self, master, a2l_file_path): try: 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) 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") sys.exit() def set_characteristic_value(self, master, value, name, a2l_file_path): try: 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) 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_value = wanted_dict.lowerLimit max_value = wanted_dict.upperLimit address = wanted_dict.address addr_ext = wanted_dict.ecu_address_extension.extension data_type = wanted_dict.datatype if min_value <= value <= max_value: 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") sys.exit() def get_characteristic_value(self, master, name, a2l_file_path): try: 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) 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 or 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") sys.exit() def get_measurement(self, master, name, a2l_file_path): try: 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) 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 or 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") sys.exit() def create_measurement_list( self, master, event_channel, measurement_name, timestamp_enable, a2l_file_path ): try: bit_offset = 0xFF wanted_dict = {} num_of_daqs = 0 daq_pos = 0 counter = 0 pos_entry_8bit, pos_entry_16bit, pos_entry_32bit, pos_entry_64bit = ( 0, 0, 0, 0, ) ( pos_odt_8bit, pos_odt_16bit, pos_odt_32bit, pos_odt_1_64bit, pos_odt_2_64bit, ) = 0, 0, 0, 0, 1 daq_8bit_exists, daq_16bit_exists, daq_32bit_exists, daq_64bit_exists = ( False, False, False, False, ) ( odt_entry_count_8bit, odt_entry_count_16bit, odt_entry_count_32bit, odt_entry_count_64bit, ) = 0, 0, 0, 0 dict_8bit, dict_16bit, dict_32bit, dict_64bit = [], [], [], [] ( odt_daq_dict_8bit, odt_daq_dict_16bit, odt_daq_dict_32bit, odt_daq_dict_64bit, ) = {}, {}, {}, {} list_8bit, list_16bit, list_32bit, list_64bit = [], [], [], [] 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) measurement_dict = session.query(model.Measurement).all() for name in measurement_name: wanted_dict[name] = {} for i in range(len(measurement_dict)): if measurement_dict[i].name == name: wanted_dict[name] = measurement_dict[i] if DATATYPES[wanted_dict[name].datatype] == 1: daq_8bit_exists = True dict_8bit.append(wanted_dict[name]) elif DATATYPES[wanted_dict[name].datatype] == 2: daq_16bit_exists = True dict_16bit.append(wanted_dict[name]) elif DATATYPES[wanted_dict[name].datatype] == 4: daq_32bit_exists = True dict_32bit.append(wanted_dict[name]) elif DATATYPES[wanted_dict[name].datatype] == 8: daq_64bit_exists = True dict_64bit.append(wanted_dict[name]) break if daq_8bit_exists: num_of_daqs += 1 first_odt_entry = True for data in dict_8bit: daq_pos = 0 datatype = data.datatype measurement_size = DATATYPES[datatype] byte_order = get_byte_order(data.byte_order) address = data.ecu_address.address addr_ext = data.ecu_address_extension.extension if odt_entry_count_8bit == 0 and timestamp_enable and daq_pos == 0: pos_entry_8bit = 4 if (pos_entry_8bit + measurement_size) > 6: odt_daq_dict_8bit.update({pos_odt_8bit: odt_entry_count_8bit}) pos_odt_8bit += 1 pos_entry_8bit = 0 odt_entry_count_8bit = 0 tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_8bit, "ODT_ENTRY_pos": pos_entry_8bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": False, } pos_entry_8bit += 1 odt_entry_count_8bit += 1 self.common_dict.append(tmp) list_8bit.append(tmp) first_odt_entry = False self.one_big_dict[f"{data.name}"] = [] odt_daq_dict_8bit.update({pos_odt_8bit: odt_entry_count_8bit}) daq_pos += 1 if daq_16bit_exists: num_of_daqs += 1 first_odt_entry = True has_two_odts = False for data in dict_16bit: datatype = data.datatype measurement_size = DATATYPES[datatype] address = data.ecu_address.address addr_ext = data.ecu_address_extension.extension byte_order = get_byte_order(data.byte_order) if odt_entry_count_16bit == 0 and timestamp_enable and daq_pos == 0: pos_entry_16bit = 4 if (pos_entry_16bit + measurement_size) > 6: odt_daq_dict_16bit.update( {pos_odt_16bit: odt_entry_count_16bit} ) pos_odt_16bit += 1 pos_entry_16bit = 0 odt_entry_count_16bit = 0 tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_16bit, "ODT_ENTRY_pos": pos_entry_16bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": has_two_odts, } pos_entry_16bit += 2 odt_entry_count_16bit += 1 self.common_dict.append(tmp) list_16bit.append(tmp) first_odt_entry = False self.one_big_dict[f"{data.name}"] = [] odt_daq_dict_16bit.update({pos_odt_16bit: odt_entry_count_16bit}) daq_pos += 1 if daq_32bit_exists: num_of_daqs += 1 first_odt_entry = True for data in dict_32bit: datatype = data.datatype measurement_size = DATATYPES[datatype] address = data.ecu_address.address addr_ext = data.ecu_address_extension.extension byte_order = get_byte_order(data.byte_order) if timestamp_enable: if first_odt_entry: odt_entry_count_32bit = 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) pos_entry_32bit = 4 tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_32bit, "ODT_1": pos_odt_32bit + 1, "ODT_ENTRY_pos_0": pos_entry_32bit, "ODT_ENTRY_pos_1": pos_entry_32bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } first_odt_entry = False pos_entry_32bit = 2 pos_odt_32bit += 1 self.common_dict.append(tmp) list_32bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: if pos_entry_32bit < 6 or pos_entry_32bit == 6: if pos_entry_32bit == 6: pos_entry_32bit = 0 odt_entry_count_32bit = 1 pos_odt_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) else: odt_entry_count_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) if pos_entry_32bit == 4: tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_32bit, "ODT_1": pos_odt_32bit + 1, "ODT_ENTRY_pos_0": pos_entry_32bit, "ODT_ENTRY_pos_1": pos_entry_32bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } pos_odt_32bit += 1 odt_entry_count_32bit = 1 else: tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_32bit, "ODT_ENTRY_pos": pos_entry_32bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": False, } pos_entry_32bit += 4 self.common_dict.append(tmp) list_32bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: odt_entry_count_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) pos_entry_32bit = pos_entry_32bit % 6 tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_32bit, "ODT_ENTRY_pos": pos_entry_32bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": False, } pos_entry_32bit += 4 self.common_dict.append(tmp) list_32bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: if pos_entry_32bit < 6 or pos_entry_32bit == 6: if pos_entry_32bit == 6: pos_entry_32bit = 0 odt_entry_count_32bit = 1 pos_odt_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) else: odt_entry_count_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) if pos_entry_32bit == 4: tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_32bit, "ODT_1": pos_odt_32bit + 1, "ODT_ENTRY_pos_0": pos_entry_32bit, "ODT_ENTRY_pos_1": pos_entry_32bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } pos_odt_32bit += 1 odt_entry_count_32bit = 1 else: tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_32bit, "ODT_ENTRY_pos": pos_entry_32bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": False, } pos_entry_32bit += 4 self.common_dict.append(tmp) list_32bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: odt_entry_count_32bit += 1 odt_daq_dict_32bit.update( {pos_odt_32bit: odt_entry_count_32bit} ) pos_entry_32bit = pos_entry_32bit % 6 tmp = { "name": data.name, "DAQ": daq_pos, "ODT": pos_odt_32bit, "ODT_ENTRY_pos": pos_entry_32bit, "address": address, "addr_ext": addr_ext, "entry_size": measurement_size, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": False, } pos_entry_32bit += 4 self.common_dict.append(tmp) list_32bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] odt_daq_dict_32bit.update({pos_odt_32bit: odt_entry_count_32bit}) daq_pos += 1 if daq_64bit_exists: num_of_daqs += 1 first_odt_entry = True for data in dict_64bit: datatype = data.datatype measurement_size = DATATYPES[datatype] address = data.ecu_address.address addr_ext = data.ecu_address_extension.extension byte_order = get_byte_order(data.byte_order) if timestamp_enable: if first_odt_entry: odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) pos_entry_64bit = 4 tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 6, "entry_size_1": 6, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } first_odt_entry = False pos_odt_1_64bit += 2 self.common_dict.append(tmp) list_64bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: if counter == 0: pos_entry_64bit = 0 odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit, "address_0": address, "address_1": address + 6, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } pos_odt_1_64bit += 1 counter = 1 elif counter == 1: pos_entry_64bit = 2 odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit + 1} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit - 2, "address_0": address, "address_1": address + 4, "addr_ext": addr_ext, "entry_size_0": measurement_size - 4, "entry_size_1": 4, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } counter = 2 pos_odt_1_64bit += 1 elif counter == 2: pos_entry_64bit = 4 odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit + 1} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 6, "entry_size_1": 6, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } counter = 0 pos_odt_1_64bit += 1 self.common_dict.append(tmp) list_64bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: if first_odt_entry: odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) pos_entry_64bit = 0 tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit, "address_0": address, "address_1": address + 6, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } first_odt_entry = False pos_odt_1_64bit += 1 self.common_dict.append(tmp) list_64bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] else: if counter == 0: pos_entry_64bit = 2 odt_entry_count_64bit += 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit} ) odt_daq_dict_64bit.update({pos_odt_1_64bit + 1: 1}) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit - 2, "address_0": address, "address_1": address + 4, "addr_ext": addr_ext, "entry_size_0": measurement_size - 4, "entry_size_1": 4, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } pos_odt_1_64bit += 1 counter = 1 elif counter == 1: pos_entry_64bit = 4 odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit + 1} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit - 4, "address_0": address, "address_1": address + 2, "addr_ext": addr_ext, "entry_size_0": measurement_size - 6, "entry_size_1": 6, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } counter = 2 pos_odt_1_64bit += 2 elif counter == 2: pos_entry_64bit = 0 odt_entry_count_64bit = 1 odt_daq_dict_64bit.update( {pos_odt_1_64bit: odt_entry_count_64bit} ) odt_daq_dict_64bit.update( {pos_odt_1_64bit + 1: odt_entry_count_64bit} ) tmp = { "name": data.name, "DAQ": daq_pos, "ODT_0": pos_odt_1_64bit, "ODT_1": pos_odt_1_64bit + 1, "ODT_ENTRY_pos_0": pos_entry_64bit, "ODT_ENTRY_pos_1": pos_entry_64bit, "address_0": address, "address_1": address + 6, "addr_ext": addr_ext, "entry_size_0": measurement_size - 2, "entry_size_1": 2, "datatype": datatype, "byte_order": byte_order, "first_odt_entry": first_odt_entry, "has_two_odts": True, } pos_odt_1_64bit += 1 counter = 0 self.common_dict.append(tmp) list_64bit.append(tmp) self.one_big_dict[f"{data.name}"] = [] odt_daq_dict_64bit.update({pos_odt_1_64bit: odt_entry_count_64bit}) daq_pos += 1 daq_pos = 0 master.freeDaq() master.allocDaq(num_of_daqs) if daq_8bit_exists: master.allocOdt(daq_pos, len(odt_daq_dict_8bit)) for j in range(len(odt_daq_dict_8bit)): num_of_odt_entry = odt_daq_dict_8bit[j] master.allocOdtEntry(daq_pos, j, num_of_odt_entry) daq_pos += 1 if daq_16bit_exists: master.allocOdt(daq_pos, len(odt_daq_dict_16bit)) for j in range(len(odt_daq_dict_16bit)): num_of_odt_entry = odt_daq_dict_16bit[j] master.allocOdtEntry(daq_pos, j, num_of_odt_entry) daq_pos += 1 if daq_32bit_exists: master.allocOdt(daq_pos, len(odt_daq_dict_32bit)) for j in range(len(odt_daq_dict_32bit)): num_of_odt_entry = odt_daq_dict_32bit[j] master.allocOdtEntry(daq_pos, j, num_of_odt_entry) daq_pos += 1 if daq_64bit_exists: master.allocOdt(daq_pos, len(odt_daq_dict_64bit)) for j in range(len(odt_daq_dict_64bit)): num_of_odt_entry = odt_daq_dict_64bit[j] master.allocOdtEntry(daq_pos, j, num_of_odt_entry) daq_pos += 1 previous_odt = 55 previous_daq = 55 if daq_8bit_exists: for data in list_8bit: daq = data["DAQ"] odt = data["ODT"] if daq != previous_daq or odt != previous_odt: master.setDaqPtr(daq, odt, 0) master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) else: master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) previous_odt = odt previous_daq = daq previous_odt = 55 previous_daq = 55 if daq_16bit_exists: for data in list_16bit: daq = data["DAQ"] odt = data["ODT"] if daq != previous_daq or odt != previous_odt: master.setDaqPtr(daq, odt, 0) master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) else: master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) previous_odt = odt previous_daq = daq previous_odt = 55 previous_daq = 55 if daq_32bit_exists: for data in list_32bit: if data["has_two_odts"]: daq = data["DAQ"] odt0 = data["ODT_0"] odt1 = data["ODT_1"] if timestamp_enable and data["first_odt_entry"]: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: if daq != previous_daq or odt0 != previous_odt: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) else: master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) previous_odt = odt1 previous_daq = daq else: daq = data["DAQ"] odt = data["ODT"] if daq != previous_daq or odt != previous_odt: master.setDaqPtr(daq, odt, 0) master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) else: master.writeDaq( bit_offset, data["entry_size"], data["addr_ext"], data["address"], ) previous_odt = odt previous_daq = daq previous_odt = 55 previous_daq = 55 if daq_64bit_exists: for data in list_64bit: if data["has_two_odts"]: daq = data["DAQ"] odt0 = data["ODT_0"] odt1 = data["ODT_1"] if timestamp_enable: if data["first_odt_entry"]: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: if daq != previous_daq or odt0 != previous_odt: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: if data["first_odt_entry"]: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: if daq != previous_daq or odt0 != previous_odt: master.setDaqPtr(data["DAQ"], data["ODT_0"], 0) master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) else: master.writeDaq( bit_offset, data["entry_size_0"], data["addr_ext"], data["address_0"], ) master.setDaqPtr(data["DAQ"], data["ODT_1"], 0) master.writeDaq( bit_offset, data["entry_size_1"], data["addr_ext"], data["address_1"], ) previous_odt = odt1 previous_daq = daq mode = 0x10 if timestamp_enable else 0x00 if timestamp_enable: set_and_start_daq( master=master, mode=mode, daq=0, event_channel=event_channel, prescalar=1, priority=1, ) for i in range(1, num_of_daqs): set_and_start_daq( master=master, mode=0x00, daq=i, event_channel=event_channel, prescalar=1, priority=1, ) else: for i in range(num_of_daqs): set_and_start_daq( master=master, mode=mode, daq=i, event_channel=event_channel, prescalar=1, priority=1, ) session.close() return self.common_dict, self.one_big_dict except Exception as e: print(e) sys.exit()