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]]] [source] 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 [source]
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
]] [source] 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
]] [source] 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 [source]
A simple helper function that constructs a
repr()
form of an object. Used widely across the library. String representations will be obtained by invokingstr()
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 [source]
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>'