REDAC Low-Level Abstractions

Transport Layer

class pyanabrid.base.transport.BaseTransport(reader: StreamReader, writer: StreamWriter, name: str = None)

Abstract base class for transports.

Transports are based on asyncio.StreamReader and asyncio.StreamWriter objects.

close()

Close the underlying asyncio.StreamWriter.

async receive_line(timeout=1) → bytes

Receive one line of data from the transport.

async send_line(data: bytes) → None

Send one line of data over the transport. Newline character ‘ ‘ is appended automatically.

class pyanabrid.base.transport.PassthroughTransport(reader: StreamReader, writer: StreamWriter, name: str = None)

A utility passthrough transport for testing purposes.

Use it to pass a reader and a writer from your code into anything that uses transports.

async classmethod create(reader: StreamReader, writer: StreamWriter, **kwargs) → PassthroughTransport

Create a new PassthroughTransport instance.

Parameters:

  • reader – A asyncio.StreamReader instance.

  • writer – A asyncio.StreamWriter instance.

  • kwargs – Keyword arguments are passed on to BaseTransport.

Returns:

A PassthroughTransport instance.

class pyanabrid.base.transport.SerialTransport(reader: StreamReader, writer: StreamWriter, name: str = None)

A transport abstraction for serial connections.

async classmethod create(device, baudrate, /, **kwargs)

Create a new SerialTransport instance.

Parameters:

  • device – The serial device’s path.

  • baudrate – The baudrate to use.

  • kwargs – Keyword arguments are passed on to BaseTransport.

Returns:

A SerialTransport instance.

class pyanabrid.base.transport.TCPTransport(reader: StreamReader, writer: StreamWriter, name: str = None)

A TCP/IP transport implementation for network communication.

async classmethod create(host, port, /, **kwargs)

Create a new TCPTransport instance for communicating over network.

Parameters:

  • host – Target hostname or IP address.

  • port – Target network port.

  • kwargs – Keyword arguments are passed on to BaseTransport.

Returns:

A TCPTransport instance.

class pyanabrid.base.transport.UnixSocketTransport(reader: StreamReader, writer: StreamWriter, name: str = None)

Transport based on unix sockets, mostly for testing purposes.

async classmethod create(path: Optional[Union[str, PathLike]], **kwargs) → UnixSocketTransport

Create a new UnixSocketTransport.

Parameters:

  • path – Path which should be used for the socket. A temporary file is created if None.

  • kwargs – Keyword arguments are passed on to BaseTransport.

Returns:

A UnixSocketTransport instance