qns.entity package

Subpackages

Submodules

qns.entity.entity module

class qns.entity.entity.Entity(name: Optional[str] = None)[source]

Bases: object

This is the basic entity class, including memories, channels and nodes.

handle(event: qns.simulator.event.Event) None[source]

handle is called to process an receiving Event.

Parameters

event (qns.simulator.event.Event) – the event that send to this entity

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

Module contents

class qns.entity.Application[source]

Bases: object

Application can be deployed on the quantum nodes.

add_handler(handler, EventTypeList: List = [], ByList: List = [])[source]

Add a handler function to the dispather.

Parameters
  • handler – The handler to process the event. It is an object method whose function signature is the same to handle function.

  • EventTypeList – a list of Event Class Type. An empty list meaning to match all events.

  • ByList – a list of Entities, QNodes or Applications, that generates this event. An empty list meaning to match all entities.

get_node()[source]

get the node that runs this application

Returns

the quantum node

get_simulator()[source]

get the simulator

Returns

the simulator

handle(node, event: qns.simulator.event.Event) Optional[bool][source]

process the event on the node.

Parameters
  • node (QNode) – the node that will handle this event

  • event (Event) – the event

Returns

if skip is True, further applications will not handle this event

Return type

skip (bool, None)

install(node, simulator: qns.simulator.simulator.Simulator)[source]

install initial events for this QNode

Parameters
  • node (QNode) – the node that will handle this event

  • simulator (Simulator) – the simulator

class qns.entity.ClassicChannel(name: Optional[str] = None, node_list: List[qns.entity.node.node.QNode] = [], bandwidth: int = 0, delay: Union[float, qns.models.delay.delay.DelayModel] = 0, drop_rate: float = 0, max_buffer_size: int = 0)[source]

Bases: qns.entity.entity.Entity

ClassicChannel is the channel for classic message

install(simulator: qns.simulator.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: qns.entity.cchannel.cchannel.ClassicPacket, next_hop: qns.entity.node.node.QNode)[source]

Send a classic packet to the next_hop

Parameters
Raises

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

class qns.entity.ClassicPacket(msg: Union[str, bytes, Any], src: Optional[qns.entity.node.node.QNode] = None, dest: Optional[qns.entity.node.node.QNode] = 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])

class qns.entity.Entity(name: Optional[str] = None)[source]

Bases: object

This is the basic entity class, including memories, channels and nodes.

handle(event: qns.simulator.event.Event) None[source]

handle is called to process an receiving Event.

Parameters

event (qns.simulator.event.Event) – the event that send to this entity

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

class qns.entity.MemoryReadRequestEvent(memory, key: Union[qns.models.core.backend.QuantumModel, str], t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

MemoryReadRequestEvent is the event that request a memory read

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.MemoryReadResponseEvent(node: qns.entity.node.node.QNode, result: Optional[qns.models.core.backend.QuantumModel] = None, request: Optional[qns.entity.memory.event.MemoryReadRequestEvent] = None, t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

MemoryReadResponseEvent is the event that returns the memory read result

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.MemoryWriteRequestEvent(memory, qubit: qns.models.core.backend.QuantumModel, t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

MemoryWriteRequestEvent is the event that request a memory write

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.MemoryWriteResponseEvent(node: qns.entity.node.node.QNode, result: Optional[qns.models.core.backend.QuantumModel] = None, request: Optional[qns.entity.memory.event.MemoryReadRequestEvent] = None, t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

MemoryWriteResponseEvent is the event that returns the memory write result

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.Monitor(name: Optional[str] = None, network: Optional[qns.network.network.QuantumNetwork] = None)[source]

Bases: qns.entity.entity.Entity

add_attribution(name: str, calculate_func: Callable[[qns.simulator.simulator.Simulator, Optional[qns.network.network.QuantumNetwork], Optional[qns.simulator.event.Event]], Any]) None[source]

Set an attribution that will be recorded. For example, an attribution could be the throughput, or the fidelity.

Parameters
  • name (str) – the column’s name, e.g., fidelity, throughput, time …

  • calculate_func (Callable[[Simulator, Optional[QuantumNetwork], Optional[Event]]) – a function to calculate the value, it has three input parameters (Simulator, QuantumNetwork, Event), and it returns the value.

Usage:

m = Monitor()

# record the event happening time m.add_attribution(“time”, lambda s,n,e: e.t)

# get the ‘name’ attribution of the last node m.add_attribution(“count”, lambda s,network,e: network.nodes[-1].name)

at_event(event_type) None[source]

Watch network status whenever the event happends

Parameters

event_type (Event) – the watching event

Usage:

# record network status when a node receives a qubit m.at_event(RecvQubitPacket)

at_finish() None[source]

Watch the final status after the simulation.

Usage:

m.at_finish()

at_period(period_time: float) None[source]

Watch network status at a constant period.

Parameters

period_time (float) – the period of watching network status [s]

Usage:

# record network status every 3 seconds. m.at_period(3)

at_start() None[source]

Watch the initial status before the simulation starts.

Usage:

m.at_start()

calculate_date(event: qns.simulator.event.Event)[source]
get_date()[source]

Get the collected data.

Returns

the collected data, as a pd.DataFrame.

handle(event: qns.simulator.event.Event) None[source]

handle is called to process an receiving Event.

Parameters

event (qns.simulator.event.Event) – the event that send to this entity

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

class qns.entity.MonitorEvent(t: Optional[qns.simulator.ts.Time], monitor, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

the event that notify the monitor to write down network status

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.OperateRequestEvent(operator, qubits: List[qns.models.core.backend.QuantumModel] = [], t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

OperateRequestEvent is the event that request a operator to handle

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.OperateResponseEvent(node: qns.entity.node.node.QNode, result: Optional[Union[int, List[int]]] = None, request: Optional[qns.entity.operator.event.OperateRequestEvent] = None, t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

OperateResponseEvent is the event that returns the operating result

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.QNode(name: Optional[str] = None, apps: Optional[List[qns.entity.node.app.Application]] = None)[source]

Bases: qns.entity.entity.Entity

QNode is a quantum node in the quantum network

add_apps(app: qns.entity.node.app.Application)[source]

Insert an Application into the app list.

Parameters

app (Application) – the inserting application.

add_cchannel(cchannel)[source]

Add a classic channel in this QNode

Parameters

cchannel (ClassicChannel) – the classic channel

add_memory(memory)[source]

Add a quantum memory in this QNode

Parameters

memory (Memory) – the quantum memory

add_network(network)[source]

add a network object to this node

Parameters

network (qns.network.network.Network) – the network object

add_operator(operator)[source]

Add a quantum operator in this node

Parameters

operator (QuantumOperator) – the quantum operator

add_qchannel(qchannel)[source]

Add a quantum channel in this QNode

Parameters

qchannel (QuantumChannel) – the quantum channel

add_request(request)[source]

add a request to this node

Parameters

request (Request) – the inserting request

clear_request()[source]

clear all requests

get_apps(app_type)[source]

Get an Application that is app_type

Parameters

app_type – the class of app_type

get_cchannel(dst: qns.entity.node.node.QNode)[source]

Get the classic channel that connects to the dst

Parameters

dst (QNode) – the destination

get_qchannel(dst: qns.entity.node.node.QNode)[source]

Get the quantum channel that connects to the dst

Parameters

dst (QNode) – the destination

handle(event: qns.simulator.event.Event) None[source]

This function will handle an Event. This event will be passed to every applications in apps list in order.

Parameters

event (Event) – the event that happens on this QNode

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

class qns.entity.QuantumChannel(name: Optional[str] = None, node_list: List[qns.entity.node.node.QNode] = [], bandwidth: int = 0, delay: Union[float, qns.models.delay.delay.DelayModel] = 0, drop_rate: float = 0, max_buffer_size: int = 0, length: float = 0, decoherence_rate: Optional[float] = 0, transfer_error_model_args: dict = {})[source]

Bases: qns.entity.entity.Entity

QuantumChannel is the channel for transmitting qubit

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

simulator (Simulator) – the simulator

send(qubit: qns.models.core.backend.QuantumModel, next_hop: qns.entity.node.node.QNode)[source]

Send a qubit to the next_hop

Parameters
  • qubit (QuantumModel) – the transmitting qubit

  • next_hop (QNode) – the next hop QNode

Raises

NextHopNotConnectionException – the next_hop is not connected to this channel

class qns.entity.QuantumMemory(name: Optional[str] = None, node: Optional[qns.entity.node.node.QNode] = None, capacity: int = 0, decoherence_rate: Optional[float] = 0, store_error_model_args: dict = {}, delay: Union[float, qns.models.delay.delay.DelayModel] = 0)[source]

Bases: qns.entity.entity.Entity

Quantum memory stores qubits or entangled pairs.

It has two modes:

Synchronous mode, users can use the read and write function to operate the memory directly without delay Asynchronous mode, users can use events to operate memories asynchronously

property count: int

return the current memory usage

get(key: Union[qns.models.core.backend.QuantumModel, str, int]) Optional[qns.models.core.backend.QuantumModel][source]

get a qubit from the memory but without removing it from the memory

Parameters

key (Union[QuantumModel, str, int]) – the key. It can be a QuantumModel object, its name or the index number.

get_store_time(key: Union[qns.models.core.backend.QuantumModel, str, int]) Optional[qns.models.core.backend.QuantumModel][source]

get the store time of a qubit from the memory

Parameters

key (Union[QuantumModel, str, int]) – the key. It can be a QuantumModel object, its name or the index number.

handle(event: qns.simulator.event.Event) None[source]

handle is called to process an receiving Event.

Parameters

event (qns.simulator.event.Event) – the event that send to this entity

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

is_full() bool[source]

check whether the memory is full

read(key: Union[qns.models.core.backend.QuantumModel, str]) Optional[qns.models.core.backend.QuantumModel][source]

The API for reading a qubit from the memory

Parameters

key (Union[QuantumModel, str]) – the key. It can be a QuantumModel object, its name or the index number.

write(qm: qns.models.core.backend.QuantumModel) bool[source]

The API for storing a qubit to the memory

Parameters

qm (QuantumModel) – the QuantumModel, could be a qubit or an entangled pair

Returns

whether the qubit is stored successfully

Return type

bool

class qns.entity.QuantumOperator(name: Optional[str] = None, node: Optional[qns.entity.node.node.QNode] = None, gate: Optional[Callable[[...], Union[None, int, List[int]]]] = None, delay: Union[float, qns.models.delay.delay.DelayModel] = 0)[source]

Bases: qns.entity.entity.Entity

Quantum operator can perfrom quantum operation or measurements on qubits. It has two modes:

Synchronous mode, users can use the operate function to operate qubits directly without delay Asynchronous mode, users will use events to operate quantum operations asynchronously

handle(event: qns.simulator.event.Event) None[source]

handle is called to process an receiving Event.

Parameters

event (qns.simulator.event.Event) – the event that send to this entity

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

operate(*qubits) Optional[Union[int, List[int]]][source]

operate on qubits and return the measure result

Parameters

qubits – the operating qubits

Returns

the measure result

set_own(node: qns.entity.node.node.QNode)[source]

set the owner of this quantum operator

class qns.entity.RecvClassicPacket(t: Optional[qns.simulator.ts.Time] = None, name: Optional[str] = None, cchannel: Optional[qns.entity.cchannel.cchannel.ClassicChannel] = None, packet: Optional[qns.entity.cchannel.cchannel.ClassicPacket] = None, dest: Optional[qns.entity.node.node.QNode] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

The event for a QNode to receive a classic packet

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.RecvQubitPacket(t: Optional[qns.simulator.ts.Time] = None, qchannel: Optional[qns.entity.qchannel.qchannel.QuantumChannel] = None, qubit: Optional[qns.models.core.backend.QuantumModel] = None, dest: Optional[qns.entity.node.node.QNode] = None, name: Optional[str] = None, by: Optional[Any] = None)[source]

Bases: qns.simulator.event.Event

The event for a QNode to receive a classic packet

invoke() None[source]

Invoke the event, should be implemented

class qns.entity.Timer(name: str, start_time: float, end_time: float = 0, step_time: float = 1, trigger_func=None)[source]

Bases: qns.entity.entity.Entity

A Timer is an Entity that trigger the function trigger_func one-shot or periodically.

install(simulator: qns.simulator.simulator.Simulator) None[source]

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

Parameters

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

trigger()[source]