27423189236

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

Build # 27423189236

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #136 from yoursunny/edispatch

simulator: EventDispatcherMixin and other cleanups

Coverage Stats

71 of 87 new or added lines in 20 files covered. (81.61%)

2 existing lines in 2 files now uncovered.

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

97.73

/mqns/entity/node/qnode.py

#    Multiverse Quantum Network Simulator: a simulator for comparative
#    evaluation of quantum routing strategies
#    Copyright (C) [2025] Amar Abane
#
#    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 TYPE_CHECKING

from mqns.entity.node.app import Application
from mqns.entity.node.node import Node
from mqns.simulator import Simulator

if TYPE_CHECKING:
    from mqns.entity.memory import QuantumMemory
    from mqns.entity.operator import QuantumOperator
    from mqns.entity.qchannel import QuantumChannel


class QNode(Node):
    """QNode is a quantum node in the quantum network. Inherits Node and add quantum elements."""

    def __init__(self, name: str, *, apps: list[Application] | None = None):
        """
        Args:
            name: node name
            apps: applications on the node.
        """
        super().__init__(name, apps=apps)
        self.qchannels: list["QuantumChannel"] = []
        """Quantum channels connected to this node."""
        self._qchannel_by_dst = dict[Node, "QuantumChannel"]()
        self._memory: "QuantumMemory|None" = None
        self.operators: list["QuantumOperator"] = []

    def install(self, simulator: Simulator) -> None:
        super().install(simulator)
        # initiate sub-entities
        from mqns.entity.memory import QuantumMemory  # noqa: PLC0415
        from mqns.entity.operator import QuantumOperator  # noqa: PLC0415
        from mqns.entity.qchannel import QuantumChannel  # noqa: PLC0415

        if self._memory is not None:
            assert isinstance(self._memory, QuantumMemory)
            self._memory.install(simulator)
        for operator in self.operators:
            assert isinstance(operator, QuantumOperator)
            operator.install(simulator)

        self._install_channels(QuantumChannel, self.qchannels, self._qchannel_by_dst)

    @property
    def memory(self) -> "QuantumMemory":
        """
        Retrieve associated QuantumMemory.

        Raises:
            LookupError: Memory does not exist.
        """
        if self._memory is None:
            raise LookupError(f"node {self} does not have memory")
        return self._memory

    @memory.setter
    def memory(self, value: "QuantumMemory"):
        """
        Assign QuantumMemory to this node.
        This setter is available prior to calling .install().
        """
        self.ensure_not_installed()
        value.node = self
        self._memory = value

    def add_operator(self, operator: "QuantumOperator"):
        """Add a quantum operator in this node

        Args:
            operator: the quantum operator

        This function is available prior to calling .install().
        """
        self.ensure_not_installed()
        operator.set_own(self)
        self.operators.append(operator)

    def add_qchannel(self, qchannel: "QuantumChannel"):
        """
        Add a quantum channel in this QNode.
        This function is available prior to calling .install().
        """
        self._add_channel(qchannel, self.qchannels)

    def get_qchannel(self, dst: "QNode") -> "QuantumChannel":
        """
        Retrieve the quantum channel that connects to ``dst``.

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

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

1	# Multiverse Quantum Network Simulator: a simulator for comparative
2	# evaluation of quantum routing strategies
3	# Copyright (C) [2025] Amar Abane
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 TYPE_CHECKING	1✔
19
20	from mqns.entity.node.app import Application	1✔
21	from mqns.entity.node.node import Node	1✔
22	from mqns.simulator import Simulator	1✔
23
24	if TYPE_CHECKING:
25	from mqns.entity.memory import QuantumMemory
26	from mqns.entity.operator import QuantumOperator
27	from mqns.entity.qchannel import QuantumChannel
28
29
30	class QNode(Node):	1✔
31	"""QNode is a quantum node in the quantum network. Inherits Node and add quantum elements."""
32
33	def __init__(self, name: str, *, apps: list[Application] \| None = None):	1✔
34	"""
35	Args:
36	name: node name
37	apps: applications on the node.
38	"""
39	super().__init__(name, apps=apps)	1✔
40	self.qchannels: list["QuantumChannel"] = []	1✔
41	"""Quantum channels connected to this node."""	1✔
42	self._qchannel_by_dst = dict[Node, "QuantumChannel"]()	1✔
43	self._memory: "QuantumMemory\|None" = None	1✔
44	self.operators: list["QuantumOperator"] = []	1✔
45
46	def install(self, simulator: Simulator) -> None:	1✔
47	super().install(simulator)	1✔
48	# initiate sub-entities
49	from mqns.entity.memory import QuantumMemory # noqa: PLC0415	1✔
50	from mqns.entity.operator import QuantumOperator # noqa: PLC0415	1✔
51	from mqns.entity.qchannel import QuantumChannel # noqa: PLC0415	1✔
52
53	if self._memory is not None:	1✔
54	assert isinstance(self._memory, QuantumMemory)	1✔
55	self._memory.install(simulator)	1✔
56	for operator in self.operators:	1✔
57	assert isinstance(operator, QuantumOperator)	1✔
58	operator.install(simulator)	1✔
59
60	self._install_channels(QuantumChannel, self.qchannels, self._qchannel_by_dst)	1✔
61
62	@property	1✔
63	def memory(self) -> "QuantumMemory":	1✔
64	"""
65	Retrieve associated QuantumMemory.
66
67	Raises:
68	LookupError: Memory does not exist.
69	"""
70	if self._memory is None:	1✔
NEW 71	raise LookupError(f"node {self} does not have memory")	×
72	return self._memory	1✔
73
74	@memory.setter	1✔
75	def memory(self, value: "QuantumMemory"):	1✔
76	"""
77	Assign QuantumMemory to this node.
78	This setter is available prior to calling .install().
79	"""
80	self.ensure_not_installed()	1✔
81	value.node = self	1✔
82	self._memory = value	1✔
83
84	def add_operator(self, operator: "QuantumOperator"):	1✔
85	"""Add a quantum operator in this node
86
87	Args:
88	operator: the quantum operator
89
90	This function is available prior to calling .install().
91	"""
92	self.ensure_not_installed()	1✔
93	operator.set_own(self)	1✔
94	self.operators.append(operator)	1✔
95
96	def add_qchannel(self, qchannel: "QuantumChannel"):	1✔
97	"""
98	Add a quantum channel in this QNode.
99	This function is available prior to calling .install().
100	"""
101	self._add_channel(qchannel, self.qchannels)	1✔
102
103	def get_qchannel(self, dst: "QNode") -> "QuantumChannel":	1✔
104	"""
105	Retrieve the quantum channel that connects to ``dst``.
106
107	Raises:
108	LookupError: channel does not exist
109	"""
110	return self._get_channel(dst, self._qchannel_by_dst)	1✔
111
112	def __repr__(self) -> str:	1✔
113	return f"<qnode {self.name}>"	1✔

yoursunny / mqns / 27423189236

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