We have seen in earlier blog articles how to create a DBC database and how to use signals from the database while sending and receiving CAN traffic.1 This time we’ll take a look at how we can use the DBC database to show signals from logged data.
Reading signals from log files
Configure the logger
In order to capture our signals we need to first configure our logger. I have a Kvaser Memorator Pro 5xHS (00778-9) connected to my PC, you should change the EAN number in the code below to match your own device.
from canlib import kvDevice
from canlib import kvamemolibxml
# Use .strip() to remove first newline, otherwise we get the error:
# XML declaration allowed only at the start of the document
CONFIG_XML = """
<?xml version="1.0" ?>
<!DOCTYPE KVASER>
<KVASER>
<VERSION>2.0</VERSION>
<BINARY_VERSION>6.0</BINARY_VERSION>
<SETTINGS>
<MODE fifo_mode="NO" log_all="YES"/>
</SETTINGS>
<CAN_BUS>
<PARAMETERS bitrate="1000000" channel="0" silent="YES" sjw="1" tseg1="5" tseg2="2"/>
<PARAMETERS bitrate="1000000" channel="1" silent="YES" sjw="1" tseg1="5" tseg2="2"/>
</CAN_BUS>
<TRIGGERBLOCK/>
<SCRIPTS/>
</KVASER>
""".strip()
def init_card(dev):
"""Format disk on device"""
# Open the device
dev.memoOpen()
# Initialize the SD card with default values
print('Formatting Card on:\n%s' % dev)
dev.memo.deviceFormatDisk()
# Convert the XML configuration to a binary configuration
config_lif = kvamemolibxml.kvaXmlToBuffer(CONFIG_XML)
# write the configuration
dev.memo.kmfWriteConfig(config_lif)
# Close the device
dev.memoClose()
def main():
# Connect to our Kvaser Memorator Pro 5xHS with EAN 00778-9
dev = kvDevice.kvDevice(ean="73-30130-00778-9")
init_card(dev)
if __name__ == '__main__':
main()Listing 2: Basic configuration to log everything on channel 0 and 1 at 1000 kbit/s using a Kvaser Memorator Pro 5xHS.
Running the code in Listing 2 prepares our Kvaser Memorator device by initializing the SD-card and downloading a “log-all” configuration.
Creating the sample data using framebox
Now we need some data, so why not create a script that generates random data based on the signal definitions available in a given database. In our case we use a database with two messages defined, EngineData and GearBoxInfo, each containing a number of signals (PetrolLevel, EngPower, and so on).
# Declare the signal name and the name of the message the signal
# can be found in. E.g. the signal 'PetrolLevel' can be found in
# the message 'EngineData'.
SIGNALS = {
'PetrolLevel': 'EngineData',
'EngPower': 'EngineData',
'EngForce': 'EngineData',
'IdleRunning': 'EngineData',
'EngTemp': 'EngineData',
'EngSpeed': 'EngineData',
'EcoMode': 'GearBoxInfo',
'ShiftRequest': 'GearBoxInfo',
'Gear': 'GearBoxInfo',
}When we later would like to give a signal a random value, we actually do not care about what message the signal is housed in, we just would like to set a value on a specific signal, whatever message that signal belongs to. The Framebox is our helper class to accomplish this. Given a database, db, we first create a framebox containing our messages mentioned in the SIGNALS definition above. We can then set a signal value through the framebox using framebox.signal('PetrolLevel').phys=35 with confidence that the value will end up in the correct message.2
Let us take a look at the function that is responsible for sending and receiving random signal values using a framebox.
def send_receive_using_framebox(db, quantity, ch_tx, ch_rx):
# Create framebox
framebox = kvadblib.FrameBox(db)
# Add our messages to the framebox
for msg_name in set(SIGNALS.values()):
framebox.add_message(msg_name)
# Randomly fill signals with values
for i in range(quantity):
set_random_framebox_signal(db, framebox)
# Send all frames contained in the framebox (one frame for each message)
for frame in framebox.frames():
# let the database interpret the frame, a bound message will be
# returned
bmsg = db.interpret(frame=frame)
# print the contents of the bound message using our own string
# converter
print('Sending %s\n' % bmessage_str(bmsg))
# send the frame, wait for Tx acknowledge
ch_tx.writeWait(frame, timeout=5000)
# Receive our CAN frames
try:
while True:
# receive one frame
frame = ch_rx.read(timeout=500)
# let the database interpret the frame
bmsg = db.interpret(frame=frame)
# print the contents of the bound messages
print('Received %s\n' % bmessage_str(bmsg))
# stop when we did not receive any more CAN frames
except canlib.CanNoMsg:
print("Received all frames!")Given the database DB, we now want to create a random value for a random signal. We do this by first randomly selecting a signal from the SIGNALS definition above and then defer the actual finding of a valid value to the function get_random_value().
def set_random_framebox_signal(db, framebox):
# pick a signal name at random
sig_name = random.choice(list(SIGNALS))
# get a random, but valid, value
value = get_random_value(db, sig_name)
print("Setting %s to %s" % (sig_name, value))
# set the signal value through our framebox
framebox.signal(sig_name).phys = valueLater on, we would like to print the contents of bound messages so we add the bmessage_str() function to convert the content of a bound message to a string. The only thing we have to add now is some configuration of the channels. Let us take a look at our complete program before running it.
import random
from canlib import canlib
from canlib import kvadblib
# Declare the signal name and the name of the message the signal
# can be found in. E.g. the signal 'PetrolLevel' can be found in
# the message 'EngineData'.
SIGNALS = {
'PetrolLevel': 'EngineData',
'EngPower': 'EngineData',
'EngForce': 'EngineData',
'IdleRunning': 'EngineData',
'EngTemp': 'EngineData',
'EngSpeed': 'EngineData',
'EcoMode': 'GearBoxInfo',
'ShiftRequest': 'GearBoxInfo',
'Gear': 'GearBoxInfo',
}
def open_channel(channel):
"""Open a new channel , set bitrate 1Mbit/s and go bus on."""
ch = canlib.openChannel(channel, canlib.canOPEN_ACCEPT_VIRTUAL)
ch.setBusOutputControl(canlib.canDRIVER_NORMAL)
ch.setBusParams(canlib.canBITRATE_1M)
ch.busOn()
return ch
def close_channel(ch):
"""Go bus off and close channel."""
ch.busOff()
ch.close()
def get_random_bound_signal(db):
"""Create a bmsg, randomly choosing a message/signal pair from SIGNALS"""
# pick a signal name at random
sig_name = random.choice(list(SIGNALS))
# get the message name
msg_name = SIGNALS[sig_name]
# get the message from our database
msg = db.get_message_by_name(msg_name)
# create a BoundSignal, containing an empty frame
# with correct ID and DLC
bsig = msg.get_signal_by_name(sig_name).bind()
# get the signal, which holds the definitions
sig = bsig.signal
# get the signal's limits
limits = sig.limits
# create a value (with many decimals) within the limits
value = random.uniform(limits.min, limits.max)
# round value depending on type...
if sig.type is kvadblib.SignalType.UNSIGNED:
# ...remove decimals if the signal was of type unsigned
value = int(round(value))
else:
# ...otherwise, round to get only one decimal
value = round(value, 1)
# Now we would like to set the physical value
bsig.phys = value
return bsig
def bsignal_str(bsignal):
"""Create a nice looking string of the contents of the bound signal"""
value = bsignal.phys
if bsignal.signal.type is kvadblib.SignalType.UNSIGNED:
# remove decimals if the signal was of type unsigned
value = int(value)
else:
# otherwise, round to get only one decimal
value = round(value, 1)
# create adjusted strings for prettier printing
name = str(bsignal.name).rjust(11)
val = str(value).rjust(5)
return '%12s: %3s %s' % (name, val, bsignal.unit)
def bmessage_str(bmsg):
"""Create a nice looking string of the contents of the bound message"""
txt = ''
for bsig in bmsg:
txt += '\n\t%s' % bsignal_str(bsig)
return txt
def get_random_value(db, sig_name):
# get the message name
msg_name = SIGNALS[sig_name]
# get the message from our database
msg = db.get_message_by_name(msg_name)
# get the signal, which holds the definitions
sig = msg.get_signal_by_name(sig_name)
# get the signal's limits
limits = sig.limits
# create a value (with many decimals) within the limits
value = random.uniform(limits.min, limits.max)
# round value depending on type...
if (sig.type is kvadblib.SignalType.UNSIGNED or
sig.type is kvadblib.SignalType.SIGNED):
# ...remove decimals if the signal was of type unsigned
value = int(round(value))
else:
# ...otherwise, round to get only one decimal
value = round(value, 1)
return value
def set_random_framebox_signal(db, framebox):
# pick a signal name at random
sig_name = random.choice(list(SIGNALS))
# get a random, but valid, value
value = get_random_value(db, sig_name)
print("Setting %s to %s" % (sig_name, value))
# set the signal value through our framebox
framebox.signal(sig_name).phys = value
def send_receive_using_framebox(db, quantity, ch_tx, ch_rx):
# Create framebox
framebox = kvadblib.FrameBox(db)
# Add our messages to the framebox
for msg_name in set(SIGNALS.values()):
framebox.add_message(msg_name)
# Randomly fill signals with values
for i in range(quantity):
set_random_framebox_signal(db, framebox)
# Send all frames contained in the framebox (one frame for each message)
for frame in framebox.frames():
# let the database interpret the frame, a bound message will be
# returned
bmsg = db.interpret(frame=frame)
# print the contents of the bound message using our own string
# converter
print('Sending %s\n' % bmessage_str(bmsg))
# send the frame, wait for Tx acknowledge
ch_tx.writeWait(frame, timeout=5000)
# Receive our CAN frames
try:
while True:
# receive one frame
frame = ch_rx.read(timeout=500)
# let the database interpret the frame
bmsg = db.interpret(frame=frame)
# print the contents of the bound messages
print('Received %s\n' % bmessage_str(bmsg))
# stop when we did not receive any more CAN frames
except canlib.CanNoMsg:
print("Received all frames!")
def main():
print("kvadblib v%s" % kvadblib.dllversion())
# open the database we would like to use
db = kvadblib.Dbc(filename='engine_example.dbc')
# Open and setup channels
ch0 = open_channel(0)
ch1 = open_channel(1)
# Loop the sending twice so we send four CAN frames in total
for i in range(2):
# quantity specifies how many time we should set a random value on a
# random signal
send_receive_using_framebox(db=db,
quantity=20,
ch_tx=ch0,
ch_rx=ch1)
close_channel(ch0)
close_channel(ch1)
if __name__ == '__main__':
main()After arming our logger device by applying CAN power and disconnecting USB, we can now finally run the program on a second device that is connected to the same CAN bus (I used a Kvaser USBcan Pro) and note that the sent and received signals are identical!
kvadblib v8.22.360
Setting EngSpeed to 6778
Setting EngForce to 250
:
Sending
EcoMode: 1
ShiftRequest: 0
Gear: 2
Sending
PetrolLevel: 0 l
EngPower: 62 kW
EngForce: 223 N
IdleRunning: 1
EngTemp: 6 degC
EngSpeed: 6778 rpm
Received
EcoMode: 1
ShiftRequest: 0
Gear: 2
Received
PetrolLevel: 0 l
EngPower: 62 kW
EngForce: 223 N
IdleRunning: 1
EngTemp: 6 degC
EngSpeed: 6778 rpm
:
Received all frames!Examining the logged data
Now that we have logged data on our Kvaser Memorator Pro, let us examine what is in there. We have gone through this in earlier blog articles3 so let us here just recap the essential part were we do the printing (full code listing will follow). When we write the raw frame contents, we use canlib’s built in string conversion
def read_raw_events(dev, fileIndx):
# read events from logfile number indicated by fileIndx
myEvents = dev.memoReadEvents(fileIndx)
for event in myEvents:
# print event with built in string conversion
print(event)
print("\n")Using the read_raw_events function results in the following output:
Found 1 file on card:
*t: - EAN:73-30130-00778-9 s/n:1023 FW:v3.8.384 LIO:v0.0
t: 0.250051287 DateTime: 2018-02-09 09:04:25
t: 0.250051287 Log Trigger Event (type: 0x1, trigno: 0x00, pre-trigger: 0, post-
trigger: -1)
t: 9.732494837 DateTime: 2018-02-09 09:04:34
t: 12.094059087 ch:0 f: 2 id: 3fc dlc: 1 d:42
t: 12.101087212 ch:0 f: 2 id: 64 dlc: 8 d:7a 1a 9c 00 df 00 38 18
t: 12.623195962 ch:0 f: 2 id: 3fc dlc: 1 d:43
t: 12.624621087 ch:0 f: 2 id: 64 dlc: 8 d:13 08 86 f5 44 01 a4 38
t: 20.232475162 DateTime: 2018-02-09 09:04:45
So let us now use the database to extract our signals from those message events and replace the printed output with the name and value of our corresponding signal. This is done by converting each read message event into a frame, then let the database interpret the frame which results in a bound message. Then we can just iterate over the bound message, as we have done before, to get hold of the contained signals. The last step is to adjust the printing to fit the built in string formatting.
def read_signal_events(dev, fileIndx):
# open the database we would like to use
db = kvadblib.Dbc(filename='engine_example.dbc')
myEvents = dev.memoReadEvents(fileIndx)
for event in myEvents:
# only message events have a data frame
if type(event) is kvmlib.events.MessageEvent:
# convert the message event into a frame
frame = event.asframe()
# let the database interpret the frame, a bound message will be
# returned
bmsg = db.interpret(frame=frame)
# generate a nice looking string from the bound message, and
# print on a format matching the built in string conversion
# used for other types of events
timestamp = event.asframe().timestamp / 1000000000
print(' t:%14s Signal %s ' % (timestamp, bmessage_str(bmsg)))
else:
print(event)
print("\n")Using the read_signal_events() function results in the following output:
Found 1 file on card:
*t: - EAN:73-30130-00778-9 s/n:1023 FW:v3.8.384 LIO:v0.0
t: 0.250051287 DateTime: 2018-02-09 09:04:25
t: 0.250051287 Log Trigger Event (type: 0x1, trigno: 0x00, pre-trigger: 0, post-
trigger: -1)
t: 9.732494837 DateTime: 2018-02-09 09:04:34
t: 12.094059087 Signal EcoMode: 1 ShiftRequest: 0 Gear: 2
t: 12.101087212 Signal PetrolLevel: 0 l EngPower: 62 kW EngForce: 223 N
IdleRunning: 1 EngTemp: 6 degC EngSpeed: 6778 rpm
t: 12.623195962 Signal EcoMode: 1 ShiftRequest: 0 Gear: 3
t: 12.624621087 Signal PetrolLevel:245 l EngPower:145 kW EngForce: 324 N
IdleRunning: 1 EngTemp: -38 degC EngSpeed: 2067 rpm
t: 20.232475162 DateTime: 2018-02-09 09:04:45Finally, here is the complete listing of our program.
from canlib import kvDevice
from canlib import kvadblib
from canlib import kvmlib
def read_raw_events(dev, fileIndx):
# read events from logfile number indicated by fileIndx
myEvents = dev.memoReadEvents(fileIndx)
for event in myEvents:
# print event with built in string conversion
print(event)
print("\n")
def bsignal_str(bsignal):
"""Create a nice looking string of the contents of the bound signal"""
value = bsignal.phys
if bsignal.signal.type is kvadblib.SignalType.UNSIGNED:
# remove decimals if the signal was of type unsigned
value = int(value)
else:
# otherwise, round to get only one decimal
value = round(value, 1)
# create adjusted strings for prettier printing
name = str(bsignal.name).rjust(11)
val = str(value).rjust(5)
return '%s: %s %s' % (name, val, bsignal.unit)
def bmessage_str(bmsg):
"""Create a nice looking string of the contents of the bound message"""
txt = ''
for bsig in bmsg:
txt += '%s' % bsignal_str(bsig)
return txt
def read_signal_events(dev, fileIndx):
# open the database we would like to use
db = kvadblib.Dbc(filename='engine_example.dbc')
myEvents = dev.memoReadEvents(fileIndx)
for event in myEvents:
# only message events have a data frame
if type(event) is kvmlib.events.MessageEvent:
# convert the message event into a frame
frame = event.asframe()
# let the database interpret the frame, a bound message will be
# returned
bmsg = db.interpret(frame=frame)
# generate a nice looking string from the bound message, and
# print on a format matching the built in string conversion
# used for other types of events
timestamp = event.asframe().timestamp / 1000000000
print(' t:%14s Signal %s ' % (timestamp, bmessage_str(bmsg)))
else:
print(event)
print("\n")
def read_log(dev, read_function):
# Connect to our Kvaser Memorator Pro 5xHS with EAN 00778-9
dev = kvDevice.kvDevice(ean="73-30130-00778-9")
dev.open()
dev.memoOpen()
fileCount = dev.memo.logFileGetCount()
print("Found %d file%s on card:" % (fileCount, "s" if fileCount > 1 else ""))
# Loop through all logfiles and write their contents to stdout
for fileIndx in range(fileCount):
read_function(dev, fileIndx)
# Delete all logfiles
#dev.memo.logFileDeleteAll()
# Close device
dev.memoClose()
dev.close()
def main():
# connect to our Kvaser Memorator Pro 5xHS with EAN 00778-9
dev = kvDevice.kvDevice(ean="73-30130-00778-9")
# read and print raw events from the device
read_log(dev, read_raw_events)
# read and print signals from the device
read_log(dev, read_signal_events)
if __name__ == '__main__':
main()Convert logged data to a signal based format
If we already have extracted our logged data from our Kvaser Memorator into e.g. a .kme50 file, we can still expose our signals by feeding the database to the converter library and convert our data to a signal based format. Here we will convert our .kme50 file into the csv signal format.4
# set output format
fmt = kvlclib.WriterFormat(kvlclib.FILE_FORMAT_CSV_SIGNAL)
# the name of the formatter is fetched using kvlcGetWriterName() internally
print("Output format is '%s'" % fmt.name)
# set resulting output filename taking advantage of the extension defined
# in the format. (Uses kvlcGetWriterExtension() under the hood.)
outfile = "myresult." + fmt.extension
print("Output filename is '%s'" % outfile)
# create converter
kc = kvlclib.Converter(outfile, fmt)
# add database file
channel_mask = kvlclib.ChannelMask.ONE | kvlclib.ChannelMask.TWO
kc.addDatabaseFile("engine_example.dbc", channel_mask)
# Set input filename and format
inputfile = "engine_example.kme50"
print("Input filename is '%s'" % inputfile)
kc.setInputFile(inputfile, file_format=kvlclib.FILE_FORMAT_KME50)The full listing of our programs is shown below.
from canlib import kvlclib
def trySetProperty(converter, property, value=None):
# Check if the format supports the given property
if converter.format.isPropertySupported(property):
# If a value was specified, set the property to this value
if value is not None:
converter.setProperty(property, value)
# get the property's default value
default = converter.format.getPropertyDefault(property)
print(" PROPERTY_%s is supported (Default: %s)" %
(property['name'], default))
# get the property's current value
value = converter.getProperty(property)
print(" Current value: %s" % value)
else:
print(" PROPERTY %s is not supported" %
(property['name']))
def convertEvents(kc):
# Get estimated number of remaining events in the input file. This can be
# useful for displaying progress during conversion.
total = kc.eventCount()
print("Converting about %d events..." % total)
while True:
try:
# Convert events from input file one by one until EOF is reached
kc.convertEvent()
if kc.isOutputFilenameNew():
print("New output filename: %s" % kc.getOutputFilename())
print("About %d events left to convert..." % kc.eventCount())
except kvlclib.KvlcEndOfFile:
if kc.isOverrunActive():
print("NOTE! The extracted data contained overrun.")
kc.resetOverrunActive()
if kc.isDataTruncated():
print("NOTE! The extracted data was truncated.")
kc.resetStatusTruncated()
break
# set output format
fmt = kvlclib.WriterFormat(kvlclib.FILE_FORMAT_CSV_SIGNAL)
# the name of the formatter is fetched using kvlcGetWriterName() internally
print("Output format is '%s'" % fmt.name)
# set resulting output filename taking advantage of the extension defined
# in the format. (Uses kvlcGetWriterExtension() under the hood.)
outfile = "myresult." + fmt.extension
print("Output filename is '%s'" % outfile)
# create converter
kc = kvlclib.Converter(outfile, fmt)
# add database file
channel_mask = kvlclib.ChannelMask.ONE | kvlclib.ChannelMask.TWO
kc.addDatabaseFile("engine_example.dbc", channel_mask)
# Set input filename and format
inputfile = "engine_example.kme50"
print("Input filename is '%s'" % inputfile)
kc.setInputFile(inputfile, file_format=kvlclib.FILE_FORMAT_KME50)
# allow output file to overwrite existing files
trySetProperty(kc, kvlclib.PROPERTY_OVERWRITE, 1)
# add nice header to the output file
trySetProperty(kc, kvlclib.PROPERTY_WRITE_HEADER, 1)
# we are converting CAN traffic with max 8 bytes, so we can minimize the
# width of the data output to 8 bytes
trySetProperty(kc, kvlclib.PROPERTY_LIMIT_DATA_BYTES, 8)
# convert all events
convertEvents(kc)Running the above program reveals that the property LIMIT_DATA_BYTES was obviously not available in the “CSV Signal” format we choose, but we were saved by our trySetProperty() function.
Output format is 'CSV Signal'
Output filename is 'myresult.csv'
Input filename is 'histogram.kme50'
PROPERTY_OVERWRITE is supported (Default: 0)
Current value: 1
PROPERTY_WRITE_HEADER is supported (Default: 0)
Current value: 1
PROPERTY LIMIT_DATA_BYTES is not supported
Converting about 42 events...
New output filename: myresult.csv
About 41 events left to convert...Looking in the resulting file myresult.csv shown in Figure 2 we can see the signals as we expected.
Figure 2: Our resulting csv file reveals the now familiar signal values.
This ends the tour on how to use a DBC database to show signals and values from events and logged data. Hopefully you learned something, or at least found it interesting. If you have any questions please feel free to contact [email protected].
Footnotes
1 We created a database in the blog entry Handling CAN databases in Python, and sent and received signals in Send and receive database signals.
2 If the signal is missing from the framebox, we get an exception.
3 We read logged messages when we started using kvmlib in https://www.kvaser.com/developer-blog/getting-started-with-kvmlib/.
4 This code was introduced in a blog series about the Converter Library (kvlclib) https://www.kvaser.com/developer-blog/converting-to-plain-ascii/.
