""" 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 datetime import logging import struct import time import zlib from typing import Any, BinaryIO, Generator, List, Optional, Tuple, Union, cast from ..message import Message from ..typechecking import StringPathLike from ..util import channel2int, dlc2len, len2dlc from .generic import BinaryIOMessageReader, FileIOMessageWriter TSystemTime = Tuple[int, int, int, int, int, int, int, int] class BLFParseError(Exception): """BLF file could not be parsed correctly.""" LOG = logging.getLogger(__name__) # 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(" TSystemTime: 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(round(timestamp, 3)) return ( t.year, t.month, t.isoweekday() % 7, t.day, t.hour, t.minute, t.second, t.microsecond // 1000, ) def systemtime_to_timestamp(systemtime: TSystemTime) -> float: try: t = datetime.datetime( systemtime[0], systemtime[1], systemtime[3], systemtime[4], systemtime[5], systemtime[6], systemtime[7] * 1000, ) return t.timestamp() except ValueError: return 0 class BLFReader(BinaryIOMessageReader): """ Iterator of CAN messages from a Binary Logging File. Only CAN messages and error frames are supported. Other object types are silently ignored. """ file: BinaryIO def __init__( self, file: Union[StringPathLike, BinaryIO], **kwargs: Any, ) -> None: """ :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(cast(TSystemTime, header[14:22])) self.stop_timestamp = systemtime_to_timestamp(cast(TSystemTime, header[22:30])) # Read rest of header self.file.read(header[1] - FILE_HEADER_STRUCT.size) self._tail = b"" self._pos = 0 def __iter__(self) -> Generator[Message, None, None]: while True: data = self.file.read(OBJ_HEADER_BASE_STRUCT.size) if not data: # EOF break signature, _, _, obj_size, obj_type = OBJ_HEADER_BASE_STRUCT.unpack(data) if signature != b"LOBJ": raise BLFParseError() obj_data = self.file.read(obj_size - OBJ_HEADER_BASE_STRUCT.size) # Read padding bytes self.file.read(obj_size % 4) if obj_type == LOG_CONTAINER: method, _ = 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: zobj = zlib.decompressobj() data = zobj.decompress(container_data) else: # Unknown compression method LOG.warning("Unknown compression method (%d)", method) continue yield from self._parse_container(data) self.stop() def _parse_container(self, data): if self._tail: data = b"".join((self._tail, data)) try: yield from self._parse_data(data) except struct.error: # There was not enough data in the container to unpack a struct pass # Save the remaining data that could not be processed self._tail = data[self._pos :] def _parse_data(self, data): """Optimized inner loop by making local copies of global variables and class members and hardcoding some values.""" unpack_obj_header_base = OBJ_HEADER_BASE_STRUCT.unpack_from obj_header_base_size = OBJ_HEADER_BASE_STRUCT.size unpack_obj_header_v1 = OBJ_HEADER_V1_STRUCT.unpack_from obj_header_v1_size = OBJ_HEADER_V1_STRUCT.size unpack_obj_header_v2 = OBJ_HEADER_V2_STRUCT.unpack_from obj_header_v2_size = OBJ_HEADER_V2_STRUCT.size unpack_can_msg = CAN_MSG_STRUCT.unpack_from unpack_can_fd_msg = CAN_FD_MSG_STRUCT.unpack_from unpack_can_fd_64_msg = CAN_FD_MSG_64_STRUCT.unpack_from can_fd_64_msg_size = CAN_FD_MSG_64_STRUCT.size unpack_can_error_ext = CAN_ERROR_EXT_STRUCT.unpack_from start_timestamp = self.start_timestamp max_pos = len(data) pos = 0 # Loop until a struct unpack raises an exception while True: self._pos = pos # Find next object after padding (depends on object type) try: pos = data.index(b"LOBJ", pos, pos + 8) except ValueError: if pos + 8 > max_pos: # Not enough data in container return raise BLFParseError("Could not find next object") from None header = unpack_obj_header_base(data, pos) # print(header) signature, _, header_version, obj_size, obj_type = header if signature != b"LOBJ": raise BLFParseError() # Calculate position of next object next_pos = pos + obj_size if next_pos > max_pos: # This object continues in the next container return pos += obj_header_base_size # Read rest of header if header_version == 1: flags, _, _, timestamp = unpack_obj_header_v1(data, pos) pos += obj_header_v1_size elif header_version == 2: flags, _, _, timestamp = unpack_obj_header_v2(data, pos) pos += obj_header_v2_size else: LOG.warning("Unknown object header version (%d)", header_version) pos = next_pos continue # Calculate absolute timestamp in seconds factor = 1e-5 if flags == 1 else 1e-9 timestamp = timestamp * factor + start_timestamp if obj_type in (CAN_MESSAGE, CAN_MESSAGE2): channel, flags, dlc, can_id, can_data = unpack_can_msg(data, pos) yield 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_rx=not bool(flags & DIR), dlc=dlc, data=can_data[:dlc], channel=channel - 1, ) elif obj_type == CAN_ERROR_EXT: members = unpack_can_error_ext(data, pos) channel = members[0] dlc = members[5] can_id = members[7] can_data = members[9] yield 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, ) elif obj_type == CAN_FD_MESSAGE: members = unpack_can_fd_msg(data, pos) ( channel, flags, dlc, can_id, _, _, fd_flags, valid_bytes, can_data, ) = members yield 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 & 0x1), is_rx=not bool(flags & DIR), bitrate_switch=bool(fd_flags & 0x2), error_state_indicator=bool(fd_flags & 0x4), dlc=dlc2len(dlc), data=can_data[:valid_bytes], channel=channel - 1, ) elif obj_type == CAN_FD_MESSAGE_64: ( channel, dlc, valid_bytes, _, can_id, _, fd_flags, _, _, _, _, _, direction, _, _, ) = unpack_can_fd_64_msg(data, pos) pos += can_fd_64_msg_size yield Message( timestamp=timestamp, arbitration_id=can_id & 0x1FFFFFFF, is_extended_id=bool(can_id & CAN_MSG_EXT), is_remote_frame=bool(fd_flags & 0x0010), is_fd=bool(fd_flags & 0x1000), is_rx=not direction, bitrate_switch=bool(fd_flags & 0x2000), error_state_indicator=bool(fd_flags & 0x4000), dlc=dlc2len(dlc), data=data[pos : pos + valid_bytes], channel=channel - 1, ) pos = next_pos class BLFWriter(FileIOMessageWriter): """ Logs CAN data to a Binary Logging File compatible with Vector's tools. """ file: BinaryIO #: Max log container size of uncompressed data max_container_size = 128 * 1024 #: Application identifier for the log writer application_id = 5 def __init__( self, file: Union[StringPathLike, BinaryIO], append: bool = False, channel: int = 1, compression_level: int = -1, **kwargs: Any, ) -> None: """ :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 mode "wb+". :param channel: Default channel to log as if not specified by the interface. :param append: Append messages to an existing log file. :param compression_level: An integer from 0 to 9 or -1 controlling the level of compression. 1 (Z_BEST_SPEED) is fastest and produces the least compression. 9 (Z_BEST_COMPRESSION) is slowest and produces the most. 0 means that data will be stored without processing. The default value is -1 (Z_DEFAULT_COMPRESSION). Z_DEFAULT_COMPRESSION represents a default compromise between speed and compression (currently equivalent to level 6). """ mode = "rb+" if append else "wb" try: super().__init__(file, mode=mode) except FileNotFoundError: # Trying to append to a non-existing file, create a new one append = False mode = "wb" super().__init__(file, mode=mode) assert self.file is not None self.channel = channel self.compression_level = compression_level self._buffer: List[bytes] = [] self._buffer_size = 0 # If max container size is located in kwargs, then update the instance if kwargs.get("max_container_size", False): self.max_container_size = kwargs["max_container_size"] if append: # Parse file header 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.uncompressed_size = header[11] self.object_count = header[12] self.start_timestamp: Optional[float] = systemtime_to_timestamp( cast(TSystemTime, header[14:22]) ) self.stop_timestamp: Optional[float] = systemtime_to_timestamp( cast(TSystemTime, header[22:30]) ) # Jump to the end of the file self.file.seek(0, 2) else: self.object_count = 0 self.uncompressed_size = FILE_HEADER_SIZE self.start_timestamp = None self.stop_timestamp = None # Write a default header which will be updated when stopped self._write_header(FILE_HEADER_SIZE) def _write_header(self, filesize): header = [b"LOGG", FILE_HEADER_SIZE, self.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.object_count, 0]) header.extend(timestamp_to_systemtime(self.start_timestamp)) header.extend(timestamp_to_systemtime(self.stop_timestamp)) self.file.write(FILE_HEADER_STRUCT.pack(*header)) # Pad to header size self.file.write(b"\x00" * (FILE_HEADER_SIZE - FILE_HEADER_STRUCT.size)) 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 if not msg.is_rx: flags |= DIR can_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 can_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, len(can_data), can_data, ) self._add_object(CAN_FD_MESSAGE, data, msg.timestamp) else: data = CAN_MSG_STRUCT.pack(channel, flags, msg.dlc, arb_id, can_data) self._add_object(CAN_MESSAGE, data, msg.timestamp) 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._buffer.append(base_header) self._buffer.append(obj_header) self._buffer.append(data) padding_size = len(data) % 4 if padding_size: self._buffer.append(b"\x00" * padding_size) self._buffer_size += obj_size + padding_size self.object_count += 1 if self._buffer_size >= self.max_container_size: self._flush() def _flush(self): """Compresses and writes data in the buffer to file.""" if self.file.closed: return buffer = b"".join(self._buffer) if not buffer: # Nothing to write return uncompressed_data = memoryview(buffer)[: self.max_container_size] # Save data that comes after max size to next container tail = buffer[self.max_container_size :] self._buffer = [tail] self._buffer_size = len(tail) if not self.compression_level: data = uncompressed_data method = NO_COMPRESSION else: data = zlib.compress(uncompressed_data, self.compression_level) method = ZLIB_DEFLATE obj_size = OBJ_HEADER_BASE_STRUCT.size + LOG_CONTAINER_STRUCT.size + len(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(method, len(uncompressed_data)) self.file.write(base_header) self.file.write(container_header) self.file.write(data) # Write padding bytes self.file.write(b"\x00" * (obj_size % 4)) self.uncompressed_size += OBJ_HEADER_BASE_STRUCT.size self.uncompressed_size += LOG_CONTAINER_STRUCT.size self.uncompressed_size += len(uncompressed_data) def file_size(self) -> int: """Return an estimate of the current file size in bytes.""" return self.file.tell() + self._buffer_size def stop(self): """Stops logging and closes the file.""" self._flush() if self.file.seekable(): filesize = self.file.tell() # Write header in the beginning of the file self.file.seek(0) self._write_header(filesize) else: LOG.error("Could not write BLF header since file is not seekable") super().stop()