[Refac] Add pyq noise

qadence/backend.py

@@ -269,6 +269,8 @@ def run(
     ) -> ArrayLike:
         """Run a circuit and return the resulting wave function.
 
+        #TODO : Add the possibility to run noisy circuit
+
         Arguments:
             circuit: A converted circuit as returned by `backend.circuit`.
             param_values: _**Already embedded**_ parameters of the circuit. See

qadence/backends/pyqtorch/convert_ops.py

@@ -27,6 +27,7 @@
 from torch.nn import Module, ParameterDict
 
 from qadence.backends.utils import (
+    block_noisy_protocols,
     finitediff,
     pyqify,
     unpyqify,
@@ -98,7 +99,7 @@ def convert_block(
     if config is None:
         config = Configuration()
 
-    if isinstance(block, ScaleBlock):
+    if isinstance(block, ScaleBlock):  #!Look to add noise
         scaled_ops = convert_block(block.block, n_qubits, config)
         scale = extract_parameter(block, config)
         return [pyq.Scale(pyq.Sequence(scaled_ops), scale)]
@@ -138,42 +139,59 @@ def convert_block(
             ]
 
     elif isinstance(block, MatrixBlock):
-        return [pyq.primitives.Primitive(block.matrix, block.qubit_support)]
+        return [pyq.primitive.Primitive(block.matrix, block.qubit_support, block.noise)]
     elif isinstance(block, CompositeBlock):
         ops = list(flatten(*(convert_block(b, n_qubits, config) for b in block.blocks)))
         if isinstance(block, AddBlock):
             return [pyq.Add(ops)]  # add
-        elif is_single_qubit_chain(block) and config.use_single_qubit_composition:
-            return [pyq.Merge(ops)]  # for chains of single qubit ops on the same qubit
+        elif (
+            is_single_qubit_chain(block)
+            and config.use_single_qubit_composition
+            and all([b.noise is None for b in block])  # type: ignore[attr-defined]
+        ):
+            return [
+                pyq.Merge(ops)
+            ]  # for chains of single qubit ops on the same qubit without noise
         else:
-            return [pyq.Sequence(ops)]  # for kron and chain
-    elif isinstance(block, tuple(non_unitary_gateset)):
+            return [pyq.Sequence(ops)]  # for kron and chain with multiple qubits/1-qubit with noise
+    elif isinstance(block, tuple(non_unitary_gateset)):  #!Look to add noise
+        pyq_noise = block_noisy_protocols(block)
         if isinstance(block, ProjectorBlock):
             projector = getattr(pyq, block.name)
             if block.name == OpName.N:
-                return [projector(target=qubit_support)]
+                return [projector(target=qubit_support, noise=pyq_noise)]
             else:
-                return [projector(qubit_support=qubit_support, ket=block.ket, bra=block.bra)]
+                return [
+                    projector(
+                        qubit_support=qubit_support,
+                        ket=block.ket,
+                        bra=block.bra,
+                        #!noise=pyq_noise Need to add noise here but got error
+                    )
+                ]
         else:
             return [getattr(pyq, block.name)(qubit_support[0])]
     elif isinstance(block, tuple(single_qubit_gateset)):
         pyq_cls = getattr(pyq, block.name)
+        pyq_noise = block_noisy_protocols(block)
         if isinstance(block, ParametricBlock):
             if isinstance(block, U):
-                op = pyq_cls(qubit_support[0], *config.get_param_name(block))
+                op = pyq_cls(qubit_support[0], *config.get_param_name(block), pyq_noise)
             else:
-                op = pyq_cls(qubit_support[0], extract_parameter(block, config))
+                op = pyq_cls(qubit_support[0], extract_parameter(block, config), pyq_noise)
         else:
-            op = pyq_cls(qubit_support[0])
+            op = pyq_cls(qubit_support[0], pyq_noise)  # type: ignore[attr-defined]
         return [op]
-    elif isinstance(block, tuple(two_qubit_gateset)):
+    elif isinstance(block, tuple(two_qubit_gateset)):  #!Look to add noise
         pyq_cls = getattr(pyq, block.name)
         if isinstance(block, ParametricBlock):
             op = pyq_cls(qubit_support[0], qubit_support[1], extract_parameter(block, config))
         else:
             op = pyq_cls(qubit_support[0], qubit_support[1])
         return [op]
-    elif isinstance(block, tuple(three_qubit_gateset) + tuple(multi_qubit_gateset)):
+    elif isinstance(
+        block, tuple(three_qubit_gateset) + tuple(multi_qubit_gateset)
+    ):  #!Look to add noise
         block_name = block.name[1:] if block.name.startswith("M") else block.name
         pyq_cls = getattr(pyq, block_name)
         if isinstance(block, ParametricBlock):

qadence/backends/utils.py

@@ -9,7 +9,9 @@
 import torch
 from numpy.typing import ArrayLike
 from pyqtorch.apply import apply_operator
+from pyqtorch.noise import NoiseProtocol
 from pyqtorch.primitives import Parametric as PyQParametric
+from pyqtorch.utils import DensityMatrix
 from torch import (
     Tensor,
     cat,
@@ -20,6 +22,8 @@
     rand,
 )
 
+from qadence.blocks import AbstractBlock
+from qadence.noise import Noise
 from qadence.types import Endianness, ParamDictType
 from qadence.utils import int_to_basis, is_qadence_shape
 
@@ -121,7 +125,13 @@ def pyqify(state: Tensor, n_qubits: int = None) -> ArrayLike:
 
 
 def unpyqify(state: Tensor) -> Tensor:
-    """Convert a state of shape [2] * n_qubits + [batch_size] to (batch_size, 2**n_qubits)."""
+    """Convert a state of shape [2] * n_qubits + [batch_size] to (batch_size, 2**n_qubits).
+
+    Convert a density matrix of shape (2**n_qubits, 2**n_qubits, batch_size)
+    to (batch_size, 2**n_qubits, 2**n_qubits)
+    """
+    if isinstance(state, DensityMatrix):
+        return torch.einsum("ijk->kij", state)
     return torch.flatten(state, start_dim=0, end_dim=-2).t()
 
 
@@ -249,3 +259,17 @@ def dydxx(
         ),
         values[op.param_name],
     )
+
+
+def block_noisy_protocols(
+    block: AbstractBlock,
+) -> NoiseProtocol | dict[str, NoiseProtocol] | None:
+    pyq_noise = block.noise  # type: ignore[attr-defined]
+    if pyq_noise:
+        if isinstance(pyq_noise, dict):
+            pyq_noise = {
+                noise: noise_instance.to_pyq() for noise, noise_instance in pyq_noise.items()
+            }
+        elif isinstance(pyq_noise, Noise):
+            pyq_noise = pyq_noise.to_pyq()
+    return pyq_noise

qadence/blocks/matrix.py

@@ -8,6 +8,7 @@
 from torch.linalg import eigvals
 
 from qadence.blocks import PrimitiveBlock
+from qadence.noise import Noise
 
 logger = getLogger(__name__)
 
@@ -64,6 +65,7 @@ def __init__(
         self,
         matrix: torch.Tensor | np.ndarray,
         qubit_support: tuple[int, ...],
+        noise: Noise | dict[str, Noise] | None = None,
         check_unitary: bool = True,
         check_hermitian: bool = False,
     ) -> None:
@@ -82,7 +84,7 @@ def __init__(
             if not self.is_unitary(matrix):
                 logger.warning("Provided matrix is not unitary.")
         self.matrix = matrix.clone()
-        super().__init__(qubit_support)
+        super().__init__(qubit_support, noise)
 
     @cached_property
     def eigenvalues_generator(self) -> torch.Tensor:

qadence/blocks/primitive.py

@@ -11,6 +11,7 @@
 from rich.tree import Tree
 
 from qadence.blocks.abstract import AbstractBlock
+from qadence.noise import Noise
 from qadence.parameters import (
     Parameter,
     ParamMap,
@@ -33,13 +34,20 @@ class PrimitiveBlock(AbstractBlock):
 
     name = "PrimitiveBlock"
 
-    def __init__(self, qubit_support: tuple[int, ...]):
+    def __init__(
+        self, qubit_support: tuple[int, ...], noise: Noise | dict[str, Noise] | None = None
+    ):
         self._qubit_support = qubit_support
+        self._noise = noise
 
     @property
     def qubit_support(self) -> Tuple[int, ...]:
         return self._qubit_support
 
+    @property
+    def noise(self) -> Noise | dict[str, Noise] | None:
+        return self._noise
+
     def digital_decomposition(self) -> AbstractBlock:
         """Decomposition into purely digital gates.
 
@@ -48,6 +56,10 @@ def digital_decomposition(self) -> AbstractBlock:
         'gates', by manual/custom knowledge of how this can be done efficiently.
         :return:
         """
+        if self.noise is None:
+            raise ValueError(
+                "Decomposition into purely digital gates is only avalaible for unitary gate"
+            )
         return self
 
     def __len__(self) -> int:
@@ -77,22 +89,56 @@ def __eq__(self, other: object) -> bool:
         if not isinstance(other, AbstractBlock):
             raise TypeError(f"Cant compare {type(self)} to {type(other)}")
         if isinstance(other, type(self)):
-            return self.qubit_support == other.qubit_support
+            return self.qubit_support == other.qubit_support and self.noise == other.noise
         return False
 
     def _to_dict(self) -> dict:
+        if self.noise is None:
+            noise_info = None
+        elif isinstance(self.noise, Noise):
+            noise_info = self.noise._to_dict()
+        elif isinstance(self.noise, dict):
+            noise_info = {k: v._to_dict() for k, v in self.noise.items()}
+        else:
+            noise_info = dict()
         return {
             "type": type(self).__name__,
             "qubit_support": self.qubit_support,
             "tag": self.tag,
+            "noise": noise_info,
         }
 
     @classmethod
     def _from_dict(cls, d: dict) -> PrimitiveBlock:
-        return cls(*d["qubit_support"])
+        noise = d.get("noise")
+        if isinstance(noise, dict):
+            noise = {k: Noise._from_dict(v) for k, v in noise.items()}
+        elif noise is not None:
+            noise = Noise._from_dict(noise)
+        return cls(tuple(d["qubit_support"]), noise)
+
+    @property
+    def _block_title(self) -> str:
+        bits = ",".join(str(i) for i in self.qubit_support)
+        s = f"{type(self).__name__}({bits})"
+        if self.noise:
+            noise_details = []
+            if isinstance(self.noise, Noise):
+                noise_details.append(
+                    f"{self.noise.protocol}({self.noise.options['error_probability']})"
+                )
+            elif isinstance(self.noise, dict):
+                for noise_instance in self.noise.values():
+                    noise_details.append(
+                        f"{noise_instance.protocol}({noise_instance.options['error_probability']})"
+                    )
+            s += ", Noise: " + ", ".join(noise_details)
+        if self.tag is not None:
+            s += rf" \[tag: {self.tag}]"
+        return s
 
     def __hash__(self) -> int:
-        return hash(self._to_json())
+        return hash(self._to_json())  # TODO: To modify
 
     @property
     def n_qubits(self) -> int:
@@ -103,6 +149,7 @@ def n_supports(self) -> int:
         return len(self.qubit_support)
 
     def dagger(self) -> PrimitiveBlock:
+        # Do not do the dagger of the noise gate. Only of the primitive one.
         return self
 
 
@@ -131,7 +178,11 @@ def _block_title(self) -> str:
                 params_str.append(val)
             else:
                 params_str.append(stringify(p))
-
+        if self.noise:
+            noise_index = s.find(", Noise")
+            primitive_part = s[:noise_index].strip()
+            noise_part = s[noise_index:].strip()
+            return f"{primitive_part} [params: {params_str}]{noise_part}"
         return s + rf" \[params: {params_str}]"
 
     @property
@@ -179,6 +230,7 @@ def __eq__(self, other: object) -> bool:
             return (
                 self.qubit_support == other.qubit_support
                 and self.parameters.parameter == other.parameters.parameter
+                and self.noise == other.noise
             )
         return False
 
@@ -191,20 +243,38 @@ def __contains__(self, other: object) -> bool:
         return False
 
     def _to_dict(self) -> dict:
+        if self.noise is None:
+            noise_info = None
+        elif isinstance(self.noise, Noise):
+            noise_info = self.noise._to_dict()
+        elif isinstance(self.noise, dict):
+            noise_info = {
+                noise: noise_instance._to_dict() for noise, noise_instance in self.noise.items()
+            }
+        else:
+            noise_info = dict()
+
         return {
             "type": type(self).__name__,
             "qubit_support": self.qubit_support,
             "tag": self.tag,
             "parameters": self.parameters._to_dict(),
+            "noise": noise_info,
         }
 
     @classmethod
     def _from_dict(cls, d: dict) -> ParametricBlock:
         params = ParamMap._from_dict(d["parameters"])
+        noise = d.get("noise")
+        if isinstance(noise, dict):
+            noise = {k: Noise._from_dict(v) for k, v in noise.items()}
+        elif noise is not None:
+            noise = Noise._from_dict(noise)
         target = d["qubit_support"][0]
-        return cls(target, params)  # type: ignore[call-arg]
+        return cls((target,), noise, params)  # type: ignore[call-arg]
 
     def dagger(self) -> ParametricBlock:
+        # Do not do the dagger of the noise gate. Only of the parametric one.
         exprs = self.parameters.expressions()
         params = tuple(-extract_original_param_entry(param) for param in exprs)
         return type(self)(*self.qubit_support, *params)  # type: ignore[arg-type]
@@ -497,13 +567,15 @@ def __init__(
         ket: str,
         bra: str,
         qubit_support: int | tuple[int, ...],
+        noise: Noise | dict[str, Noise] | None = None,
     ) -> None:
         """
         Arguments:
 
             ket (str): The ket given as a bitstring.
             bra (str): The bra given as a bitstring.
             qubit_support (int | tuple[int]): The qubit_support of the block.
+            noise (Noise | dict[str, Noise] | None): Optional noise protocols to apply.
         """
         if isinstance(qubit_support, int):
             qubit_support = (qubit_support,)
@@ -522,4 +594,4 @@ def __init__(
 
         self.ket = ket
         self.bra = bra
-        super().__init__(qubit_support)
+        super().__init__(qubit_support, noise)