From b42d3ef145fab15cf1dfa8fcbfecf7f8e6fc004e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Sainz-Maza=20Serna?= Date: Fri, 28 Aug 2026 11:52:47 +0200 Subject: [PATCH] feat: add J1939Message for subscriber callbacks with destination address _notify_subscribers() has always received the destination address from the data link layer but never passed it to subscriber callbacks, which only get (priority, pgn, sa, timestamp, data). This makes it impossible for a subscriber to know which destination address a message was addressed to. Add j1939.J1939Message, a NamedTuple carrying all five existing fields plus dest_address. Subscribers can opt in by declaring a single-argument callback: def on_message(msg: j1939.J1939Message): ... The calling convention is detected once, from the callback's signature, at subscribe() time (ElectronicControlUnit._callback_takes_message), so existing 5-argument callbacks throughout the codebase, examples, and downstream users keep working unchanged -- this is intentionally additive rather than a hard breaking change, since callbacks in this kind of protocol stack often run on embedded/field systems with slower upgrade cycles. Internal library subscribers (DM1, DM14/16, memory_access) are left on the legacy form since none of them currently use the destination address; migrating them would be pure churn. Updated the two simplest examples and the README quick-start to demonstrate the new style. Fixes #61 Co-Authored-By: Claude Sonnet 5 --- README.md | 25 ++++---- examples/simple_receive_global.py | 18 +----- examples/simple_receive_peer_to_peer.py | 18 +----- j1939/__init__.py | 1 + j1939/electronic_control_unit.py | 49 +++++++++++++++- j1939/message.py | 31 ++++++++++ test/test_ecu.py | 77 +++++++++++++++++++++++++ 7 files changed, 171 insertions(+), 48 deletions(-) create mode 100644 j1939/message.py diff --git a/README.md b/README.md index bde0939..1363761 100644 --- a/README.md +++ b/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: diff --git a/examples/simple_receive_global.py b/examples/simple_receive_global.py index eb7ac96..36d2ded 100644 --- a/examples/simple_receive_global.py +++ b/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") diff --git a/examples/simple_receive_peer_to_peer.py b/examples/simple_receive_peer_to_peer.py index 546dce7..250a4b0 100644 --- a/examples/simple_receive_peer_to_peer.py +++ b/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") diff --git a/j1939/__init__.py b/j1939/__init__.py index fec139a..13ff6e5 100644 --- a/j1939/__init__.py +++ b/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 diff --git a/j1939/electronic_control_unit.py b/j1939/electronic_control_unit.py index 7637240..6d3d333 100644 --- a/j1939/electronic_control_unit.py +++ b/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 diff --git a/j1939/message.py b/j1939/message.py new file mode 100644 index 0000000..d3007d2 --- /dev/null +++ b/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 diff --git a/test/test_ecu.py b/test/test_ecu.py index 7a8a101..00af138 100644 --- a/test/test_ecu.py +++ b/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 = []