feat: add J1939Message for subscriber callbacks with destination address

README.md

@@ -106,21 +106,11 @@ import j1939
 logging.getLogger('j1939').setLevel(logging.DEBUG)
 logging.getLogger('can').setLevel(logging.DEBUG)
 
-def on_message(priority, pgn, sa, timestamp, data):
-    """Receive incoming messages from the bus
-
-    :param int priority:
-        Priority of the message
-    :param int pgn:
-        Parameter Group Number of the message
-    :param int sa:
-        Source Address of the message
-    :param int timestamp:
-        Timestamp of the message
-    :param bytearray data:
-        Data of the PDU
-    """
-    print("PGN {} length {}".format(pgn, len(data)))
+def on_message(msg: j1939.J1939Message):
+    """Receive incoming messages from the bus"""
+    print("PGN {} from {:#04x} to {:#04x}, length {}".format(
+        msg.pgn, msg.source_address, msg.dest_address, len(msg.data)
+    ))
 
 def main():
     print("Initializing")
@@ -150,6 +140,11 @@ if __name__ == '__main__':
     main()
 ```
 
+The legacy 5-argument callback form,
+`on_message(priority, pgn, sa, timestamp, data)` (without the destination
+address), is also still supported and detected automatically from the
+callback's signature — existing subscribers don't need to change.
+
 A more sophisticated example in which the CA class was overloaded to
 include its own functionality:
 

examples/simple_receive_global.py

@@ -6,21 +6,9 @@
 logging.getLogger('j1939').setLevel(logging.DEBUG)
 logging.getLogger('can').setLevel(logging.DEBUG)
 
-def on_message(priority, pgn, sa, timestamp, data):
-    """Receive incoming messages from the bus
-
-    :param int priority:
-        Priority of the message
-    :param int pgn:
-        Parameter Group Number of the message
-    :param int sa:
-        Source Address of the message
-    :param int timestamp:
-        Timestamp of the message
-    :param bytearray data:
-        Data of the PDU
-    """
-    print(f"PGN {pgn} length {len(data)}")
+def on_message(msg: j1939.J1939Message):
+    """Receive incoming messages from the bus"""
+    print(f"PGN {msg.pgn} from {msg.source_address:#04x} length {len(msg.data)}")
 
 def main():
     print("Initializing")

examples/simple_receive_peer_to_peer.py

@@ -6,21 +6,9 @@
 logging.getLogger('j1939').setLevel(logging.DEBUG)
 logging.getLogger('can').setLevel(logging.DEBUG)
 
-def on_message(priority, pgn, sa, timestamp, data):
-    """Receive incoming messages from the bus
-
-    :param int priority:
-        Priority of the message
-    :param int pgn:
-        Parameter Group Number of the message
-    :param int sa:
-        Source Address of the message
-    :param int timestamp:
-        Timestamp of the message
-    :param bytearray data:
-        Data of the PDU
-    """
-    print(f"PGN {hex(pgn)} length {len(data)}")
+def on_message(msg: j1939.J1939Message):
+    """Receive incoming messages from the bus"""
+    print(f"PGN {hex(msg.pgn)} to {hex(msg.dest_address)} length {len(msg.data)}")
 
 def main():
     print("Initializing")

j1939/__init__.py

@@ -5,6 +5,7 @@
 from .electronic_control_unit import ElectronicControlUnit as ElectronicControlUnit
 from .error_info import *  # noqa: F403
 from .memory_access import *  # noqa: F403
+from .message import J1939Message as J1939Message
 from .message_id import MessageId as MessageId
 from .name import Name as Name
 from .parameter_group_number import ParameterGroupNumber as ParameterGroupNumber

j1939/electronic_control_unit.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import heapq
+import inspect
 import logging
 import queue
 import threading
@@ -13,6 +14,7 @@
 from .controller_application import ControllerApplication
 from .j1939_21 import J1939_21
 from .j1939_22 import J1939_22
+from .message import J1939Message
 from .message_id import FrameFormat
 from .parameter_group_number import ParameterGroupNumber
 
@@ -324,11 +326,44 @@ def disconnect(self):
             self._bus_created = False
         self._bus = None
 
+    @staticmethod
+    def _callback_takes_message(callback):
+        """Detect whether ``callback`` expects the new single-argument
+        ``J1939Message`` calling convention rather than the legacy
+        ``(priority, pgn, sa, timestamp, data)`` positional arguments.
+
+        A callback is treated as new-style if it accepts exactly one
+        positional parameter (besides ``self`` for bound methods), or if its
+        signature cannot be inspected (e.g. some C-implemented callables) —
+        in which case it falls back to the legacy calling convention.
+        """
+        try:
+            sig = inspect.signature(callback)
+        except (TypeError, ValueError):
+            return False
+        positional = [
+            p
+            for p in sig.parameters.values()
+            if p.kind
+            in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
+        ]
+        has_var_positional = any(
+            p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
+        )
+        if has_var_positional:
+            return False
+        return len(positional) == 1
+
     def subscribe(self, callback, device_address=None, owner=None):
         """Add the given callback to the message notification stream.
 
         :param callback:
-            Function to call when message is received.
+            Function to call when message is received. Either the legacy
+            5-argument form ``callback(priority, pgn, sa, timestamp, data)``
+            or the new single-argument form ``callback(msg: J1939Message)``,
+            which also carries the destination address. The calling
+            convention is detected once from the callback's signature at
+            subscribe time.
         :param int device_address:
             Device address of the application.
             This is a simple way for peer-to-peer reception without adding a controller-application.
@@ -345,7 +380,12 @@ def subscribe(self, callback, device_address=None, owner=None):
         """
         with self._subscribers_lock:
             self._subscribers.append(
-                {"cb": callback, "dev_adr": device_address, "owner": owner}
+                {
+                    "cb": callback,
+                    "dev_adr": device_address,
+                    "owner": owner,
+                    "takes_message": self._callback_takes_message(callback),
+                }
             )
 
     def unsubscribe(self, callback, owner=None):
@@ -728,7 +768,10 @@ def _notify_subscribers(self, priority, pgn, sa, dest, timestamp, data):
                 or (callable(dic["dev_adr"]) and dic["dev_adr"](dest))
                 or (dest == dic["dev_adr"])
             ):
-                dic["cb"](priority, pgn, sa, timestamp, data)
+                if dic["takes_message"]:
+                    dic["cb"](J1939Message(priority, pgn, sa, timestamp, data, dest))
+                else:
+                    dic["cb"](priority, pgn, sa, timestamp, data)
 
     def _is_message_acceptable(self, dest):
         # Ownership / active-participation check only: does a subscriber own this

j1939/message.py

@@ -0,0 +1,31 @@
+from typing import NamedTuple
+
+
+class J1939Message(NamedTuple):
+    """A received J1939 message, passed to single-argument subscriber callbacks.
+
+    Also unpacks positionally like the legacy 5-tuple
+    ``(priority, pgn, source_address, timestamp, data)`` for callbacks that
+    don't need ``dest_address``, since it's appended last.
+
+    :ivar int priority:
+        Priority of the message.
+    :ivar int pgn:
+        Parameter Group Number of the message.
+    :ivar int source_address:
+        Source address of the message.
+    :ivar int timestamp:
+        Timestamp of the CAN message.
+    :ivar bytearray data:
+        Data of the PDU.
+    :ivar int dest_address:
+        Destination address of the message. ``ParameterGroupNumber.Address.GLOBAL``
+        for broadcast (PDU2) messages.
+    """
+
+    priority: int
+    pgn: int
+    source_address: int
+    timestamp: int
+    data: bytearray
+    dest_address: int

test/test_ecu.py

@@ -438,6 +438,83 @@ def callback(priority: int, pgn: int, sa: int, timestamp: int, data: bytearray):
     assert call_count == 1
 
 
+def test_subscribe_new_style_receives_j1939_message(feeder):
+    """A single-argument callback receives a J1939Message with dest_address."""
+    received = []
+
+    def callback(msg: j1939.J1939Message):
+        received.append(msg)
+
+    feeder.ecu.subscribe(callback)
+
+    feeder.can_messages = [
+        (Feeder.MsgType.CANRX, 0x00FEB201, [1, 2, 3, 4, 5, 6, 7, 8], 0.0),
+    ]
+    feeder.pdus = [(Feeder.MsgType.PDU, 65202, [1, 2, 3, 4, 5, 6, 7, 8])]
+    feeder.receive()
+
+    assert len(received) == 1
+    msg = received[0]
+    assert isinstance(msg, j1939.J1939Message)
+    assert msg.pgn == 65202
+    assert msg.source_address == 1
+    assert msg.dest_address == j1939.ParameterGroupNumber.Address.GLOBAL
+    assert list(msg.data) == [1, 2, 3, 4, 5, 6, 7, 8]
+
+
+def test_subscribe_new_style_receives_peer_to_peer_dest_address(feeder):
+    """A single-argument callback observes the destination address of a
+    directed (PDU1) message, including when it's not one of this node's own
+    addresses (passive wildcard observation, see #59).
+    """
+    received = []
+
+    def callback(msg: j1939.J1939Message):
+        received.append(msg)
+
+    feeder.ecu.subscribe(callback)
+
+    # Destination-specific (PDU1) single-frame message (PGN 0xDC00, ATS) to a
+    # third node (0x21), not owned by this ECU.
+    feeder.can_messages = [
+        (Feeder.MsgType.CANRX, 0x00DC2101, [1, 2, 3, 4, 5, 6, 7, 8], 0.0),
+    ]
+    feeder.accept_all_messages()
+    feeder._inject_messages_into_ecu()
+
+    deadline = time.monotonic() + 2.0
+    while not received and time.monotonic() < deadline:
+        time.sleep(0.01)
+
+    assert len(received) == 1
+    assert received[0].dest_address == 0x21
+
+
+def test_subscribe_both_calling_conventions_side_by_side(feeder):
+    """Legacy 5-arg and new J1939Message-style subscribers can coexist."""
+    legacy_calls = []
+    new_calls = []
+
+    def legacy_callback(priority, pgn, sa, timestamp, data):
+        legacy_calls.append((priority, pgn, sa, timestamp, data))
+
+    def new_callback(msg):
+        new_calls.append(msg)
+
+    feeder.ecu.subscribe(legacy_callback)
+    feeder.ecu.subscribe(new_callback)
+
+    feeder.can_messages = [
+        (Feeder.MsgType.CANRX, 0x00FEB201, [1, 2, 3, 4, 5, 6, 7, 8], 0.0),
+    ]
+    feeder.pdus = [(Feeder.MsgType.PDU, 65202, [1, 2, 3, 4, 5, 6, 7, 8])]
+    feeder.receive()
+
+    assert len(legacy_calls) == 1
+    assert len(new_calls) == 1
+    assert new_calls[0].pgn == legacy_calls[0][1]
+
+
 def test_remove_ca_cleans_ca_subscriptions_but_preserves_ecu_subscriptions(feeder):
     """Removing a CA removes only subscriptions registered through that CA."""
     received = []