candump-bitrate

#!/usr/bin/env python3
"""Compute estimated CAN bus bitrate

Consumes a CSV file generated by 'can2csv'.
"""
import argparse
import collections
import csv
import logging
import pathlib
import shutil
import sys
import tempfile

from csvutils import column_name_to_index, detect_dialect


def parse_args():
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--log-level",
        "-l",
        type=str,
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
        help="Set the logging output level. Defaults to INFO.",
    )
    parser.add_argument(
        "input",
        default="-",
        nargs="?",
        help="Candump CSV file input. Defaults to stdin.",
    )
    parser.add_argument(
        "--bitrate",
        "-b",
        default=250000,
        type=int,
        help="The bus bitrate",
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--inplace",
        "-i",
        action="store_true",
        help="Whether to edit the input file inplace. Requires a file input, not stdin",
    )
    group.add_argument(
        "--output",
        "-o",
        default="-",
        help="Script output. Defaults to stdout.",
    )

    group = parser.add_argument_group("CSV options")
    group.add_argument(
        "--no-header",
        action="store_true",
        default=False,
        help="Whether the CSV file has a header",
    )
    group.add_argument(
        "--delimiter",
        "-d",
        default=",",
        help="Specify the column delimiter. Default is ','",
    )

    return parser.parse_args()


def main(args):
    # Handle inplace input/output, and stdin/file input/output
    input = None
    if not args.input or args.input == "-":
        input = sys.stdin
    else:
        input = pathlib.Path(args.input)
        if not input.exists():
            logging.critical("Input file '%s' does not exists", args.input)
            sys.exit(1)
        input = input.open("r", encoding="utf-8")

    output = None
    tempoutput = None
    if args.inplace:
        output = tempfile.NamedTemporaryFile(mode="w", delete=False)
        tempoutput = pathlib.Path(output.name)
    elif not args.output or args.output == "-":
        output = sys.stdout
    else:
        output = pathlib.Path(args.output)
        output = output.open("w", encoding="utf-8")

    # Detect CSV dialect, sanity check header
    dialect, has_header, input = detect_dialect(args, input)
    if args.no_header:
        has_header = False
    reader = csv.reader(input, dialect)
    header = None
    if has_header:
        header = next(reader)

    timestamp_idx = column_name_to_index("timestamp", header) if has_header else 0
    data_idx = column_name_to_index("data", header) if has_header else 3
    if has_header:
        if data_idx >= len(header):
            logging.critical("'data' column not found in header '%s'", ",".join(header))
            sys.exit(1)

    # Write the new header
    writer = csv.writer(output, dialect)
    if has_header:
        header += [
            "raw-bits",
            "data-bits",
            "raw-bitrate",
            "data-bitrate",
            "raw-busload",
            "data-busload",
        ]
        writer.writerow(header)

    # SOF + SRR + IDE + RTR + CF + CRC + ACK + EOF + IFS
    overhead_bits = 1 + 2 + 1 + 6 + 16 + 2 + 7 + 3

    # This isn't a true measure of bitrate, just an estimate, since it can't accomodate Remote or
    # Error CAN frames, and assumes 29-bit IDs (as opposed to 11-bit). Additionally, it doesn't
    # account for the time not spent sending data (in between messages), nor can it account for
    # arbitration and ECU negotiation.
    #
    # The purpose of this script is to give an estimate, and a way to compare two different CAN logs
    # to understand bus load.
    window_size = 100
    timestamp_ring = collections.deque(maxlen=window_size)
    data_bits_ring = collections.deque(maxlen=window_size)
    for row in reader:
        if timestamp_idx >= len(row) or data_idx >= len(row):
            logging.critical(
                "Row not long enough to find field 'data' or 'timestamp' in row '%s'",
                ", ".join(row),
            )
            sys.exit(1)
        value = row[timestamp_idx]
        timestamp = None
        if not value:
            continue
        try:
            timestamp = float(value)
        except ValueError:
            logging.critical("Could not process non-float timestamp '%s'", value)
            sys.exit(1)
        value = row[data_idx]
        # 4 bits per nibble with 1 char nibbles
        data_bits = len(value) * 4
        # Include the CAN ID in the data bits
        data_bits += 29
        raw_bits = data_bits + overhead_bits

        timestamp_ring.append(timestamp)
        duration = timestamp_ring[-1] - timestamp_ring[0]
        data_bits_ring.append(data_bits)
        data_bitrate = sum(data_bits_ring) / duration if duration else args.bitrate
        raw_bitrate = (
            (sum(data_bits_ring) + overhead_bits * len(data_bits_ring)) / duration
            if duration
            else args.bitrate
        )
        # The sliding window average takes some time to settle, so filter out extremes
        data_bitrate = min(data_bitrate, args.bitrate)
        raw_bitrate = min(raw_bitrate, args.bitrate)

        raw_percent = raw_bitrate * 100 / args.bitrate
        data_percent = data_bitrate * 100 / args.bitrate

        row += [raw_bits, data_bits, raw_bitrate, data_bitrate, raw_percent, data_percent]
        writer.writerow(row)

    # Move the temp output file over the top of the input file
    output.close()
    input.close()
    if args.inplace:
        dst = pathlib.Path(args.input)
        shutil.move(tempoutput, dst)


if __name__ == "__main__":
    args = parse_args()
    fmt = "%(asctime)s - %(module)s - %(levelname)s: %(message)s"
    logging.basicConfig(
        format=fmt,
        level=args.log_level,
        stream=sys.stderr,
    )
    # Color log output if possible, because I'm a sucker
    try:
        import coloredlogs

        coloredlogs.install(fmt=fmt, level=args.log_level)
    except ImportError:
        pass
    main(args)