Add auto-modifying cyclic tasks

[bessman]

This is a proposal to add an optional callback function to ModifiableCyclicTaskABC which manipulates each message automatically before sending it. This feature can be used to, for example, increment a counter or compute a checksum, as shown in examples/cyclic_checksum.py.

If you think this functionality is appropriate for merging, I will of course add it to all interfaces which override the fallback ThreadBasedCyclicSendTask.

[codecov]

Codecov Report

Merging #703 (433b91f) into develop (4b1acde) will increase coverage by 3.18%.
The diff coverage is 64.28%.

❗ Current head 433b91f differs from pull request most recent head 93ef698. Consider uploading reports for the commit 93ef698 to get more accurate results

Additional details and impacted files

@@             Coverage Diff             @@
##           develop     #703      +/-   ##
===========================================
+ Coverage    65.16%   68.34%   +3.18%     
===========================================
  Files           81       69      -12     
  Lines         8853     6211    -2642     
===========================================
- Hits          5769     4245    -1524     
+ Misses        3084     1966    -1118     

[christiansandberg]

I think this is something that is much needed. Things like checksums and alive counters are getting more common as safety is becoming a major driving force. Unfortunately it's only going to work with the thread based software method. Passing the task entirely to the kernel or hardware also means you can't do any processing in user space.

Should we provide an interface to allow the user to choose the thread based transmission at the expense of poor timing?

[bessman]

I've thought about this some more, and I'm no longer sure a callback is the way to go. At least for the issue of counters and checksums, a better solution is to pre-calculate all CRCs and send them using CyclicSendTask's already existing ability to send a sequence of messages, e.g.:

def crc(msg, counter):
    #  checksum calculation
    return checksum


msgs = []

for counter in range(16):
    m = Message(arbitration_id=0x123, data=[1, 2, 3, 4, 5, 6, 7, 0])
    m.data[7] = counter + (crc(m, counter) << 4)
    msgs.append(m)

bus.send_periodic(msgs, 0.01)

This has the added benefit that the user does not have to keep track of the counter when the message data changes. With the callback method, the counter will reset when the data changes unless the user manually checks the counter of the most recently sent message and adds it to the message.

The callback is more flexible, and there could be applications where pre-calculating all messages is not possible (though I can't think of any at the moment). But if that flexibility comes at the cost of performance, perhaps it isn't worth it.

[zariiii9003]

Maybe we could implement this in the Message class? The data attribute could be a generator that calculates the next bytearray for the message.
The user could implement his calculation by subclassing Message and reimplementing the data attribute. Otherwise it would just return a private _data variable.

[bessman]

Maybe we could implement this in the Message class? The data attribute could be a generator that calculates the next bytearray for the message.
The user could implement his calculation by subclassing Message and reimplementing the data attribute. Otherwise it would just return a private _data variable.

How would this work, exactly? If data is a generator, wouldn't we need to call next(data) all over the place? That seems like a big change.

[hardbyte]

This is a good proposal, and a clean default implementation. I also don't see how a callback would interact nicely with the backends that already offload periodic messages (without falling back to the threading approach), which isn't a great user experience.

However if this is useful for a few people (e.g. #809) and we can implement it in a way that doesn't harm the performance when not being used then I'd be happy to merge it in. I'm thinking that in the socketcan case we make it very clear in the docs that giving a callback will mean looser timing of messages being sent based on Python threads rather than the kernel.

[zorce]

Would love to see something like this, but if possible a kernel implementation to handle similar issues.
Also I have a use case where in an industry use case "CAN in Automation (CiA)" where you might want to read multiple muxed signals. And hence want to change the mux value between every frame, the frames have an inhibit time of let say 100ms between the messages.

[hartkopp]

Would love to see something like this, but if possible a kernel implementation to handle similar issues.
Also I have a use case where in an industry use case "CAN in Automation (CiA)" where you might want to read multiple muxed signals. And hence want to change the mux value between every frame, the frames have an inhibit time of let say 100ms between the messages.

The CAN_BCM (in-kernel Broadcast Manager) can send a sequence of up to 256 CAN(FD) frames, e.g. every 100ms the next frame in the row is transmitted. The nframes element in the bcm_msg_head is usually 1 but can be extended to 2 .. 256 to send a sequence.
So you can send mux messages with just one TX_SETUP command:
https://www.kernel.org/doc/html/latest/networking/can.html#broadcast-manager-message-sequence-transmission

Same on the receiving side: When you want to filter on data changes on a specific MUX ID you define the mux mask (to define where the MUX ID sits in the CAN frame data) and put the MUX ID in that place in the RX_SETUP filter.
https://www.kernel.org/doc/html/latest/networking/can.html#broadcast-manager-multiplex-message-receive-filter

[zorce]

Would love to see something like this, but if possible a kernel implementation to handle similar issues.
Also I have a use case where in an industry use case "CAN in Automation (CiA)" where you might want to read multiple muxed signals. And hence want to change the mux value between every frame, the frames have an inhibit time of let say 100ms between the messages.

The CAN_BCM (in-kernel Broadcast Manager) can send a sequence of up to 256 CAN(FD) frames, e.g. every 100ms the next frame in the row is transmitted. The nframes element in the bcm_msg_head is usually 1 but can be extended to 2 .. 256 to send a sequence.
So you can send mux messages with just one TX_SETUP command:
https://www.kernel.org/doc/html/latest/networking/can.html#broadcast-manager-message-sequence-transmission

Same on the receiving side: When you want to filter on data changes on a specific MUX ID you define the mux mask (to define where the MUX ID sits in the CAN frame data) and put the MUX ID in that place in the RX_SETUP filter.
https://www.kernel.org/doc/html/latest/networking/can.html#broadcast-manager-multiplex-message-receive-filter

Didn't know that, thanks for the information. Is this also exposed with other backends, noticed there is a cyclic_multiple example added 2 years ago. but I was running on version 3.3.4. will try to update and try it out.

[hartkopp]

Didn't know that, thanks for the information. Is this also exposed with other backends, noticed there is a cyclic_multiple example added 2 years ago. but I was running on version 3.3.4. will try to update and try it out.

I've just programmed the CAN_BCM module and started to look into the API that python-can provides. Unfortunately I'm not that good in Python to get behind the current API concept for handling (and on-the-fly updating) a sequence of cyclic messages ;-)

[leu1dy]

I am also interested in about this feature. AliveCounter & Checksums are part of the daily work in automotive domain.

Any chance to see this feature at a release state ?

Thank you

[djnnvx]

Hello,

I too am interested in this feature. Is there anything to correct for a merge, besides fixing the merge conflicts ?

I would love to see it merged (or even help with the work left for a merge), as I am currently building a tool that requires such a need

kind regards

[djnnvx]

For those of you looking to have a callback like I had, I found a way to do it. Here's how:

import asyncio
import can

 # Initiate listeners
logger = can.Logger('example.asc')

async_reader = can.AsyncBufferedReader()
listeners: List[can.notifier.MessageRecipient] = [
    async_reader, logger,
    print_msg,  # print callback function
]

 with can.ThreadSafeBus(

            # fetching from config
            name=some_name,
            bustype=some_bustype,
            channel=some_channel,

            # Maybe it can work without this, actually. Have not tested yet
            receive_own_messages=True,
    ) as bus:

    # Create Notifier with an explicit loop to use for scheduling of callbacks
    notifier = can.Notifier(bus, listeners, loop=asyncio.get_running_loop())

    # send message periodically
    task = bus.send_periodic(some_message, 0.2)

    while True:

        # read all messages asynchronously
        async for msg in async_reader:

            msg.data = crc_function_or_whatever(msg)
            task.modify_data(msg)

Since we read messages asynchronously, we can modify the tasks when we receive a new version of the message. This should then be sent on the bus to other ECUs. To ensure that we refresh the message periodically, we require the bus to receive our own message, which should trigger the callback.

have not run extensive tests yet so there may be better way/things to fix, but it seems to work.

Hope this helps :)~

[zariiii9003]

I resolved the merge conflicts. There are still a few problems though:

• IXXATBus._send_periodic_internal must be adapted
• IXXATBus and socketcan should warn, if modifier_callback is not None and use the thread based cyclic send task as fallback

Personally i would prefer a mutating callback with no return value though (Callable[[Message], None])

[zariiii9003]

I think this can be merged if everyone agrees with the changes.