Index: leahi_dialin/utils/abstract_classes.py =================================================================== diff -u -rc41ad46a9dea345a8feb4e674b6753999e381629 -r4c11f61e1e328407ed98a1b81a3e8b0adb278bd5 --- leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision c41ad46a9dea345a8feb4e674b6753999e381629) +++ leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision 4c11f61e1e328407ed98a1b81a3e8b0adb278bd5) @@ -24,6 +24,16 @@ from leahi_dialin.common.constants import MSG_HEADER_SIZE +class LocalVars: + """ + Reference class for local variables + """ + + def __init__(self, name): + self.name = name + self.value = None + + class AbstractObserver(ABC): """ Publicly accessible parent class for all observers. @@ -66,7 +76,7 @@ self._observers.remove(observer) - def process_into_vars(self, decoder_list: List[Tuple], message, start_from_byte: int=0, debug: bool=False) -> dict: + def process_into_vars(self, decoder_list: List[Tuple], message, start_from_byte: int=0, debug: bool=False) -> None: """ Process the CAN message with the help of the decoder list into variables and a dictionary. Note: updating variables will only be done when it's class wide one, aka "self.attr_name". @@ -79,102 +89,49 @@ :return: (Dictionary) A dictionary for the variable_name and value pair """ start_pos = MSG_HEADER_SIZE + start_from_byte - results = {} + result = {} # Needed if datatype is a string and indicates an earlier entry's value. Entry will be stored here and then it's value is read from here if debug: - print(f'\n\ndecoder_list: {decoder_list}') - print(f'full message: {message["message"]}') + print(f'\n\nDecoder_list: {decoder_list}') for decode_details in decoder_list: - # Content of the decode list - variable_name = decode_details[0] - # If last position is multichar length if isinstance(decode_details[-1], int): length = decode_details[-1] datatype: DataTypes = decode_details[-2] - else: + base_list_length = 3 + # If it's the normal Datatype + elif isinstance(decode_details[-1], DataTypes): length = 1 datatype: DataTypes = decode_details[-1] + base_list_length = 2 + # If it's a string containing the name of the datatype + else: + length = 1 + datatype = DataTypes(result[decode_details[-1]]) + base_list_length = 2 + # Check if the decoder list contains keys indicating the target is a dictionary + key_list_length = len(decode_details) - base_list_length + key_1 = decode_details[1] if len(decode_details) >= base_list_length + 1 else None + key_2 = decode_details[2] if len(decode_details) >= base_list_length + 2 else None + key_3 = decode_details[3] if len(decode_details) >= base_list_length + 3 else None end_pos = start_pos + datatype.size() - - for i in range(0, length): - end_pos = start_pos + datatype.size() - try: - new_value = struct.unpack(datatype.unpack_attrib(), bytearray(message['message'][start_pos:end_pos]))[0] - if length == 1: - value = new_value - elif datatype == DataTypes.U08 and length > 1: - if i == 0: - value = chr(new_value) - else: - value += chr(new_value) - if debug: - print(f'Message Part: {message["message"][start_pos:end_pos]}') - print(f'Input Data: {bytearray(message["message"][start_pos:end_pos])}') - print(f'New value: {new_value} + -> Value: {value}') - except Exception as e: - value = None - break - start_pos = end_pos - - if 'nan' in str(value).lower(): - value = None - break - # raise ValueError(f'{value} is not an accepted value!') - if datatype in [DataTypes.BOOL, DataTypes.BOOL_U08]: - value = True if value == 1 else False - results[variable_name] = value - # If it's a instance variable (self.) then set it's value - if variable_name.startswith('self'): - attr_path = variable_name[5:].split('.') - obj = self - for attr in attr_path[:-1]: - obj = getattr(obj, attr) - setattr(obj, attr_path[-1], value) - start_pos = end_pos - if debug: - print('Finished cycle\n') - if debug: - print(f'results: {results}\n') - return results - - - def process_into_dict(self, dict_to_update: dict, decoder_list: List[Tuple], message, start_from_byte: int=0, debug: bool=False): - """ - Process the CAN message with the help of the decoder list into a dictionary. - - :param decoder_list: (List[Tuple[DialEnum, DialEnum, DataTypes]]) Contains the dictioarny key names and DataType of the indexed message - :param message: (Bytearray) The raw CAN message - :param start_from_byte: (Integer) Start from the nth byte after the header - :param debug: (Boolean) Prints for debugging - :return: (Dictionary) The updated dictionary - """ - start_pos = MSG_HEADER_SIZE + start_from_byte - if debug: - print(f'\n\ndecoder_list: {decoder_list}') - print(f'full message: {message["message"]}') - for decode_details in decoder_list: - # Content of the decode list - key_1 = decode_details[0] - key_2 = decode_details[1] if len(decode_details) >= 3 else None - key_3 = decode_details[2] if len(decode_details) >= 4 else None + print(f'Len: {len(decode_details)}') + print(f'key_list_length: {key_list_length}') + if key_list_length > 0: + print(f'key_1: {key_1}') + print(f'key_2: {key_2}') + print(f'key_3: {key_3}') + print(f'datatype: {datatype} - {length}') - # If last position is multichar length - if isinstance(decode_details[-1], int): - length = decode_details[-1] - datatype: DataTypes = decode_details[-2] - key_list_length = len(decode_details) - 2 - else: - length = 1 - datatype: DataTypes = decode_details[-1] - key_list_length = len(decode_details) - 1 - + # Extract the value and convert it to correct type for i in range(0, length): end_pos = start_pos + datatype.size() try: new_value = struct.unpack(datatype.unpack_attrib(), bytearray(message['message'][start_pos:end_pos]))[0] + # If not multilength just save it if length == 1: value = new_value + # If multilength entry (length > 1), convert it to character and add append it to the value until fully read elif datatype == DataTypes.U08 and length > 1: if i == 0: value = chr(new_value) @@ -185,6 +142,8 @@ print(f'Input Data: {bytearray(message["message"][start_pos:end_pos])}') print(f'New value: {new_value} + -> Value: {value}') except Exception as e: + if debug: + print('Message extraction exception occured!') value = None break start_pos = end_pos @@ -195,25 +154,49 @@ # If the type is Bool, convert the value from Integer to Boolean if datatype in [DataTypes.BOOL, DataTypes.BOOL_U08]: value = True if value == 1 else False - # Save the value into the Dictionary - if key_list_length == 1: - dict_to_update[key_1] = value - elif key_list_length == 2: - if key_1 not in dict_to_update: - dict_to_update[key_1] = {} - dict_to_update[key_1][key_2] = value - elif key_list_length == 3: - if key_1 not in dict_to_update: - dict_to_update[key_1] = {} - if key_2 not in dict_to_update[key_1]: - dict_to_update[key_1][key_2] = {} - dict_to_update[key_1][key_2][key_3] = value + # Save processed value into the target + # If the target is a property + if isinstance(decode_details[0], property): + property_name = decode_details[0].fget.__name__ + if debug: + print(f'Property name: {property_name} <- {value}') + decode_details[0].__set__(self, value) + result[property_name] = value + # If the target is a dictionary use the extracted keys to add it under the correct position + elif isinstance(decode_details[0], dict): + dict_to_update = decode_details[0] + if key_list_length == 1: + dict_to_update[key_1] = value + if debug: + print(f'Dictionary key [{key_1}] <- {value}') + elif key_list_length == 2: + if key_1 not in dict_to_update: + dict_to_update[key_1] = {} + dict_to_update[key_1][key_2] = value + if debug: + print(f'Dictionary key [{key_1}][{key_2}] <- {value}') + elif key_list_length == 3: + if key_1 not in dict_to_update: + dict_to_update[key_1] = {} + if key_2 not in dict_to_update[key_1]: + dict_to_update[key_1][key_2] = {} + dict_to_update[key_1][key_2][key_3] = value + if debug: + print(f'Dictionary key [{key_1}][{key_2}][{key_3}] <- {value}') + # If the target is a list, append the current value to the end + elif isinstance(decode_details[0], list): + decode_details[0].append(value) + # If the target is a local variables + else: + variable_name = decode_details[0].name + if debug: + print(f'Local variable name: {variable_name} <- {value}') + decode_details[0].value = value + result[variable_name] = value start_pos = end_pos if debug: print('Finished cycle\n') - if debug: - print('done\n') def crc16(self, data: bytes) -> int: