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 = []