20960597173

Committed 13 Jan 2026 02:32PM UTC coverage: 80.367% (-0.03%) from 80.395%

Build # 20960597173

Build Type

push

github

Committed by

yoursunny

Commit Message

node: deduplicate app.node and app.own

Run Details

171 of 183 new or added lines in 22 files covered. (93.44%)

3 existing lines in 3 files now uncovered.

4077 of 5073 relevant lines covered (80.37%)

0.8 hits per line

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

91.67

/mqns/network/protocol/node_process_delay.py

#    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 mqns.entity.node import Application, Node
from mqns.simulator import Event


class NodeProcessDelayApp(Application[Node]):
    """This application will add an addition delay whenever the node received an event.
    It is used to represent the processing delay on quantum nodes.
    """

    def __init__(self, delay: float = 0, delay_event_list: type | tuple[type, ...] | None = None):
        """Args:
        delay (float): the processing delay
        delay_event_list: a list of Event classic list that will add a delay.
            If `delay_event_list` is None, all events will be added a delay.

        """
        super().__init__()
        self.delay = delay
        self.delay_event_list = delay_event_list
        self.wait_rehandle_event_list = []

    def check_in_delay_event_list(self, event) -> bool:
        if self.delay_event_list is None:
            return True
        return isinstance(event, self.delay_event_list)

    def handle(self, event: Event) -> bool:
        if not self.check_in_delay_event_list(event):
            return False

        if event in self.wait_rehandle_event_list:
            self.wait_rehandle_event_list.remove(event)
            return False

        # first handle the event
        # add to list
        self.wait_rehandle_event_list.append(event)
        # get the delay time
        t = self.simulator.tc + self.delay
        # reset event's occur time
        event.t = t
        event.by = self
        self.simulator.add_event(event)
        return True

1	# SimQN: a discrete-event simulator for the quantum networks
2	# Copyright (C) 2021-2022 Lutong Chen, Jian Li, Kaiping Xue
3	# University of Science and Technology of China, USTC.
4	#
5	# This program is free software: you can redistribute it and/or modify
6	# it under the terms of the GNU General Public License as published by
7	# the Free Software Foundation, either version 3 of the License, or
8	# (at your option) any later version.
9	#
10	# This program is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU General Public License for more details.
14	#
15	# You should have received a copy of the GNU General Public License
16	# along with this program. If not, see <https://www.gnu.org/licenses/>.
17
18	from mqns.entity.node import Application, Node	1✔
19	from mqns.simulator import Event	1✔
20
21
22	class NodeProcessDelayApp(Application[Node]):	1✔
23	"""This application will add an addition delay whenever the node received an event.
24	It is used to represent the processing delay on quantum nodes.
25	"""
26
27	def __init__(self, delay: float = 0, delay_event_list: type \| tuple[type, ...] \| None = None):	1✔
28	"""Args:
29	delay (float): the processing delay
30	delay_event_list: a list of Event classic list that will add a delay.
31	If `delay_event_list` is None, all events will be added a delay.
32
33	"""
34	super().__init__()	1✔
35	self.delay = delay	1✔
36	self.delay_event_list = delay_event_list	1✔
37	self.wait_rehandle_event_list = []	1✔
38
39	def check_in_delay_event_list(self, event) -> bool:	1✔
40	if self.delay_event_list is None:	1✔
41	return True	×
42	return isinstance(event, self.delay_event_list)	1✔
43
44	def handle(self, event: Event) -> bool:	1✔
45	if not self.check_in_delay_event_list(event):	1✔
UNCOV 46	return False	×
47
48	if event in self.wait_rehandle_event_list:	1✔
49	self.wait_rehandle_event_list.remove(event)	1✔
50	return False	1✔
51
52	# first handle the event
53	# add to list
54	self.wait_rehandle_event_list.append(event)	1✔
55	# get the delay time
56	t = self.simulator.tc + self.delay	1✔
57	# reset event's occur time
58	event.t = t	1✔
59	event.by = self	1✔
60	self.simulator.add_event(event)	1✔
61	return True	1✔

yoursunny / mqns / 20960597173

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