fix(j1939): decode by source address (#13), split shared messages, stop 11-bit ID cross-matching

core/dbc_decoder.py

@@ -8,7 +8,7 @@
 
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, Hashable, Iterable
 import inspect
 import xml.etree.ElementTree as ET
 
@@ -233,19 +233,49 @@ def load_database_file(
         return db, load_messages
 
 
+def source_address_message_name(message_name: str, source_address: int) -> str:
+    """Name the series of one J1939 sender of a message that several send."""
+    return f"{message_name} [SA 0x{source_address:02X}]"
+
+
+def multi_sender_messages(matches: Iterable[tuple[Hashable, int]]) -> set[Hashable]:
+    """
+    Return the keys matched by extended frames from more than one source address.
+
+    *matches* pairs a key identifying a database message with the ID of a
+    frame that decoded into it. A J1939 message matched through its PGN
+    placeholder can be sent by several ECUs; their series are kept apart
+    instead of interleaving, e.g. one ECU reporting a switch and another
+    reporting it as not available.
+    """
+    sources: dict[Hashable, set[int]] = {}
+    for key, frame_id in matches:
+        if frame_id > 0x7FF:
+            sources.setdefault(key, set()).add(frame_id & 0xFF)
+    return {key for key, addresses in sources.items() if len(addresses) > 1}
+
+
 class DBCDecoder:
     def __init__(self, dbc_path: str | Path) -> None:
         self.dbc_path = Path(dbc_path)
         self.database, self.load_messages = self._load_database(self.dbc_path)
         self._decode_signature = None
         self._decode_kwargs_cache: dict[str, Any] | None = None   # perf: built once
 
-        # Primary lookup: arbitration_id → [message, ...]
-        self._messages_exact: dict[int, list[Any]] = {}
+        # Primary lookup: (is_extended, arbitration_id) → [message, ...]. A
+        # standard and an extended frame never match each other's messages.
+        self._messages_exact: dict[tuple[bool, int], list[Any]] = {}
         self._messages_pgn:   dict[int, list[Any]] = {}
+        # (PGN, source address) → [message, ...]. Used for the PGNs in
+        # _pgn_sa_specific, which the database defines at more than one SA:
+        # their frames only decode into the message for their own SA, and an
+        # SA the database does not define stays undecoded instead of
+        # borrowing another node's message.
+        self._messages_pgn_sa: dict[tuple[int, int], list[Any]] = {}
+        self._pgn_sa_specific: set[int] = set()
 
         # Perf: per-frame candidate cache (same ID seen repeatedly → reuse result)
-        self._candidate_cache: dict[int, list[Any]] = {}
+        self._candidate_cache: dict[tuple[int, bool], list[Any]] = {}
 
         # Perf: per-signal choices dict cached at build time (avoids getattr per sample)
         # key = (message_name, signal_name) → {int_key: label_str}
@@ -281,22 +311,30 @@ def _build_indexes(self) -> None:
                 self._dbc_message_ids_preview.append(
                     f"{message.name} | {frame_id_text} | len={getattr(message, 'length', '?')}"
                 )
-            # Register under all masked variants (exact, 29-bit, 11-bit)
-            for fid in (frame_id, frame_id & 0x1FFFFFFF, frame_id & 0x7FF):
+            is_extended = bool(getattr(message, "is_extended_frame", False)) or frame_id > 0x7FF
+            # Register under the exact ID and the ID without the extended flag bit
+            for fid in (frame_id, frame_id & 0x1FFFFFFF):
                 if fid >= 0:
-                    self._messages_exact.setdefault(fid, []).append(message)
+                    self._messages_exact.setdefault((is_extended, fid), []).append(message)
             # J1939 PGN index
-            is_extended = bool(getattr(message, "is_extended_frame", False)) or frame_id > 0x7FF
             if is_extended:
                 pgn = self._extract_j1939_pgn(frame_id)
                 if pgn is not None:
                     self._messages_pgn.setdefault(pgn, []).append(message)
+                    self._messages_pgn_sa.setdefault(
+                        (pgn, frame_id & 0xFF), []
+                    ).append(message)
 
             # Perf: pre-cache signal choices so decode_frame avoids getattr per sample
             for signal in getattr(message, "signals", []):
                 choices = getattr(signal, "choices", None) or {}
                 self._choices_cache[(message.name, signal.name)] = dict(choices)
 
+        self._pgn_sa_specific = {
+            pgn for pgn, messages in self._messages_pgn.items()
+            if len({int(m.frame_id) & 0xFF for m in messages}) > 1
+        }
+
     # ── Decode kwargs — built once, reused every frame ────────────────────
 
     def _get_decode_kwargs(self) -> dict[str, Any]:
@@ -329,16 +367,17 @@ def _extract_j1939_pgn(frame_id: int) -> int | None:
         ps = (can_id >> 8) & 0xFF
         return (pf << 8) if pf < 240 else ((pf << 8) | ps)
 
-    def _get_candidates(self, frame: RawFrame) -> list[Any]:
+    def candidates_for(self, arb_id: int, is_extended: bool) -> list[Any]:
         """
-        Return message candidates for this frame's arbitration_id.
-        Result is cached after first lookup — same ID seen in every periodic frame.
+        Return the messages that may decode ``arb_id``, best match first.
+
+        Exact ID matches come first, among messages of the frame's own type:
+        a standard frame only matches standard messages and an extended frame
+        only extended ones. Extended frames then fall back to J1939 PGN
+        matching: a PGN the database defines at a single source address
+        matches any SA (the DBC's SA is a placeholder), while a PGN defined at
+        several SAs only matches the message for the frame's SA.
         """
-        arb_id = frame.arbitration_id
-        cached = self._candidate_cache.get(arb_id)
-        if cached is not None:
-            return cached
-
         seen: set[tuple[str, int]] = set()
         candidates: list[Any] = []
 
@@ -348,19 +387,38 @@ def add(msg: Any) -> None:
                 seen.add(key)
                 candidates.append(msg)
 
-        # Exact + masked lookups
-        for lookup_id in (arb_id, arb_id & 0x1FFFFFFF, arb_id & 0x7FF):
-            for msg in self._messages_exact.get(lookup_id, []):
+        extended = is_extended or arb_id > 0x7FF
+
+        # Exact lookups, with and without the extended flag bit
+        for lookup_id in (arb_id, arb_id & 0x1FFFFFFF):
+            for msg in self._messages_exact.get((extended, lookup_id), []):
                 add(msg)
 
         # J1939 PGN fallback
-        if frame.is_extended_id or arb_id > 0x7FF:
+        if extended:
             pgn = self._extract_j1939_pgn(arb_id)
             if pgn is not None:
-                for msg in self._messages_pgn.get(pgn, []):
+                if pgn in self._pgn_sa_specific:
+                    matches = self._messages_pgn_sa.get((pgn, arb_id & 0xFF), [])
+                else:
+                    matches = self._messages_pgn.get(pgn, [])
+                for msg in matches:
                     add(msg)
 
-        self._candidate_cache[arb_id] = candidates
+        return candidates
+
+    def _get_candidates(self, frame: RawFrame) -> list[Any]:
+        """
+        Return message candidates for this frame's arbitration_id.
+        Result is cached after first lookup — same ID seen in every periodic frame.
+        """
+        cache_key = (frame.arbitration_id, bool(frame.is_extended_id))
+        cached = self._candidate_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
+        candidates = self.candidates_for(*cache_key)
+        self._candidate_cache[cache_key] = candidates
         return candidates
 
     # ── Frame decode ──────────────────────────────────────────────────────

core/debug_inspector.py

@@ -1408,23 +1408,7 @@ def inspect_measurement(
 
 
 def _database_candidates(decoder, frame_id: int) -> list[object]:
-    candidates: list[object] = []
-    seen: set[int] = set()
-    for lookup in (frame_id, frame_id & 0x1FFFFFFF, frame_id & 0x7FF):
-        for message in decoder._messages_exact.get(lookup, []):
-            marker = id(message)
-            if marker not in seen:
-                seen.add(marker)
-                candidates.append(message)
-    if frame_id > 0x7FF:
-        pgn = decoder._extract_j1939_pgn(frame_id)
-        if pgn is not None:
-            for message in decoder._messages_pgn.get(pgn, []):
-                marker = id(message)
-                if marker not in seen:
-                    seen.add(marker)
-                    candidates.append(message)
-    return candidates
+    return decoder.candidates_for(frame_id, frame_id > 0x7FF)
 
 
 def _inspect_ldf(

core/load_worker.py

@@ -22,6 +22,7 @@
     store_key_prefix,
 )
 from core.channel_config import ChannelConfig
+from core.dbc_decoder import multi_sender_messages, source_address_message_name
 from core.signal_store import as_channel_key
 
 # ── Streaming constants ───────────────────────────────────────────────────
@@ -434,11 +435,12 @@ def flush_batch() -> None:
             f"Bulk decoding {n:,} frames across {total_groups:,} CAN ID groups..."
         )
 
+        # Resolve every group's message before decoding any: whether a
+        # message gets one series per J1939 source address depends on all of
+        # the frame IDs that matched it.
+        group_matches: list[tuple | None] = []
         for g in range(total_groups):
-            start, end = int(boundaries[g]), int(boundaries[g + 1])
-            group_idx  = sort_idx[start:end]
-
-            first    = group_idx[0]
+            first    = sort_idx[int(boundaries[g])]
             ch_byte  = int(channels_np[first])
             arb_id   = int(arb_ids_np[first])
             bus      = BusType.LIN if is_lin_np[first] else BusType.CAN
@@ -450,6 +452,7 @@ def flush_batch() -> None:
                 or _decoder_map.get((bus, ALL_CHANNELS_NUMBER))
             )
             if decoder is None:
+                group_matches.append(None)
                 continue
 
             vec = vec_dbcs.get(id(decoder))
@@ -458,13 +461,31 @@ def flush_batch() -> None:
                 vec_dbcs[id(decoder)] = vec
 
             candidates = vec.get_candidates(arb_id, is_extended=(arb_id > 0x7FF))
-            if not candidates:
+            # Match existing single-decoder behaviour: pick the first candidate.
+            group_matches.append(
+                (ch_key, arb_id, decoder, vec, candidates[0]) if candidates else None
+            )
+
+        split_messages = multi_sender_messages(
+            ((match[0], id(match[4])), match[1])
+            for match in group_matches if match is not None
+        )
+
+        for g in range(total_groups):
+            start, end = int(boundaries[g]), int(boundaries[g + 1])
+            group_idx  = sort_idx[start:end]
+
+            match = group_matches[g]
+            if match is None:
                 continue
+            ch_key, arb_id, decoder, vec, message = match
 
-            # Match existing single-decoder behaviour: pick the first candidate.
-            message = candidates[0]
-            msg_name = message.name
-            msg_id   = int(getattr(message, 'frame_id', arb_id))
+            if (ch_key, id(message)) in split_messages:
+                msg_name = source_address_message_name(message.name, arb_id & 0xFF)
+                msg_id   = arb_id
+            else:
+                msg_name = message.name
+                msg_id   = int(getattr(message, 'frame_id', arb_id))
             msg_dec  = vec.get_message_decoder(message)
 
             # Frames shorter than the message they matched decode from the
@@ -601,6 +622,9 @@ def flush_batch() -> None:
         # unmatched because decoding has not happened yet).
         store.decoded_frames   = decoded_total
         store.unmatched_frames = n - decoded_total
+        # Groups are decoded one frame ID at a time, so a signal fed by
+        # several IDs holds one sweep per ID until it is reordered.
+        store.sort_merged_series()
         decode_elapsed = time.perf_counter() - decode_started
         n_sigs = len(store._series_by_key)
         hint = (
@@ -1158,6 +1182,8 @@ def on_metadata_ready(metadata_rows):
             store.decoded_frames = trace_decoded_frames
             store.unmatched_frames = trace_frames - trace_decoded_frames
 
+        # asammdf yields one group per frame ID; see sort_merged_series().
+        store.sort_merged_series()
         import_elapsed = time.perf_counter() - import_start
         self.progress.emit(
             f"Bulk import complete: {total:,} signals | samples: "
@@ -1254,6 +1280,7 @@ def on_metadata_ready(metadata_rows):
                     f"Loaded {ch_count:,} channels | samples: {store.total_samples:,}"
                 )
 
+        store.sort_merged_series()
         self.tree_update.emit(store.build_tree_payload())
         self.partial_ready.emit()
         if metadata_first: