Merge pull request #1121 from qiboteam/restrict_topology

Restrict connectivity qubits

src/qibo/transpiler/blocks.py

@@ -3,7 +3,7 @@
 from qibo import Circuit, gates
 from qibo.config import raise_error
 from qibo.gates import Gate
-from qibo.transpiler.exceptions import BlockingError
+from qibo.transpiler._exceptions import BlockingError
 
 
 class Block:

src/qibo/transpiler/pipeline.py

@@ -7,8 +7,8 @@
 from qibo.config import raise_error
 from qibo.models import Circuit
 from qibo.quantum_info.random_ensembles import random_statevector
+from qibo.transpiler._exceptions import TranspilerPipelineError
 from qibo.transpiler.abstract import Optimizer, Placer, Router
-from qibo.transpiler.exceptions import TranspilerPipelineError
 from qibo.transpiler.optimizer import Preprocessing
 from qibo.transpiler.placer import Trivial, assert_placement
 from qibo.transpiler.router import ConnectivityError, assert_connectivity
@@ -120,8 +120,12 @@ def assert_transpiling(
     if original_circuit.nqubits != transpiled_circuit.nqubits:
         qubit_matcher = Preprocessing(connectivity=connectivity)
         original_circuit = qubit_matcher(circuit=original_circuit)
-    assert_placement(circuit=original_circuit, layout=initial_layout)
-    assert_placement(circuit=transpiled_circuit, layout=final_layout)
+    assert_placement(
+        circuit=original_circuit, layout=initial_layout, connectivity=connectivity
+    )
+    assert_placement(
+        circuit=transpiled_circuit, layout=final_layout, connectivity=connectivity
+    )
     if check_circuit_equivalence:
         assert_circuit_equivalence(
             original_circuit=original_circuit,
@@ -131,53 +135,91 @@ def assert_transpiling(
         )
 
 
+def restrict_connectivity_qubits(connectivity: nx.Graph, qubits: list):
+    """Restrict the connectivity to selected qubits.
+
+    Args:
+        connectivity (:class:`networkx.Graph`): chip connectivity.
+        qubits (list): list of physical qubits to be used.
+
+    Returns:
+        (:class:`networkx.Graph`): restricted connectivity.
+    """
+    if not set(qubits).issubset(set(connectivity.nodes)):
+        raise_error(
+            ConnectivityError, "Some qubits are not in the original connectivity."
+        )
+
+    new_connectivity = nx.Graph()
+    new_connectivity.add_nodes_from(qubits)
+    new_edges = [
+        edge for edge in connectivity.edges if edge[0] in qubits and edge[1] in qubits
+    ]
+    new_connectivity.add_edges_from(new_edges)
+
+    if not nx.is_connected(new_connectivity):
+        raise_error(ConnectivityError, "New connectivity graph is not connected.")
+
+    return new_connectivity
+
+
 class Passes:
     """Define a transpiler pipeline consisting of smaller transpiler steps that are applied sequentially:
 
     Args:
-        passes (list): list of passes to be applied sequentially,
-            if None default transpiler will be used, it requires hardware connectivity.
-        connectivity (nx.Graph): hardware qubit connectivity.
+        passes (list, optional): list of passes to be applied sequentially.
+            If ``None``, default transpiler will be used.
+            Defaults to ``None``.
+        connectivity (:class:`networkx.Graph`, optional): physical qubits connectivity.
+            If ``None``, :class:`` is used.
+            Defaults to ``None``.
+        native_gates (:class:`qibo.transpiler.unroller.NativeGates`, optional): native gates.
+            Defaults to :math:`qibo.transpiler.unroller.NativeGates.default`.
+        on_qubits (list, optional): list of physical qubits to be used.
+            If "None" all qubits are used. Defaults to ``None``.
     """
 
     def __init__(
         self,
         passes: list = None,
         connectivity: nx.Graph = None,
         native_gates: NativeGates = NativeGates.default(),
+        on_qubits: list = None,
     ):
-        self.native_gates = native_gates
-        if passes is None:
-            self.passes = self.default(connectivity)
-        else:
-            self.passes = passes
+        if on_qubits is not None:
+            connectivity = restrict_connectivity_qubits(connectivity, on_qubits)
         self.connectivity = connectivity
+        self.native_gates = native_gates
+        self.passes = self.default() if passes is None else passes
 
-    def default(self, connectivity: nx.Graph):
+    def default(self):
         """Return the default transpiler pipeline for the required hardware connectivity."""
-        if not isinstance(connectivity, nx.Graph):
+        if not isinstance(self.connectivity, nx.Graph):
             raise_error(
                 TranspilerPipelineError,
                 "Define the hardware chip connectivity to use default transpiler",
             )
         default_passes = []
         # preprocessing
-        default_passes.append(Preprocessing(connectivity=connectivity))
+        default_passes.append(Preprocessing(connectivity=self.connectivity))
         # default placer pass
-        default_passes.append(Trivial(connectivity=connectivity))
+        default_passes.append(Trivial(connectivity=self.connectivity))
         # default router pass
         default_passes.append(StarConnectivity())
         # default unroller pass
         default_passes.append(Unroller(native_gates=self.native_gates))
+
         return default_passes
 
     def __call__(self, circuit):
         self.initial_layout = None
         final_layout = None
         for transpiler_pass in self.passes:
             if isinstance(transpiler_pass, Optimizer):
+                transpiler_pass.connectivity = self.connectivity
                 circuit = transpiler_pass(circuit)
             elif isinstance(transpiler_pass, Placer):
+                transpiler_pass.connectivity = self.connectivity
                 if self.initial_layout == None:
                     self.initial_layout = transpiler_pass(circuit)
                 else:
@@ -186,6 +228,7 @@ def __call__(self, circuit):
                         "You are defining more than one placer pass.",
                     )
             elif isinstance(transpiler_pass, Router):
+                transpiler_pass.connectivity = self.connectivity
                 if self.initial_layout is not None:
                     circuit, final_layout = transpiler_pass(
                         circuit, self.initial_layout
@@ -201,14 +244,17 @@ def __call__(self, circuit):
                     TranspilerPipelineError,
                     f"Unrecognised transpiler pass: {transpiler_pass}",
                 )
+
         return circuit, final_layout
 
-    def is_satisfied(self, circuit):
-        """Return True if the circuit respects the hardware connectivity and native gates, False otherwise.
+    def is_satisfied(self, circuit: Circuit):
+        """Returns ``True`` if the circuit respects the hardware connectivity and native gates, ``False`` otherwise.
 
         Args:
-            circuit (qibo.models.Circuit): circuit to be checked.
-            native_gates (NativeGates): two qubit native gates.
+            circuit (:class:`qibo.models.circuit.Circuit`): circuit to be checked.
+
+        Returns:
+            (bool): satisfiability condition.
         """
         try:
             assert_connectivity(circuit=circuit, connectivity=self.connectivity)