6. Other noteworthy APIs

class trainlib_async.messages.SnapCommand(iterable=(), /)

Snap command is a tuple of 4 colors (SnapColorValue).

It always starts with white or cyan. If the snap sequence is shorter, it is padded with black. Since white and cyan always start a new snap command, these colors will never appear on other position.

__eq__(o: object) → bool

Compare (exact match) the snap command to a tuple of colors.

Since snap commands are always 4 colors, this allows to exact match with shorter tuples.

Example

Note: SnapColorValue is imported as C in this example.

>>> msg.colors == (C.WHITE, C.RED, C.BLACK, C.BLACK)
>>> # exact match with implicit blacks
>>> msg.colors == (C.WHITE, C.RED)

start_with(colors: Tuple[SnapColorValue, ...] | SnapColorValue, *args: SnapColorValue)

Partial / prefix match.

This function is aliased as start_with() and starts_with() to be used depending on the context.

Example

Note: SnapColorValue is imported as C in this example.

>>> snap_command.starts_with(C.WHITE, C.RED)
>>> msg.colors.start_with(C.CYAN, C.BLUE)

Parameters:: colors (SnapColorValue) – It accepts either tuples or multiple parameters of SnapColorValue.

starts_with(colors: Tuple[SnapColorValue, ...] | SnapColorValue, *args: SnapColorValue)

Partial / prefix match.

This function is aliased as start_with() and starts_with() to be used depending on the context.

Example

Note: SnapColorValue is imported as C in this example.

>>> snap_command.starts_with(C.WHITE, C.RED)
>>> msg.colors.start_with(C.CYAN, C.BLUE)

Parameters:: colors (SnapColorValue) – It accepts either tuples or multiple parameters of SnapColorValue.

class trainlib_async.helpers.AsyncObserver(on_next: Callable[[Any], None] | Callable[[_T], Coroutine[Any, Any, None]] | None = None, on_error: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None, on_completed: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]] | None = None): Observer implementation that accepts async functions as callbacks wrapping them in asyncio.create_task.