# coding: utf-8
"""
Implements support for BLF (Binary Logging Format) which is a proprietary
CAN log format from Vector Informatik GmbH (Germany).
No official specification of the binary logging format is available.
This implementation is based on Toby Lorenz' C++ library "Vector BLF" which is
licensed under GPLv3. https://bitbucket.org/tobylorenz/vector_blf.
The file starts with a header. The rest is one or more "log containers"
which consists of a header and some zlib compressed data, usually up to 128 kB
of uncompressed data each. This data contains the actual CAN messages and other
objects types.
"""
import struct
import zlib
import datetime
import time
import logging
from can.message import Message
from can.listener import Listener
from can.util import len2dlc, dlc2len, channel2int
from .generic import BaseIOHandler
class BLFParseError(Exception):
"""BLF file could not be parsed correctly."""
LOG = logging.getLogger(__name__)
# 0 = unknown, 2 = CANoe
APPLICATION_ID = 5
# signature ("LOGG"), header size,
# application ID, application major, application minor, application build,
# bin log major, bin log minor, bin log build, bin log patch,
# file size, uncompressed size, count of objects, count of objects read,
# time start (SYSTEMTIME), time stop (SYSTEMTIME)
FILE_HEADER_STRUCT = struct.Struct("<4sLBBBBBBBBQQLL8H8H")
# Pad file header to this size
FILE_HEADER_SIZE = 144
# signature ("LOBJ"), header size, header version, object size, object type
OBJ_HEADER_BASE_STRUCT = struct.Struct("<4sHHLL")
# flags, client index, object version, timestamp
OBJ_HEADER_V1_STRUCT = struct.Struct("<LHHQ")
# flags, timestamp status, object version, timestamp, original timestamp
OBJ_HEADER_V2_STRUCT = struct.Struct("<LBxHQQ")
# compression method, size uncompressed
LOG_CONTAINER_STRUCT = struct.Struct("<H6xL4x")
# channel, flags, dlc, arbitration id, data
CAN_MSG_STRUCT = struct.Struct("<HBBL8s")
# channel, flags, dlc, arbitration id, frame length, bit count, FD flags,
# valid data bytes, data
CAN_FD_MSG_STRUCT = struct.Struct("<HBBLLBBB5x64s")
# channel, dlc, valid payload length of data, tx count, arbitration id,
# frame length, flags, bit rate used in arbitration phase,
# bit rate used in data phase, time offset of brs field,
# time offset of crc delimiter field, bit count, direction,
# offset if extDataOffset is used, crc
CAN_FD_MSG_64_STRUCT = struct.Struct("<BBBBLLLLLLLHBBL")
# channel, length, flags, ecc, position, dlc, frame length, id, flags ext, data
CAN_ERROR_EXT_STRUCT = struct.Struct("<HHLBBBxLLH2x8s")
# commented event type, foreground color, background color, relocatable,
# group name length, marker name length, description length
GLOBAL_MARKER_STRUCT = struct.Struct("<LLL3xBLLL12x")
CAN_MESSAGE = 1
CAN_ERROR = 2
LOG_CONTAINER = 10
CAN_ERROR_EXT = 73
CAN_MESSAGE2 = 86
GLOBAL_MARKER = 96
CAN_FD_MESSAGE = 100
CAN_FD_MESSAGE_64 = 101
NO_COMPRESSION = 0
ZLIB_DEFLATE = 2
CAN_MSG_EXT = 0x80000000
REMOTE_FLAG = 0x80
EDL = 0x1
BRS = 0x2
ESI = 0x4
# CAN FD 64 Flags
REMOTE_FLAG_64 = 0x0010
EDL_64 = 0x1000
BRS_64 = 0x2000
ESI_64 = 0x4000
TIME_TEN_MICS = 0x00000001
TIME_ONE_NANS = 0x00000002
def timestamp_to_systemtime(timestamp):
if timestamp is None or timestamp < 631152000:
# Probably not a Unix timestamp
return (0, 0, 0, 0, 0, 0, 0, 0)
t = datetime.datetime.fromtimestamp(timestamp)
return (
t.year,
t.month,
t.isoweekday() % 7,
t.day,
t.hour,
t.minute,
t.second,
int(round(t.microsecond / 1000.0)),
)
def systemtime_to_timestamp(systemtime):
try:
t = datetime.datetime(
systemtime[0],
systemtime[1],
systemtime[3],
systemtime[4],
systemtime[5],
systemtime[6],
systemtime[7] * 1000,
)
return time.mktime(t.timetuple()) + systemtime[7] / 1000.0
except ValueError:
return 0
[docs]class BLFReader(BaseIOHandler):
"""
Iterator of CAN messages from a Binary Logging File.
Only CAN messages and error frames are supported. Other object types are
silently ignored.
"""
def __init__(self, file):
"""
:param file: a path-like object or as file-like object to read from
If this is a file-like object, is has to opened in binary
read mode, not text read mode.
"""
super().__init__(file, mode="rb")
data = self.file.read(FILE_HEADER_STRUCT.size)
header = FILE_HEADER_STRUCT.unpack(data)
if header[0] != b"LOGG":
raise BLFParseError("Unexpected file format")
self.file_size = header[10]
self.uncompressed_size = header[11]
self.object_count = header[12]
self.start_timestamp = systemtime_to_timestamp(header[14:22])
self.stop_timestamp = systemtime_to_timestamp(header[22:30])
# Read rest of header
self.file.read(header[1] - FILE_HEADER_STRUCT.size)
def __iter__(self):
tail = b""
while True:
data = self.file.read(OBJ_HEADER_BASE_STRUCT.size)
if not data:
# EOF
break
header = OBJ_HEADER_BASE_STRUCT.unpack(data)
if header[0] != b"LOBJ":
raise BLFParseError()
obj_type = header[4]
obj_data_size = header[3] - OBJ_HEADER_BASE_STRUCT.size
obj_data = self.file.read(obj_data_size)
# Read padding bytes
self.file.read(obj_data_size % 4)
if obj_type == LOG_CONTAINER:
method, uncompressed_size = LOG_CONTAINER_STRUCT.unpack_from(obj_data)
container_data = obj_data[LOG_CONTAINER_STRUCT.size :]
if method == NO_COMPRESSION:
data = container_data
elif method == ZLIB_DEFLATE:
data = zlib.decompress(container_data, 15, uncompressed_size)
else:
# Unknown compression method
LOG.warning("Unknown compression method (%d)", method)
continue
if tail:
data = tail + data
pos = 0
while pos + OBJ_HEADER_BASE_STRUCT.size < len(data):
header = OBJ_HEADER_BASE_STRUCT.unpack_from(data, pos)
# print(header)
if header[0] != b"LOBJ":
raise BLFParseError()
obj_size = header[3]
obj_type = header[4]
# Calculate position of next object
next_pos = pos + obj_size
if obj_type != CAN_FD_MESSAGE_64:
next_pos += obj_size % 4
if next_pos > len(data):
# Object continues in next log container
break
pos += OBJ_HEADER_BASE_STRUCT.size
# Read rest of header
header_version = header[2]
if header_version == 1:
flags, _, _, timestamp = OBJ_HEADER_V1_STRUCT.unpack_from(
data, pos
)
pos += OBJ_HEADER_V1_STRUCT.size
elif header_version == 2:
flags, _, _, timestamp, _ = OBJ_HEADER_V2_STRUCT.unpack_from(
data, pos
)
pos += OBJ_HEADER_V2_STRUCT.size
else:
# Unknown header version
LOG.warning(
"Unknown object header version (%d)", header_version
)
pos = next_pos
continue
if flags == TIME_TEN_MICS:
factor = 10 * 1e-6
else:
factor = 1e-9
timestamp = timestamp * factor + self.start_timestamp
# Both CAN message types have the same starting content
if obj_type in (CAN_MESSAGE, CAN_MESSAGE2):
(
channel,
flags,
dlc,
can_id,
can_data,
) = CAN_MSG_STRUCT.unpack_from(data, pos)
msg = Message(
timestamp=timestamp,
arbitration_id=can_id & 0x1FFFFFFF,
is_extended_id=bool(can_id & CAN_MSG_EXT),
is_remote_frame=bool(flags & REMOTE_FLAG),
dlc=dlc,
data=can_data[:dlc],
channel=channel - 1,
)
yield msg
elif obj_type == CAN_FD_MESSAGE:
(
channel,
flags,
dlc,
can_id,
_,
_,
fd_flags,
_,
can_data,
) = CAN_FD_MSG_STRUCT.unpack_from(data, pos)
length = dlc2len(dlc)
msg = Message(
timestamp=timestamp,
arbitration_id=can_id & 0x1FFFFFFF,
is_extended_id=bool(can_id & CAN_MSG_EXT),
is_remote_frame=bool(flags & REMOTE_FLAG),
is_fd=bool(fd_flags & EDL),
bitrate_switch=bool(fd_flags & BRS),
error_state_indicator=bool(fd_flags & ESI),
dlc=length,
data=can_data[:length],
channel=channel - 1,
)
yield msg
elif obj_type == CAN_FD_MESSAGE_64:
(
channel,
dlc,
_,
_,
can_id,
_,
fd_flags,
) = CAN_FD_MSG_64_STRUCT.unpack_from(data, pos)[:7]
length = dlc2len(dlc)
can_data = struct.unpack_from(
"<{}s".format(length), data, pos + CAN_FD_MSG_64_STRUCT.size
)[0]
msg = Message(
timestamp=timestamp,
arbitration_id=can_id & 0x1FFFFFFF,
is_extended_id=bool(can_id & CAN_MSG_EXT),
is_remote_frame=bool(fd_flags & REMOTE_FLAG_64),
is_fd=bool(fd_flags & EDL_64),
bitrate_switch=bool(fd_flags & BRS_64),
error_state_indicator=bool(fd_flags & ESI_64),
dlc=length,
data=can_data[:length],
channel=channel - 1,
)
yield msg
elif obj_type == CAN_ERROR_EXT:
(
channel,
_,
_,
_,
_,
dlc,
_,
can_id,
_,
can_data,
) = CAN_ERROR_EXT_STRUCT.unpack_from(data, pos)
msg = Message(
timestamp=timestamp,
is_error_frame=True,
is_extended_id=bool(can_id & CAN_MSG_EXT),
arbitration_id=can_id & 0x1FFFFFFF,
dlc=dlc,
data=can_data[:dlc],
channel=channel - 1,
)
yield msg
# else:
# LOG.warning("Unknown object type (%d)", obj_type)
pos = next_pos
# save the remaining data that could not be processed
tail = data[pos:]
self.stop()
[docs]class BLFWriter(BaseIOHandler, Listener):
"""
Logs CAN data to a Binary Logging File compatible with Vector's tools.
"""
#: Max log container size of uncompressed data
MAX_CACHE_SIZE = 128 * 1024
#: ZLIB compression level
COMPRESSION_LEVEL = 9
def __init__(self, file, channel=1):
"""
:param file: a path-like object or as file-like object to write to
If this is a file-like object, is has to opened in binary
write mode, not text write mode.
"""
super().__init__(file, mode="wb")
self.channel = channel
# Header will be written after log is done
self.file.write(b"\x00" * FILE_HEADER_SIZE)
self.cache = []
self.cache_size = 0
self.count_of_objects = 0
self.uncompressed_size = FILE_HEADER_SIZE
self.start_timestamp = None
self.stop_timestamp = None
[docs] def on_message_received(self, msg):
channel = channel2int(msg.channel)
if channel is None:
channel = self.channel
else:
# Many interfaces start channel numbering at 0 which is invalid
channel += 1
arb_id = msg.arbitration_id
if msg.is_extended_id:
arb_id |= CAN_MSG_EXT
flags = REMOTE_FLAG if msg.is_remote_frame else 0
data = bytes(msg.data)
if msg.is_error_frame:
data = CAN_ERROR_EXT_STRUCT.pack(
channel,
0, # length
0, # flags
0, # ecc
0, # position
len2dlc(msg.dlc),
0, # frame length
arb_id,
0, # ext flags
data,
)
self._add_object(CAN_ERROR_EXT, data, msg.timestamp)
elif msg.is_fd:
fd_flags = EDL
if msg.bitrate_switch:
fd_flags |= BRS
if msg.error_state_indicator:
fd_flags |= ESI
data = CAN_FD_MSG_STRUCT.pack(
channel, flags, len2dlc(msg.dlc), arb_id, 0, 0, fd_flags, msg.dlc, data
)
self._add_object(CAN_FD_MESSAGE, data, msg.timestamp)
else:
data = CAN_MSG_STRUCT.pack(channel, flags, msg.dlc, arb_id, data)
self._add_object(CAN_MESSAGE, data, msg.timestamp)
[docs] def log_event(self, text, timestamp=None):
"""Add an arbitrary message to the log file as a global marker.
:param str text:
The group name of the marker.
:param float timestamp:
Absolute timestamp in Unix timestamp format. If not given, the
marker will be placed along the last message.
"""
try:
# Only works on Windows
text = text.encode("mbcs")
except LookupError:
text = text.encode("ascii")
comment = b"Added by python-can"
marker = b"python-can"
data = GLOBAL_MARKER_STRUCT.pack(
0, 0xFFFFFF, 0xFF3300, 0, len(text), len(marker), len(comment)
)
self._add_object(GLOBAL_MARKER, data + text + marker + comment, timestamp)
def _add_object(self, obj_type, data, timestamp=None):
if timestamp is None:
timestamp = self.stop_timestamp or time.time()
if self.start_timestamp is None:
self.start_timestamp = timestamp
self.stop_timestamp = timestamp
timestamp = int((timestamp - self.start_timestamp) * 1e9)
header_size = OBJ_HEADER_BASE_STRUCT.size + OBJ_HEADER_V1_STRUCT.size
obj_size = header_size + len(data)
base_header = OBJ_HEADER_BASE_STRUCT.pack(
b"LOBJ", header_size, 1, obj_size, obj_type
)
obj_header = OBJ_HEADER_V1_STRUCT.pack(TIME_ONE_NANS, 0, 0, max(timestamp, 0))
self.cache.append(base_header)
self.cache.append(obj_header)
self.cache.append(data)
padding_size = len(data) % 4
if padding_size:
self.cache.append(b"\x00" * padding_size)
self.cache_size += obj_size + padding_size
self.count_of_objects += 1
if self.cache_size >= self.MAX_CACHE_SIZE:
self._flush()
def _flush(self):
"""Compresses and writes data in the cache to file."""
if self.file.closed:
return
cache = b"".join(self.cache)
if not cache:
# Nothing to write
return
uncompressed_data = cache[: self.MAX_CACHE_SIZE]
# Save data that comes after max size to next round
tail = cache[self.MAX_CACHE_SIZE :]
self.cache = [tail]
self.cache_size = len(tail)
compressed_data = zlib.compress(uncompressed_data, self.COMPRESSION_LEVEL)
obj_size = (
OBJ_HEADER_V1_STRUCT.size + LOG_CONTAINER_STRUCT.size + len(compressed_data)
)
base_header = OBJ_HEADER_BASE_STRUCT.pack(
b"LOBJ", OBJ_HEADER_BASE_STRUCT.size, 1, obj_size, LOG_CONTAINER
)
container_header = LOG_CONTAINER_STRUCT.pack(
ZLIB_DEFLATE, len(uncompressed_data)
)
self.file.write(base_header)
self.file.write(container_header)
self.file.write(compressed_data)
# Write padding bytes
self.file.write(b"\x00" * (obj_size % 4))
self.uncompressed_size += OBJ_HEADER_V1_STRUCT.size + LOG_CONTAINER_STRUCT.size
self.uncompressed_size += len(uncompressed_data)
[docs] def stop(self):
"""Stops logging and closes the file."""
self._flush()
filesize = self.file.tell()
super().stop()
# Write header in the beginning of the file
header = [b"LOGG", FILE_HEADER_SIZE, APPLICATION_ID, 0, 0, 0, 2, 6, 8, 1]
# The meaning of "count of objects read" is unknown
header.extend([filesize, self.uncompressed_size, self.count_of_objects, 0])
header.extend(timestamp_to_systemtime(self.start_timestamp))
header.extend(timestamp_to_systemtime(self.stop_timestamp))
with open(self.file.name, "r+b") as f:
f.write(FILE_HEADER_STRUCT.pack(*header))