30315099891

Committed 27 Jul 2026 11:34PM UTC coverage: 85.792% (+0.008%) from 85.784%

Build # 30315099891

Build Type

push

github

Committed by

yoursunny

Commit Message

proactive: RoutingPath(m_v="auto")

Coverage Stats

53 of 57 new or added lines in 6 files covered. (92.98%)

85 existing lines in 22 files now uncovered.

6014 of 7010 relevant lines covered (85.79%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.24

/mqns/entity/node/node.py

#    Modified by Amar Abane for Multiverse Quantum Network Simulator
#    Date: 05/17/2025
#    Summary of changes: Adapted logic to support dynamic approaches.
#
#    This file is based on a snapshot of SimQN (https://github.com/QNLab-USTC/SimQN),
#    which is licensed under the GNU General Public License v3.0.
#
#    The original SimQN header is included below.


#    SimQN: a discrete-event simulator for the quantum networks
#    Copyright (C) 2021-2022 Lutong Chen, Jian Li, Kaiping Xue
#    University of Science and Technology of China, USTC.
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program.  If not, see <https://www.gnu.org/licenses/>.

from collections import defaultdict
from collections.abc import Iterable
from typing import TYPE_CHECKING, cast, override

from mqns.entity.entity import Entity
from mqns.entity.node.app import Application
from mqns.simulator import Event, Simulator

if TYPE_CHECKING:
    from mqns.entity.base_channel import BaseChannel
    from mqns.entity.cchannel import ClassicChannel, ClassicPacket
    from mqns.network.network import QuantumNetwork, TimingMode


class Node(Entity):
    """Node is a generic node in the quantum network"""

    def __init__(self, name: str, *, apps: list[Application] | None = None):
        """
        Args:
            name: Node name.
            apps: Applications on the node.
        """
        super().__init__(name)
        self.cchannels: list["ClassicChannel"] = []
        """Classic channels connected to this node."""
        self._cchannel_by_dst = dict["Node", "ClassicChannel"]()
        self.apps: list[Application] = [] if apps is None else apps
        """Applications on this node."""
        self._app_by_type: dict[type, Application] | None = None

    @override
    def install(self, simulator: Simulator) -> None:
        """Attach ``Simulator`` to entities within node."""
        super().install(simulator)
        self._install_channels(self.cchannels, self._cchannel_by_dst)
        self._node_install()

        apps_by_type = defaultdict[type, list[Application]](lambda: [])
        for app in self.apps:
            apps_by_type[type(app)].append(app)
            app.install(self)

        self._app_by_type = {}
        for typ, apps in apps_by_type.items():
            if len(apps) == 1:
                self._app_by_type[typ] = apps[0]

    def _node_install(self) -> None:
        pass

    @override
    def handle(self, event: Event) -> None:
        """
        Dispatch an ``Event`` that happens on this Node.
        This event is passed to every application in apps list in order.

        Args:
            event: the event that happens on this Node.
        """
        for app in self.apps:
            skip = app.handle(event)
            if skip:
                break

    def add_apps(self, app: Application | Iterable[Application]):
        """
        Insert one or more applications into the app list.

        Args:
            app: an application or a list of applications.
                 The caller is responsible for ``deepcopy`` if needed, so that each node has a separate instance.

        """
        self.ensure_not_installed()
        if isinstance(app, Application):
            self.apps.append(app)
        else:
            self.apps.extend(app)

    def get_apps[A: Application](self, app_type: type[A]) -> list[A]:
        """
        Retrieve applications of given type.

        Args:
            app_type: Application type/class.
        """
        return [app for app in self.apps if isinstance(app, app_type)]

    def get_app[A: Application](self, app_type: type[A]) -> A:
        """
        Retrieve an application of given type.
        There must be exactly one instance of this application.

        Args:
            app_type: Application type/class.

        Raises:
            LookupError: Application does not exist, or there are multiple instances.
        """
        if self._app_by_type is None:  # this is called before self.install() populates _app_by_type
            self.ensure_not_installed()
            return self._get_app_from_apps(app_type)

        try:
            return cast(A, self._app_by_type[app_type])
        except KeyError:
            # app_type is a base class
            return self._get_app_from_apps(app_type)

    def _get_app_from_apps[A: Application](self, app_type: type[A]) -> A:
        apps = self.get_apps(app_type)
        if len(apps) != 1:
            raise LookupError(f"node does not have exactly one instance of {app_type}")
        return apps[0]

    def _add_channel[C: "BaseChannel"](self, channel: C, channels: list[C]) -> None:
        self.ensure_not_installed()
        channel.node_list.append(self)
        channels.append(channel)

    def _install_channels[C: "BaseChannel"](self, channels: list[C], by_neighbor: dict["Node", C]) -> None:
        for ch in channels:
            for dst in ch.node_list:
                if dst is not self:
                    by_neighbor[dst] = ch
            ch.install(self.simulator)

    @staticmethod
    def _get_channel[C: "BaseChannel"](dst: "Node", by_neighbor: dict["Node", C]) -> C:
        return by_neighbor[dst]

    def add_cchannel(self, cchannel: "ClassicChannel"):
        """
        Add a classic channel in this Node.
        This function is available prior to calling .install().
        """
        self.ensure_not_installed()
        self._add_channel(cchannel, self.cchannels)

    def get_cchannel(self, dst: "Node") -> "ClassicChannel":
        """
        Retrieve the classic channel that connects to ``dst``.

        Raises:
            LookupError: channel does not exist
        """
        return self._get_channel(dst, self._cchannel_by_dst)

    def send_cpacket(self, next_hop: "Node", pkt: "ClassicPacket") -> None:
        """
        Send a classic packet to an adjacent node.

        Args:
            next_hop: an adjacent node that shares a classic channel with this node.
            pkt: the packet.
        """
        self.get_cchannel(next_hop).send(pkt, next_hop)

    def add_network(self, network: "QuantumNetwork"):
        """
        Assign a network object to this node.
        """
        self.network = network
        """Quantum network that contains this node."""

    @property
    def timing(self) -> "TimingMode":
        """
        Access the network-wide application timing mode.
        """
        return self.network.timing

    def __repr__(self) -> str:
        return f"<node {self.name}>"

1	# Modified by Amar Abane for Multiverse Quantum Network Simulator
2	# Date: 05/17/2025
3	# Summary of changes: Adapted logic to support dynamic approaches.
4	#
5	# This file is based on a snapshot of SimQN (https://github.com/QNLab-USTC/SimQN),
6	# which is licensed under the GNU General Public License v3.0.
7	#
8	# The original SimQN header is included below.
9
10
11	# SimQN: a discrete-event simulator for the quantum networks
12	# Copyright (C) 2021-2022 Lutong Chen, Jian Li, Kaiping Xue
13	# University of Science and Technology of China, USTC.
14	#
15	# This program is free software: you can redistribute it and/or modify
16	# it under the terms of the GNU General Public License as published by
17	# the Free Software Foundation, either version 3 of the License, or
18	# (at your option) any later version.
19	#
20	# This program is distributed in the hope that it will be useful,
21	# but WITHOUT ANY WARRANTY; without even the implied warranty of
22	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23	# GNU General Public License for more details.
24	#
25	# You should have received a copy of the GNU General Public License
26	# along with this program. If not, see <https://www.gnu.org/licenses/>.
27
28	from collections import defaultdict	1✔
29	from collections.abc import Iterable	1✔
30	from typing import TYPE_CHECKING, cast, override	1✔
31
32	from mqns.entity.entity import Entity	1✔
33	from mqns.entity.node.app import Application	1✔
34	from mqns.simulator import Event, Simulator	1✔
35
36	if TYPE_CHECKING:
37	from mqns.entity.base_channel import BaseChannel
38	from mqns.entity.cchannel import ClassicChannel, ClassicPacket
39	from mqns.network.network import QuantumNetwork, TimingMode
40
41
42	class Node(Entity):	1✔
43	"""Node is a generic node in the quantum network"""
44
45	def __init__(self, name: str, *, apps: list[Application] \| None = None):	1✔
46	"""
47	Args:
48	name: Node name.
49	apps: Applications on the node.
50	"""
51	super().__init__(name)	1✔
52	self.cchannels: list["ClassicChannel"] = []	1✔
53	"""Classic channels connected to this node."""	1✔
54	self._cchannel_by_dst = dict["Node", "ClassicChannel"]()	1✔
55	self.apps: list[Application] = [] if apps is None else apps	1✔
56	"""Applications on this node."""	1✔
57	self._app_by_type: dict[type, Application] \| None = None	1✔
58
59	@override	1✔
60	def install(self, simulator: Simulator) -> None:	1✔
61	"""Attach ``Simulator`` to entities within node."""
62	super().install(simulator)	1✔
63	self._install_channels(self.cchannels, self._cchannel_by_dst)	1✔
64	self._node_install()	1✔
65
66	apps_by_type = defaultdict[type, list[Application]](lambda: [])	1✔
67	for app in self.apps:	1✔
68	apps_by_type[type(app)].append(app)	1✔
69	app.install(self)	1✔
70
71	self._app_by_type = {}	1✔
72	for typ, apps in apps_by_type.items():	1✔
73	if len(apps) == 1:	1✔
74	self._app_by_type[typ] = apps[0]	1✔
75
76	def _node_install(self) -> None:	1✔
77	pass	1✔
78
79	@override	1✔
80	def handle(self, event: Event) -> None:	1✔
81	"""
82	Dispatch an ``Event`` that happens on this Node.
83	This event is passed to every application in apps list in order.
84
85	Args:
86	event: the event that happens on this Node.
87	"""
88	for app in self.apps:	1✔
89	skip = app.handle(event)	1✔
90	if skip:	1✔
91	break	1✔
92
93	def add_apps(self, app: Application \| Iterable[Application]):	1✔
94	"""
95	Insert one or more applications into the app list.
96
97	Args:
98	app: an application or a list of applications.
99	The caller is responsible for ``deepcopy`` if needed, so that each node has a separate instance.
100
101	"""
102	self.ensure_not_installed()	1✔
103	if isinstance(app, Application):	1✔
104	self.apps.append(app)	1✔
105	else:
106	self.apps.extend(app)	1✔
107
108	def get_apps[A: Application](self, app_type: type[A]) -> list[A]:	1✔
109	"""
110	Retrieve applications of given type.
111
112	Args:
113	app_type: Application type/class.
114	"""
115	return [app for app in self.apps if isinstance(app, app_type)]	1✔
116
117	def get_app[A: Application](self, app_type: type[A]) -> A:	1✔
118	"""
119	Retrieve an application of given type.
120	There must be exactly one instance of this application.
121
122	Args:
123	app_type: Application type/class.
124
125	Raises:
126	LookupError: Application does not exist, or there are multiple instances.
127	"""
128	if self._app_by_type is None: # this is called before self.install() populates _app_by_type	1✔
129	self.ensure_not_installed()	×
130	return self._get_app_from_apps(app_type)	×
131
132	try:	1✔
133	return cast(A, self._app_by_type[app_type])	1✔
134	except KeyError:	1✔
135	# app_type is a base class
136	return self._get_app_from_apps(app_type)	1✔
137
138	def _get_app_from_apps[A: Application](self, app_type: type[A]) -> A:	1✔
139	apps = self.get_apps(app_type)	1✔
140	if len(apps) != 1:	1✔
141	raise LookupError(f"node does not have exactly one instance of {app_type}")	×
142	return apps[0]	1✔
143
144	def _add_channel[C: "BaseChannel"](self, channel: C, channels: list[C]) -> None:	1✔
145	self.ensure_not_installed()	1✔
146	channel.node_list.append(self)	1✔
147	channels.append(channel)	1✔
148
149	def _install_channels[C: "BaseChannel"](self, channels: list[C], by_neighbor: dict["Node", C]) -> None:	1✔
150	for ch in channels:	1✔
151	for dst in ch.node_list:	1✔
152	if dst is not self:	1✔
153	by_neighbor[dst] = ch	1✔
154	ch.install(self.simulator)	1✔
155
156	@staticmethod	1✔
157	def _get_channel[C: "BaseChannel"](dst: "Node", by_neighbor: dict["Node", C]) -> C:	1✔
158	return by_neighbor[dst]	1✔
159
160	def add_cchannel(self, cchannel: "ClassicChannel"):	1✔
161	"""
162	Add a classic channel in this Node.
163	This function is available prior to calling .install().
164	"""
165	self.ensure_not_installed()	1✔
166	self._add_channel(cchannel, self.cchannels)	1✔
167
168	def get_cchannel(self, dst: "Node") -> "ClassicChannel":	1✔
169	"""
170	Retrieve the classic channel that connects to ``dst``.
171
172	Raises:
173	LookupError: channel does not exist
174	"""
175	return self._get_channel(dst, self._cchannel_by_dst)	1✔
176
177	def send_cpacket(self, next_hop: "Node", pkt: "ClassicPacket") -> None:	1✔
178	"""
179	Send a classic packet to an adjacent node.
180
181	Args:
182	next_hop: an adjacent node that shares a classic channel with this node.
183	pkt: the packet.
184	"""
185	self.get_cchannel(next_hop).send(pkt, next_hop)	1✔
186
187	def add_network(self, network: "QuantumNetwork"):	1✔
188	"""
189	Assign a network object to this node.
190	"""
191	self.network = network	1✔
192	"""Quantum network that contains this node."""	1✔
193
194	@property	1✔
195	def timing(self) -> "TimingMode":	1✔
196	"""
197	Access the network-wide application timing mode.
198	"""
199	return self.network.timing	1✔
200
201	def __repr__(self) -> str:	1✔
UNCOV 202	return f"<node {self.name}>"	×

yoursunny / mqns / 30315099891

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous