27369341227

Committed 11 Jun 2026 06:39PM UTC coverage: 84.78% (+1.2%) from 83.556%

Build # 27369341227

Build Type

push

github

Committed by

yoursunny

Commit Message

docs: add CITATION.cff file

Coverage Stats

5587 of 6590 relevant lines covered (84.78%)

0.85 hits per line

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

92.0

/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 typing import override

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: the processing delay.
            delay_event_list: a list of Event classic list that will add a delay.
                              If 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)

    @override
    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
        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 typing import override	1✔
19
20	from mqns.entity.node import Application, Node	1✔
21	from mqns.simulator import Event	1✔
22
23
24	class NodeProcessDelayApp(Application[Node]):	1✔
25	"""This application will add an addition delay whenever the node received an event.
26	It is used to represent the processing delay on quantum nodes.
27	"""
28
29	def __init__(self, delay: float = 0, delay_event_list: type \| tuple[type, ...] \| None = None):	1✔
30	"""
31	Args:
32	delay: the processing delay.
33	delay_event_list: a list of Event classic list that will add a delay.
34	If None, all events will be added a delay.
35
36	"""
37	super().__init__()	1✔
38	self.delay = delay	1✔
39	self.delay_event_list = delay_event_list	1✔
40	self.wait_rehandle_event_list = []	1✔
41
42	def check_in_delay_event_list(self, event) -> bool:	1✔
43	if self.delay_event_list is None:	1✔
44	return True	×
45	return isinstance(event, self.delay_event_list)	1✔
46
47	@override	1✔
48	def handle(self, event: Event) -> bool:	1✔
49	if not self.check_in_delay_event_list(event):	1✔
50	return False	×
51
52	if event in self.wait_rehandle_event_list:	1✔
53	self.wait_rehandle_event_list.remove(event)	1✔
54	return False	1✔
55
56	# first handle the event
57	# add to list
58	self.wait_rehandle_event_list.append(event)	1✔
59	# get the delay time
60	t = self.simulator.tc + self.delay	1✔
61	# reset event's occur time
62	event.t = t	1✔
63	self.simulator.add_event(event)	1✔
64	return True	1✔

yoursunny / mqns / 27369341227

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