From cfff53de7544241215c795dd4078e2c1aead4431 Mon Sep 17 00:00:00 2001 From: kipp-ing Date: Wed, 8 Jul 2026 22:24:47 +0200 Subject: [PATCH 1/5] add dynamic AUTOSAR PDU-container extraction to CAN bus logging Container (multiplexed PDU) frames carry a variable sequence of contained PDUs, each prefixed by a Header_ID/Header_DLC header, so a PDU's byte offset depends on the lengths of the PDUs before it. extract_mux() only handled is_multiplexed frames, so container frames were mis-decoded. - extract_pdus() in bus_logging_utils.py walks the container headers per frame (vectorized across frames, one pass per PDU slot), gathers each contained PDU payload and extracts its signals. Each contained PDU becomes its own channel group, reusing the existing channel-group machinery; a PDU's signal start bits are PDU-relative so extract_signal applies unchanged. canmatrix.Frame.unpack is the reference algorithm. - route is_pdu_container frames through extract_pdus in _extract_can_logging; all other frames keep using extract_mux. - fix extract_signal signedness: a signed signal with a standard bit width (8/16/32/64) at a non-byte-aligned offset was viewed as i{std_size} on the padded width instead of being sign-extended from its real bit width (e.g. an 8-bit signed field at bit offset 1 returned 225 instead of -31). This affected the normal extract_mux path too. - add offline test/test_CAN_pdu_extraction.py validating extract_pdus against canmatrix.unpack (big/little-endian headers, 0x00/0xFF padding, unique multi-PDU frames, signed bit-packed signals) plus a full extract_bus_logging end-to-end case and a static-container skip. --- src/asammdf/blocks/bus_logging_utils.py | 236 +++++++++++++++++++++++- src/asammdf/blocks/mdf_v4.py | 38 ++-- test/test_CAN_pdu_extraction.py | 164 ++++++++++++++++ 3 files changed, 425 insertions(+), 13 deletions(-) create mode 100644 test/test_CAN_pdu_extraction.py diff --git a/src/asammdf/blocks/bus_logging_utils.py b/src/asammdf/blocks/bus_logging_utils.py index 54bb68a24..78e0432a5 100644 --- a/src/asammdf/blocks/bus_logging_utils.py +++ b/src/asammdf/blocks/bus_logging_utils.py @@ -226,7 +226,11 @@ def extract_signal( vals &= (2**bit_count) - 1 if signed and not is_float: - if extra_bytes or bit_count not in (8, 16, 32, 64): + # A plain ``view("i{std_size}")`` only sign-extends correctly when the + # value fills exactly ``std_size`` byte-aligned bytes. A non-byte-aligned + # signal (``bit_offset``) has been shifted/masked into a wider unsigned + # container, so it must be sign-extended from its real bit width instead. + if extra_bytes or bit_offset or bit_count not in (8, 16, 32, 64): vals = as_non_byte_sized_signed_int(vals, bit_count) else: vals = vals.view(f"i{std_size}") @@ -461,6 +465,236 @@ def extract_mux( return extracted_signals +# Reserved synthetic signal names that canmatrix injects into a PDU-container +# frame to describe the per-PDU header. They are not user payload signals. +PDU_CONTAINER_HEADER_ID: Final = "Header_ID" +PDU_CONTAINER_HEADER_DLC: Final = "Header_DLC" + + +def extract_pdus( + payload: NDArray[Any], + message: Frame, + message_id: int | None, + bus: int | None, + t: NDArray[Any], + original_message_id: int | None = None, + raw: bool = False, + include_message_name: bool = False, + ignore_value2text_conversion: bool = True, + is_j1939: bool = False, + is_extended: bool = False, +) -> dict[tuple[int | None, int | None, bool, int | None, str | None, int, int], dict[str, ExtractedSignal]]: + """Extract signals from an AUTOSAR dynamic PDU-container CAN frame. + + A dynamic container frame carries a variable sequence of contained PDUs, + each prefixed by a header (``Header_ID`` + ``Header_DLC``). Because a PDU's + byte offset depends on the lengths of the PDUs before it, the container is + walked header-by-header per frame; the contained PDU payloads are then + gathered per PDU id and their signals extracted vectorized. + + The reference algorithm is ``canmatrix.Frame.unpack`` for a + ``is_pdu_container`` frame. A contained PDU's signal ``start_bit`` values are + relative to the PDU payload (not the frame), so once a PDU payload slice is + isolated the regular :func:`extract_signal` machinery applies unchanged. + + Only dynamic containers (those exposing ``Header_ID``/``Header_DLC``) are + handled; static containers are skipped (empty result). + + Parameters + ---------- + payload : np.ndarray + Raw CAN payload as 2D numpy array of shape ``(n_frames, n_bytes)``. + message : canmatrix.Frame + Container frame description parsed by canmatrix. + message_id : int + Message id of the container frame. + bus : int + Bus channel number. + t : np.ndarray + Timestamps for the raw payload. + original_message_id : int, optional + Original message id. + ignore_value2text_conversion : bool, default True + Ignore value to text conversions. + + Returns + ------- + extracted_signals : dict + Same structure as :func:`extract_mux`: keyed by an entry tuple, each + value is the dict of signals for one contained PDU. The PDU identity is + carried in the ``muxer`` slot of the entry so every contained PDU maps + to its own channel group. + """ + + extracted_signals: dict[ + tuple[int | None, int | None, bool, int | None, str | None, int, int], dict[str, ExtractedSignal] + ] = {} + + header_id_signal = message.signal_by_name(PDU_CONTAINER_HEADER_ID) + header_dlc_signal = message.signal_by_name(PDU_CONTAINER_HEADER_DLC) + # Static containers (no per-PDU header) are out of scope. + if header_id_signal is None or header_dlc_signal is None: + return extracted_signals + + if payload.shape[1] == 0 or len(payload) == 0: + return extracted_signals + + n_frames = payload.shape[0] + frame_bytes = payload.shape[1] + + # Header geometry, derived from the header signals (short header: 24 + 8; + # long header: 32 + 32). Header is assumed byte aligned at the frame start, + # matching canmatrix's container decoder. + header_size = (header_id_signal.size + header_dlc_signal.size + 7) // 8 + + if header_size == 0 or header_size > frame_bytes: + return extracted_signals + + # Contained PDU payload sizes (fall back to the maximum signal extent when + # the PDU length is not populated by the parser). + def _pdu_size(pdu: Any) -> int: + size = getattr(pdu, "size", 0) or 0 + if size: + return int(size) + extent = 0 + for sig in pdu.signals: + extent = max(extent, sig.get_startbit(bit_numbering=1) + sig.size) + return (extent + 7) // 8 + + pdu_sizes = {pdu.id: _pdu_size(pdu) for pdu in message.pdus} + max_pdu_size = max(pdu_sizes.values(), default=0) + + # Pad on the right with 0xFF so a truncated trailing PDU can still be read + # (mirrors canmatrix's allow_truncated behaviour); the header walk itself is + # bounded by the original frame width. + if max_pdu_size: + padded = np.column_stack([payload, np.full(n_frames, 0xFF, dtype=f"({max_pdu_size},)u1")]) + else: + padded = payload + + # --- Walk the headers across all frames, one header "slot" per iteration. + # Frames diverge in offset after the first PDU, so we advance a per-frame + # byte offset and gather each frame's current header window vectorized. + offset = np.zeros(n_frames, dtype=np.int64) + rows_all: list[NDArray[Any]] = [] + ids_all: list[NDArray[Any]] = [] + starts_all: list[NDArray[Any]] = [] + + max_slots = frame_bytes // header_size + 1 + for _ in range(max_slots): + active = np.nonzero((offset + header_size) <= frame_bytes)[0] + if active.size == 0: + break + + offs = offset[active] + window_cols = offs[:, None] + np.arange(header_size)[None, :] + windows = padded[active[:, None], window_cols] + + ids = extract_signal(header_id_signal, windows, raw=True).astype(" +header_size; 0xFF padding -> past end). + offset[active] = payload_start + dlcs + + if not rows_all: + return extracted_signals + + rows_flat = np.concatenate(rows_all) + ids_flat = np.concatenate(ids_all) + starts_flat = np.concatenate(starts_all) + + # --- Per contained PDU: gather its payload slices and extract its signals. + for pdu in message.pdus: + pdu_size = pdu_sizes[pdu.id] + if pdu_size == 0: + continue + + mask = ids_flat == pdu.id + if not mask.any(): + continue + + sel_rows = rows_flat[mask] + sel_starts = starts_flat[mask] + + # Keep timestamp order (a frame may contain several PDUs / repeats). + order = np.argsort(sel_rows, kind="stable") + sel_rows = sel_rows[order] + sel_starts = sel_starts[order] + + gather_cols = sel_starts[:, None] + np.arange(pdu_size)[None, :] + pdu_payload = padded[sel_rows[:, None], gather_cols] + t_ = t[sel_rows] + + entry = ( + bus, + message_id, + is_extended, + original_message_id, + f"ContainedPDU:0x{pdu.id:X}:{pdu.name}", + 0, + 0, + ) + signals = extracted_signals.setdefault(entry, {}) + + for sig in pdu.signals: + if sig.name in (PDU_CONTAINER_HEADER_ID, PDU_CONTAINER_HEADER_DLC): + continue + + samples = extract_signal( + sig, + pdu_payload, + ignore_value2text_conversion=ignore_value2text_conversion, + raw=True, + ) + if len(samples) == 0 and len(t_): + continue + + if include_message_name: + sig_name = f"{message.name}.{sig.name}" + else: + sig_name = sig.name + + try: + scale_ranges = getattr(sig, "scale_ranges", None) + if scale_ranges: + unit = scale_ranges[0]["unit"] or "" + else: + unit = sig.unit or "" + + signals[sig_name] = { + "name": sig_name, + "comment": sig.comment or "", + "unit": unit, + "samples": samples if raw else apply_conversion(samples, sig, ignore_value2text_conversion), + "conversion": get_conversion(sig) if raw else None, + "t": t_, + "invalidation_bits": None, + } + + if is_j1939: + signals[sig_name]["invalidation_bits"] = samples > MAX_VALID_J1939[defined_j1939_bit_count(sig)] + + except: + print(format_exc()) + print(message, pdu, sig) + print(samples, set(samples), samples.dtype, samples.shape) + raise + + # Drop the PDU entirely if none of its signals produced samples. + if not signals: + del extracted_signals[entry] + + return extracted_signals + + def get_conversion(signal: Signal) -> v4b.ChannelConversion: conv: v4b.ChannelConversionKwargs = {} diff --git a/src/asammdf/blocks/mdf_v4.py b/src/asammdf/blocks/mdf_v4.py index df71d52b1..5861caea2 100644 --- a/src/asammdf/blocks/mdf_v4.py +++ b/src/asammdf/blocks/mdf_v4.py @@ -10437,18 +10437,32 @@ def _extract_can_logging( t = bus_t[idx] try: - extracted_signals = bus_logging_utils.extract_mux( - payload, - message, - msg_id, - bus, - t, - original_message_id=source_address if is_j1939 else None, - ignore_value2text_conversion=ignore_value2text_conversion, - is_j1939=is_j1939, - is_extended=is_extended, - raw=True, - ) + if message.is_pdu_container: + extracted_signals = bus_logging_utils.extract_pdus( + payload, + message, + msg_id, + bus, + t, + original_message_id=source_address if is_j1939 else None, + ignore_value2text_conversion=ignore_value2text_conversion, + is_j1939=is_j1939, + is_extended=is_extended, + raw=True, + ) + else: + extracted_signals = bus_logging_utils.extract_mux( + payload, + message, + msg_id, + bus, + t, + original_message_id=source_address if is_j1939 else None, + ignore_value2text_conversion=ignore_value2text_conversion, + is_j1939=is_j1939, + is_extended=is_extended, + raw=True, + ) except: print(format_exc()) raise diff --git a/test/test_CAN_pdu_extraction.py b/test/test_CAN_pdu_extraction.py new file mode 100644 index 000000000..aa38efcee --- /dev/null +++ b/test/test_CAN_pdu_extraction.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python +"""Tests for AUTOSAR dynamic PDU-container extraction from CAN bus logging. + +Fully offline: a container is built in-memory with canmatrix and +``canmatrix.Frame.unpack`` is used as the decoding oracle. +""" + +import random +import unittest + +import canmatrix +import numpy as np + +from asammdf import MDF, Signal +from asammdf.blocks import v4_constants as v4c +from asammdf.blocks.bus_logging_utils import extract_pdus +from asammdf.blocks.source_utils import Source + +FRAME_BYTES = 32 +CONTAINER_ID = 0x555 + + +def build_container(header_big_endian: bool = True) -> tuple[canmatrix.Frame, dict[int, canmatrix.Pdu]]: + cont = canmatrix.Frame(name="Cont", size=FRAME_BYTES) + cont.arbitration_id = canmatrix.ArbitrationId(id=CONTAINER_ID, extended=False) + cont.add_signal(canmatrix.Signal(name="Header_ID", start_bit=0, size=24, is_little_endian=not header_big_endian)) + cont.add_signal(canmatrix.Signal(name="Header_DLC", start_bit=24, size=8, is_little_endian=not header_big_endian)) + + pA = canmatrix.Pdu(name="PduA", id=0x100, size=3) + pA.add_signal(canmatrix.Signal(name="A_u8", start_bit=0, size=8, is_little_endian=True, is_signed=False)) + pA.add_signal(canmatrix.Signal(name="A_i16", start_bit=8, size=16, is_little_endian=True, is_signed=True)) + pB = canmatrix.Pdu(name="PduB", id=0x2AB, size=2) + pB.add_signal(canmatrix.Signal(name="B_u16", start_bit=0, size=16, is_little_endian=False, is_signed=False)) + pC = canmatrix.Pdu(name="PduC", id=0x77, size=1) + pC.add_signal(canmatrix.Signal(name="C_u8", start_bit=0, size=8, is_little_endian=True, is_signed=False)) + # PduD exercises bit-packed, non-byte-aligned *signed* signals (the geometry + # that real OEM containers use heavily and that the signedness handling of + # extract_signal must get right regardless of alignment). + pD = canmatrix.Pdu(name="PduD", id=0x3C0, size=3) + pD.add_signal(canmatrix.Signal(name="D_nib", start_bit=0, size=4, is_little_endian=True, is_signed=False)) + pD.add_signal(canmatrix.Signal(name="D_i8_be", start_bit=4, size=8, is_little_endian=False, is_signed=True)) + pD.add_signal(canmatrix.Signal(name="D_i8_le", start_bit=12, size=8, is_little_endian=True, is_signed=True)) + pD.add_signal(canmatrix.Signal(name="D_end", start_bit=20, size=4, is_little_endian=True, is_signed=False)) + for p in (pA, pB, pC, pD): + cont.add_pdu(p) + return cont, {p.id: p for p in (pA, pB, pC, pD)} + + +def _encode_header(pdu_id: int, dlc: int, big_endian: bool) -> bytes: + if big_endian: + return bytes([(pdu_id >> 16) & 0xFF, (pdu_id >> 8) & 0xFF, pdu_id & 0xFF, dlc & 0xFF]) + return bytes([pdu_id & 0xFF, (pdu_id >> 8) & 0xFF, (pdu_id >> 16) & 0xFF, dlc & 0xFF]) + + +def build_frames(pdus: dict[int, canmatrix.Pdu], n_frames: int, big_endian: bool, pad: int, seed: int) -> list[bytes]: + random.seed(seed) + ids = list(pdus) + frames = [] + for _ in range(n_frames): + buf, used = bytearray(), 0 + for _ in range(random.randint(0, 4)): + pid = random.choice(ids) + need = 4 + pdus[pid].size + if used + need > FRAME_BYTES: + break + buf += _encode_header(pid, pdus[pid].size, big_endian) + buf += bytes(random.randrange(256) for _ in range(pdus[pid].size)) + used += need + frames.append(bytes(buf).ljust(FRAME_BYTES, bytes([pad]))) + return frames + + +def oracle(cont: canmatrix.Frame, frames: list[bytes], t: np.ndarray): + """Ground truth via canmatrix.Frame.unpack, accumulated in frame order.""" + values: dict[tuple[str, str], list] = {} + times: dict[str, list] = {} + for i, frame in enumerate(frames): + for pdu_entry in cont.unpack(frame)["pdus"]: + if pdu_entry is None: + continue + for pdu_name, sigdict in pdu_entry.items(): + times.setdefault(pdu_name, []).append(t[i]) + for sig_name, decoded in sigdict.items(): + values.setdefault((pdu_name, sig_name), []).append(decoded.raw_value) + return values, times + + +class TestPduContainerExtraction(unittest.TestCase): + def test_extract_pdus_matches_canmatrix_unpack(self) -> None: + for big_endian in (True, False): + for pad in (0x00, 0xFF): + with self.subTest(header_big_endian=big_endian, pad=pad): + cont, pdus = build_container(big_endian) + frames = build_frames(pdus, n_frames=200, big_endian=big_endian, pad=pad, seed=1) + payload = np.array([list(f) for f in frames], dtype="u1") + t = np.arange(len(frames), dtype="f8") + + extracted = extract_pdus(payload, cont, message_id=CONTAINER_ID, bus=1, t=t, raw=True) + + exp_values, exp_times = oracle(cont, frames, t) + got = {entry[4].split(":")[-1]: sigs for entry, sigs in extracted.items()} + + self.assertEqual(set(got), set(exp_times)) + for (pdu_name, sig_name), exp in exp_values.items(): + samples = got[pdu_name][sig_name]["samples"] + self.assertTrue( + np.array_equal(samples, np.array(exp)), + f"{pdu_name}.{sig_name} mismatch", + ) + self.assertTrue( + np.array_equal(got[pdu_name][sig_name]["t"], np.array(exp_times[pdu_name], dtype="f8")) + ) + + def test_extract_bus_logging_container_e2e(self) -> None: + cont, pdus = build_container(header_big_endian=True) + db = canmatrix.CanMatrix() + db.add_frame(cont) + + frames = build_frames(pdus, n_frames=150, big_endian=True, pad=0x00, seed=7) + payload = np.array([list(f) for f in frames], dtype="u1") + t = np.arange(len(frames), dtype="f8") * 0.01 + + dtype = np.dtype( + [ + ("CAN_DataFrame.BusChannel", "u1"), + ("CAN_DataFrame.ID", "u4"), + ("CAN_DataFrame.IDE", "u1"), + ("CAN_DataFrame.DataBytes", "u1", (FRAME_BYTES,)), + ] + ) + rec = np.zeros(len(frames), dtype=dtype) + rec["CAN_DataFrame.ID"] = CONTAINER_ID + rec["CAN_DataFrame.DataBytes"] = payload + + acq_source = Source(name="CAN", path="CAN", comment="", source_type=v4c.SOURCE_BUS, bus_type=v4c.BUS_TYPE_CAN) + with MDF(version="4.10") as mdf: + cg_nr = mdf.append(Signal(samples=rec, timestamps=t, name="CAN_DataFrame"), acq_source=acq_source) + mdf.groups[cg_nr].channel_group.flags = v4c.FLAG_CG_BUS_EVENT + + out = mdf.extract_bus_logging({"CAN": [(db, 0)]}, ignore_value2text_conversion=True) + + # one channel group per contained PDU that actually appeared + exp_values, _exp_times = oracle(cont, frames, t) + self.assertEqual(len(out.groups), len({p for (p, _s) in exp_values})) + + for (pdu_name, sig_name), exp in exp_values.items(): + sig = out.get(sig_name, raw=True) + self.assertTrue(np.array_equal(sig.samples, np.array(exp)), f"{pdu_name}.{sig_name} mismatch") + + def test_static_container_without_header_is_skipped(self) -> None: + """A container lacking Header_ID/Header_DLC (static) yields no signals.""" + cont = canmatrix.Frame(name="StaticCont", size=8) + pdu = canmatrix.Pdu(name="Pdu", id=0x1, size=2) + pdu.add_signal(canmatrix.Signal(name="S", start_bit=0, size=16, is_little_endian=True)) + cont.add_pdu(pdu) + self.assertTrue(cont.is_pdu_container) + + payload = np.zeros((5, 8), dtype="u1") + t = np.arange(5, dtype="f8") + self.assertEqual(extract_pdus(payload, cont, message_id=1, bus=0, t=t, raw=True), {}) + + +if __name__ == "__main__": + unittest.main() From 523a7086facd7349b9f63e511764d23aa49691a4 Mon Sep 17 00:00:00 2001 From: kipp-ing Date: Thu, 9 Jul 2026 19:52:31 +0200 Subject: [PATCH 2/5] support static PDU containers and cover the CAN-FD path Static (header-less) AUTOSAR containers have a fixed layout: canmatrix rebases each contained PDU's signal start bits to be frame-relative and Frame.unpack itself refuses them, so they were previously skipped. Handle them by decoding every contained PDU straight from the full frame payload, one channel group each (pdu.id is None -> identity keyed on the PDU name). - factor the per-PDU signal emission out of extract_pdus into _emit_pdu_signals, shared by the dynamic (PDU-relative payload) and static (full-frame payload) paths; _contained_pdu_muxer builds the channel-group identity and tolerates a None header id. - test_extract_pdus_static_container validates static extraction against a flat frame carrying the same frame-relative signals (byte-aligned LE/BE plus a non-byte-aligned signed field). - test_extract_bus_logging_canfd_container_e2e runs the full pipeline on genuine CAN-FD container frames (64-byte payload, EDL flag + DataLength members set) to exercise the real-world container transport. --- src/asammdf/blocks/bus_logging_utils.py | 198 ++++++++++++++++-------- test/test_CAN_pdu_extraction.py | 139 ++++++++++++++--- 2 files changed, 255 insertions(+), 82 deletions(-) diff --git a/src/asammdf/blocks/bus_logging_utils.py b/src/asammdf/blocks/bus_logging_utils.py index 78e0432a5..dd991f13f 100644 --- a/src/asammdf/blocks/bus_logging_utils.py +++ b/src/asammdf/blocks/bus_logging_utils.py @@ -471,6 +471,99 @@ def extract_mux( PDU_CONTAINER_HEADER_DLC: Final = "Header_DLC" +def _contained_pdu_muxer(pdu: Any) -> str: + """Stable per-PDU identity for the entry ``muxer`` slot so every contained + PDU maps to its own channel group. Static (header-less) containers carry no + header id, so ``pdu.id`` is ``None`` there and the name alone identifies it. + """ + if pdu.id is None: + return f"ContainedPDU:{pdu.name}" + return f"ContainedPDU:0x{pdu.id:X}:{pdu.name}" + + +def _emit_pdu_signals( + extracted_signals: dict[ + tuple[int | None, int | None, bool, int | None, str | None, int, int], dict[str, ExtractedSignal] + ], + message: Frame, + pdu: Any, + pdu_payload: NDArray[Any], + t_: NDArray[Any], + *, + bus: int | None, + message_id: int | None, + is_extended: bool, + original_message_id: int | None, + raw: bool, + include_message_name: bool, + ignore_value2text_conversion: bool, + is_j1939: bool, +) -> None: + """Extract one contained PDU's signals from its payload slice into a + dedicated channel-group entry. Shared by the dynamic and static paths of + :func:`extract_pdus`; ``pdu_payload`` is the PDU-relative payload for a + dynamic container and the full frame payload for a static one. + """ + entry = ( + bus, + message_id, + is_extended, + original_message_id, + _contained_pdu_muxer(pdu), + 0, + 0, + ) + signals = extracted_signals.setdefault(entry, {}) + + for sig in pdu.signals: + if sig.name in (PDU_CONTAINER_HEADER_ID, PDU_CONTAINER_HEADER_DLC): + continue + + samples = extract_signal( + sig, + pdu_payload, + ignore_value2text_conversion=ignore_value2text_conversion, + raw=True, + ) + if len(samples) == 0 and len(t_): + continue + + if include_message_name: + sig_name = f"{message.name}.{sig.name}" + else: + sig_name = sig.name + + try: + scale_ranges = getattr(sig, "scale_ranges", None) + if scale_ranges: + unit = scale_ranges[0]["unit"] or "" + else: + unit = sig.unit or "" + + signals[sig_name] = { + "name": sig_name, + "comment": sig.comment or "", + "unit": unit, + "samples": samples if raw else apply_conversion(samples, sig, ignore_value2text_conversion), + "conversion": get_conversion(sig) if raw else None, + "t": t_, + "invalidation_bits": None, + } + + if is_j1939: + signals[sig_name]["invalidation_bits"] = samples > MAX_VALID_J1939[defined_j1939_bit_count(sig)] + + except: + print(format_exc()) + print(message, pdu, sig) + print(samples, set(samples), samples.dtype, samples.shape) + raise + + # Drop the PDU entirely if none of its signals produced samples. + if not signals: + extracted_signals.pop(entry, None) + + def extract_pdus( payload: NDArray[Any], message: Frame, @@ -497,8 +590,11 @@ def extract_pdus( relative to the PDU payload (not the frame), so once a PDU payload slice is isolated the regular :func:`extract_signal` machinery applies unchanged. - Only dynamic containers (those exposing ``Header_ID``/``Header_DLC``) are - handled; static containers are skipped (empty result). + Static containers (no ``Header_ID``/``Header_DLC``, i.e. no per-PDU header) + are also handled: they have a fixed layout and canmatrix has already rebased + each contained PDU's signal ``start_bit`` values to be frame-relative, so + every PDU decodes straight from the full frame payload. ``Frame.unpack`` + itself refuses these, but the fixed-layout metadata is complete. Parameters ---------- @@ -530,13 +626,32 @@ def extract_pdus( tuple[int | None, int | None, bool, int | None, str | None, int, int], dict[str, ExtractedSignal] ] = {} + if payload.shape[1] == 0 or len(payload) == 0: + return extracted_signals + header_id_signal = message.signal_by_name(PDU_CONTAINER_HEADER_ID) header_dlc_signal = message.signal_by_name(PDU_CONTAINER_HEADER_DLC) - # Static containers (no per-PDU header) are out of scope. - if header_id_signal is None or header_dlc_signal is None: - return extracted_signals - if payload.shape[1] == 0 or len(payload) == 0: + # Static container (no per-PDU header): fixed layout, every contained PDU is + # present in every frame and its signals are already frame-relative, so each + # PDU decodes straight from the full frame payload. + if header_id_signal is None or header_dlc_signal is None: + for pdu in message.pdus: + _emit_pdu_signals( + extracted_signals, + message, + pdu, + payload, + t, + bus=bus, + message_id=message_id, + is_extended=is_extended, + original_message_id=original_message_id, + raw=raw, + include_message_name=include_message_name, + ignore_value2text_conversion=ignore_value2text_conversion, + is_j1939=is_j1939, + ) return extracted_signals n_frames = payload.shape[0] @@ -633,64 +748,21 @@ def _pdu_size(pdu: Any) -> int: pdu_payload = padded[sel_rows[:, None], gather_cols] t_ = t[sel_rows] - entry = ( - bus, - message_id, - is_extended, - original_message_id, - f"ContainedPDU:0x{pdu.id:X}:{pdu.name}", - 0, - 0, + _emit_pdu_signals( + extracted_signals, + message, + pdu, + pdu_payload, + t_, + bus=bus, + message_id=message_id, + is_extended=is_extended, + original_message_id=original_message_id, + raw=raw, + include_message_name=include_message_name, + ignore_value2text_conversion=ignore_value2text_conversion, + is_j1939=is_j1939, ) - signals = extracted_signals.setdefault(entry, {}) - - for sig in pdu.signals: - if sig.name in (PDU_CONTAINER_HEADER_ID, PDU_CONTAINER_HEADER_DLC): - continue - - samples = extract_signal( - sig, - pdu_payload, - ignore_value2text_conversion=ignore_value2text_conversion, - raw=True, - ) - if len(samples) == 0 and len(t_): - continue - - if include_message_name: - sig_name = f"{message.name}.{sig.name}" - else: - sig_name = sig.name - - try: - scale_ranges = getattr(sig, "scale_ranges", None) - if scale_ranges: - unit = scale_ranges[0]["unit"] or "" - else: - unit = sig.unit or "" - - signals[sig_name] = { - "name": sig_name, - "comment": sig.comment or "", - "unit": unit, - "samples": samples if raw else apply_conversion(samples, sig, ignore_value2text_conversion), - "conversion": get_conversion(sig) if raw else None, - "t": t_, - "invalidation_bits": None, - } - - if is_j1939: - signals[sig_name]["invalidation_bits"] = samples > MAX_VALID_J1939[defined_j1939_bit_count(sig)] - - except: - print(format_exc()) - print(message, pdu, sig) - print(samples, set(samples), samples.dtype, samples.shape) - raise - - # Drop the PDU entirely if none of its signals produced samples. - if not signals: - del extracted_signals[entry] return extracted_signals diff --git a/test/test_CAN_pdu_extraction.py b/test/test_CAN_pdu_extraction.py index aa38efcee..d13b7872b 100644 --- a/test/test_CAN_pdu_extraction.py +++ b/test/test_CAN_pdu_extraction.py @@ -1,8 +1,11 @@ #!/usr/bin/env python -"""Tests for AUTOSAR dynamic PDU-container extraction from CAN bus logging. +"""Tests for AUTOSAR PDU-container extraction from CAN bus logging. -Fully offline: a container is built in-memory with canmatrix and -``canmatrix.Frame.unpack`` is used as the decoding oracle. +Fully offline: containers are built in-memory with canmatrix and +``canmatrix.Frame.unpack`` is used as the decoding oracle for dynamic +containers. Static (header-less) containers are decoded against a flat frame +carrying the same frame-relative signals, because ``Frame.unpack`` itself +refuses static containers. """ import random @@ -17,11 +20,15 @@ from asammdf.blocks.source_utils import Source FRAME_BYTES = 32 +FD_FRAME_BYTES = 64 CONTAINER_ID = 0x555 +STATIC_CONTAINER_ID = 0x556 -def build_container(header_big_endian: bool = True) -> tuple[canmatrix.Frame, dict[int, canmatrix.Pdu]]: - cont = canmatrix.Frame(name="Cont", size=FRAME_BYTES) +def build_container( + header_big_endian: bool = True, frame_bytes: int = FRAME_BYTES +) -> tuple[canmatrix.Frame, dict[int, canmatrix.Pdu]]: + cont = canmatrix.Frame(name="Cont", size=frame_bytes) cont.arbitration_id = canmatrix.ArbitrationId(id=CONTAINER_ID, extended=False) cont.add_signal(canmatrix.Signal(name="Header_ID", start_bit=0, size=24, is_little_endian=not header_big_endian)) cont.add_signal(canmatrix.Signal(name="Header_DLC", start_bit=24, size=8, is_little_endian=not header_big_endian)) @@ -46,13 +53,54 @@ def build_container(header_big_endian: bool = True) -> tuple[canmatrix.Frame, di return cont, {p.id: p for p in (pA, pB, pC, pD)} +def build_static_container(frame_bytes: int = 8) -> tuple[canmatrix.Frame, canmatrix.Frame]: + """A header-less (static) container plus the equivalent flat frame oracle. + + Contained PDUs sit at fixed byte offsets and their signal ``start_bit`` + values are frame-relative, exactly as canmatrix emits after applying each + PDU's ``OFFSET``; there is no header id, so ``pdu.id`` is ``None``. The flat + frame carries the same signals so ``canmatrix.Frame.unpack`` (which refuses + static containers) can serve as the decoding oracle. + """ + cont = canmatrix.Frame(name="StaticCont", size=frame_bytes) + cont.arbitration_id = canmatrix.ArbitrationId(id=STATIC_CONTAINER_ID, extended=False) + + # PDU at frame byte 0 (byte-aligned little-endian, incl. signed multi-byte). + p1 = canmatrix.Pdu(name="StatA", id=None, size=3) + p1.add_signal(canmatrix.Signal(name="SA_u8", start_bit=0, size=8, is_little_endian=True, is_signed=False)) + p1.add_signal(canmatrix.Signal(name="SA_i16", start_bit=8, size=16, is_little_endian=True, is_signed=True)) + # PDU at frame byte 3 (byte-aligned big-endian). + p2 = canmatrix.Pdu(name="StatB", id=None, size=2) + p2.add_signal(canmatrix.Signal(name="SB_u16_be", start_bit=24, size=16, is_little_endian=False, is_signed=False)) + # PDU at frame byte 5 (bit-packed, non-byte-aligned signed -> the path the + # extract_signal signedness fix must handle at a frame offset). + p3 = canmatrix.Pdu(name="StatC", id=None, size=3) + p3.add_signal(canmatrix.Signal(name="SC_nib", start_bit=40, size=4, is_little_endian=True, is_signed=False)) + p3.add_signal(canmatrix.Signal(name="SC_i8", start_bit=44, size=8, is_little_endian=True, is_signed=True)) + p3.add_signal(canmatrix.Signal(name="SC_end", start_bit=52, size=12, is_little_endian=True, is_signed=False)) + + flat = canmatrix.Frame(name="StaticFlat", size=frame_bytes) + for p in (p1, p2, p3): + cont.add_pdu(p) + for s in p.signals: + flat.add_signal(s) + return cont, flat + + def _encode_header(pdu_id: int, dlc: int, big_endian: bool) -> bytes: if big_endian: return bytes([(pdu_id >> 16) & 0xFF, (pdu_id >> 8) & 0xFF, pdu_id & 0xFF, dlc & 0xFF]) return bytes([pdu_id & 0xFF, (pdu_id >> 8) & 0xFF, (pdu_id >> 16) & 0xFF, dlc & 0xFF]) -def build_frames(pdus: dict[int, canmatrix.Pdu], n_frames: int, big_endian: bool, pad: int, seed: int) -> list[bytes]: +def build_frames( + pdus: dict[int, canmatrix.Pdu], + n_frames: int, + big_endian: bool, + pad: int, + seed: int, + frame_bytes: int = FRAME_BYTES, +) -> list[bytes]: random.seed(seed) ids = list(pdus) frames = [] @@ -61,12 +109,12 @@ def build_frames(pdus: dict[int, canmatrix.Pdu], n_frames: int, big_endian: bool for _ in range(random.randint(0, 4)): pid = random.choice(ids) need = 4 + pdus[pid].size - if used + need > FRAME_BYTES: + if used + need > frame_bytes: break buf += _encode_header(pid, pdus[pid].size, big_endian) buf += bytes(random.randrange(256) for _ in range(pdus[pid].size)) used += need - frames.append(bytes(buf).ljust(FRAME_BYTES, bytes([pad]))) + frames.append(bytes(buf).ljust(frame_bytes, bytes([pad]))) return frames @@ -111,6 +159,29 @@ def test_extract_pdus_matches_canmatrix_unpack(self) -> None: np.array_equal(got[pdu_name][sig_name]["t"], np.array(exp_times[pdu_name], dtype="f8")) ) + def test_extract_pdus_static_container(self) -> None: + """A static (header-less) container: every contained PDU is present in + every frame and decodes at its fixed frame offset, one channel group + each. Oracle is the equivalent flat frame.""" + cont, flat = build_static_container(frame_bytes=8) + rng = np.random.default_rng(3) + payload = rng.integers(0, 256, size=(120, 8), dtype="u1") + frames = [bytes(row) for row in payload] + t = np.arange(len(frames), dtype="f8") + + extracted = extract_pdus(payload, cont, message_id=STATIC_CONTAINER_ID, bus=2, t=t, raw=True) + + got = {entry[4].split(":")[-1]: sigs for entry, sigs in extracted.items()} + self.assertEqual(set(got), {p.name for p in cont.pdus}) + + decoded = [flat.unpack(f) for f in frames] + for pdu in cont.pdus: + for sig in pdu.signals: + exp = np.array([d[sig.name].raw_value for d in decoded]) + samples = got[pdu.name][sig.name]["samples"] + self.assertTrue(np.array_equal(samples, exp), f"{pdu.name}.{sig.name} mismatch") + self.assertTrue(np.array_equal(got[pdu.name][sig.name]["t"], t)) + def test_extract_bus_logging_container_e2e(self) -> None: cont, pdus = build_container(header_big_endian=True) db = canmatrix.CanMatrix() @@ -147,17 +218,47 @@ def test_extract_bus_logging_container_e2e(self) -> None: sig = out.get(sig_name, raw=True) self.assertTrue(np.array_equal(sig.samples, np.array(exp)), f"{pdu_name}.{sig_name} mismatch") - def test_static_container_without_header_is_skipped(self) -> None: - """A container lacking Header_ID/Header_DLC (static) yields no signals.""" - cont = canmatrix.Frame(name="StaticCont", size=8) - pdu = canmatrix.Pdu(name="Pdu", id=0x1, size=2) - pdu.add_signal(canmatrix.Signal(name="S", start_bit=0, size=16, is_little_endian=True)) - cont.add_pdu(pdu) - self.assertTrue(cont.is_pdu_container) - - payload = np.zeros((5, 8), dtype="u1") - t = np.arange(5, dtype="f8") - self.assertEqual(extract_pdus(payload, cont, message_id=1, bus=0, t=t, raw=True), {}) + def test_extract_bus_logging_canfd_container_e2e(self) -> None: + """Full pipeline on genuine CAN-FD container frames: 64-byte payload + with the EDL (extended data length) flag set and a DataLength member, + the real-world transport for AUTOSAR containers.""" + cont, pdus = build_container(header_big_endian=True, frame_bytes=FD_FRAME_BYTES) + db = canmatrix.CanMatrix() + db.add_frame(cont) + + frames = build_frames(pdus, n_frames=200, big_endian=True, pad=0x00, seed=11, frame_bytes=FD_FRAME_BYTES) + payload = np.array([list(f) for f in frames], dtype="u1") + t = np.arange(len(frames), dtype="f8") * 0.01 + + dtype = np.dtype( + [ + ("CAN_DataFrame.BusChannel", "u1"), + ("CAN_DataFrame.ID", "u4"), + ("CAN_DataFrame.IDE", "u1"), + ("CAN_DataFrame.EDL", "u1"), + ("CAN_DataFrame.DataLength", "u1"), + ("CAN_DataFrame.DataBytes", "u1", (FD_FRAME_BYTES,)), + ] + ) + rec = np.zeros(len(frames), dtype=dtype) + rec["CAN_DataFrame.ID"] = CONTAINER_ID + rec["CAN_DataFrame.EDL"] = 1 # CAN-FD frame + rec["CAN_DataFrame.DataLength"] = FD_FRAME_BYTES + rec["CAN_DataFrame.DataBytes"] = payload + + acq_source = Source(name="CAN", path="CAN", comment="", source_type=v4c.SOURCE_BUS, bus_type=v4c.BUS_TYPE_CAN) + with MDF(version="4.10") as mdf: + cg_nr = mdf.append(Signal(samples=rec, timestamps=t, name="CAN_DataFrame"), acq_source=acq_source) + mdf.groups[cg_nr].channel_group.flags = v4c.FLAG_CG_BUS_EVENT + + out = mdf.extract_bus_logging({"CAN": [(db, 0)]}, ignore_value2text_conversion=True) + + exp_values, _exp_times = oracle(cont, frames, t) + self.assertEqual(len(out.groups), len({p for (p, _s) in exp_values})) + + for (pdu_name, sig_name), exp in exp_values.items(): + sig = out.get(sig_name, raw=True) + self.assertTrue(np.array_equal(sig.samples, np.array(exp)), f"{pdu_name}.{sig_name} mismatch") if __name__ == "__main__": From 767e6b7ae86ddc0db9d7124e267403c5fea2f322 Mon Sep 17 00:00:00 2001 From: kipp-ing Date: Thu, 9 Jul 2026 19:55:42 +0200 Subject: [PATCH 3/5] doc: developer note on AUTOSAR PDU-container extraction --- doc/pdu_container_extraction.md | 101 ++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 doc/pdu_container_extraction.md diff --git a/doc/pdu_container_extraction.md b/doc/pdu_container_extraction.md new file mode 100644 index 000000000..4b37f6479 --- /dev/null +++ b/doc/pdu_container_extraction.md @@ -0,0 +1,101 @@ +# AUTOSAR PDU-container extraction (CAN bus logging) + +Developer note for the container-PDU support in +`src/asammdf/blocks/bus_logging_utils.py` (`extract_pdus`) wired into +`MDF._extract_can_logging` (`mdf_v4.py`). + +## What + +AUTOSAR **container I-PDUs** pack several contained PDUs into one CAN(-FD) +frame. `extract_mux` only understood `is_multiplexed` frames, so container +frames (`message.is_pdu_container`) were routed through it and mis-decoded. +`extract_pdus` decodes them and gives **each contained PDU its own channel +group**, reusing the existing channel-group machinery unchanged. + +Two container layouts are supported: + +- **Dynamic** — each contained PDU is prefixed by a per-PDU header + (`Header_ID` + `Header_DLC`). A PDU's byte offset depends on the lengths of + the PDUs before it, so the frame is walked header-by-header. +- **Static** — no per-PDU header; contained PDUs sit at fixed byte offsets. + +## How + +`_extract_can_logging` routes `message.is_pdu_container` frames to +`extract_pdus`; everything else keeps using `extract_mux`. `extract_pdus` +branches on whether the frame exposes the `Header_ID`/`Header_DLC` synthetic +signals that canmatrix injects. + +### Dynamic containers + +1. Header geometry is derived from the header signals (short header = 24 + 8 + bits, long header = 32 + 32); it is **not** hardcoded. +2. Headers are walked across all frames one "slot" per iteration. Frames + diverge in offset after the first PDU, so a per-frame byte offset is + advanced independently and each frame's current header window is gathered + vectorized. Unknown/padding ids advance by their DLC (matching canmatrix), + which guarantees termination. +3. Per contained PDU, its payload slices are gathered into a 2-D array and its + signals extracted. A contained PDU's signal `start_bit` values are + **PDU-payload-relative**, so once the byte-aligned PDU payload slice is + isolated the regular `extract_signal` machinery applies unchanged. + +### Static containers + +canmatrix cannot decode static containers at runtime — `Frame.unpack` raises +`DecodingConatainerPdu` on them. But its ARXML parser has already rebased each +contained PDU's signal `start_bit` values to be **frame-relative** (via the +PDU's `OFFSET`). So every contained PDU decodes straight from the full frame +payload; each still becomes its own channel group. Static PDUs carry no header +id (`pdu.id is None`), so the channel-group identity keys on the PDU name. + +### Shared emission + +Both paths call `_emit_pdu_signals(...)`, which extracts a single PDU's signals +into a dedicated channel-group entry (the PDU identity is carried in the entry's +`muxer` slot via `_contained_pdu_muxer`). The only difference is the payload it +is handed: the sliced PDU payload (dynamic) or the full frame (static). + +### CAN-FD note + +Container frames ride on CAN-FD in practice (they need > 8 bytes; canmatrix +marks every container frame `is_fd = True`). Extraction itself is purely +`CAN_DataFrame.DataBytes`-width driven — `_extract_can_logging` does not read +`EDL`/`DataLength`/`DLC` — so the FD flag has no functional effect on decoding; +the wide `DataBytes` array is all that matters. + +## Not handled + +- **PDU-internal multiplexing** — a multiplexed contained PDU is not modelled by + canmatrix (it only reads `I-SIGNAL-TO-I-PDU-MAPPING`, no `DYNAMIC/STATIC-PART` + under a contained PDU), so it is out of reach for this metadata-only approach. +- **LIN container frames** — container I-PDUs are a CAN-FD/FlexRay/Ethernet + mechanism; bus logging support here is scoped to CAN. + +## Testing + +Fully offline in `test/test_CAN_pdu_extraction.py`; containers are built +in-memory with canmatrix. + +- `test_extract_pdus_matches_canmatrix_unpack` — dynamic containers vs the + `canmatrix.Frame.unpack` oracle, across big/little-endian headers and + 0x00/0xFF padding, with unique multi-PDU frames including bit-packed, + non-byte-aligned **signed** signals. +- `test_extract_pdus_static_container` — static containers vs an equivalent + **flat** canmatrix frame carrying the same frame-relative signals (needed + because `Frame.unpack` refuses static containers); covers byte-aligned LE/BE + and a non-byte-aligned signed field, and asserts one channel group per PDU. +- `test_extract_bus_logging_container_e2e` — full `MDF.extract_bus_logging` + pipeline on a 32-byte container. +- `test_extract_bus_logging_canfd_container_e2e` — full pipeline on genuine + CAN-FD frames: 64-byte payload with the `EDL` flag and `DataLength` members + set. + +Run them with: + +```bash +python -m unittest test.test_CAN_pdu_extraction -v +``` + +The existing `test/test_CAN_bus_logging.py` (real OBD/J1939 data, downloaded) +continues to pass, confirming no regression to the `extract_mux` path. From 595cc5f59c9326b3c9d33414a7f3ad41785dbf75 Mon Sep 17 00:00:00 2001 From: kipp-ing Date: Wed, 29 Jul 2026 21:00:59 +0200 Subject: [PATCH 4/5] fix container extraction against real OEM measurements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validated `extract_pdus` against two real OEM CAN-FD bus logs (5.1 M and 2.3 M CAN frames) and three production ARXML databases, using `canmatrix.Frame.unpack` as the oracle. That uncovered three defects that the in-memory tests could not reach: 1. `OverflowError` aborted the whole extraction. Real containers carry opaque 216/288/400-bit blobs declared *signed*; those are kept as a byte matrix, so `as_non_byte_sized_signed_int` computed `1 << 216` on them. Signals wider than an integer dtype now skip two's complement, like unsigned ones already did. 2. Container padding was reported as measured data. A sender may transmit a contained PDU shorter than its declared size — the header DLC is the authority. Signals reaching past the transmitted length are now flagged through `invalidation_bits` instead of surfacing padding as a value. Real data hits this on 2.4 % of contained-PDU occurrences. 3. The header walk could run backwards. canmatrix's ARXML parser marks the synthetic `Header_ID`/`Header_DLC` signals *signed*, so a 0xFF padding byte decodes as a DLC of -1; the offset then moved backwards over a padded tail, rescanning the frame misaligned and inventing contained PDUs out of padding. Both header fields are now read unsigned. Two regression tests added for 1. and 2.; 3. is covered by the existing oracle test now that absolute values are asserted. --- doc/pdu_container_extraction.md | 45 ++++++++++- src/asammdf/blocks/bus_logging_utils.py | 77 +++++++++++++++++- test/test_CAN_pdu_extraction.py | 102 ++++++++++++++++++++++++ 3 files changed, 219 insertions(+), 5 deletions(-) diff --git a/doc/pdu_container_extraction.md b/doc/pdu_container_extraction.md index 4b37f6479..4aa27efb8 100644 --- a/doc/pdu_container_extraction.md +++ b/doc/pdu_container_extraction.md @@ -33,12 +33,23 @@ signals that canmatrix injects. 2. Headers are walked across all frames one "slot" per iteration. Frames diverge in offset after the first PDU, so a per-frame byte offset is advanced independently and each frame's current header window is gathered - vectorized. Unknown/padding ids advance by their DLC (matching canmatrix), - which guarantees termination. + vectorized. Unknown/padding ids advance by their DLC, which guarantees + termination. + `Header_ID`/`Header_DLC` are read **unsigned** even when the database marks + them signed — canmatrix's ARXML parser does mark them signed, which makes a + 0xFF padding byte decode as a DLC of −1 and walks the offset *backwards* + over a padded tail, rescanning the frame misaligned and inventing contained + PDUs out of padding. 3. Per contained PDU, its payload slices are gathered into a 2-D array and its signals extracted. A contained PDU's signal `start_bit` values are **PDU-payload-relative**, so once the byte-aligned PDU payload slice is isolated the regular `extract_signal` machinery applies unchanged. +4. A sender may transmit a contained PDU **shorter than its declared size** (the + header DLC is the authority). The gather is fixed-width, so signals reaching + past the transmitted length are decoded from the next PDU or from container + padding — those samples are flagged through `invalidation_bits` rather than + reported as measured data. Real OEM data hits this on ~2.4 % of + contained-PDU occurrences. ### Static containers @@ -90,6 +101,13 @@ in-memory with canmatrix. - `test_extract_bus_logging_canfd_container_e2e` — full pipeline on genuine CAN-FD frames: 64-byte payload with the `EDL` flag and `DataLength` members set. +- `test_extract_pdus_wide_signed_signal` — a contained-PDU signal wider than 64 + bits and declared **signed** (real OEM containers carry 216/288/400-bit + signed blobs) comes back as raw bytes instead of raising `OverflowError` in + `as_non_byte_sized_signed_int`. +- `test_extract_pdus_short_header_dlc_invalidates` — header DLC shorter than the + declared PDU size: transmitted signals decode normally, signals past the + transmitted length are flagged invalid. Run them with: @@ -99,3 +117,26 @@ python -m unittest test.test_CAN_pdu_extraction -v The existing `test/test_CAN_bus_logging.py` (real OBD/J1939 data, downloaded) continues to pass, confirming no regression to the `extract_mux` path. + +## Validation on real measurements + +Validated against two real OEM CAN-FD bus logs (5.1 M and 2.3 M CAN frames) and +three production ARXML databases covering three CAN channels, with +`canmatrix.Frame.unpack` as the oracle: + +- **1 237 247** decoded signal values across **15** real container messages, + **0** mismatches. +- Full `extract_bus_logging`: master yields only `Header_ID`/`Header_DLC` for + container messages; this branch yields 789 additional real signals on the + first measurement (58 container-derived channel groups) and 203 + container-derived groups on the second. +- All 541 non-container signals are bit-identical to master, and 676 008 + non-container values were separately confirmed against the oracle. +- Decoded values are physically plausible: HV DC-link 400 V mean / 794 V peak on + an 800 V platform, inverter and coolant temperatures 25–33 °C, 14.5 V rail. +- The same drive was also recorded **signal-based** (decoded on the fly by the + logger toolchain). Cross-checking our container decode against that recording + — an oracle sharing no code with asammdf or canmatrix — 298 of 303 comparable + signals agree on every one of 114 251 samples; the remaining five are + free-running sequence counters / CRCs whose sample instants differ between the + two recordings by more than the comparison window. diff --git a/src/asammdf/blocks/bus_logging_utils.py b/src/asammdf/blocks/bus_logging_utils.py index dd991f13f..bd308e3e7 100644 --- a/src/asammdf/blocks/bus_logging_utils.py +++ b/src/asammdf/blocks/bus_logging_utils.py @@ -225,7 +225,13 @@ def extract_signal( vals = vals >> bit_offset vals &= (2**bit_count) - 1 - if signed and not is_float: + # ``std_size > 8`` means the signal does not fit any integer dtype, so it was + # kept as a raw byte matrix (``({std_size},)u1``) above; two's complement does + # not apply to it. AUTOSAR container PDUs do carry such signals (opaque blobs + # of 216/288/400 bits declared signed), and feeding one to + # ``as_non_byte_sized_signed_int`` raises OverflowError on the ``1 << bit_count`` + # mask. Leave those as bytes, exactly like their unsigned counterparts. + if signed and not is_float and std_size <= 8: # A plain ``view("i{std_size}")`` only sign-extends correctly when the # value fills exactly ``std_size`` byte-aligned bytes. A non-byte-aligned # signal (``bit_offset``) has been shifted/masked into a wider unsigned @@ -471,6 +477,34 @@ def extract_mux( PDU_CONTAINER_HEADER_DLC: Final = "Header_DLC" +def _signal_byte_extent(signal: Signal) -> int: + """Number of payload bytes a signal needs, i.e. the 1-based index of the last + byte it touches. Mirrors the addressing :func:`extract_signal` uses. + """ + start_bit = signal.get_startbit(bit_numbering=1) + bit_count = signal.size + + if signal.is_little_endian: + start_byte, bit_offset = divmod(start_bit, 8) + byte_size, r = divmod(bit_offset + bit_count, 8) + if r: + byte_size += 1 + return start_byte + byte_size + + byte_pos = start_bit // 8 + 1 + start_pos = start_bit + bits = bit_count + while True: + pos = start_pos % 8 + 1 + if pos < bits: + byte_pos += 1 + bits -= pos + start_pos = 7 + else: + break + return byte_pos + + def _contained_pdu_muxer(pdu: Any) -> str: """Stable per-PDU identity for the entry ``muxer`` slot so every contained PDU maps to its own channel group. Static (header-less) containers carry no @@ -498,11 +532,19 @@ def _emit_pdu_signals( include_message_name: bool, ignore_value2text_conversion: bool, is_j1939: bool, + transmitted_bytes: NDArray[Any] | None = None, ) -> None: """Extract one contained PDU's signals from its payload slice into a dedicated channel-group entry. Shared by the dynamic and static paths of :func:`extract_pdus`; ``pdu_payload`` is the PDU-relative payload for a dynamic container and the full frame payload for a static one. + + ``transmitted_bytes`` is the per-occurrence ``Header_DLC``. A sender may + transmit a contained PDU shorter than its declared size, in which case the + bytes past the header DLC belong to the *next* contained PDU or to the + container padding. Signals reaching into that region are decoded (the + extraction is vectorized over a fixed-width slice) but flagged invalid, so + padding never surfaces as a measured value. """ entry = ( bus, @@ -533,6 +575,15 @@ def _emit_pdu_signals( else: sig_name = sig.name + # Samples whose bytes were not actually transmitted (header DLC shorter + # than the declared PDU size) are read from the neighbouring PDU or from + # the container padding, so mark them invalid. + invalidation_bits: NDArray[np.bool] | None = None + if transmitted_bytes is not None: + not_transmitted = transmitted_bytes < _signal_byte_extent(sig) + if not_transmitted.any(): + invalidation_bits = not_transmitted + try: scale_ranges = getattr(sig, "scale_ranges", None) if scale_ranges: @@ -547,11 +598,14 @@ def _emit_pdu_signals( "samples": samples if raw else apply_conversion(samples, sig, ignore_value2text_conversion), "conversion": get_conversion(sig) if raw else None, "t": t_, - "invalidation_bits": None, + "invalidation_bits": invalidation_bits, } if is_j1939: - signals[sig_name]["invalidation_bits"] = samples > MAX_VALID_J1939[defined_j1939_bit_count(sig)] + j1939_invalid = samples > MAX_VALID_J1939[defined_j1939_bit_count(sig)] + if invalidation_bits is not None: + j1939_invalid = j1939_invalid | invalidation_bits + signals[sig_name]["invalidation_bits"] = j1939_invalid except: print(format_exc()) @@ -694,6 +748,7 @@ def _pdu_size(pdu: Any) -> int: rows_all: list[NDArray[Any]] = [] ids_all: list[NDArray[Any]] = [] starts_all: list[NDArray[Any]] = [] + dlcs_all: list[NDArray[Any]] = [] max_slots = frame_bytes // header_size + 1 for _ in range(max_slots): @@ -708,11 +763,23 @@ def _pdu_size(pdu: Any) -> int: ids = extract_signal(header_id_signal, windows, raw=True).astype(" int: rows_flat = np.concatenate(rows_all) ids_flat = np.concatenate(ids_all) starts_flat = np.concatenate(starts_all) + dlcs_flat = np.concatenate(dlcs_all) # --- Per contained PDU: gather its payload slices and extract its signals. for pdu in message.pdus: @@ -738,11 +806,13 @@ def _pdu_size(pdu: Any) -> int: sel_rows = rows_flat[mask] sel_starts = starts_flat[mask] + sel_dlcs = dlcs_flat[mask] # Keep timestamp order (a frame may contain several PDUs / repeats). order = np.argsort(sel_rows, kind="stable") sel_rows = sel_rows[order] sel_starts = sel_starts[order] + sel_dlcs = sel_dlcs[order] gather_cols = sel_starts[:, None] + np.arange(pdu_size)[None, :] pdu_payload = padded[sel_rows[:, None], gather_cols] @@ -762,6 +832,7 @@ def _pdu_size(pdu: Any) -> int: include_message_name=include_message_name, ignore_value2text_conversion=ignore_value2text_conversion, is_j1939=is_j1939, + transmitted_bytes=sel_dlcs, ) return extracted_signals diff --git a/test/test_CAN_pdu_extraction.py b/test/test_CAN_pdu_extraction.py index d13b7872b..df27fba75 100644 --- a/test/test_CAN_pdu_extraction.py +++ b/test/test_CAN_pdu_extraction.py @@ -118,6 +118,38 @@ def build_frames( return frames +def build_wide_signal_container(frame_bytes: int = FD_FRAME_BYTES) -> canmatrix.Frame: + """Container whose contained PDU carries an opaque blob wider than 64 bits and + declared *signed*, the shape real AUTOSAR databases use for key/ID payloads. + """ + cont = canmatrix.Frame(name="WideCont", size=frame_bytes) + cont.arbitration_id = canmatrix.ArbitrationId(id=CONTAINER_ID, extended=False) + cont.add_signal(canmatrix.Signal(name="Header_ID", start_bit=0, size=24, is_little_endian=False)) + cont.add_signal(canmatrix.Signal(name="Header_DLC", start_bit=24, size=8, is_little_endian=False)) + + pdu = canmatrix.Pdu(name="WidePdu", id=0x123, size=28) + pdu.add_signal(canmatrix.Signal(name="W_blob", start_bit=0, size=216, is_little_endian=True, is_signed=True)) + pdu.add_signal(canmatrix.Signal(name="W_u8", start_bit=216, size=8, is_little_endian=True, is_signed=False)) + cont.add_pdu(pdu) + return cont + + +def build_short_dlc_container(frame_bytes: int = FRAME_BYTES) -> canmatrix.Frame: + """Container whose PDU is transmitted shorter (header DLC) than declared.""" + cont = canmatrix.Frame(name="ShortCont", size=frame_bytes) + cont.arbitration_id = canmatrix.ArbitrationId(id=CONTAINER_ID, extended=False) + cont.add_signal(canmatrix.Signal(name="Header_ID", start_bit=0, size=24, is_little_endian=False)) + cont.add_signal(canmatrix.Signal(name="Header_DLC", start_bit=24, size=8, is_little_endian=False)) + + pdu = canmatrix.Pdu(name="ShortPdu", id=0x321, size=6) + pdu.add_signal(canmatrix.Signal(name="S_early", start_bit=0, size=16, is_little_endian=True, is_signed=False)) + pdu.add_signal(canmatrix.Signal(name="S_mid", start_bit=16, size=16, is_little_endian=True, is_signed=False)) + # lives in bytes 4..5, i.e. past a 4-byte transmitted length + pdu.add_signal(canmatrix.Signal(name="S_late", start_bit=32, size=16, is_little_endian=True, is_signed=False)) + cont.add_pdu(pdu) + return cont + + def oracle(cont: canmatrix.Frame, frames: list[bytes], t: np.ndarray): """Ground truth via canmatrix.Frame.unpack, accumulated in frame order.""" values: dict[tuple[str, str], list] = {} @@ -159,6 +191,76 @@ def test_extract_pdus_matches_canmatrix_unpack(self) -> None: np.array_equal(got[pdu_name][sig_name]["t"], np.array(exp_times[pdu_name], dtype="f8")) ) + def test_extract_pdus_wide_signed_signal(self) -> None: + """A contained PDU signal wider than 64 bits and declared signed must not + blow up: it is kept as raw bytes, exactly like an unsigned one would be. + Real OEM containers carry 216/288/400-bit signed blobs, and + two's complement cannot be applied to a byte-matrix sample.""" + cont = build_wide_signal_container() + pdu = cont.pdus[0] + + rng = np.random.default_rng(11) + blobs = rng.integers(0, 256, size=(50, 27), dtype="u1") + frames = [] + for row in blobs: + frames.append( + (_encode_header(pdu.id, pdu.size, True) + bytes(row) + bytes([0xA5])).ljust(FD_FRAME_BYTES, b"\x00") + ) + payload = np.array([list(f) for f in frames], dtype="u1") + t = np.arange(len(frames), dtype="f8") + + extracted = extract_pdus(payload, cont, message_id=CONTAINER_ID, bus=1, t=t, raw=True) + + got = {entry[4].split(":")[-1]: sigs for entry, sigs in extracted.items()} + self.assertEqual(set(got), {pdu.name}) + blob = got[pdu.name]["W_blob"]["samples"] + self.assertEqual(len(blob), len(frames)) + # the 216-bit blob comes back as its raw bytes, low byte first + for i, row in enumerate(blobs): + self.assertEqual(bytes(np.asarray(blob[i]).tobytes()[:27]), bytes(row)) + self.assertTrue(np.array_equal(got[pdu.name]["W_u8"]["samples"], np.full(len(frames), 0xA5))) + + def test_extract_pdus_short_header_dlc_invalidates(self) -> None: + """When the header DLC is shorter than the declared PDU size the trailing + bytes belong to the next PDU / container padding, so signals reaching into + them are flagged invalid instead of reporting padding as measured data.""" + cont = build_short_dlc_container() + pdu = cont.pdus[0] + sent = 4 # only 4 of the declared 6 bytes are transmitted + + frames = [] + for i in range(30): + body = bytes([i, 0x00, i + 1, 0x00]) + frames.append((_encode_header(pdu.id, sent, True) + body).ljust(FRAME_BYTES, b"\xee")) + payload = np.array([list(f) for f in frames], dtype="u1") + t = np.arange(len(frames), dtype="f8") + + extracted = extract_pdus(payload, cont, message_id=CONTAINER_ID, bus=1, t=t, raw=True) + sigs = next(iter(extracted.values())) + + # transmitted signals: real values, no invalidation + self.assertIsNone(sigs["S_early"]["invalidation_bits"]) + self.assertIsNone(sigs["S_mid"]["invalidation_bits"]) + self.assertTrue(np.array_equal(sigs["S_early"]["samples"], np.arange(30))) + self.assertTrue(np.array_equal(sigs["S_mid"]["samples"], np.arange(1, 31))) + + # the signal past the transmitted length is fully invalidated + invalid = sigs["S_late"]["invalidation_bits"] + self.assertIsNotNone(invalid) + self.assertTrue(invalid.all()) + + # a full-length transmission keeps everything valid + full = [] + for i in range(30): + full.append( + (_encode_header(pdu.id, pdu.size, True) + bytes([i, 0, i + 1, 0, i + 2, 0])).ljust(FRAME_BYTES, b"\xee") + ) + payload = np.array([list(f) for f in full], dtype="u1") + extracted = extract_pdus(payload, cont, message_id=CONTAINER_ID, bus=1, t=t, raw=True) + sigs = next(iter(extracted.values())) + self.assertIsNone(sigs["S_late"]["invalidation_bits"]) + self.assertTrue(np.array_equal(sigs["S_late"]["samples"], np.arange(2, 32))) + def test_extract_pdus_static_container(self) -> None: """A static (header-less) container: every contained PDU is present in every frame and decodes at its fixed frame offset, one channel group From 21909078dd39640d32872037fe2e5cc7252d889e Mon Sep 17 00:00:00 2001 From: kipp-ing Date: Wed, 29 Jul 2026 21:39:46 +0200 Subject: [PATCH 5/5] doc: document container I-PDU support in the bus logging page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `doc/buslogging.rst` is the user-facing bus logging documentation and said nothing about container I-PDUs. Add a section covering what gets decoded (one channel group per contained PDU, addressed like any other bus logging signal), that both the dynamic and static layouts are handled, that samples whose bytes were not transmitted are marked through `invalidation_bits`, and the two known limitations (multiplexed contained PDUs, LIN). Also point the developer note at it, and note there that it is deliberately not part of the built docs — sphinx is configured for `.rst` only. --- doc/buslogging.rst | 39 ++++++++++++++++++++++++++++++--- doc/pdu_container_extraction.md | 5 +++++ 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/doc/buslogging.rst b/doc/buslogging.rst index f44153f37..31c1f15a8 100644 --- a/doc/buslogging.rst +++ b/doc/buslogging.rst @@ -41,6 +41,39 @@ There multiple ways to address this channel in this situation: #. CAN bus ID, ASAM conformant message ID and short signal name, delimited by dot .. code:: python - - mdf.get('CAN1.CAN_DataFrame_123.EngineStatus') - + + mdf.get('CAN1.CAN_DataFrame_123.EngineStatus') + + +Container I-PDUs +================ + +AUTOSAR **container I-PDUs** pack several smaller PDUs into a single CAN(-FD) +frame. When the database describes a frame as a PDU container, each contained +PDU is decoded and gets **its own channel group**, so its signals are addressed +exactly like any other bus logging signal: + +.. code:: python + + mdf.get('EngineStatus') # short signal name + mdf.get('CAN1.VehicleStatus.EngineStatus') # container frame name + +Both container layouts are supported: *dynamic* containers, where every +contained PDU is prefixed by a header (``Header_ID`` + ``Header_DLC``) and its +offset therefore changes from frame to frame, and *static* containers, where the +contained PDUs sit at fixed offsets. + +A sender may transmit a contained PDU shorter than the length declared in the +database — the header DLC is the authority. Signals that reach past the +transmitted length would otherwise be decoded out of the neighbouring PDU or out +of the container padding, so those samples are marked invalid: + +.. code:: python + + sig = mdf.get('SomeSignal') + sig.invalidation_bits # None, or True where the bytes were not transmitted + +Two limitations are worth noting: a multiplexed *contained* PDU is not modelled +by **canmatrix**, so it cannot be decoded, and container I-PDUs are only handled +for CAN — not for LIN. + diff --git a/doc/pdu_container_extraction.md b/doc/pdu_container_extraction.md index 4aa27efb8..4dfc810c0 100644 --- a/doc/pdu_container_extraction.md +++ b/doc/pdu_container_extraction.md @@ -4,6 +4,11 @@ Developer note for the container-PDU support in `src/asammdf/blocks/bus_logging_utils.py` (`extract_pdus`) wired into `MDF._extract_can_logging` (`mdf_v4.py`). +The user-facing documentation is the "Container I-PDUs" section of +`doc/buslogging.rst`. This note is not part of the built docs (sphinx is +configured for `.rst` only) — it records *why* the decoder looks the way it +does, for whoever touches it next. + ## What AUTOSAR **container I-PDUs** pack several contained PDUs into one CAN(-FD)