pycyphal.util package

Module contents

The util package contains various entities that are commonly useful in PyCyphal-based applications.

pycyphal.util.broadcast(functions: Iterable[Callable[[...], pycyphal.util.R]]) → Callable[[...], List[Union[pycyphal.util.R, Exception]]]

Returns a function that invokes each supplied function in series with the specified arguments following the specified order. If a function is executed successfully, its result is added to the output list. If it raises an exception, the exception is suppressed, logged, and added to the output list instead of the result.

This function is mostly intended for invoking various handlers.

>>> def add(a, b):
...     return a + b
>>> def fail(a, b):
...     raise ValueError(f'Arguments: {a}, {b}')
>>> broadcast([add, fail])(4, b=5)
[9, ValueError('Arguments: 4, 5')]
>>> broadcast([print])('Hello', 'world!')
Hello world!
[None]
>>> broadcast([])()
[]

pycyphal.util.import_submodules(root_module: module, error_handler: Optional[Callable[[str, ImportError], None]] = None) → None

Recursively imports all submodules and subpackages of the specified Python module or package. This is mostly intended for automatic import of all available specialized implementations of a certain functionality when they are spread out through several submodules which are not auto-imported.

Parameters

• root_module – The module to start the recursive descent from.

• error_handler –

  If None (default), any ImportError is raised normally, thereby terminating the import process after the first import error (e.g., a missing dependency). Otherwise, this would be a function that is invoked whenever an import error is encountered instead of raising the exception. The arguments are:

  • the name of the parent module whose import could not be completed due to the error;

  • the culprit of type ImportError.

>>> import pycyphal
>>> pycyphal.util.import_submodules(pycyphal.transport)  # One missing dependency would fail everything.
>>> pycyphal.transport.loopback.LoopbackTransport
<class 'pycyphal.transport.loopback...LoopbackTransport'>

>>> import tests.util.import_error  # For demo purposes, this package contains a missing import.
>>> pycyphal.util.import_submodules(tests.util.import_error)  # Yup, it fails.
Traceback (most recent call last):
  ...
ModuleNotFoundError: No module named 'nonexistent_module_should_raise_import_error'
>>> pycyphal.util.import_submodules(tests.util.import_error,  # The handler allows us to ignore ImportError.
...                                 lambda parent, ex: print(parent, ex.name))
tests.util.import_error._subpackage nonexistent_module_should_raise_import_error

pycyphal.util.iter_descendants(ty: Type[pycyphal.util.T]) → Iterable[Type[pycyphal.util.T]]

Returns a recursively descending iterator over all subclasses of the argument.

>>> class A: pass
>>> class B(A): pass
>>> class C(B): pass
>>> class D(A): pass
>>> set(iter_descendants(A)) == {B, C, D}
True
>>> list(iter_descendants(D))
[]
>>> bool in set(iter_descendants(int))
True

Practical example – discovering what transports are available:

>>> import pycyphal
>>> pycyphal.util.import_submodules(pycyphal.transport)
>>> list(sorted(map(lambda t: t.__name__, pycyphal.util.iter_descendants(pycyphal.transport.Transport))))
[...'CANTransport'...'RedundantTransport'...'SerialTransport'...]

pycyphal.util.mark_last(it: Iterable[pycyphal.util.T]) → Iterable[Tuple[bool, pycyphal.util.T]]

This is an iteration helper like enumerate(). It amends every item with a boolean flag which is False for all items except the last one. If the input iterable is empty, yields nothing.

>>> list(mark_last([]))
[]
>>> list(mark_last([123]))
[(True, 123)]
>>> list(mark_last([123, 456]))
[(False, 123), (True, 456)]
>>> list(mark_last([123, 456, 789]))
[(False, 123), (False, 456), (True, 789)]

pycyphal.util.repr_attributes(obj: object, *anonymous_elements: object, **named_elements: object) → str

A simple helper function that constructs a repr() form of an object. Used widely across the library. String representations will be obtained by invoking str() on each value.

>>> class Aa: pass
>>> assert repr_attributes(Aa()) == 'Aa()'
>>> assert repr_attributes(Aa(), 123) == 'Aa(123)'
>>> assert repr_attributes(Aa(), foo=123) == 'Aa(foo=123)'
>>> assert repr_attributes(Aa(), 456, foo=123, bar=repr('abc')) == "Aa(456, foo=123, bar='abc')"

pycyphal.util.repr_attributes_noexcept(obj: object, *anonymous_elements: object, **named_elements: object) → str

A robust version of repr_attributes() that never raises exceptions.

>>> class Aa: pass
>>> repr_attributes_noexcept(Aa(), 456, foo=123, bar=repr('abc'))
"Aa(456, foo=123, bar='abc')"
>>> class Bb:
...     def __repr__(self) -> str:
...         raise Exception('Ford, you are turning into a penguin')
>>> repr_attributes_noexcept(Aa(), foo=Bb())
"<REPR FAILED: Exception('Ford, you are turning into a penguin')>"
>>> class Cc(Exception):
...     def __str__(self) -> str:  raise Cc()  # Infinite recursion
...     def __repr__(self) -> str: raise Cc()  # Infinite recursion
>>> repr_attributes_noexcept(Aa(), foo=Cc())
'<REPR FAILED: UNKNOWN ERROR>'