pycyphal.application.register package

Subpackages

Module contents

Implementation of the Cyphal register interface as defined in the Cyphal Specification (section 5.3 Application-layer functions).

pycyphal.application.register.Empty[source]

alias of Empty_1_0

pycyphal.application.register.String[source]

alias of String_1_0

pycyphal.application.register.Unstructured[source]

alias of Unstructured_1_0

pycyphal.application.register.Bit[source]

alias of Bit_1_0

pycyphal.application.register.Integer64[source]

alias of Integer64_1_0

pycyphal.application.register.Integer32[source]

alias of Integer32_1_0

pycyphal.application.register.Integer16[source]

alias of Integer16_1_0

pycyphal.application.register.Integer8[source]

alias of Integer8_1_0

pycyphal.application.register.Natural64[source]

alias of Natural64_1_0

pycyphal.application.register.Natural32[source]

alias of Natural32_1_0

pycyphal.application.register.Natural16[source]

alias of Natural16_1_0

pycyphal.application.register.Natural8[source]

alias of Natural8_1_0

pycyphal.application.register.Real64[source]

alias of Real64_1_0

pycyphal.application.register.Real32[source]

alias of Real32_1_0

pycyphal.application.register.Real16[source]

alias of Real16_1_0

pycyphal.application.register.Value[source]

alias of Value_1_0

class pycyphal.application.register.ValueProxy(v: ValueProxy | Value_1_0 | String_1_0 | Unstructured_1_0 | Bit_1_0 | Integer8_1_0 | Integer16_1_0 | Integer32_1_0 | Integer64_1_0 | Natural8_1_0 | Natural16_1_0 | Natural32_1_0 | Natural64_1_0 | Real16_1_0 | Real32_1_0 | Real64_1_0 | str | bytes | bool | int | float | Iterable[bool] | Iterable[int] | Iterable[float] | ndarray[Any, dtype[Any]])[source]

Bases: object

This a wrapper over the standard uavcan.register.Value (transpiled into Value) with convenience accessors added that enable automatic conversion (with implicit casting) between native Python types and DSDL types.

It is possible to create a new instance from native types, in which case the most suitable regiter type will be deduced automatically. Do not rely on this behavior if a specific register type needs to be ensured.

>>> from pycyphal.application.register import Real64, Bit, String, Unstructured
>>> p = ValueProxy(Value(bit=Bit([True, False])))   # Specify explicit type.
>>> p.bools
[True, False]
>>> p.ints
[1, 0]
>>> p.floats
[1.0, 0.0]
>>> p.assign([0, 1.0])
>>> p.bools
[False, True]
>>> p = ValueProxy([0, 1.5, 2.3, -9])               # Use deduction.
>>> p.floats
[0.0, 1.5, 2.3, -9.0]
>>> p.ints
[0, 2, 2, -9]
>>> p.bools
[False, True, True, True]
>>> p.assign([False, True, False, True])
>>> p.floats
[0.0, 1.0, 0.0, 1.0]
>>> p = ValueProxy(False)
>>> bool(p)
False
>>> int(p)
0
>>> float(p)
0.0
>>> p.assign(1)
>>> bool(p), int(p), float(p)
(True, 1, 1.0)
>>> p = ValueProxy("Hello world!")                  # Create string-typed register value.
>>> str(p)
'Hello world!'
>>> bytes(p)
b'Hello world!'
>>> p.assign('Another string')
>>> str(p)
'Another string'
>>> bytes(p)
b'Another string'
>>> p = ValueProxy(b"ab01")                         # Create unstructured-typed register value.
>>> str(p)
'ab01'
>>> bytes(p)
b'ab01'
>>> p.assign("String implicitly converted to bytes")
>>> bytes(p)
b'String implicitly converted to bytes'
__init__(v: ValueProxy | Value_1_0 | String_1_0 | Unstructured_1_0 | Bit_1_0 | Integer8_1_0 | Integer16_1_0 | Integer32_1_0 | Integer64_1_0 | Natural8_1_0 | Natural16_1_0 | Natural32_1_0 | Natural64_1_0 | Real16_1_0 | Real32_1_0 | Real64_1_0 | str | bytes | bool | int | float | Iterable[bool] | Iterable[int] | Iterable[float] | ndarray[Any, dtype[Any]]) None[source]

Accepts a wide set of native and generated types. Passing native values is not recommended because the type deduction logic may be changed in the future. To ensure stability, pass only values of uavcan.primitive.*, or Value, or ValueProxy.

>>> list(ValueProxy(Value(natural16=Natural16([123, 456]))).value.natural16.value)  # Explicit Value.
[123, 456]
>>> list(ValueProxy(Natural16([123, 456])).value.natural16.value)                   # Same as above.
[123, 456]
>>> ValueProxy(-123).value.integer64.value[0]               # Integers default to 64-bit.
-123
>>> list(ValueProxy([-1.23, False]).value.real64.value)     # Floats also default to 64-bit.
[-1.23, 0.0]
>>> list(ValueProxy([True, False]).value.bit.value)         # Booleans default to bits.
[True, False]
>>> ValueProxy(b"Hello unstructured!").value.unstructured.value.tobytes()   # Bytes to unstructured.
b'Hello unstructured!'

And so on…

Raises:

ValueConversionError if the conversion is impossible or ambiguous.

property value: Value_1_0[source]

Access to the underlying standard DSDL type uavcan.register.Value.

assign(source: ValueProxy | Value_1_0 | String_1_0 | Unstructured_1_0 | Bit_1_0 | Integer8_1_0 | Integer16_1_0 | Integer32_1_0 | Integer64_1_0 | Natural8_1_0 | Natural16_1_0 | Natural32_1_0 | Natural64_1_0 | Real16_1_0 | Real32_1_0 | Real64_1_0 | str | bytes | bool | int | float | Iterable[bool] | Iterable[int] | Iterable[float] | ndarray[Any, dtype[Any]]) None[source]

Converts the value from the source into the type of the current instance, and updates this instance.

Raises:

ValueConversionError if the source value cannot be converted to the register’s type.

assign_environment_variable(environment_variable_value: str | bytes) None[source]

This is like assign(), but the argument is the value of an environment variable. The conversion rules are documented in the standard RPC-service specification uavcan.register.Access. See also: pycyphal.application.register.get_environment_variable_name().

Parameters:

environment_variable_value – E.g., 1 2 3.

Raises:

ValueConversionError if the value cannot be converted.

property floats: List[float][source]

Converts the value to a list of floats, or raises ValueConversionError if not possible.

property ints: List[int][source]

Converts the value to a list of ints, or raises ValueConversionError if not possible.

property bools: List[bool][source]

Converts the value to a list of bools, or raises ValueConversionError if not possible.

__float__() float[source]

Takes the first item from floats.

__int__() int[source]

Takes the first item from ints.

__bool__() bool[source]

Takes the first item from bools.

__str__() str[source]
__bytes__() bytes[source]
__repr__() str[source]
exception pycyphal.application.register.ValueConversionError[source]

Bases: ValueError

Raised when there is no known conversion between the argument and the specified register.

class pycyphal.application.register.Registry[source]

Bases: MutableMapping[str, ValueProxy]

The registry (register repository) is the main access point for the application to its registers (configuration). It is a facade that provides user-friendly API on top of multiple underlying register backends (see backend.Backend). Observe that it implements MutableMapping.

The user is not expected to instantiate this class manually; instead, it is provided as a member of pycyphal.application.Node, or via pycyphal.application.make_node().

>>> import pycyphal.application
>>> registry = pycyphal.application.make_registry(environment_variables={})

Create static registers (stored in the register file):

>>> from pycyphal.application.register import Natural16, Real32
>>> registry["p.a"] = Natural16([1234])                 # Assign or create.
>>> registry.setdefault("p.b", Real32([12.34]))         # Update or create. 
ValueProxyWithFlags(uavcan.register.Value...(real32=uavcan.primitive.array.Real32...(value=[12.34])),
                    mutable=True,
                    persistent=False)

Create dynamic registers (getter/setter invoked at every access; existing entries overwritten automatically):

>>> registry["d.a"] = lambda: [1.0, 2.0, 3.0]           # Immutable (read-only), deduced type: real64[3].
>>> list(registry["d.a"].value.real64.value)            # Yup, deduced as expected, real64.
[1.0, 2.0, 3.0]
>>> registry["d.a"] = lambda: Real32([1.0, 2.0, 3.0])   # Like above, but now it is "real32[3]".
>>> list(registry["d.a"].value.real32.value)
[1.0, 2.0, 3.0]
>>> d_b = [True, False, True]                   # Suppose we have some internal object.
>>> def set_d_b(v: Value):                      # Define a setter for it.
...     global d_b
...     d_b = ValueProxy(v).bools
>>> registry["d.b"] = (lambda: d_b), set_d_b    # Expose the object via mutable register with deduced type "bit[3]".

Read/write/delete using the same dict-like API:

>>> list(registry)  # Sorted lexicographically per backend. Altering backends affects register ordering.
['p.a', 'p.b', 'd.a', 'd.b']
>>> len(registry)
4
>>> int(registry["p.a"])
1234
>>> registry["p.a"] = 88                        # Automatic type conversion to "natural16[1]" (defined above).
>>> int(registry["p.a"])
88
>>> registry["d.b"].bools
[True, False, True]
>>> registry["d.b"] = [-1, 5, 0.0]              # Automatic type conversion to "bit[3]".
>>> registry["d.b"].bools
[True, True, False]
>>> del registry["*.a"]                         # Use wildcards to remove multiple at the same time.
>>> list(registry)
['p.b', 'd.b']
>>> registry["d.b"].ints                        # Type conversion by ValueProxy.
[1, 1, 0]
>>> registry["d.b"].floats
[1.0, 1.0, 0.0]
>>> registry["d.b"].value.bit                   
uavcan.primitive.array.Bit...(value=[ True, True,False])

Registers created by setdefault() are always initialized from environment variables:

>>> registry.environment_variables["P__C"] = b"999 +888.3"
>>> registry.environment_variables["D__C"] = b"Hello world!"
>>> registry.setdefault("p.c", Natural16([111, 222])).ints  # Value from environment is used here!
[999, 888]
>>> registry.setdefault("p.d", Natural16([111, 222])).ints  # No environment variable for this one.
[111, 222]
>>> d_c = 'Coffee'
>>> def set_d_c(v: Value):
...     global d_c
...     d_c = str(ValueProxy(v))
>>> str(registry.setdefault("d.c", (lambda: d_c, set_d_c))) # Setter is invoked immediately.
'Hello world!'
>>> registry["d.c"] = "New text"                            # Change the value again.
>>> d_c                                                     # Yup, changed.
'New text'
>>> str(registry.setdefault("d.c", lambda: d_c))            # Environment var ignored because no setter.
'New text'

If such behavior is undesirable, one can either clear the environment variable dict or remove specific entries. See also: pycyphal.application.make_node().

Variables created by direct assignment are (obviously) not affected by environment variables:

>>> registry["p.c"] = [111, 222]                            # Direct assignment instead of setdefault().
>>> registry["p.c"].ints                                    # Environment variables ignored!
[111, 222]

Closing the registry will close all underlying backends.

>>> registry.close()

TODO: Add modification notification callbacks to allow applications implement hot reloading.

Assignable

An instance of any type from this union can be used to assign or create a register. Creation is handled depending on the type:

  • If a single callable, it will be invoked whenever this register is read; such register is called “dynamic”. Such register will be reported as immutable. The registry file is not affected and therefore this change is not persistent. environment_variables are always ignored in this case since the register cannot be written. The result of the callable is converted to the register value using ValueProxy.

  • If a tuple of two callables, then the first one is a getter that is invoked on read (see above), and the second is a setter that is invoked on write with a single argument of type Value. It is guaranteed that the type of the value passed into the setter is always the same as that which is returned by the getter. The type conversion is performed automatically by polling the getter beforehand to discover the type. The registry file is not affected and therefore this change is not persistent.

  • Any other type (e.g., Value, Natural16, native, etc.): a static register will be created and stored in the registry file. Conversion logic is implemented by ValueProxy.

Dynamic registers (callables) overwrite existing entries unconditionally. It is not recommended to create dynamic registers with same names as existing static registers, as it may cause erratic behaviors.

alias of Union[ValueProxy, Value_1_0, String_1_0, Unstructured_1_0, Bit_1_0, Integer8_1_0, Integer16_1_0, Integer32_1_0, Integer64_1_0, Natural8_1_0, Natural16_1_0, Natural32_1_0, Natural64_1_0, Real16_1_0, Real32_1_0, Real64_1_0, str, bytes, bool, int, float, Iterable[bool], Iterable[int], Iterable[float], ndarray[Any, dtype[Any]], Callable[[], Union[ValueProxy, Value_1_0, String_1_0, Unstructured_1_0, Bit_1_0, Integer8_1_0, Integer16_1_0, Integer32_1_0, Integer64_1_0, Natural8_1_0, Natural16_1_0, Natural32_1_0, Natural64_1_0, Real16_1_0, Real32_1_0, Real64_1_0, str, bytes, bool, int, float, Iterable[bool], Iterable[int], Iterable[float], ndarray[Any, dtype[Any]]]], Tuple[Callable[[], Union[ValueProxy, Value_1_0, String_1_0, Unstructured_1_0, Bit_1_0, Integer8_1_0, Integer16_1_0, Integer32_1_0, Integer64_1_0, Natural8_1_0, Natural16_1_0, Natural32_1_0, Natural64_1_0, Real16_1_0, Real32_1_0, Real64_1_0, str, bytes, bool, int, float, Iterable[bool], Iterable[int], Iterable[float], ndarray[Any, dtype[Any]]]], Callable[[Value_1_0], None]]]

abstract property backends: Sequence[Backend][source]

If a register exists in more than one registry, only the first copy will be used; however, the count will include all redundant registers.

abstract property environment_variables: Dict[str, bytes][source]

When a new register is created using setdefault(), its default value will be overridden from this dict. This is done to let the registry use values passed over to this node via environment variables or a similar mechanism.

close() None[source]

Closes all storage backends.

index(index: int) str | None[source]

Get register name by index. The ordering is like __iter__(). Returns None if index is out of range.

setdefault(key: str, default: Assignable | None = None) ValueProxyWithFlags[source]

This is the preferred method for creating new registers.

If the register exists, its value will be returned an no further action will be taken. If the register doesn’t exist, it will be created and immediately updated from environment_variables (using ValueProxy.assign_environment_variable()). The register value instance is created using ValueProxy.

Parameters:
  • key – Register name.

  • default – If exists, this value is ignored; otherwise created as described in Assignable.

Returns:

Resulting value.

Raises:

See ValueProxy.assign_environment_variable() and ValueProxy().

__getitem__(name: str) ValueProxyWithFlags[source]
Returns:

ValueProxyWithFlags (ValueProxy) if exists.

Raises:

MissingRegisterError (KeyError) if no such register.

__setitem__(name: str, value: Assignable) None[source]

Assign a new value to the register if it exists and the type of the value is matching or can be converted to the register’s type. The mutability flag may be ignored depending on which backend the register is stored at. The conversion is implemented by ValueProxy.assign().

If the register does not exist, a new one will be created. However, unlike setdefault(), ValueProxy.assign_environment_variable() is not invoked. The register value instance is created using ValueProxy.

Raises:

ValueConversionError if the register exists but the value cannot be converted to its type or (in case of creation) the environment variable contains an invalid value.

__delitem__(wildcard: str) None[source]

Remove registers that match the specified wildcard from all backends. Matching is case-sensitive. Count and keys are invalidated. If no matching keys are found, no exception is raised.

__iter__() Iterator[str][source]

Iterator over register names. They may not be unique if different backends redefine the same register! The ordering is defined by backend ordering, then lexicographically.

__len__() int[source]

Number of registers in all backends.

__repr__() str[source]
class pycyphal.application.register.ValueProxyWithFlags(msg: Value_1_0, mutable: bool, persistent: bool)[source]

Bases: ValueProxy

This is like ValueProxy but extended with register flags.

__init__(msg: Value_1_0, mutable: bool, persistent: bool) None[source]
property mutable: bool[source]
property persistent: bool[source]
__repr__() str[source]
exception pycyphal.application.register.MissingRegisterError[source]

Bases: KeyError

Raised when the user attempts to access a register that is not defined.

pycyphal.application.register.get_environment_variable_name(register_name: str) str[source]

Convert the name of the register to the name of the environment variable that assigns it.

>>> get_environment_variable_name("m.motor.inductance_dq")
'M__MOTOR__INDUCTANCE_DQ'