runtimepy.net.udp package

Subpackages

• runtimepy.net.udp.tftp package
  • Submodules
  • runtimepy.net.udp.tftp.base module
    • BaseTftpConnection
      • BaseTftpConnection.default_auto_restart
      • BaseTftpConnection.endpoint()
      • BaseTftpConnection.init()
      • BaseTftpConnection.log_alias
      • BaseTftpConnection.path
      • BaseTftpConnection.process_datagram()
      • BaseTftpConnection.send_ack()
      • BaseTftpConnection.send_data()
      • BaseTftpConnection.send_error()
      • BaseTftpConnection.send_rrq()
      • BaseTftpConnection.send_wrq()
      • BaseTftpConnection.set_root()
      • BaseTftpConnection.should_connect
  • runtimepy.net.udp.tftp.endpoint module
    • TftpEndpoint
      • TftpEndpoint.chunk_sender()
      • TftpEndpoint.get_path()
      • TftpEndpoint.handle_ack()
      • TftpEndpoint.handle_data()
      • TftpEndpoint.handle_error()
      • TftpEndpoint.handle_read_request()
      • TftpEndpoint.handle_write_request()
      • TftpEndpoint.ingest_file()
      • TftpEndpoint.serve_file()
      • TftpEndpoint.set_root()
      • TftpEndpoint.update_from_other()
  • runtimepy.net.udp.tftp.enums module
    • TftpErrorCode
      • TftpErrorCode.ACCESS_VIOLATION
      • TftpErrorCode.DISK_FULL
      • TftpErrorCode.FILE_EXISTS
      • TftpErrorCode.FILE_NOT_FOUND
      • TftpErrorCode.ILLEGAL_OPERATION
      • TftpErrorCode.NO_USER
      • TftpErrorCode.OPTION_NEGOTIATION
      • TftpErrorCode.UNKNOWN
      • TftpErrorCode.UNKNOWN_ID
    • TftpOpCode
      • TftpOpCode.ACK
      • TftpOpCode.DATA
      • TftpOpCode.ERROR
      • TftpOpCode.INVALID
      • TftpOpCode.RRQ
      • TftpOpCode.WRQ
    • encode_filename_mode()
    • parse_filename_mode()
  • runtimepy.net.udp.tftp.io module
    • tftp_chunks()
  • Module contents
    • TftpConnection
      • TftpConnection.request_read()
      • TftpConnection.request_write()
    • tftp()
    • tftp_read()
    • tftp_write()

Submodules

runtimepy.net.udp.connection module

A module implementing a UDP connection interface.

class runtimepy.net.udp.connection.EchoUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection, EchoConnection

An echo connection for UDP.

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.connection.NullUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection, NullConnection

A null UDP connection.

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.connection.UdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: Connection, TransportMixin

A UDP connection interface.

async close() → None

Close this connection.

async classmethod create_connection(markdown: str = None, **kwargs) → T

Create a UDP connection.

async classmethod create_pair() → tuple[T, T]

Create a connection pair.

env: ChannelEnvironment

latest_rx_address: tuple[str, int] | None

log_alias = 'UDP'

logger: LoggerType

markdown: str

abstractmethod async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message to send.

send_text(data: str) → None

Enqueue a text message to send.

sendto(data: bytes | bytearray | memoryview, addr: IPv4Host | IPv6Host | tuple[str, int] | tuple[str, int, int, int] = None) → None

Send to a specific address.

set_remote_address(addr: IPv4Host | IPv6Host) → None

Set a new remote address. Note that this doesn’t interact with or attempt to ‘connect’ the underlying socket. That should be done at creation time.

should_connect: bool = True

uses_binary_tx_queue = False

uses_text_tx_queue = False

runtimepy.net.udp.create module

A module for instantiating the underlying networking resources for UdpConnection.

async runtimepy.net.udp.create.try_udp_transport_protocol(callback: Callable[[tuple[DatagramTransport, UdpQueueProtocol]], None] = None, **kwargs) → tuple[DatagramTransport, UdpQueueProtocol] | None

Attempt to create a transport and protocol pair.

async runtimepy.net.udp.create.udp_transport_protocol(**kwargs) → tuple[DatagramTransport, UdpQueueProtocol]

Create a transport and protocol pair relevant for this class’s implementation.

async runtimepy.net.udp.create.udp_transport_protocol_backoff(backoff: ExponentialBackoff = None, **kwargs) → tuple[DatagramTransport, UdpQueueProtocol]

Create a transport and protocol pair relevant for this class’s implementation.

runtimepy.net.udp.protocol module

A module implementing a DatagramProtocol for UdpConnection.

class runtimepy.net.udp.protocol.UdpQueueProtocol

Bases: DatagramProtocol

A simple UDP protocol that populates a message queue.

conn: Connection

datagram_received(data: bytes, addr: tuple[str, int]) → None

Handle incoming data.

error_received(exc: Exception) → None

Log any received errors.

logger: Logger | LoggerAdapter[Any]

runtimepy.net.udp.queue module

A module implementing a simple queue-based UDP interface.

class runtimepy.net.udp.queue.QueueUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection

An echo connection for UDP.

datagrams: Queue[tuple[bytes | bytearray | memoryview, tuple[str, int]]]

init() → None

Initialize this instance.

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

Module contents

A module implementing networking utilities relevant to UDP.

class runtimepy.net.udp.EchoUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection, EchoConnection

An echo connection for UDP.

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.NullUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection, NullConnection

A null UDP connection.

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.QueueUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection

An echo connection for UDP.

datagrams: Queue[tuple[bytes | bytearray | memoryview, tuple[str, int]]]

env: ChannelEnvironment

init() → None

Initialize this instance.

latest_rx_address: _Optional[tuple[str, int]]

logger: LoggerType

markdown: str

async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

class runtimepy.net.udp.UdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: Connection, TransportMixin

A UDP connection interface.

async close() → None

Close this connection.

async classmethod create_connection(markdown: str = None, **kwargs) → T

Create a UDP connection.

async classmethod create_pair() → tuple[T, T]

Create a connection pair.

env: ChannelEnvironment

latest_rx_address: tuple[str, int] | None

log_alias = 'UDP'

logger: LoggerType

markdown: str

abstractmethod async process_datagram(data: bytes | bytearray | memoryview, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message to send.

send_text(data: str) → None

Enqueue a text message to send.

sendto(data: bytes | bytearray | memoryview, addr: IPv4Host | IPv6Host | tuple[str, int] | tuple[str, int, int, int] = None) → None

Send to a specific address.

set_remote_address(addr: IPv4Host | IPv6Host) → None

Set a new remote address. Note that this doesn’t interact with or attempt to ‘connect’ the underlying socket. That should be done at creation time.

should_connect: bool = True

uses_binary_tx_queue = False

uses_text_tx_queue = False