VcanDecoder.py

import can
import cantools

class VcanDecoder:
 messages = []
 engineDataset = {}
 engineInfoDecoded = {}
 UseDBC = False
 MyDBCFile= 'EngineSimCanDecoder/engine_ecu.dbc'
 DBCFile = None 
 # initialize class stuff
 def __init__(self, channel='vcan0'):
 self.canbus = can.interface.Bus(channel=channel, interface='socketcan')
 self.use_dbc_file(self=self, dbc_file_name=self.MyDBCFile)
 # Get data
 def grab_data(self):
 while True:
 try:
 message = self.canbus.recv()
 except Exception as e:
 print(f"Error receiving CAN message: {e}")
 continue
 if message is not None:
 self.messages.append(message)
 if len(self.messages) >= 100:
 break
 self.write_to_file('engine_data.txt')

 self.shutdown()

 # Use to decipher and reverse engineer your own dbc
 @staticmethod
 def print_messages(messages):
 for msg in messages:
 print(msg)
 def getKeys(self):
 return self.engineDataset.keys()
 # Save logs
 def write_to_file(self, filename):
 try:
 with open(filename, 'w') as file:
 for key, value in self.engineDataset.items():
 byte_strings = [str(byte) for byte in value]
 line = f"{key}: {','.join(byte_strings)}\n"
 file.write(line)
 except Exception as e:
 print(f"Error writing to file: {e}")
 def get_engine_dataset(self):
 for msg in self.messages:
 self.engineDataset[msg.arbitration_id] = msg.data
 def get_engine_byteinfo(self):
 for key, value in self.engineDataset.items():
 byte_strings = [str(byte) for byte in value]
 self.engineInfoDecoded[key] = ",".join(byte_strings)
 print(self.engineInfoDecoded)
 # When dbc is available, use it to decode messages
 @staticmethod
 def use_dbc_file(self, dbc_file_name):
 if not dbc_file_name.endswith('.dbc'):
 print("Invalid DBC file format.")
 else:
 try:
 self.DBCFile = cantools.database.load_file(filename=dbc_file_name)
 if self.DBCFile is None:
 print("DBC support is not available in the current CAN library.")
 self.UseDBC = False
 else:
 print(f"DBC file '{dbc_file_name}' loaded successfully.")
 self.UseDBC = True
 except Exception as e:
 print(f"Error loading DBC file: {e}")
 self.UseDBC = False
 except:
 print("An unexpected error occurred while loading the DBC file.")
 self.UseDBC = False
 def decode_message(self, message):
 if self.UseDBC:
 try:
 decoded = self.DBCFile.decode_message(message.arbitration_id, message.data)
 return decoded
 except Exception as e:
 print(f"Error decoding message with DBC: {e}")
 return None
 else:
 print("DBC file not loaded. Cannot decode message.")
 return None
 # shutdown
 def shutdown(self):
 self.canbus.shutdown() 
 
VcanDecoder().grab_data()
VcanDecoder().get_engine_dataset()
VcanDecoder().get_engine_byteinfo()
VcanDecoder().print_messages(VcanDecoder().decode_message(msg) for msg in VcanDecoder().messages)
VcanDecoder().shutdown()