22241103399

Committed 20 Feb 2026 09:02PM UTC coverage: 82.114% (-0.04%) from 82.155%

Build # 22241103399

Build Type

push

github

Committed by

yoursunny

Commit Message

examples: ReactiveForwarder in 3_nodes_thruput

Run Details

3 of 10 new or added lines in 1 file covered. (30.0%)

82 existing lines in 1 file now uncovered.

4839 of 5893 relevant lines covered (82.11%)

0.82 hits per line

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

90.0

/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 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
    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:
        """Called from Network.install()"""
        super().install(simulator)
        # initiate sub-entities
        from mqns.entity.cchannel import ClassicChannel  # noqa: PLC0415

        self._install_channels(ClassicChannel, self.cchannels, self._cchannel_by_dst)

        # initiate applications
        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]

    @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 (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 | list[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, list):
            self.apps += app
        else:
            self.apps.append(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:
            IndexError - 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 IndexError("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, typ: type[C], channels: list[C], by_neighbor: dict["Node", C]) -> None:
        for ch in channels:
            assert isinstance(ch, typ)
            for dst in ch.node_list:
                if dst != 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:
            IndexError - channel does not exist
        """
        return self._get_channel(dst, self._cchannel_by_dst)

    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 typing import TYPE_CHECKING, cast, override	1✔
30
31	from mqns.entity.entity import Entity	1✔
32	from mqns.entity.node.app import Application	1✔
33	from mqns.simulator import Event, Simulator	1✔
34
35	if TYPE_CHECKING:
36	from mqns.entity.base_channel import BaseChannel
37	from mqns.entity.cchannel import ClassicChannel
38	from mqns.network.network import QuantumNetwork, TimingMode
39
40
41	class Node(Entity):	1✔
42	"""Node is a generic node in the quantum network"""
43
44	def __init__(self, name: str, *, apps: list[Application] \| None = None):	1✔
45	"""
46	Args:
47	name: node name.
48	apps: applications on the node.
49	"""
50	super().__init__(name)	1✔
51	self.cchannels: list["ClassicChannel"] = []	1✔
52	"""Classic channels connected to this node."""	1✔
53	self._cchannel_by_dst = dict["Node", "ClassicChannel"]()	1✔
54	self.apps: list[Application] = [] if apps is None else apps	1✔
55	"""Applications on this node."""	1✔
56	self._app_by_type: dict[type, Application] \| None = None	1✔
57
58	@override	1✔
59	def install(self, simulator: Simulator) -> None:	1✔
60	"""Called from Network.install()"""
61	super().install(simulator)	1✔
62	# initiate sub-entities
63	from mqns.entity.cchannel import ClassicChannel # noqa: PLC0415	1✔
64
65	self._install_channels(ClassicChannel, self.cchannels, self._cchannel_by_dst)	1✔
66
67	# initiate applications
68	apps_by_type = defaultdict[type, list[Application]](lambda: [])	1✔
69	for app in self.apps:	1✔
70	apps_by_type[type(app)].append(app)	1✔
71	app.install(self)	1✔
72
73	self._app_by_type = {}	1✔
74	for typ, apps in apps_by_type.items():	1✔
75	if len(apps) == 1:	1✔
76	self._app_by_type[typ] = apps[0]	1✔
77
78	@override	1✔
79	def handle(self, event: Event) -> None:	1✔
80	"""
81	Dispatch an `Event` that happens on this Node.
82	This event is passed to every application in apps list in order.
83
84	Args:
85	event (Event): the event that happens on this Node
86
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 \| list[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, list):	1✔
104	self.apps += app	1✔
105	else:
106	self.apps.append(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	IndexError - 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()	×
NEW 130	return self._get_app_from_apps(app_type)	×
131
132	try:	1✔
133	return cast(A, self._app_by_type[app_type])	1✔
NEW 134	except KeyError:	×
135	# app_type is a base class
NEW 136	return self._get_app_from_apps(app_type)	×
137
138	def _get_app_from_apps[A: Application](self, app_type: type[A]) -> A:	1✔
NEW 139	apps = self.get_apps(app_type)	×
NEW 140	if len(apps) != 1:	×
NEW 141	raise IndexError("node does not have exactly one instance of {app_type}")	×
NEW 142	return apps[0]	×
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, typ: type[C], channels: list[C], by_neighbor: dict["Node", C]) -> None:	1✔
150	for ch in channels:	1✔
151	assert isinstance(ch, typ)	1✔
152	for dst in ch.node_list:	1✔
153	if dst != self:	1✔
154	by_neighbor[dst] = ch	1✔
155	ch.install(self.simulator)	1✔
156
157	@staticmethod	1✔
158	def _get_channel[C: "BaseChannel"](dst: "Node", by_neighbor: dict["Node", C]) -> C:	1✔
159	return by_neighbor[dst]	1✔
160
161	def add_cchannel(self, cchannel: "ClassicChannel"):	1✔
162	"""
163	Add a classic channel in this Node.
164	This function is available prior to calling .install().
165	"""
166	self.ensure_not_installed()	1✔
167	self._add_channel(cchannel, self.cchannels)	1✔
168
169	def get_cchannel(self, dst: "Node") -> "ClassicChannel":	1✔
170	"""
171	Retrieve the classic channel that connects to `dst`.
172
173	Raises:
174	IndexError - channel does not exist
175	"""
176	return self._get_channel(dst, self._cchannel_by_dst)	1✔
177
178	def add_network(self, network: "QuantumNetwork"):	1✔
179	"""
180	Assign a network object to this node.
181	"""
182	self.network = network	1✔
183	"""Quantum network that contains this node."""	1✔
184
185	@property	1✔
186	def timing(self) -> "TimingMode":	1✔
187	"""
188	Access the network-wide application timing mode.
189	"""
190	return self.network.timing	1✔
191
192	def __repr__(self) -> str:	1✔
193	return f"<node {self.name}>"	1✔

yoursunny / mqns / 22241103399

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