qns.entity.cchannel package

Submodules

qns.entity.cchannel.cchannel module

class qns.entity.cchannel.cchannel.ClassicChannel(name: str | None = None, node_list: List[QNode] = [], bandwidth: int = 0, delay: float | DelayModel = 0, length: float | None = 0, drop_rate: float = 0, max_buffer_size: int = 0)[source]

Bases: Entity

ClassicChannel is the channel for classic message

install(simulator: Simulator) → None[source]

install is called before simulator runs to initialize or set initial events

Parameters:: simulator (qns.simulator.simulator.Simulator) – the simulator

send(packet: ClassicPacket, next_hop: QNode)[source]

Send a classic packet to the next_hop

Parameters:

packet (ClassicPacket) – the packet
next_hop (QNode) – the next hop QNode

Raises:

qns.entity.cchannel.cchannel.NextHopNotConnectionException – the next_hop is not connected to this channel

class qns.entity.cchannel.cchannel.ClassicPacket(msg: str | bytes | Any, src: QNode | None = None, dest: QNode | None = None)[source]

Bases: object

ClassicPacket is the message that transfer on a ClassicChannel

encode() → bytes[source]

encode the self.msg if it is a str

Returns:: (bytes) a bytes object

get()[source]

get the message from packet

Returns:: (Union[str, bytes, Any])

exception qns.entity.cchannel.cchannel.NextHopNotConnectionException[source]: Bases: Exception

Bases: Event

The event for a QNode to receive a classic packet

invoke() → None[source]: Invoke the event, should be implemented

Module contents

class qns.entity.cchannel.ClassicChannel(name: str | None = None, node_list: List[QNode] = [], bandwidth: int = 0, delay: float | DelayModel = 0, length: float | None = 0, drop_rate: float = 0, max_buffer_size: int = 0)[source]