pycyphal.transport.loopback package

Module contents

class pycyphal.transport.loopback.LoopbackTransport(local_node_id: int | None, *, allow_anonymous_transfers: bool = True, loop: AbstractEventLoop | None = None)[source]

Bases: Transport

The loopback transport is intended for basic testing and API usage demonstrations. It works by short-circuiting input and output sessions together as if there was an underlying network.

It is not possible to exchange data between different nodes using this transport. The only valid usage is sending and receiving same data on the same node.

__init__(local_node_id: int | None, *, allow_anonymous_transfers: bool = True, loop: AbstractEventLoop | None = None)[source]

property protocol_parameters: ProtocolParameters[source]

property local_node_id: int | None[source]

property spoof_result: bool | Exception[source]: Test rigging. If True, spoof() will always succeed (this is the default). If False, it will always time out. If Exception, it will be raised.

property send_delay: float[source]: Test rigging. If positive, this delay will be inserted for each sent transfer. If after the delay the transfer deadline is in the past, it is assumed to have timed out. Zero by default (no delay is inserted, deadline not checked).

close() → None[source]

get_input_session(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata) → LoopbackInputSession[source]

get_output_session(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata) → LoopbackOutputSession[source]

sample_statistics() → pycyphal.transport.loopback.LoopbackTransportStatistics[source]

property input_sessions: Sequence[LoopbackInputSession][source]

property output_sessions: Sequence[LoopbackOutputSession][source]

begin_capture(handler: Callable[[Capture], None]) → None[source]

property capture_active: bool[source]

static make_tracer() → LoopbackTracer[source]: See LoopbackTracer.

async spoof(transfer: AlienTransfer, monotonic_deadline: float) → bool[source]: Spoofed transfers can be observed using begin_capture(). Also see spoof_result.

property capture_handlers: Sequence[Callable[[Capture], None]][source]

class pycyphal.transport.loopback.LoopbackInputSession(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, closer: Callable[[], None])[source]

Bases: InputSession

DEFAULT_TRANSFER_ID_TIMEOUT = 2

__init__(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, closer: Callable[[], None])[source]

async receive(monotonic_deadline: float) → TransferFrom | None[source]

async push(transfer: TransferFrom) → None[source]: Inserts a transfer into the receive queue of this loopback session.

property transfer_id_timeout: float[source]

property specifier: InputSessionSpecifier[source]

property payload_metadata: PayloadMetadata[source]

sample_statistics() → SessionStatistics[source]

close() → None[source]

class pycyphal.transport.loopback.LoopbackOutputSession(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata, closer: Callable[[], None], router: Callable[[Transfer, float], Awaitable[bool]])[source]

Bases: OutputSession

__init__(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata, closer: Callable[[], None], router: Callable[[Transfer, float], Awaitable[bool]])[source]

enable_feedback(handler: Callable[[Feedback], None]) → None[source]

disable_feedback() → None[source]

async send(transfer: Transfer, monotonic_deadline: float) → bool[source]

property specifier: OutputSessionSpecifier[source]

property payload_metadata: PayloadMetadata[source]

sample_statistics() → SessionStatistics[source]

close() → None[source]

property exception: Exception | None[source]: This is a test rigging. Use this property to configure an exception object that will be raised when send() is invoked. Set None to remove the injected exception (None is the default value). Useful for testing error handling logic.