pycyphal.dsdl package

Module contents

This module is used for automatic generation of Python classes from DSDL type definitions and also for various manipulations on them. Auto-generated classes have a high-level application-facing API and built-in auto-generated serialization and deserialization routines.

The serialization code heavily relies on NumPy and the data alignment analysis implemented in PyDSDL. Some of the technical details are covered in the following posts:

The main entity of this module is the function compile().

pycyphal.dsdl.compile(root_namespace_directory: _AnyPath, lookup_directories: Optional[list[_AnyPath]] = None, output_directory: Optional[_AnyPath] = None, allow_unregulated_fixed_port_id: bool = False) → Optional[GeneratedPackageInfo][source]

This function runs the DSDL compiler, converting a specified DSDL root namespace into a Python package. In the generated package, nested DSDL namespaces are represented as Python subpackages, DSDL types as Python classes, type version numbers as class name suffixes separated via underscores (like Type_1_0), constants as class attributes, fields as properties. For a more detailed information on how to use generated types, just generate them and read the resulting code – it is made to be human-readable and contains docstrings.

Generated packages can be freely moved around the file system or even deployed on other systems as long as their dependencies are satisfied, which are numpy and pydsdl.

Generated packages do not automatically import their nested subpackages. For example, if the application needs to use uavcan.node.Heartbeat.1.0, it has to import uavcan.node explicitly; doing just import uavcan is not sufficient.

If the source definition contains identifiers, type names, namespace components, or other entities whose names are listed in nunavut.lang.py.PYTHON_RESERVED_IDENTIFIERS, the compiler applies stropping by suffixing such entities with an underscore _. A small subset of applications may require access to a generated entity without knowing in advance whether its name is a reserved identifier or not (i.e., whether it’s stropped or not). To simplify usage, this submodule provides helper functions pycyphal.dsdl.get_attribute() and pycyphal.dsdl.set_attribute() that provide access to generated class/object attributes using their original names before stropping. Likewise, the function pycyphal.dsdl.get_model() can find a generated type even if any of its name components are stropped; e.g., a DSDL type str.Type.1.0 would be imported as str_.Type_1_0. None of it, however, is relevant for an application that does not require genericity (vast majority of applications don’t), so a much easier approach in that case is just to look at the generated code and see if there are any stropped identifiers in it, and then just use appropriate names statically.

Tip

Production applications should compile their DSDL namespaces as part of the package build process. This can be done by overriding the build_py command in setup.py and invoking this function from there.

Tip

Configure your IDE to index the compilation output directory as a source directory to enable code completion. For PyCharm: right click the directory –> “Mark Directory as” ->”Sources Root”.

Parameters

root_namespace_directory – The source DSDL root namespace directory path. The last component of the path is the name of the root namespace. For example, to generate package for the root namespace uavcan, the path would be like foo/bar/uavcan.
lookup_directories – An iterable of DSDL root namespace directory paths where to search for referred DSDL definitions. The format of each path is the same as for the previous parameter; i.e., the last component of each path is a DSDL root namespace name. If you are generating code for a vendor-specific DSDL root namespace, make sure to provide at least the path to the standard uavcan namespace directory here.
output_directory – The generated Python package directory will be placed into this directory. If not specified or None, the current working directory is used. For example, if this argument equals foo/bar, and the DSDL root namespace name is uavcan, the top-level __init__.py of the generated package will end up in foo/bar/uavcan/__init__.py. The directory tree will be created automatically if it does not exist (like mkdir -p). If the destination exists, it will be silently written over. Applications that compile DSDL lazily are recommended to shard the output directory by the library version number to avoid compatibility issues with code generated by older versions of the library. Don’t forget to add the output directory to PYTHONPATH.
allow_unregulated_fixed_port_id – If True, the compiler will not reject unregulated data types with fixed port-ID. If you are not sure what it means, do not use it, and read the Cyphal specification first.

Returns

An instance of GeneratedPackageInfo describing the generated package, unless the root namespace is empty, in which case it’s None.

Raises

OSError if required operations on the file system could not be performed; pydsdl.InvalidDefinitionError if the source DSDL definitions are invalid; pydsdl.InternalError if there is a bug in the DSDL processing front-end; ValueError if any of the arguments are otherwise invalid.

The following table is an excerpt from the Cyphal specification. Observe that unregulated fixed port identifiers are prohibited by default, but it can be overridden.

Scope	Regulated	Unregulated
Public	Standard and contributed (e.g., vendor-specific) definitions. Fixed port identifiers are allowed; they are called “regulated port-IDs”.	Definitions distributed separately from the Cyphal specification. Fixed port identifiers are not allowed.
Private	Nonexistent category.	Definitions that are not available to anyone except their authors. Fixed port identifiers are permitted (although not recommended); they are called “unregulated fixed port-IDs”.

pycyphal.dsdl.compile_all(root_namespace_directories: Iterable[_AnyPath], output_directory: Optional[_AnyPath] = None, *, allow_unregulated_fixed_port_id: bool = False) → list[GeneratedPackageInfo][source]

This is a simple convenience wrapper over compile() that addresses a very common use case where the application needs to compile multiple inter-dependent namespaces.

Parameters

root_namespace_directories – compile() will be invoked once for each directory in the list, using all of them as look-up dirs for each other. They may be ordered arbitrarily. Directories that contain no DSDL definitions are ignored.
output_directory – See compile().
allow_unregulated_fixed_port_id – See compile().

Returns

A list of of GeneratedPackageInfo, one per non-empty root namespace directory.

>>> import sys
>>> import pathlib
>>> import importlib
>>> import pycyphal
>>> compiled_dsdl_dir = pathlib.Path(".lazy_compiled", pycyphal.__version__)
>>> compiled_dsdl_dir.mkdir(parents=True, exist_ok=True)
>>> sys.path.insert(0, str(compiled_dsdl_dir))
>>> try:
...     import sirius_cyber_corp
...     import uavcan.si.sample.volumetric_flow_rate
... except (ImportError, AttributeError):
...     _ = pycyphal.dsdl.compile_all(
...         [
...             DEMO_DIR / "custom_data_types/sirius_cyber_corp",
...             DEMO_DIR / "public_regulated_data_types/uavcan",
...             DEMO_DIR / "public_regulated_data_types/reg/",
...         ],
...         output_directory=compiled_dsdl_dir,
...     )
...     importlib.invalidate_caches()
...     import sirius_cyber_corp
...     import uavcan.si.sample.volumetric_flow_rate

class pycyphal.dsdl.GeneratedPackageInfo(path: 'pathlib.Path', models: 'Sequence[pydsdl.CompositeType]', name: 'str')[source]

Bases: object

path: pathlib.Path: Path to the directory that contains the top-level __init__.py.

models: Sequence[pydsdl._serializable._composite.CompositeType]: List of PyDSDL objects describing the source DSDL definitions. This can be used for arbitrarily complex introspection and reflection.

name: str: The name of the generated package, which is the same as the name of the DSDL root namespace unless the name had to be stropped. See nunavut.lang.py.PYTHON_RESERVED_IDENTIFIERS.

__delattr__(name)[source]

__eq__(other)[source]

__hash__()[source]

__init__(path: pathlib.Path, models: Sequence[pydsdl._serializable._composite.CompositeType], name: str) → None[source]

__repr__()[source]

__setattr__(name, value)[source]

pycyphal.dsdl.serialize(obj: Any) → Iterable[memoryview][source]: Constructs a serialized representation of the provided top-level object. The resulting serialized representation is padded to one byte in accordance with the Cyphal specification. The constructed serialized representation is returned as a sequence of byte-aligned fragments which must be concatenated in order to obtain the final representation. The objective of this model is to avoid copying data into a temporary buffer when possible. Each yielded fragment is of type memoryview pointing to raw unsigned bytes. It is guaranteed that at least one fragment is always returned (which may be empty).

pycyphal.dsdl.deserialize(dtype: Type[pycyphal.dsdl.T], fragmented_serialized_representation: Sequence[memoryview]) → Optional[pycyphal.dsdl.T][source]

Constructs an instance of the supplied DSDL-generated data type from its serialized representation. Returns None if the provided serialized representation is invalid.

This function will never raise an exception for invalid input data; the only possible outcome of an invalid data being supplied is None at the output. A raised exception can only indicate an error in the deserialization logic.

Important

The constructed object may contain arrays referencing the memory allocated for the serialized representation. Therefore, in order to avoid unintended data corruption, the caller should destroy all references to the serialized representation after the invocation.

Important

The supplied fragments of the serialized representation should be writeable. If they are not, some of the array-typed fields of the constructed object may be read-only.

pycyphal.dsdl.get_fixed_port_id(class_or_instance: Any) → Optional[int][source]: Returns None if the supplied type has no fixed port-ID.

pycyphal.dsdl.get_model(class_or_instance: Any) → pydsdl._serializable._composite.CompositeType[source]: Obtains a PyDSDL model of the supplied DSDL-generated class or its instance. This is the inverse of get_class().

pycyphal.dsdl.get_class(model: pydsdl._serializable._composite.CompositeType) → type[source]

Returns a generated native class implementing the specified DSDL type represented by its PyDSDL model object. Promotes the model to delimited type automatically if necessary. This is the inverse of get_model().

Raises

ImportError if the generated package or subpackage cannot be found.
AttributeError if the package is found but it does not contain the requested type.
TypeError if the requested type is found, but its model does not match the input argument. This error may occur if the DSDL source has changed since the type was generated. To fix this, regenerate the package and make sure that all components of the application use identical or compatible DSDL source files.

pycyphal.dsdl.get_extent_bytes(class_or_instance: Any) → int[source]

pycyphal.dsdl.get_attribute(obj: Any, name: str) → Any[source]

DSDL type attributes whose names can’t be represented in Python (such as def or type) are suffixed with an underscore. This function allows the caller to read arbitrary attributes referring to them by their original DSDL names, e.g., def instead of def_.

This function behaves like getattr() if the attribute does not exist.

pycyphal.dsdl.set_attribute(obj: Any, name: str, value: Any) → None[source]

DSDL type attributes whose names can’t be represented in Python (such as def or type) are suffixed with an underscore. This function allows the caller to assign arbitrary attributes referring to them by their original DSDL names, e.g., def instead of def_.

If the attribute does not exist, raises AttributeError.

pycyphal.dsdl.is_serializable(dtype: Any) → bool[source]

pycyphal.dsdl.is_message_type(dtype: Any) → bool[source]

pycyphal.dsdl.is_service_type(dtype: Any) → bool[source]

pycyphal.dsdl.to_builtin(obj: object) → Dict[str, Any][source]

Accepts a DSDL object (an instance of a Python class auto-generated from a DSDL definition), returns its value represented using only native built-in types: dict, list, bool, int, float, str. Ordering of dict elements is guaranteed to match the field ordering of the source definition. Keys of dicts representing DSDL objects use the original unstropped names from the source DSDL definition; e.g., if, not if_.

This is intended for use with JSON, YAML, and other serialization formats.

>>> import json
>>> import uavcan.primitive.array
>>> json.dumps(to_builtin(uavcan.primitive.array.Integer32_1_0([-123, 456, 0])))
'{"value": [-123, 456, 0]}'
>>> import uavcan.register
>>> request = uavcan.register.Access_1_0.Request(
...     uavcan.register.Name_1_0('my.register'),
...     uavcan.register.Value_1_0(integer16=uavcan.primitive.array.Integer16_1_0([1, 2, +42, -10_000]))
... )
>>> to_builtin(request)  
{'name':  {'name': 'my.register'},
 'value': {'integer16': {'value': [1, 2, 42, -10000]}}}

pycyphal.dsdl.update_from_builtin(destination: pycyphal.dsdl.T, source: Any) → pycyphal.dsdl.T[source]

Updates the provided DSDL object (an instance of a Python class auto-generated from a DSDL definition) with the values from a native representation, where DSDL objects are represented as dicts, arrays are lists, and primitives are represented as int/float/bool. This is the reverse of to_builtin(). Values that are not specified in the source are not updated (left at their original values), so an empty source will leave the input object unchanged.

Source field names shall match the original unstropped names provided in the DSDL definition; e.g., if, not if_. If there is more than one variant specified for a union type, the last specified variant takes precedence. If the structure of the source does not match the destination object, the correct representation may be deduced automatically as long as it can be done unambiguously.

Parameters

destination – The object to update. The update will be done in-place. If you don’t want the source object modified, clone it beforehand.
source – The dict instance containing the values to update the destination object with.

Returns

A reference to destination (not a copy).

Raises

ValueError if the provided source values cannot be applied to the destination object, or if the source contains fields that are not present in the destination object. TypeError if an entity of the source cannot be converted into the type expected by the destination.

>>> import tests; tests.dsdl.compile()  # DSDL package generation not shown in this example.
[...]
>>> import json
>>> import uavcan.primitive.array
>>> import uavcan.register
>>> request = uavcan.register.Access_1_0.Request(
...     uavcan.register.Name_1_0('my.register'),
...     uavcan.register.Value_1_0(string=uavcan.primitive.String_1_0('Hello world!'))
... )
>>> request
uavcan.register.Access.Request...name='my.register'...value='Hello world!'...
>>> update_from_builtin(request, {  # Switch the Value union from string to int16; keep the name unchanged.
...     'value': {
...         'integer16': {
...             'value': [1, 2, 42, -10000]
...         }
...     }
... })  
uavcan.register.Access.Request...name='my.register'...value=[ 1, 2, 42,-10000]...

The following examples showcase positional initialization:

>>> from uavcan.node import Heartbeat_1
>>> update_from_builtin(Heartbeat_1(), [123456, 1, 2])  
uavcan.node.Heartbeat.1.0(uptime=123456,
                          health=uavcan.node.Health.1.0(value=1),
                          mode=uavcan.node.Mode.1.0(value=2),
                          vendor_specific_status_code=0)
>>> update_from_builtin(Heartbeat_1(), 123456)  
uavcan.node.Heartbeat.1.0(uptime=123456,
                          health=uavcan.node.Health.1.0(value=0),
                          mode=uavcan.node.Mode.1.0(value=0),
                          vendor_specific_status_code=0)
>>> update_from_builtin(Heartbeat_1(), [0, 0, 0, 0, 0])  
Traceback (most recent call last):
    ...
TypeError: ...

>>> update_from_builtin(uavcan.primitive.array.Real64_1(), 123.456) 
uavcan.primitive.array.Real64.1.0(value=[123.456])
>>> update_from_builtin(uavcan.primitive.array.Real64_1(), [123.456]) 
uavcan.primitive.array.Real64.1.0(value=[123.456])
>>> update_from_builtin(uavcan.primitive.array.Real64_1(), [123.456, -9]) 
uavcan.primitive.array.Real64.1.0(value=[123.456, -9. ])

>>> update_from_builtin(uavcan.register.Access_1_0.Request(), ["X", {"integer8": 99}])  # Same as the next one!
uavcan.register.Access.Request...name='X'...value=[99]...
>>> update_from_builtin(uavcan.register.Access_1_0.Request(), {'name': 'X', 'value': {'integer8': {'value': [99]}}})
uavcan.register.Access.Request...name='X'...value=[99]...

pycyphal.dsdl.install_import_hook(lookup_directories: Optional[Iterable[Union[pathlib.Path, str]]] = None, output_directory: Union[None, str, pathlib.Path] = None, allow_unregulated_fixed_port_id: Optional[bool] = None) → None[source]

Installs python import hook, which automatically compiles any DSDL if package is not found.

A default import hook is automatically installed when pycyphal is imported. To opt out, set environment variable PYCYPHAL_NO_IMPORT_HOOK=True before importing pycyphal.

Parameters

lookup_directories – List of directories where to look for DSDL sources. If not provided, it is sourced from CYPHAL_PATH environment variable.
output_directory – Directory to output compiled DSDL packages. If not provided, PYCYPHAL_PATH environment variable is used. If that is not available either, a default ~/.pycyphal (or other OS equivalent) directory is used.
allow_unregulated_fixed_port_id – If True, the compiler will not reject unregulated data types with fixed port-ID. If not provided, it will be sourced from CYPHAL_ALLOW_UNREGULATED_FIXED_PORT_ID variable or default to False.

pycyphal.dsdl.generate_package(*args, **kwargs)[source]: Deprecated alias of compile().