Merge pull request #46 from QuTech-Delft/bugfix-depolarise-multithread

Bugfix depolarise link models multithread variant

CHANGELOG.md

@@ -1,7 +1,12 @@
 CHANGELOG
 =========
 
-2024-09-26 (0.13.1)
+2024-10-21 (0.13.2)
+------------------
+- Bugfix for link noise models depolarise and discrete depolarise in multithread variant
+- Added tests for link noise and hardware models for multithread variant
+
+2024-10-20 (0.13.1)
 ------------------
 - Bugfix for pydantic 2.x versions
 

squidasm/sim/network/network.py

@@ -1,4 +1,5 @@
 import copy
+import dataclasses
 import os
 from typing import Dict, List, Optional, Tuple
 
@@ -37,6 +38,7 @@
 from netsquid_magic.model_parameters import (
     BitFlipModelParameters,
     DepolariseModelParameters,
+    IModelParameters,
     PerfectModelParameters,
 )
 from netsquid_magic.state_delivery_sampler import HeraldedStateDeliverySamplerFactory
@@ -218,18 +220,22 @@ def _create_link_distributor(
                 noise_type == NoiseType.Depolarise
             ):  # use Depolarise distributor defined in this module
                 noise = 1 - link.fidelity
+                model_params = LinearDepolariseModelParameters(
+                    cycle_time=state_delay, prob_success=1, prob_max_mixed=noise
+                )
                 return LinearDepolariseMagicDistributor(
-                    nodes=[node1, node2], depolar_noise=noise, state_delay=state_delay
+                    nodes=[node1, node2], model_params=model_params
                 )
             elif (
                 noise_type == NoiseType.DiscreteDepolarise
             ):  # use Depolarise distributor defined in netsquid_magic
                 noise = 1 - link.fidelity
-                model_params = DepolariseModelParameters(prob_max_mixed=noise)
+                model_params = DepolariseModelParameters(
+                    prob_max_mixed=noise, cycle_time=state_delay
+                )
                 return DepolariseMagicDistributor(
                     nodes=[node1, node2],
                     model_params=model_params,
-                    state_delay=state_delay,
                 )
             elif noise_type == NoiseType.Bitflip:
                 flip_prob = 1 - link.fidelity
@@ -555,35 +561,45 @@ def __init__(
         )
 
 
+@dataclasses.dataclass
+class LinearDepolariseModelParameters(IModelParameters):
+    """Data class for the parameters of the linear depolarising entanglement generation model."""
+
+    prob_max_mixed: float = 0
+    """Fraction of maximally mixed state in the EPR state generated."""
+    prob_success: float = 1
+    """Probability of successfully generating an EPR state per cycle."""
+
+    def verify(self):
+        super().verify()
+        self.verify_between_0_and_1("prob_max_mixed")
+        self.verify_between_0_and_1("prob_success")
+
+
 class LinearDepolariseMagicDistributor(MagicDistributor):
     """
     Distributes (noisy) EPR pairs to 2 connected nodes, using samplers created
     by a :class:`LinearDepolariseStateSamplerFactory`.
     """
 
-    def __init__(self, nodes, depolar_noise, **kwargs):
+    def __init__(self, nodes, model_params: LinearDepolariseModelParameters, **kwargs):
         """
         Parameters
         ----------
         nodes : list of :obj:`~netsquid.nodes.node.Node`
             Pair of nodes to which noisy EPR pairs will be distributed.
-        depolar_noise : float
-            Depolarizing noise.
+        model_params : LinearDepolariseModelParameters
+            Model parameters for LinearDepolariseModel model, but will not use `prob_success`.
         """
-        self.depolar_noise = depolar_noise
         super().__init__(
             delivery_sampler_factory=LinearDepolariseStateSamplerFactory(),
+            model_params=model_params,
+            label_delay=0,
+            state_delay=model_params.cycle_time,
             nodes=nodes,
             **kwargs,
         )
 
-    def add_delivery(self, memory_positions, **kwargs):
-        return super().add_delivery(
-            memory_positions=memory_positions,
-            depolar_noise=self.depolar_noise,
-            **kwargs,
-        )
-
 
 class LinearDepolariseStateSamplerFactory(HeraldedStateDeliverySamplerFactory):
     """
@@ -597,13 +613,12 @@ def __init__(self):
         )
 
     @staticmethod
-    def _delivery_func(depolar_noise, **kwargs):
+    def _delivery_func(model_params: LinearDepolariseModelParameters, **kwargs):
         """
         Parameters
         ----------
-        depolar_noise : float
-            Used to calculate the linear combination of the original state
-            and the maximally mixed state.
+        model_params : LinearDepolariseModelParameters
+            Model parameters for LinearDepolariseModel model, but will not use `prob_success`.
 
         Returns
         -------
@@ -620,7 +635,8 @@ def _delivery_func(depolar_noise, **kwargs):
         return (
             StateSampler(
                 qreprs=[
-                    (1 - depolar_noise) * epr_state + depolar_noise * maximally_mixed
+                    (1 - model_params.prob_max_mixed) * epr_state
+                    + model_params.prob_max_mixed * maximally_mixed
                 ],
                 probabilities=[1],
             ),

tests/test_network.py

@@ -0,0 +1,134 @@
+import os
+import tempfile
+import unittest
+from typing import List
+
+from netqasm.examples.apps.teleport.app_receiver import main as receiver_main
+from netqasm.examples.apps.teleport.app_sender import main as sender_main
+from netqasm.runtime.application import Application, ApplicationInstance, Program
+from netqasm.runtime.env import get_timed_log_dir
+from netqasm.runtime.interface.config import (
+    Link,
+    NetworkConfig,
+    Node,
+    NoiseType,
+    QuantumHardware,
+    Qubit,
+)
+from netqasm.sdk.config import LogConfig
+
+from squidasm.run.multithread.runtime_mgr import SquidAsmRuntimeManager
+
+_DEFAULT_NUM_QUBITS = 5
+
+
+def get_network_config(
+    node_names: List[str],
+    hardware: QuantumHardware = QuantumHardware.Generic,
+    link_noise_type: NoiseType = NoiseType.NoNoise,
+) -> NetworkConfig:
+    """Create a config for a fully connected network of nodes with the given names"""
+    nodes = []
+    links = []
+    for name in node_names:
+        qubits = [Qubit(id=i, t1=0, t2=0) for i in range(_DEFAULT_NUM_QUBITS)]
+        node = Node(name=name, hardware=hardware, qubits=qubits, gate_fidelity=1)
+        nodes += [node]
+
+        for other_name in node_names:
+            if other_name == name:
+                continue
+            link = Link(
+                name=f"link_{name}_{other_name}",
+                node_name1=name,
+                node_name2=other_name,
+                noise_type=link_noise_type,
+                fidelity=1,
+            )
+            links += [link]
+
+    return NetworkConfig(nodes, links)
+
+
+class TestNetwork(unittest.TestCase):
+    @staticmethod
+    def run_network_test(
+        hardware: QuantumHardware = QuantumHardware.Generic,
+        link_noise_type: NoiseType = NoiseType.NoNoise,
+    ):
+        with tempfile.TemporaryDirectory() as dirpath:
+
+            mgr = SquidAsmRuntimeManager()
+            network_cfg = get_network_config(
+                ["delft", "amsterdam"],
+                hardware=hardware,
+                link_noise_type=link_noise_type,
+            )
+
+            prog_sender = Program(
+                party="sender", entry=sender_main, args=["app_config"], results=[]
+            )
+            prog_receiver = Program(
+                party="receiver", entry=receiver_main, args=["app_config"], results=[]
+            )
+
+            log_cfg = LogConfig(
+                track_lines=False, log_subroutines_dir=dirpath, comm_log_dir=dirpath
+            )
+
+            app = Application(programs=[prog_sender, prog_receiver], metadata=None)
+            app_instance = ApplicationInstance(
+                app=app,
+                program_inputs={
+                    "sender": {},
+                    "receiver": {},
+                },
+                network=None,
+                party_alloc={
+                    "sender": "delft",
+                    "receiver": "amsterdam",
+                },
+                logging_cfg=log_cfg,
+            )
+
+            mgr.set_network(network_cfg)
+            backend_log_dir = os.path.join(dirpath, "backend_log")
+            if not os.path.exists(backend_log_dir):
+                os.mkdir(backend_log_dir)
+            mgr.backend_log_dir = backend_log_dir
+            mgr.start_backend()
+            for i in range(1):
+                print(f"\niteration {i}")
+                if app_instance.logging_cfg is not None:
+                    log_dir = get_timed_log_dir(dirpath)
+                    app_instance.logging_cfg.log_subroutines_dir = log_dir
+                    app_instance.logging_cfg.comm_log_dir = log_dir
+                results = mgr.run_app(app_instance)
+                print(f"results: {results}")
+            mgr.stop_backend()
+
+    def test_generic(self):
+        self.run_network_test(
+            hardware=QuantumHardware.Generic, link_noise_type=NoiseType.NoNoise
+        )
+
+    def test_nv(self):
+        self.run_network_test(
+            hardware=QuantumHardware.NV, link_noise_type=NoiseType.NoNoise
+        )
+
+    def test_depolarise(self):
+        self.run_network_test(
+            hardware=QuantumHardware.Generic, link_noise_type=NoiseType.Depolarise
+        )
+
+    def test_discrete_depolarise(self):
+        self.run_network_test(
+            hardware=QuantumHardware.Generic,
+            link_noise_type=NoiseType.DiscreteDepolarise,
+        )
+
+    def test_bitflip(self):
+        self.run_network_test(
+            hardware=QuantumHardware.Generic, link_noise_type=NoiseType.Bitflip
+        )