Index: leahi_dialin/utils/abstract_classes.py =================================================================== diff -u -r8f1ef718ce3c23a6cc5a22ae13bc1a0627008f24 -rb167367cb2256f8b4a35bc2c9cde6cde2129d05d --- leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision 8f1ef718ce3c23a6cc5a22ae13bc1a0627008f24) +++ leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision b167367cb2256f8b4a35bc2c9cde6cde2129d05d) @@ -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. @@ -117,6 +127,124 @@ return results + def process_into_vars_2(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". + For local attributes to avoid namespace issues use the returned dictionary. Format: {attr_name : value} + + :param decoder_list: (List[Tuple[String, DataTypes]]) Contains the variable name and DataType pair 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) A dictionary for the variable_name and value pair + """ + start_pos = MSG_HEADER_SIZE + start_from_byte + result = {} + if debug: + print(f'\n\nDecoder_list: {decoder_list}') + for decode_details in decoder_list: + # If last position is multichar length + if isinstance(decode_details[-1], int): + length = decode_details[-1] + datatype: DataTypes = decode_details[-2] + 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 + # If it's a dictionary and the decoder list contains keys + 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() + if debug: + 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}') + + # 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] + print('new_value') + 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: + if debug: + print('Message extraction exception occured!') + value = None + break + start_pos = end_pos + + if 'nan' in str(value).lower(): + value = None + # raise ValueError(f'{value} is not an accepted value!') + # 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 processed value into the input + 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 + 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}') + elif isinstance(decode_details[0], list): + decode_details[0].append(value) + 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') + + 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.