synchros2.callables module

class synchros2.callables.ComposableCallable

Bases: GeneralizedCallable

A generalized callable that can be composed.

compose(func: Callable, starred: bool = False) → ComposedCallable

Compose this callable with the given func.

Parameters:

• func – callable to be composed, assumed synchronous.

• starred – whether the func call return value requires

• not. (a star expansion for composition or)

Returns:

the composed generalized callable.

class synchros2.callables.ComposedCallable

Bases: GeneralizedDecorator, ComposableCallable, VectorizingCallable

A composition decorator that combines a generalized callable and a regular callable.

__init__(wrapped_callable: GeneralizedCallable, composed_callable: Callable, starred: bool = False) → None

Initializes composed callable.

Parameters:

• wrapped_callable – the left hand side (or outer) callable.

• composed_callable – the right hand side (or inner) callable.

• starred – whether the inner call return value requires a star

• not. (expansion for composition or)

asynchronous(*args: Any, **kwargs: Any) → Any

Invoke callable asynchronously with the result of invoking the composed callable.

Keyword arguments that match the composed callable signature will be forwarded to it.

synchronous(*args: Any, **kwargs: Any) → Any

Invoke callable synchronously with the result of invoking the composed callable.

Keyword arguments that match the composed callable signature will be forwarded to it.

class synchros2.callables.GeneralizedCallable

Bases: ABC

A generalized callable that allows synchronous and asynchronous execution.

abstractmethod asynchronous(*args: Any, **kwargs: Any) → Any

Invoke callable asynchronously, returning a future-like object.

asynchronously(*args: Any, **kwargs: Any) → Any

Alias of GeneralizedCallable.asynchronous.

abstractmethod synchronous(*args: Any, **kwargs: Any) → Any

Invoke callable synchronously (ie. potentially blocking).

synchronously(*args: Any, **kwargs: Any) → Any

Alias of GeneralizedCallable.synchronous.

class synchros2.callables.GeneralizedDecorator

Bases: GeneralizedCallable

A decorator for generalized callables.

__init__(wrapped_callable: GeneralizedCallable) → None

Initializes the decorator, wrapping another callable.

class synchros2.callables.GeneralizedFunction

Bases: GeneralizedCallable

A generalized callable defined by parts.

__init__(synchronous_callable: Callable | None = None, asynchronous_callable: Callable | None = None) → None

Initialize generalized function.

Parameters:

• synchronous_callable – optional synchronous body.

• asynchronous_callable – optional asynchronous body.

asynchronous(*args: Any, **kwargs: Any) → Any

Invoke function asynchronously, returning a future-like object.

synchronous(*args: Any, **kwargs: Any) → Any

Invoke function synchronously (ie. potentially blocking).

class synchros2.callables.GeneralizedGuard

Bases: GeneralizedDecorator

A decorator that guards generalized callable invocations.

__init__(condition: Callable[[], bool], wrapped_callable: GeneralizedCallable, message: str | None = None) → None

Initializes generalized guard.

Parameters:

• condition – boolean predicate to guard invocations.

• wrapped_callable – the guarded callable.

• message – optional human-readable message to raise whenever

• hold. (the guard condition does not)

asynchronous(*args: Any, **kwargs: Any) → Any

Invokes callable asynchronously if the guarded condition holds true, raises otherwise.

synchronous(*args: Any, **kwargs: Any) → Any

Invokes callable synchronously if the guarded condition holds true, raises otherwise.

class synchros2.callables.GeneralizedMethod

Bases: object

A data descriptor for generalized callables bound to class instances.

class Bound

Bases: VectorizingCallable, ComposableCallable

A bound generalized method callable.

__init__(body: GeneralizedCallable, default_callable: Callable | None = None) → None

Initialize bound method callable.

Parameters:

• body – method body as a generalized callable

• default_callable – optionally override default plain calls, defaults to synchronous calls.

asynchronous(*args: Any, **kwargs: Any) → Any

Invoke method asynchronously.

synchronous(*args: Any, **kwargs: Any) → Any

Invoke method synchronously.

class Unbound

Bases: object

An unbound generalized method descriptor.

__init__(method: GeneralizedMethod) → None

Initialize unbound descriptor.

Parameters:

method – associated generalized method.

__init__(prototype: Callable, transitional: bool) → None

Initializes the generalized method.

Parameters:

• prototype – method prototype, usually just a signature but

• the (simplifying)

• overload). (function type matches the missing)

• transitional – a transitional method will stick to its

• invocations (prototype for default)

• the

• codebases. (adoption of generalized methods in existing)

async_overload(func: Callable) → Callable

Register func as this method asynchronous overload.

asynchronous(func: Callable) → Callable

Register func as this method asynchronous overload.

legacy(func: Callable) → Callable

Register func as this method legacy overload.

rebind(instance: Any, body: GeneralizedCallable) → None

Change this method’s body for the given instance.

sync_overload(func: Callable) → Callable

Register func as this method synchronous overload.

synchronous(func: Callable) → Callable

Register func as this method synchronous overload.

property transitional: bool

Check whether this method is transitional or not.

class synchros2.callables.GeneralizedMethodLike

Bases: Generic[P], GeneralizedMethod

A generalized method that can be type annotated via user-defined protocols.

class synchros2.callables.VectorizedCallable

Bases: GeneralizedDecorator, ComposableCallable, VectorizingCallable

A vectorization decorator that aggregates multiple invocations sequentially.

asynchronous(*vargs: Iterable[Any], **kwargs: Any) → rclpy.task.Future

Invoke callable asynchronously over a sequence of a argument tuples (zipped).

A future to a sequence of results is returned.

synchronous(*vargs: Iterable[Any], **kwargs: Any) → List

Invoke callable synchronously over a sequence of a argument tuples (zipped).

A sequence of results is returned.

class synchros2.callables.VectorizingCallable

Bases: GeneralizedCallable

A generalized callable that can be vectorized.

property vectorized: VectorizedCallable

Get a vectorized version of this callable.

synchros2.callables.generalized_method(func: Callable, *, spec: Type[GeneralizedMethodLike[P]]) → GeneralizedMethodLike[P]

synchros2.callables.generalized_method(func: Literal[None] = None, *, spec: Type[GeneralizedMethodLike[P]], transitional: bool = False) → Callable[[Callable], GeneralizedMethodLike[P]]

synchros2.callables.generalized_method(func: Callable, *, transitional: bool = False) → GeneralizedMethod

synchros2.callables.generalized_method(func: Literal[None] = None, *, transitional: bool = False) → Callable[[Callable], GeneralizedMethod]

Define a generalized method by decoration.

Parameters:

• func – method function, usually just a signature but

• convenience. (may also be used as an overload for)

• spec – optional type annotated specification.

• transitional – a transitional method will stick to its

• invocations. (prototype for default)

synchros2.callables.starmap_async(func: Callable[[...], FutureLike | FutureConvertible], iterable: Iterable[Tuple[Any, ...]]) → rclpy.task.Future

Transform an iterable of tuples by star expanding them and invoking an asynchronous func.

Parameters:

• func – a callable with a future-like return value.

• iterable – an iterable of tuples of positional arguments for func.

Returns:

a future-like object to the transformed iterable.