pycyphal.transport.udp package

Module contents

Cyphal/UDP transport overview

Please refer to the appropriate section of the Cyphal Specification for the definition of the Cyphal/UDP transport.

This transport module contains no media sublayers because the media abstraction is handled directly by the standard UDP/IP stack of the underlying operating system.

Forward error correction (FEC)

For unreliable networks, optional forward error correction (FEC) is supported by this implementation. This measure is only available for service transfers, not for message transfers due to their different semantics. If the probability of a frame loss exceeds the desired reliability threshold, the transport can be configured to repeat every outgoing service transfer a specified number of times, on the assumption that the probability of losing any given frame is uncorrelated (or weakly correlated) with that of its neighbors. Assuming that the probability of transfer loss P is time-invariant, the influence of the FEC multiplier M can be approximated as P' = P^M.

Duplicates are emitted immediately following the original transfer. For example, suppose that a service transfer contains three frames, F0 to F2, and the service transfer multiplication factor is two, then the resulting frame sequence would be as follows:

F0      F1      F2      F0      F1      F2
\_______________/       \_______________/
   main copy             redundant copy
 (TX timestamp)      (never TX-timestamped)

------------------ time ------------------>

As shown on the diagram, if the transmission timestamping is requested, only the first copy is timestamped. Further, any errors occurring during the transmission of redundant copies may be silently ignored by the stack, provided that the main copy is transmitted successfully.

The resulting behavior in the provided example is that the transport network may lose up to three unique frames without affecting the application. In the following example, the frames F0 and F2 of the main copy are lost, but the transfer survives:

F0 F1 F2 F0 F1 F2
|  |  |  |  |  |
x  |  x  |  |  \_____ F2 __________________________
   |     |  \________ F1 (redundant, discarded) x  \
   |     \___________ F0 ________________________  |
   \_________________ F1 ______________________  \ |
                                               \ | |
----- time ----->                              v v v
                                            reassembled
                                            multi-frame
                                             transfer

Removal of duplicate transfers at the opposite end of the link is natively guaranteed by the Cyphal protocol; no special activities are needed there (refer to the Cyphal Specification for background).

For time-deterministic (real-time) networks this strategy is preferred over the conventional confirmation-retry approach (e.g., the TCP model) because it results in more predictable network load, lower worst-case latency, and is stateless (participants do not make assumptions about the state of other agents involved in data exchange).

Usage

Create two transport instances – one with a node-ID, one anonymous:

>>> import asyncio
>>> import pycyphal
>>> import pycyphal.transport.udp
>>> tr_0 = pycyphal.transport.udp.UDPTransport(local_ip_address='127.0.0.1', local_node_id=10)
>>> tr_0.local_ip_address
IPv4Address('127.0.0.1')
>>> tr_0.local_node_id
10
>>> tr_1 = pycyphal.transport.udp.UDPTransport(local_ip_address='127.0.0.1',
...                                            local_node_id=None) # Anonymous is only for listening.
>>> tr_1.local_node_id is None
True

Create an output and an input session:

>>> pm = pycyphal.transport.PayloadMetadata(1024)
>>> ds = pycyphal.transport.MessageDataSpecifier(42)
>>> pub = tr_0.get_output_session(pycyphal.transport.OutputSessionSpecifier(ds, None), pm)
>>> pub.socket.getpeername()   # UDP port is fixed, and the multicast group address is computed as shown above.
('239.0.0.42', 9382)
>>> sub = tr_1.get_input_session(pycyphal.transport.InputSessionSpecifier(ds, None), pm)

Send a transfer from one instance to the other:

>>> doctest_await(pub.send(pycyphal.transport.Transfer(pycyphal.transport.Timestamp.now(),
...                                                    pycyphal.transport.Priority.LOW,
...                                                    transfer_id=1111,
...                                                    fragmented_payload=[]),
...                        asyncio.get_event_loop().time() + 1.0))
True
>>> doctest_await(sub.receive(asyncio.get_event_loop().time() + 1.0))
TransferFrom(..., transfer_id=1111, ...)
>>> tr_0.close()
>>> tr_1.close()

Tooling

Run Cyphal networks on the local loopback interface (127.0.0.1) or create virtual interfaces for testing.

Use Wireshark for monitoring and inspection.

Use netcat for trivial monitoring; e.g., listen to a UDP port like this: nc -ul 48469.

List all open UDP ports on the local machine: netstat -vpaun (GNU/Linux).

Inheritance diagram

class pycyphal.transport.udp.UDPTransport(local_ip_address: IPv4Address | IPv6Address | str, local_node_id: int | None = 0, *, mtu: int = 1408, service_transfer_multiplier: int = 1, loop: AbstractEventLoop | None = None, anonymous: bool = False)

Bases: Transport

The Cyphal/UDP (IP v4/v6) transport is designed for low-latency, high-throughput, high-reliability vehicular networks based on Ethernet. Please read the module documentation for details.

TRANSFER_ID_MODULO = 18446744073709551616

MTU_MIN = 4

This is the application-level MTU, not including the Cyphal/UDP header and other overheads.

The Cyphal/UDP protocol does not limit the maximum MTU value, but the minimum is restricted to 4 bytes because it is necessary provide space at least for the transfer-CRC.

A conventional Ethernet jumbo frame can carry up to 9 KiB (9216 bytes).

MTU_DEFAULT = 1408

This is the application-level MTU, not including the Cyphal/UDP header and other overheads. The value derived as:

1500B Ethernet MTU (RFC 894) - 60B IPv4 max header - 8B UDP Header - 24B Cyphal header = 1408B payload.

VALID_SERVICE_TRANSFER_MULTIPLIER_RANGE = (1, 5)

__init__(local_ip_address: IPv4Address | IPv6Address | str, local_node_id: int | None = 0, *, mtu: int = 1408, service_transfer_multiplier: int = 1, loop: AbstractEventLoop | None = None, anonymous: bool = False)

Parameters:

• local_ip_address –

  Specifies which local network interface to use for this transport.

  Using INADDR_ANY here (i.e., 0.0.0.0 for IPv4) is not expected to work reliably or be portable because this configuration is, generally, incompatible with multicast sockets (even in the anonymous mode). In order to set up even a listening multicast socket, it is necessary to specify the correct local address such that the underlying IP stack is aware of which interface to receive multicast packets from.

  When the anonymous mode is enabled, it is quite possible to snoop on the network even if there is another node running locally on the same interface (because sockets are initialized with SO_REUSEADDR and SO_REUSEPORT, when available).

• local_node_id –

  As explained previously, the node-ID is part of the UDP Frame.

  • If the value is None, an anonymous instance will be constructed. Emitted UDP frames will then report its source_node_id as None.

  • If the value is a non-negative integer, then we can setup both input and output sessions.

• mtu – The application-level MTU for outgoing packets. In other words, this is the maximum number of serialized bytes per Cyphal/UDP frame. Transfers where the number of payload bytes does not exceed this value minus 4 bytes for the CRC will be single-frame transfers; otherwise, multi-frame transfers will be used. This setting affects only outgoing frames; incoming frames of any MTU are always accepted.

• service_transfer_multiplier – Forward error correction is disabled by default. This parameter specifies the number of times each outgoing service transfer will be repeated. This setting does not affect message transfers.

• loop – Deprecated.

• anonymous – DEPRECATED and scheduled for removal; replace with local_node_id=None.

property protocol_parameters: ProtocolParameters

property local_node_id: int | None

close() → None

get_input_session(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata) → UDPInputSession

get_output_session(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata) → UDPOutputSession

sample_statistics() → UDPTransportStatistics

property input_sessions: Sequence[UDPInputSession]

property output_sessions: Sequence[UDPOutputSession]

property local_ip_address: IPv4Address | IPv6Address

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

Reported events are of type UDPCapture.

In order for the network capture to work, the local machine should be connected to a SPAN port of the switch. See https://en.wikipedia.org/wiki/Port_mirroring and read the documentation for your networking hardware. Additional preconditions must be met depending on the platform:

• On GNU/Linux, network capture requires that either the process is executed by root, or the raw packet capture capability CAP_NET_RAW is enabled. For more info read man 7 capabilities and consider checking the docs for Wireshark/libpcap.

• On Windows, Npcap needs to be installed and configured; see https://nmap.org/npcap/.

Packets that do not originate from the current Cyphal/UDP subnet (configured on this transport instance) are not reported via this interface. This restriction is critical because there may be other Cyphal/UDP networks running on the same physical L2 network segregated by different subnets, so that if foreign packets were not dropped, conflicts would occur.

property capture_active: bool

static make_tracer() → UDPTracer

See UDPTracer.

async spoof(transfer: AlienTransfer, monotonic_deadline: float) → bool

Not implemented yet. Always raises NotImplementedError. When implemented, this method will rely on libpcap to emit spoofed link-layer packets.

class pycyphal.transport.udp.UDPTransportStatistics(received_datagrams: 'typing.Dict[pycyphal.transport.InputSessionSpecifier, UDPInputSessionStatistics]' = <factory>)

Bases: TransportStatistics

received_datagrams: Dict[InputSessionSpecifier, UDPInputSessionStatistics]

Basic input session statistics: instances of UDPInputSessionStatistics keyed by their data specifier.

__eq__(other)

__hash__ = None

__init__(received_datagrams: ~typing.Dict[~pycyphal.transport._session.InputSessionSpecifier, ~pycyphal.transport.udp._session._input.UDPInputSessionStatistics] = <factory>) → None

__match_args__ = ('received_datagrams',)

__repr__()

class pycyphal.transport.udp.UDPInputSession(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None] | None, local_node_id: int | None)

Bases: InputSession

The input session logic is simple because most of the work is handled by the UDP/IP stack of the operating system.

Here we just wait for the frames to arrive (from the socket), reassemble them, and pass the resulting transfer.

[Socket] —> [Input session] —> [UDP API]

(The plurality notation is supposed to resemble UML: 1 - one, * - many.)

A UDP datagram is an atomic unit of workload for the stack. Unlike, say, the serial transport, the operating system does the low-level work of framing and CRC checking for us (thank you), so we get our stuff sorted up to the OSI layer 4 inclusive. The processing pipeline per datagram is as follows:

• The socket obtains the datagram from the socket using recvfrom(). The contents of the Cyphal UDP frame instance is parsed which, among others, contains the source node-ID. If anything goes wrong here (like if the datagram does not contain a valid Cyphal frame or whatever), the datagram is dropped and the appropriate statistical counters are updated.

• Upon reception of the frame, the input session updates its reassembler state machine(s) (many in case of PromiscuousInputSession) and runs all that meticulous bookkeeping you can’t get away from if you need to receive multi-frame transfers.

• If the received frame happened to complete a transfer, the input session passes it up to the higher layer.

The input session logic is extremely simple because most of the work is handled by the UDP/IP stack of the operating system. Here we just need to reconstruct the transfer from the frames and pass it up to the higher layer.

DEFAULT_TRANSFER_ID_TIMEOUT = 2.0

Units are seconds. Can be overridden after instantiation if needed.

__init__(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None] | None, local_node_id: int | None)

Parent class of PromiscuousInputSession and SelectiveInputSession.

async receive(monotonic_deadline: float) → TransferFrom | None

This method will wait for self._reader_thread to put a frame in the queue. If a frame is available, it will retrieved and used to construct a transfer. Once a complete transfer can be constructed from the frames, it will be returned.

The method will block until a transfer is available or the deadline is reached.

If the deadline is reached, the method will return None. If the session is closed, the method will raise ResourceClosedError.

property transfer_id_timeout: float

property specifier: InputSessionSpecifier

property payload_metadata: PayloadMetadata

close() → None

Closes the instance and its socket, waits for the thread to terminate (which should happen instantly).

Once closed, new listeners can no longer be added. Raises RuntimeError instead of closing if there is at least one active listener.

abstract sample_statistics() → UDPInputSessionStatistics

class pycyphal.transport.udp.PromiscuousUDPInputSession(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None], local_node_id: int | None, statistics: PromiscuousUDPInputSessionStatistics)

Bases: UDPInputSession

__init__(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None], local_node_id: int | None, statistics: PromiscuousUDPInputSessionStatistics)

Do not call this directly, use the factory method instead.

sample_statistics() → PromiscuousUDPInputSessionStatistics

class pycyphal.transport.udp.SelectiveUDPInputSession(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None], local_node_id: int | None, statistics: SelectiveUDPInputSessionStatistics)

Bases: UDPInputSession

__init__(specifier: InputSessionSpecifier, payload_metadata: PayloadMetadata, socket: socket, finalizer: Callable[[], None], local_node_id: int | None, statistics: SelectiveUDPInputSessionStatistics)

Do not call this directly, use the factory method instead.

sample_statistics() → SelectiveUDPInputSessionStatistics

class pycyphal.transport.udp.UDPInputSessionStatistics(transfers: int = 0, frames: int = 0, payload_bytes: int = 0, errors: int = 0, drops: int = 0)

Bases: SessionStatistics

class pycyphal.transport.udp.PromiscuousUDPInputSessionStatistics(transfers: 'int' = 0, frames: 'int' = 0, payload_bytes: 'int' = 0, errors: 'int' = 0, drops: 'int' = 0, reassembly_errors_per_source_node_id: 'typing.Dict[int, typing.Dict[TransferReassembler.Error, int]]' = <factory>)

Bases: UDPInputSessionStatistics

reassembly_errors_per_source_node_id: Dict[int, Dict[Error, int]]

Keys are source node-IDs; values are dicts where keys are error enum members and values are counts.

__eq__(other)

__hash__ = None

__init__(transfers: int = 0, frames: int = 0, payload_bytes: int = 0, errors: int = 0, drops: int = 0, reassembly_errors_per_source_node_id: ~typing.Dict[int, ~typing.Dict[~pycyphal.transport.commons.high_overhead_transport._transfer_reassembler.TransferReassembler.Error, int]] = <factory>) → None

__match_args__ = ('transfers', 'frames', 'payload_bytes', 'errors', 'drops', 'reassembly_errors_per_source_node_id')

__repr__()

class pycyphal.transport.udp.SelectiveUDPInputSessionStatistics(transfers: 'int' = 0, frames: 'int' = 0, payload_bytes: 'int' = 0, errors: 'int' = 0, drops: 'int' = 0, reassembly_errors: 'typing.Dict[TransferReassembler.Error, int]' = <factory>)

Bases: UDPInputSessionStatistics

reassembly_errors: Dict[Error, int]

Keys are error enum members and values are counts.

__eq__(other)

__hash__ = None

__init__(transfers: int = 0, frames: int = 0, payload_bytes: int = 0, errors: int = 0, drops: int = 0, reassembly_errors: ~typing.Dict[~pycyphal.transport.commons.high_overhead_transport._transfer_reassembler.TransferReassembler.Error, int] = <factory>) → None

__match_args__ = ('transfers', 'frames', 'payload_bytes', 'errors', 'drops', 'reassembly_errors')

__repr__()

class pycyphal.transport.udp.UDPOutputSession(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata, mtu: int, multiplier: int, sock: socket, source_node_id: int | None, finalizer: Callable[[], None])

Bases: OutputSession

The output session logic is extremely simple because most of the work is handled by the UDP/IP stack of the operating system. Here we just split the transfer into frames, encode the frames, and write them into the socket one by one. If the transfer multiplier is greater than one (for unreliable networks), we repeat that the required number of times.

__init__(specifier: OutputSessionSpecifier, payload_metadata: PayloadMetadata, mtu: int, multiplier: int, sock: socket, source_node_id: int | None, finalizer: Callable[[], None])

Do not call this directly. Instead, use the factory method. Instances take ownership of the socket.

async send(transfer: Transfer, monotonic_deadline: float) → bool

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

disable_feedback() → None

property specifier: OutputSessionSpecifier

property payload_metadata: PayloadMetadata

sample_statistics() → SessionStatistics

close() → None

property socket: socket

Provides access to the underlying UDP socket.

class pycyphal.transport.udp.UDPFeedback(original_transfer_timestamp: Timestamp, first_frame_transmission_timestamp: Timestamp)

Bases: Feedback

__init__(original_transfer_timestamp: Timestamp, first_frame_transmission_timestamp: Timestamp)

property original_transfer_timestamp: Timestamp

property first_frame_transmission_timestamp: Timestamp

class pycyphal.transport.udp.UDPFrame(priority: Priority, transfer_id: int, index: int, end_of_transfer: bool, payload: memoryview, source_node_id: int | None, destination_node_id: int | None, data_specifier: DataSpecifier, user_data: int)

Bases: Frame

An important thing to keep in mind is that the minimum size of an UDP/IPv4 payload when transferred over 100M Ethernet is 18 bytes, due to the minimum Ethernet frame size limit. That is, if the application payload requires less space, the missing bytes will be padded out to the minimum size.

The current header format enables encoding by trivial memory aliasing on any conventional little-endian platform.

MAC header	IP header	UDP header	Cyphal header	Cyphal payload
			Layers modeled by this type

NODE_ID_MASK = 65535

SUBJECT_ID_MASK = 32767

SERVICE_ID_MASK = 16383

TRANSFER_ID_MASK = 18446744073709551615

INDEX_MASK = 2147483647

NODE_ID_MAX = 65534

Cyphal/UDP supports 65535 nodes per logical network, from 0 to 65534 inclusive. 65535 is reserved for the anonymous/broadcast ID.

source_node_id: int | None

destination_node_id: int | None

data_specifier: DataSpecifier

user_data: int

compile_header_and_payload() → Tuple[memoryview, memoryview]

Compiles the UDP frame header and returns it as a read-only memoryview along with the payload, separately. The caller is supposed to handle the header and the payload independently. The reason is to avoid unnecessary data copying in the user space, allowing the caller to rely on the vectorized IO API instead (sendmsg).

static parse(image: memoryview) → UDPFrame | None

__delattr__(name)

__eq__(other)

__hash__()

__init__(priority: Priority, transfer_id: int, index: int, end_of_transfer: bool, payload: memoryview, source_node_id: int | None, destination_node_id: int | None, data_specifier: DataSpecifier, user_data: int) → None

__match_args__ = ('priority', 'transfer_id', 'index', 'end_of_transfer', 'payload', 'source_node_id', 'destination_node_id', 'data_specifier', 'user_data')

__setattr__(name, value)

pycyphal.transport.udp.message_data_specifier_to_multicast_group(data_specifier: MessageDataSpecifier, ipv6_addr: bool = False, cy_addr_version: int = 0) → IPv4Address | IPv6Address

Takes a (Message) data_specifier; returns the corresponding multicast address. For IPv4, the resulting address is constructed as follows:

          fixed          subject-ID (Service)
      (15 bits)     res. (15 bits)
   ______________   | _____________
  /              \  v/             \
  11101111.00000000.znnnnnnn.nnnnnnnn
  \__/      ^     ^
(4 bits)  Cyphal SNM
  IPv4     UDP
multicast address
 prefix   version

>>> from pycyphal.transport import MessageDataSpecifier
>>> from ipaddress import ip_address
>>> str(message_data_specifier_to_multicast_group(MessageDataSpecifier(123)))
'239.0.0.123'
>>> str(message_data_specifier_to_multicast_group(MessageDataSpecifier(456)))
'239.0.1.200'
>>> str(message_data_specifier_to_multicast_group(MessageDataSpecifier(2**14)))
Traceback (most recent call last):
  ...
ValueError: Invalid subject-ID...
>>> msg_ip = message_data_specifier_to_multicast_group(MessageDataSpecifier(123))
>>> assert (int(msg_ip) & SNM_BIT_MASK) != SNM_BIT_MASK, "SNM bit is 0 for message"

pycyphal.transport.udp.service_node_id_to_multicast_group(destination_node_id: int | None, ipv6_addr: bool = False, cy_addr_version: int = 0) → IPv4Address | IPv6Address

Takes a destination node_id; returns the corresponding multicast address (for Service). For IPv4, the resulting address is constructed as follows:

        fixed
      (15 bits)
   ______________
  /              \
  11101111.00000001.nnnnnnnn.nnnnnnnn
  \__/      ^     ^ \_______________/
(4 bits)  Cyphal SNM     (16 bits)
  IPv4     UDP           destination node-ID (Service)
multicast address
 prefix   version

>>> from ipaddress import ip_address
>>> str(service_node_id_to_multicast_group(123))
'239.1.0.123'
>>> str(service_node_id_to_multicast_group(456))
'239.1.1.200'
>>> str(service_node_id_to_multicast_group(None))
'239.1.255.255'
>>> str(service_node_id_to_multicast_group(int(0xFFFF)))
Traceback (most recent call last):
  ...
ValueError: Invalid node-ID...
>>> str(service_node_id_to_multicast_group(65536))
Traceback (most recent call last):
  ...
ValueError: Invalid node-ID...
>>> srvc_ip = service_node_id_to_multicast_group(123)
>>> assert (int(srvc_ip) & SNM_BIT_MASK) == SNM_BIT_MASK, "SNM bit is 1 for service"

class pycyphal.transport.udp.LinkLayerPacket(protocol: AddressFamily, source: memoryview, destination: memoryview, payload: memoryview)

Bases: object

OSI L2 packet representation. The addresses are represented here in the link-native byte order (big endian for Ethernet).

protocol: AddressFamily

The protocol encapsulated inside this link-layer packet; e.g., IPv6.

source: memoryview

destination: memoryview

Link-layer addresses, if applicable. Empty if not supported by the link layer.

payload: memoryview

The packet of the specified protocol.

__repr__() → str

The repr displays only the first 100 bytes of the payload. If the payload is longer, its string representation is appended with an ellipsis.

__delattr__(name)

__eq__(other)

__hash__()

__init__(protocol: AddressFamily, source: memoryview, destination: memoryview, payload: memoryview) → None

__match_args__ = ('protocol', 'source', 'destination', 'payload')

__setattr__(name, value)

class pycyphal.transport.udp.IPPacket(protocol: 'int', payload: 'memoryview')

Bases: object

protocol: int

payload: memoryview

property source_destination: Tuple[IPv4Address, IPv4Address] | Tuple[IPv6Address, IPv6Address]

static parse(link_layer_packet: LinkLayerPacket) → IPPacket | None

__delattr__(name)

__eq__(other)

__hash__()

__init__(protocol: int, payload: memoryview) → None

__match_args__ = ('protocol', 'payload')

__repr__()

__setattr__(name, value)

class pycyphal.transport.udp.IPv4Packet(protocol: 'int', payload: 'memoryview', source: 'IPv4Address', destination: 'IPv4Address')

Bases: IPPacket

source: IPv4Address

destination: IPv4Address

property source_destination: Tuple[IPv4Address, IPv4Address]

static parse_payload(link_layer_payload: memoryview) → IPv4Packet | None

__delattr__(name)

__eq__(other)

__hash__()

__init__(protocol: int, payload: memoryview, source: IPv4Address, destination: IPv4Address) → None

__match_args__ = ('protocol', 'payload', 'source', 'destination')

__repr__()

__setattr__(name, value)

class pycyphal.transport.udp.IPv6Packet(protocol: 'int', payload: 'memoryview', source: 'IPv6Address', destination: 'IPv6Address')

Bases: IPPacket

source: IPv6Address

destination: IPv6Address

property source_destination: Tuple[IPv6Address, IPv6Address]

static parse_payload(link_layer_payload: memoryview) → IPv6Packet | None

__delattr__(name)

__eq__(other)

__hash__()

__init__(protocol: int, payload: memoryview, source: IPv6Address, destination: IPv6Address) → None

__match_args__ = ('protocol', 'payload', 'source', 'destination')

__repr__()

__setattr__(name, value)

class pycyphal.transport.udp.UDPIPPacket(source_port: 'int', destination_port: 'int', payload: 'memoryview')

Bases: object

source_port: int

destination_port: int

payload: memoryview

static parse(ip_packet: IPPacket) → UDPIPPacket | None

__delattr__(name)

__eq__(other)

__hash__()

__init__(source_port: int, destination_port: int, payload: memoryview) → None

__match_args__ = ('source_port', 'destination_port', 'payload')

__repr__()

__setattr__(name, value)

class pycyphal.transport.udp.UDPCapture(timestamp: Timestamp, link_layer_packet: LinkLayerPacket)

Bases: Capture

The UDP transport does not differentiate between sent and received packets. See pycyphal.transport.udp.UDPTransport.begin_capture() for details.

link_layer_packet: LinkLayerPacket

parse() → Tuple[AlienSessionSpecifier, UDPFrame] | None

The parsed representation is only defined if the packet is a valid Cyphal/UDP frame. The source node-ID can be None in the case of anonymous messages.

static get_transport_type() → Type[UDPTransport]

__delattr__(name)

__eq__(other)

__hash__()

__init__(timestamp: Timestamp, link_layer_packet: LinkLayerPacket) → None

__match_args__ = ('timestamp', 'link_layer_packet')

__repr__()

__setattr__(name, value)

class pycyphal.transport.udp.UDPTracer

Bases: Tracer

This is like a Wireshark dissector but Cyphal-focused. Return types from update():

• pycyphal.transport.TransferTrace

• UDPErrorTrace

__init__() → None

update(cap: Capture) → Trace | None

class pycyphal.transport.udp.UDPErrorTrace(timestamp: 'pycyphal.transport.Timestamp', error: 'TransferReassembler.Error')

Bases: ErrorTrace

error: Error

__delattr__(name)

__eq__(other)

__hash__()

__init__(timestamp: Timestamp, error: Error) → None

__match_args__ = ('timestamp', 'error')

__repr__()

__setattr__(name, value)