capkit

capkit is a Python library that reads CAN bus capture logs into one common frame stream. Every supported format parses into the same frozen Frame dataclass, so code that consumes frames never depends on which tool captured the log.

Use it to:

• read captures from different tools as one lazy stream of typed Frame objects
• filter frame streams lazily by arbitration ID, channel, and timestamp
• rebase timestamps lazily and explicitly without changing reader behavior
• merge already-time-ordered frame streams lazily across files or buses
• decompose J1939 arbitration IDs into priority, PGN, source, and optional PDU1 destination addresses
• probe a file for header metadata without scanning the frame body
• detect the log format from the file extension or the file content
• skip real-world log noise by default, or reject it with strict=True
• feed frames into dbckit for DBC signal decoding

See format support for the exact dialect each reader accepts, and the roadmap for sequencing.

Install

pip install capkit

Requires Python >=3.11. capkit has no runtime dependencies.

Design

• Frame and LogMeta are frozen, slotted dataclasses.
• read() is lazy and keeps constant parser state, so file size does not matter.
• read() returns timestamps exactly as recorded in the source; the separate rebase_timestamps() operation changes them only when explicitly requested.
• decompose_j1939_id() is pure per-ID arithmetic with no DBC or signal awareness and no dependency on dbckit.
• A format is added only when a real captured fixture pins its dialect under tests/fixtures/; unsupported dialects fail clearly instead of parsing approximately.

Quick Start

import capkit

# stream frames
for frame in capkit.read("trace.txt"):
    print(frame.timestamp, hex(frame.arbitration_id), frame.data.hex())

# inspect a J1939 ID without a DBC
j1939 = capkit.decompose_j1939_id(0x18EF20A5)
print(j1939.priority, hex(j1939.pgn), j1939.source_address, j1939.destination_address)

# compose lazy stream filters with inclusive time bounds
filtered = capkit.filter_frames(
    capkit.read("trace.txt"),
    arbitration_ids={0x123, 0x456},
    channels={1, 2},
    start_time=10.0,
    end_time=20.0,
)

# lazily make the first recorded timestamp zero
relative = capkit.rebase_timestamps(capkit.read("capture.log"))

# merge ordered captures that share a time base
merged = capkit.merge_frames(
    capkit.read("powertrain.asc"),
    capkit.read("body.asc"),
)

# header metadata only
meta = capkit.probe("trace.txt")
print(meta.format, meta.start_time)

# registered reader names
print(capkit.available_formats())   # ['candump', 'kvaser-txt', 'vector-asc']

The public API is eleven names: read, probe, available_formats, register_reader, decompose_j1939_id, filter_frames, merge_frames, rebase_timestamps, Frame, LogMeta, and J1939Fields.

Features

Inspect J1939 identifiers

decompose_j1939_id() validates a clean 29-bit arbitration ID and returns a frozen J1939Fields value containing its priority, PGN, source address, and optional PDU1 destination address. PDU1 destination bytes are excluded from the PGN; PDU2 group extensions remain part of it. This operation is immediate, dependency-free arithmetic and does not require a Frame or DBC.

Format detection

An explicit format= names a reader and takes precedence over the file extension:

frames = capkit.read("capture.bin", format="kvaser-txt")

Without format=, capkit matches the extension against registered readers and sniffs the first 4 KiB when the extension is unknown or ambiguous.

Add your own reader

Register a zero-argument reader class to make it available to read(), probe(), and format detection:

from collections.abc import Iterator
from pathlib import Path

import capkit


class MyReader:
    name: str = "my-format"
    extensions: tuple[str, ...] = (".mylog",)

    def __init__(self, *, strict: bool = False) -> None:
        self.strict = strict

    def sniff(self, sample: str) -> bool:
        return sample.startswith("MYLOG")

    def probe(self, path: Path) -> capkit.LogMeta:
        return capkit.LogMeta(format=self.name)

    def read(self, path: Path) -> Iterator[capkit.Frame]:
        # Parse path lazily and yield capkit.Frame objects here.
        yield from ()


capkit.register_reader(MyReader)

Registration is process-global. Installed packages can also advertise reader classes through the capkit.readers entry-point group; capkit discovers and caches them on the first read(), probe(), or available_formats() call. dbckit's .txt entry point sniffs among all registered readers, so a reader whose sniff() uniquely matches the content of a .txt log is used there too, regardless of the extensions it claims.

Dirty logs and strict mode

Readers skip headers, trailers, comments, blank lines, and unrelated noise by default. Pass strict=True to raise a line-numbered ValueError on the first unrecognized nonblank line instead:

frames = capkit.read("trace.txt", strict=True)

A frame record whose DLC disagrees with its data bytes raises in both modes; corrupt frames are never silently dropped.

Use with dbckit

dbckit decodes CAN frames against a DBC database. capkit and dbckit are separate packages — neither depends on or imports the other — with adjacent jobs: capkit turns bytes on disk into frames, dbckit turns frames plus a DBC into signals.

For J1939, capkit.decompose_j1939_id() exposes fields from each raw frame ID. dbckit remains responsible for using derived PGNs to match DBC messages and decode signals.

import capkit
import dbckit

db = dbckit.load("truck.dbc")
for decoded in dbckit.decode_frames(db, capkit.read("trace.txt")):
    print(decoded.timestamp, decoded.signals)

capkit also publishes exactly three extension-keyed entries in dbckit's dbckit.readers group: txt and log use capkit's sniffing DispatchReader, while asc uses VectorAscReader directly. These keys are file extensions, not capkit reader names: the corresponding capkit formats are kvaser-txt, candump, and vector-asc. With both packages installed, dbckit.decode_log() therefore reads .txt, .log, and .asc logs through capkit without manual registration:

for decoded in dbckit.decode_log(db, "trace.txt"):
    print(decoded.signals)

Scope and Caveats

• Kvaser dialects with absolute start-time headers are not supported; probe() returns start_time=None for kvaser-txt.
• candump error-flag records are skipped by default and rejected in strict mode; decoding them as CAN error frames is not claimed.
• Vector ASC relative timestamp directives and non-English month names are rejected instead of being interpreted approximately.
• capkit reads frames only: no DBC or signal awareness (that is dbckit's job), no hardware I/O, no log writing, no dataframe export, no CLI.

Documentation

• Format support — supported formats and the exact dialect each reader accepts
• API reference — the public API contract
• Recipes — counting and decomposing IDs, filtering and merging frame streams, cycle-time estimation, CSV export, and dataframes

Development

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pytest

The dev extra includes dbckit so the entry-point integration tests run; the core and contract suites pass without it.

License

MIT