Index: leahi_dialin/utils/abstract_classes.py =================================================================== diff -u -rb167367cb2256f8b4a35bc2c9cde6cde2129d05d -r3f4937e339925dde0b95f08e49969f8983c5cba4 --- leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision b167367cb2256f8b4a35bc2c9cde6cde2129d05d) +++ leahi_dialin/utils/abstract_classes.py (.../abstract_classes.py) (revision 3f4937e339925dde0b95f08e49969f8983c5cba4) @@ -76,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". @@ -89,59 +89,8 @@ :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'len: {len(decoder_list[0])}') - for decode_details in decoder_list: - # Content of the decode list - variable_name = decode_details[0] - datatype: DataTypes = decode_details[-1] - end_pos = start_pos + datatype.size() - try: - value = struct.unpack(datatype.unpack_attrib(), bytearray(message['message'][start_pos:end_pos]))[0] - except Exception as e: - value = None - if debug: - print(f'{variable_name}: {value} ({datatype.name})') - print(f'pos: {start_pos} - {end_pos}') - if 'nan' in str(value).lower(): - value = None - # 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_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 @@ -159,7 +108,7 @@ length = 1 datatype = DataTypes(result[decode_details[-1]]) base_list_length = 2 - # If it's a dictionary and the decoder list contains keys + # 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 @@ -179,9 +128,10 @@ 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 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) @@ -205,13 +155,15 @@ if datatype in [DataTypes.BOOL, DataTypes.BOOL_U08]: value = True if value == 1 else False - # Save processed value into the input + # 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, key should exist in the dictionary already elif isinstance(decode_details[0], dict): dict_to_update = decode_details[0] if key_list_length == 1: @@ -232,8 +184,10 @@ 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: @@ -243,48 +197,3 @@ 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. - - :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'len: {len(decoder_list[0])}') - 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 - datatype: DataTypes = decode_details[-1] - - end_pos = start_pos + datatype.size() - value = struct.unpack(datatype.unpack_attrib(), bytearray(message['message'][start_pos:end_pos]))[0] - if debug: - print(f'key_1: {key_1}') - print(f'key_2: {key_2}') - print(f'value: {value} ({datatype.name})') - print(f'pos: {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 the value into the Dictionary - if len(decode_details) == 2: - dict_to_update[key_1] = value - elif len(decode_details) == 3: - dict_to_update[key_1][key_2] = value - start_pos = end_pos - if debug: - print('Finished cycle\n') - if debug: - print('done\n')