PyUAVCAN design thread

I have moved my rewrite of PyUAVCAN v1.0 from my personal fork back into the UAVCAN org:

At this point, the most critical parts of the library seem more or less finished and reasonably well tested (branch coverage ~94%; or 90% if you consider the auto-generated code which is not that well-tested), although they are still far from being API-stable:

  • Python package generation from DSDL namespaces using Nunavut.
  • Virtually zero-copy data serialization and deserialization.
  • Extensible transport layer architecture that is sufficiently generic to support at least CAN 2.0, CAN FD, UDP, and IEEE 802.15.4, including heterogeneous redundant configurations.
  • A concrete implementation of the CAN transport (2.0/FD universal).

One can generate code from DSDL, construct an object, serialize it, and send it over a simulated CAN bus (the reverse is also possible). Higher-level abstractions that would tie these operations together and provide a convenient application-level API are still missing. My next goal is to implement the so called “presentation layer” which provides the usual pub/sub and RPC abstractions; this seems straightforward and I don’t anticipate any issues here. Actually, I don’t anticipate any issues at all since the hardcore parts (namely, the serialization and the extensible transport layer arch) are already finished.

As I wrote earlier on this forum, I could really use help with development. At the moment, the most pressing issues are the lack of CAN media drivers and also the lack of adequate test rigging for them. I would like someone (perhaps @adolfogc and @ldg since they have expressed interest earlier, or whoever is available) to help move the project forward by contributing CAN media drivers for:

  • SocketCAN
  • Generic SLCAN adapters, with the loopback extensions as documented in the Zubax Babel datasheet
  • Python-CAN library

Some code may be borrowed from the old library; however, much of it will have to be rewritten from scratch because the old library has certain fatal issues in its architecture and does not provide asynchronous API. Media implementations are abstracted through an interface which seems kinda stable so far since I managed to write a working CAN transport implementation and a mock driver without changing anything in it:

class Media:
    ReceivedFramesHandler = Callable[[Iterable[TimestampedDataFrame]], None]
    def interface_name(self) -> str: ...
    def max_data_field_length(self) -> int: ...
    def number_of_acceptance_filters(self) -> int: ...
    def set_received_frames_handler(self, handler: ReceivedFramesHandler) -> None: ...
    async def configure_acceptance_filters(self, configuration: Sequence[FilterConfiguration]) -> None: ...
    async def enable_automatic_retransmission(self) -> None: ...
    async def send(self, frames: Iterable[DataFrame]) -> None: ...
    async def close(self) -> None: ...

There is a stub for SocketCAN located under which would be a good place to start with.

Besides CAN media drivers, there is also an outstanding issue of increasing the coverage of auto-generated code. As far as serialization and deserialization goes, the existing tests seem adequate: there are some basic carefully crafted manual serialization and deserialization tests and also a colossal randomized test which just generates tons of random objects and random serialized representations and throws that at the marshaling logic, ensuring that the random data survives roundtrip serialization/deserialization without distortions and that the serializer is able to handle random (insecure) inputs from the network correctly (it’s not allowed to throw exceptions). What’s missing is the error handling on the side of internal API (such as trying to initialize a union with more than one active field or trying to assign a string where numpy.array(dtype=uint8) is expected), and also perhaps it’s worth exploring whether it would be possible to generate an automatic test suite which would compare generated serialization/deserialization code against its source DSDL definition and prove its correctness. I am far from certain that it is necessary for PyUAVCAN since Python is not applicable in high-integrity applications (it’s not a high-integrity platform itself), but this could be a way to gather relevant experience which could be applied then in other implementations such as libcanard, libuavcan, or (nudge nudge @kjetilkjeka) which are designed for high(er)-reliability systems. I will spare you the details, but the general idea is that the current code generator emits special hooks in the comments that could potentially allow a sufficiently intelligent piece of software to automatically trace every line of generated code back to its DSDL specification (such requirement tracing either in automatic or manual form is one of the cornerstones of safety-critical software development). Here is an example for uavcan.node.port.GetInfo.0.1.Request, the annotations follow the form BEGIN .+ (DE)?SERIALIZATION:

        def _serialize_aligned_(self, _ser_: GetInfo_0_1.Request._SerializerTypeVar_) -> None:
            assert _ser_.current_bit_length % 8 == 0, 'Serializer is not byte-aligned'
            _orig_bit_length_ = _ser_.current_bit_length
            # BEGIN COMPOSITE SERIALIZATION: uavcan.node.port.GetInfo.Request.0.1
            assert _ser_.current_bit_length % 8 == 0, 'self.: void7'
            # BEGIN STRUCTURE FIELD SERIALIZATION: uavcan.node.port.ID.1.0 port_id
            # Object self.port_id is not always byte-aligned, serializing in-place.
            # BEGIN COMPOSITE SERIALIZATION: uavcan.node.port.ID.1.0
            # Tag field byte-aligned: False; values byte-aligned: True
            # BEGIN UNION FIELD SERIALIZATION: uavcan.node.port.ServiceID.1.0 service_id
            if self.port_id.service_id is not None:
                _ser_.add_unaligned_unsigned(0, 1)  # Tag 0
                assert _ser_.current_bit_length % 8 == 0, 'self.port_id.service_id: uavcan.node.port.ServiceID.1.0'
                self.port_id.service_id._serialize_aligned_(_ser_)  # Delegating because this object is always byte-aligned.
            # END UNION FIELD SERIALIZATION: uavcan.node.port.ServiceID.1.0 service_id
            # BEGIN UNION FIELD SERIALIZATION: uavcan.node.port.SubjectID.1.0 subject_id
            elif self.port_id.subject_id is not None:
                _ser_.add_unaligned_unsigned(1, 1)  # Tag 1
                assert _ser_.current_bit_length % 8 == 0, 'self.port_id.subject_id: uavcan.node.port.SubjectID.1.0'
                self.port_id.subject_id._serialize_aligned_(_ser_)  # Delegating because this object is always byte-aligned.
            # END UNION FIELD SERIALIZATION: uavcan.node.port.SubjectID.1.0 subject_id
                raise RuntimeError('Malformed union uavcan.node.port.ID.1.0')
            # END COMPOSITE SERIALIZATION: uavcan.node.port.ID.1.0
            # END STRUCTURE FIELD SERIALIZATION: uavcan.node.port.ID.1.0 port_id
            # END COMPOSITE SERIALIZATION: uavcan.node.port.GetInfo.Request.0.1
            assert 24 <= (_ser_.current_bit_length - _orig_bit_length_) <= 24, \
                'Bad serialization of uavcan.node.port.GetInfo.Request.0.1'

        def _deserialize_aligned_(_des_: GetInfo_0_1.Request._DeserializerTypeVar_) -> GetInfo_0_1.Request:
            assert _des_.consumed_bit_length % 8 == 0, 'Deserializer is not byte-aligned'
            _bit_length_base_ = _des_.consumed_bit_length
            # BEGIN COMPOSITE DESERIALIZATION: uavcan.node.port.GetInfo.Request.0.1
            assert _des_.consumed_bit_length % 8 == 0, 'void7'
            # BEGIN STRUCTURE FIELD DESERIALIZATION: uavcan.node.port.ID.1.0 port_id
            # The temporary _f0_ holds the value of the field "port_id"
            # The object is not always byte-aligned, deserializing in-place.
            # BEGIN COMPOSITE DESERIALIZATION: uavcan.node.port.ID.1.0
            # Tag field byte-aligned: False; values byte-aligned: True
            _tag1_ = _des_.fetch_unaligned_unsigned(1)
            # BEGIN UNION FIELD DESERIALIZATION: uavcan.node.port.ServiceID.1.0 service_id
            if _tag1_ == 0:
                assert _des_.consumed_bit_length % 8 == 0, 'uavcan.node.port.ServiceID.1.0'
                _uni2_ = uavcan.node.port.ServiceID_1_0._deserialize_aligned_(_des_)
                _f0_ = uavcan.node.port.ID_1_0(service_id=_uni2_)
            # END UNION FIELD DESERIALIZATION: uavcan.node.port.ServiceID.1.0 service_id
            # BEGIN UNION FIELD DESERIALIZATION: uavcan.node.port.SubjectID.1.0 subject_id
            elif _tag1_ == 1:
                assert _des_.consumed_bit_length % 8 == 0, 'uavcan.node.port.SubjectID.1.0'
                _uni3_ = uavcan.node.port.SubjectID_1_0._deserialize_aligned_(_des_)
                _f0_ = uavcan.node.port.ID_1_0(subject_id=_uni3_)
            # END UNION FIELD DESERIALIZATION: uavcan.node.port.SubjectID.1.0 subject_id
                raise _des_.FormatError(f'uavcan.node.port.ID.1.0: Union tag value {_tag1_} is invalid')
            # END COMPOSITE DESERIALIZATION: uavcan.node.port.ID.1.0
            # END STRUCTURE FIELD DESERIALIZATION: uavcan.node.port.ID.1.0 port_id
            self = GetInfo_0_1.Request(port_id=_f0_)
            # END COMPOSITE DESERIALIZATION: uavcan.node.port.GetInfo.Request.0.1
            assert 24 <= (_des_.consumed_bit_length - _bit_length_base_) <= 24, \
                'Bad deserialization of uavcan.node.port.GetInfo.Request.0.1'
            assert isinstance(self, GetInfo_0_1.Request)
            return self

Also, some information on DSDL code generation (particularly related to in-depth data layout analysis for generation of highly efficient (de)serialization code) is provided here:

Perhaps it’s also worth looking into the mutability concerns in the context of (near) zero-copy serialization. Generally, zero-copy stacks require immutability of objects due to the issues with shared memory, which the following snippet (not pseudocode but an actual example that can be reproduced in REPL) demonstrates:

>>> import pyuavcan
>>> import uavcan.primitive.array  # Assuming that the package is already generated
>>> b = bytearray([2, 1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> msg = pyuavcan.dsdl.try_deserialize(uavcan.primitive.array.Natural32_1_0, b)
>>> msg
uavcan.primitive.array.Natural32.1.0(value=[67305985, 134678021])
>>> msg.value[0] = 0xFFFFFFFF
>>> list(b)                        # Source array has been updated
[2, 255, 255, 255, 255, 5, 6, 7, 8, 9]

Since immutability cannot be statically enforced in Python, this could potentially cause someone to shoot themselves in the foot by mutating objects whose data is still being processed by the stack, leading to potentially hard-to-trace data corruption issues. I am not yet sure how important this property is, but any feedback and ideas would be appreciated.

I am referring to the serialization logic as “near zero-copy” because the current implementation actually does copy the data during deserialization if the underlying transport emits the data in several fragments rather than one contiguous chunk. The primitive serialization classes should be updated to support fragmented inputs to eliminate this special case, so this is a yet another aspect where contributions would be welcome.

Let me summarize. The library is a work-in-progress, but it is nearing a usable state fast and at this point the hardest parts seem to be done. It could use help with (ordered from most critical to least critical):

  • Async CAN media drivers for SocketCAN, SLCAN, and Python-CAN.
  • Support for fragmented serialization buffers.
  • Automatic verification of generated serialization code by tracing it back to the source DSDL definition.

I would like PyUAVCAN to become a full-featured implementation of the stack, a first-class citizen of the UAVCAN ecosystem (which can’t be said about the old implementation due to its architectural shortcomings and limitations) which could be used for testing new protocol features and building sophisticated monitoring and testing tools on top of, such as Yukon.

There will be more opportunities for contribution once I’m done with the higher-level API which I am currently working on.

Paging @adolfogc @ldg @Zarkopafilis @scottdixon @erik.rainey


Would you mind if Integrated tox to ensure we are supporting the full 3.5 - 3.8 range of Python and setup a release on If you leave documentation to the end it’ll be drudgery. It’s better to do it as you go.

I would not mind at all. I am targeting Python 3.7/3.8+ though so 3.5/3.6 will not work. We could extend the support but it might require changes I would not be entirely happy about; also keep in mind that 3.5 will be EOLed next year anyway, it’s pretty old already. I think that it is more or less safe-ish to require 3.7 seeing as major operating systems support it or at least are going to support it by the time UAVCAN v1 is out.

Specifically, PyUAVCAN requires the following features that are available since Python 3.6:

And also these which are available since Python 3.7:

Yes I agree. I had to rewrite the abstract transport model several times though which I expected will happen so it made sense to postpone documenting things but at this point it should be safe to proceed. I already got as far as sketching out diagrams:

Non-redundant transport:


Redundant transport:


Redundancy does not make sense for PyUAVCAN from the reliability standpoint but it is necessary for diagnostic and monitoring purposes.

My attempt to get the first pre-release out before the summit is hereby declared unsuccessful (but just barely). PyUAVCAN generally works, you can connect to a CAN bus via SocketCAN (both 2.0 and FD supported), you can pub/sub and rpc, but there is still nobody in the world who knows how to use it except myself and making anything released without the docs seems pointless.

Nudge @Zarkopafilis – once you’re done with your exams, ping me please so that we could discuss integration of PyUAVCAN into Yukon (finally).

For testing we’ll have to switch from Travis to something else because Travis doesn’t let me load the vcan kernel module which is necessary for testing (which is super weird because the module is shipped with Ubuntu by default); also we need testing against Windows which is again less than trivial to set up on Travis.

OK. If you want to start messing around with the backend, you can take a look at how the devserver serves the responses with quart, or the scaffolds on the backend folder.

You may start with some basic integrations on endpoints, or just plain broadcasting to all the clients connected to a SSE socket, or just, having a shared/populated state which can then be served.

Perhaps AppVeyor could work? It supports both Linux and Windows.

Just use a docker container. We have a docker hub organization setup already and Nunavut and libuavcan already build in containers.

PyUAVCAN v0.2.0 is now documented and usable, although it is still far from being API-stable. I don’t plan on breaking anything, but I know that something will end up being broken on our way towards v1.0.

Here’s the link once again: (the branch will replace master soon-ish, and the current master will become uavcan-v0). The docs are at and the package is released on PyPI.

I would like to invite everyone to readthedocs and submit constructive criticism here or on GitHub. The following questions are of particular interest:

  • Does the overall architecture seem sensible?
  • Is the public API sufficient and consistent?
  • Is the documentation sufficiently detailed?
  • Are there any obvious issues with the CAN transport implementation?
  • Are there any issues with the proposed experimental serial transport implementation (only a stub is available at the moment)? I am now noticing that I have completely neglected to provide justification for its existence or to define business requirements for it. Part of the reason for such neglect is that the serial transport is based entirely on the PoPCoP protocol, which has been successfully deployed in certain projects I am involved in for real-time point-to-point communication over serial links: The objective of the UAVCAN serial transport is to obviate the need for such ad-hoc solutions and to address the need for transport log storage discussed on this forum earlier. If there is interest, we can split this into a separate topic where I could detail my vision of it.

I am now going to switch from my current cowboy workflow (works well for early-stage development) to a more structured approach so it’ll be possible to observe the development in progress and submit feedback in real-time.

1 Like