20963379303

Committed 13 Jan 2026 03:57PM UTC coverage: 84.329% (+4.0%) from 80.28%

Build # 20963379303

Build Type

push

github

Committed by

yoursunny

Commit Message

simulator: eliminate default_accuracy

Run Details

9 of 9 new or added lines in 5 files covered. (100.0%)

43 existing lines in 8 files now uncovered.

4278 of 5073 relevant lines covered (84.33%)

0.84 hits per line

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

92.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, ApplicationT
from mqns.simulator import Event, Simulator

if TYPE_CHECKING:
    from mqns.entity.base_channel import ChannelT
    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(self, app_type: type[ApplicationT]) -> list[ApplicationT]:
        """
        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(self, app_type: type[ApplicationT]) -> ApplicationT:
        """
        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()
            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]

        return cast(ApplicationT, self._app_by_type[app_type])

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

    def _install_channels(
        self, typ: type["ChannelT"], channels: list["ChannelT"], by_neighbor: dict["Node", "ChannelT"]
    ) -> 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(dst: "Node", by_neighbor: dict["Node", "ChannelT"]) -> "ChannelT":
        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, ApplicationT	1✔
33	from mqns.simulator import Event, Simulator	1✔
34
35	if TYPE_CHECKING:
36	from mqns.entity.base_channel import ChannelT
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(self, app_type: type[ApplicationT]) -> list[ApplicationT]:	1✔
109	"""
110	Retrieve applications of given type.
111
112	Args:
113	app_type: Application type/class.
114	"""
UNCOV 115	return [app for app in self.apps if isinstance(app, app_type)]	×
116
117	def get_app(self, app_type: type[ApplicationT]) -> ApplicationT:	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✔
UNCOV 129	self.ensure_not_installed()	×
UNCOV 130	apps = self.get_apps(app_type)	×
UNCOV 131	if len(apps) != 1:	×
UNCOV 132	raise IndexError("node does not have exactly one instance of {app_type}")	×
UNCOV 133	return apps[0]	×
134
135	return cast(ApplicationT, self._app_by_type[app_type])	1✔
136
137	def _add_channel(self, channel: "ChannelT", channels: list["ChannelT"]) -> None:	1✔
138	self.ensure_not_installed()	1✔
139	channel.node_list.append(self)	1✔
140	channels.append(channel)	1✔
141
142	def _install_channels(	1✔
143	self, typ: type["ChannelT"], channels: list["ChannelT"], by_neighbor: dict["Node", "ChannelT"]
144	) -> None:
145	for ch in channels:	1✔
146	assert isinstance(ch, typ)	1✔
147	for dst in ch.node_list:	1✔
148	if dst != self:	1✔
149	by_neighbor[dst] = ch	1✔
150	ch.install(self.simulator)	1✔
151
152	@staticmethod	1✔
153	def _get_channel(dst: "Node", by_neighbor: dict["Node", "ChannelT"]) -> "ChannelT":	1✔
154	return by_neighbor[dst]	1✔
155
156	def add_cchannel(self, cchannel: "ClassicChannel"):	1✔
157	"""
158	Add a classic channel in this Node.
159	This function is available prior to calling .install().
160	"""
161	self.ensure_not_installed()	1✔
162	self._add_channel(cchannel, self.cchannels)	1✔
163
164	def get_cchannel(self, dst: "Node") -> "ClassicChannel":	1✔
165	"""
166	Retrieve the classic channel that connects to `dst`.
167
168	Raises:
169	IndexError - channel does not exist
170	"""
171	return self._get_channel(dst, self._cchannel_by_dst)	1✔
172
173	def add_network(self, network: "QuantumNetwork"):	1✔
174	"""
175	Assign a network object to this node.
176	"""
177	self.network = network	1✔
178	"""Quantum network that contains this node."""	1✔
179
180	@property	1✔
181	def timing(self) -> "TimingMode":	1✔
182	"""
183	Access the network-wide application timing mode.
184	"""
185	return self.network.timing	1✔
186
187	def __repr__(self) -> str:	1✔
188	return f"<node {self.name}>"	1✔

yoursunny / mqns / 20963379303

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