Update dynamical decoupling pass to support staggered pulses

qiskit_ibm_provider/transpiler/passes/scheduling/dynamical_decoupling.py

@@ -16,6 +16,7 @@
 from typing import Dict, List, Optional, Union
 
 import numpy as np
+import rustworkx as rx
 from qiskit.circuit import Qubit, Gate
 from qiskit.circuit.delay import Delay
 from qiskit.circuit.library.standard_gates import IGate, UGate, U3Gate
@@ -26,6 +27,7 @@
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.instruction_durations import InstructionDurations
 from qiskit.transpiler.passes.optimization import Optimize1qGates
+from qiskit.transpiler import CouplingMap
 
 from .block_base_padder import BlockBasePadder
 
@@ -117,6 +119,8 @@ def __init__(
         extra_slack_distribution: str = "middle",
         sequence_min_length_ratios: Optional[Union[int, List[int]]] = None,
         insert_multiple_cycles: bool = False,
+        coupling_map: CouplingMap = None,
+        alt_spacings: Optional[Union[List[List[float]], List[float]]] = None,
     ):
         """Dynamical decoupling initializer.
 
@@ -133,7 +137,9 @@ def __init__(
                 The available slack will be divided according to this.
                 The list length must be one more than the length of dd_sequence,
                 and the elements must sum to 1. If None, a balanced spacing
-                will be used [d/2, d, d, ..., d, d, d/2].
+                will be used [d/2, d, d, ..., d, d, d/2]. This spacing only
+                applies to the first subcircuit, if a ``coupling_map`` is
+                specified
             skip_reset_qubits: If True, does not insert DD on idle periods that
                 immediately follow initialized/reset qubits
                 (as qubits in the ground state are less susceptible to decoherence).
@@ -161,14 +167,22 @@ def __init__(
             insert_multiple_cycles: If the available duration exceeds
                 2*sequence_min_length_ratio*duration(dd_sequence) enable the insertion of multiple
                 rounds of the dynamical decoupling sequence in that delay.
+            coupling_map: directed graph representing the coupling map for the device. Specifying a
+                coupling map partitions the device into subcircuits, in order to apply DD sequences
+                with different pulse spacings within each. Currently support 2 subcircuits.
+            alt_spacings: A list of lists of spacings between the DD gates, for the second subcircuit,
+                as determined by the coupling map. If None, a balanced spacing that is staggered with
+                respect to the first subcircuit will be used [d, d, d, ..., d, d, 0].
         Raises:
             TranspilerError: When invalid DD sequence is specified.
             TranspilerError: When pulse gate with the duration which is
                 non-multiple of the alignment constraint value is found.
+            TranspilerError: When the coupling map is not supported (i.e., if degree > 3)
         """
 
         super().__init__()
         self._durations = durations
+
         # Enforce list of DD sequences
         if dd_sequences:
             try:
@@ -179,19 +193,37 @@ def __init__(
         self._qubits = qubits
         self._skip_reset_qubits = skip_reset_qubits
         self._alignment = pulse_alignment
+        self._coupling_map = coupling_map
+        self._coupling_coloring = None
 
         if spacings is not None:
             try:
                 iter(spacings[0])  # type: ignore
             except TypeError:
                 spacings = [spacings]  # type: ignore
+        if alt_spacings is not None:
+            try:
+                iter(alt_spacings[0])  # type: ignore
+            except TypeError:
+                alt_spacings = [alt_spacings]  # type: ignore
         self._spacings = spacings
+        self._alt_spacings = alt_spacings
 
         if self._spacings and len(self._spacings) != len(self._dd_sequences):
             raise TranspilerError(
                 "Number of sequence spacings must equal number of DD sequences."
             )
 
+        if self._alt_spacings:
+            if not self._coupling_map:
+                warnings.warn(
+                    "Alternate spacings are ignored because a coupling map was not provided"
+                )
+            elif len(self._alt_spacings) != len(self._dd_sequences):
+                raise TranspilerError(
+                    "Number of alternate sequence spacings must equal number of DD sequences."
+                )
+
         self._extra_slack_distribution = extra_slack_distribution
 
         self._dd_sequence_lengths: Dict[Qubit, List[List[Gate]]] = {}
@@ -217,9 +249,26 @@ def __init__(
     def _pre_runhook(self, dag: DAGCircuit) -> None:
         super()._pre_runhook(dag)
 
+        if self._coupling_map:
+            physical_qubits = [dag.qubits.index(q) for q in dag.qubits]
+            sub_coupling_map = self._coupling_map.reduce(physical_qubits)
+            self._coupling_coloring = rx.graph_greedy_color(
+                sub_coupling_map.graph.to_undirected()
+            )
+            if any(c > 1 for c in self._coupling_coloring.values()):
+                raise TranspilerError(
+                    "This circuit topology is not supported for staggered dynamical decoupling."
+                    "The maximum connectivity is 3 nearest neighbors per qubit."
+                )
+
         spacings_required = self._spacings is None
         if spacings_required:
             self._spacings = []  # type: ignore
+        alt_spacings_required = (
+            self._alt_spacings is None and self._coupling_map is not None
+        )
+        if alt_spacings_required:
+            self._alt_spacings = []  # type: ignore
 
         for seq_idx, seq in enumerate(self._dd_sequences):
             num_pulses = len(self._dd_sequences[seq_idx])
@@ -242,6 +291,19 @@ def _pre_runhook(self, dag: DAGCircuit) -> None:
                         "of the slack period and sum to 1."
                     )
 
+            if self._coupling_map:
+                if alt_spacings_required:
+                    mid = 1 / num_pulses
+                    self._alt_spacings.append([mid] * num_pulses + [0])  # type: ignore
+                else:
+                    if sum(self._alt_spacings[seq_idx]) != 1 or any(  # type: ignore
+                        a < 0 for a in self._alt_spacings[seq_idx]  # type: ignore
+                    ):
+                        raise TranspilerError(
+                            "The spacings must be given in terms of fractions "
+                            "of the slack period and sum to 1."
+                        )
+
             # Check if DD sequence is identity
             if num_pulses != 1:
                 if num_pulses % 2 != 0:
@@ -363,6 +425,11 @@ def _pad(
             seq_length = np.sum(seq_lengths)
             seq_ratio = self._sequence_min_length_ratios[sequence_idx]
             spacings = self._spacings[sequence_idx]
+            alt_spacings = (
+                np.asarray(self._alt_spacings[sequence_idx])
+                if self._coupling_map
+                else None
+            )
 
             # Verify the delay duration exceeds the minimum time to insert
             if time_interval / seq_length <= seq_ratio:
@@ -383,9 +450,9 @@ def _pad(
             # multiple dd sequences may be inserted
             if num_sequences > 1:
                 dd_sequence = list(dd_sequence) * num_sequences
-                spacings = spacings * num_sequences
                 seq_lengths = seq_lengths * num_sequences
                 seq_length = np.sum(seq_lengths)
+                spacings = spacings * num_sequences
 
             spacings = np.asarray(spacings) / num_sequences
             slack = time_interval - seq_length
@@ -431,8 +498,16 @@ def _pad(
             def _constrained_length(values: np.array) -> np.array:
                 return self._alignment * np.floor(values / self._alignment)
 
+            if self._coupling_map:
+                if self._coupling_coloring[qubit.index] == 0:
+                    sub_spacings = spacings
+                else:
+                    sub_spacings = alt_spacings
+            else:
+                sub_spacings = spacings
+
             # (1) Compute DD intervals satisfying the constraint
-            taus = _constrained_length(slack * spacings)
+            taus = _constrained_length(slack * sub_spacings)
             extra_slack = slack - np.sum(taus)
             # (2) Distribute extra slack
             if self._extra_slack_distribution == "middle":

releasenotes/notes/staggered-dd-95ebe09ebc46f60f.yaml

@@ -0,0 +1,9 @@
+---
+upgrade:
+  - |
+    For each qubit, add the ability to select one of two different sets of spacings between dynamical decoupling pulses, 
+    based on the input coupling map. 
+fixes:
+  - |
+    Staggered inter-pulse spacings allow the user to disable certain interactions between neighboring qubits by
+    Refer to `#539 <https://github.com/Qiskit/qiskit-ibm-provider/issues/539>` for more details

test/unit/transpiler/passes/scheduling/test_dynamical_decoupling.py

@@ -22,6 +22,7 @@
 from qiskit.quantum_info import Operator
 from qiskit.transpiler.passmanager import PassManager
 from qiskit.transpiler.exceptions import TranspilerError
+from qiskit.transpiler.coupling import CouplingMap
 
 from qiskit_ibm_provider.transpiler.passes.scheduling.dynamical_decoupling import (
     PadDynamicalDecoupling,
@@ -75,6 +76,8 @@ def setUp(self):
             ]
         )
 
+        self.coupling_map = CouplingMap([[0, 1], [1, 2], [2, 3]])
+
     def test_insert_dd_ghz(self):
         """Test DD gates are inserted in correct spots."""
         dd_sequence = [XGate(), XGate()]
@@ -853,3 +856,116 @@ def test_multiple_dd_sequence_cycles(self):
         expected.x(0)
         expected.delay(225, 0)
         self.assertEqual(qc_dd, expected)
+
+    def test_staggered_dd(self):
+        """Test that timing on DD can be staggered if coupled with each other"""
+        dd_sequence = [XGate(), XGate()]
+        pm = PassManager(
+            [
+                ASAPScheduleAnalysis(self.durations),
+                PadDynamicalDecoupling(
+                    self.durations,
+                    dd_sequence,
+                    coupling_map=self.coupling_map,
+                    alt_spacings=[0.1, 0.8, 0.1],
+                ),
+            ]
+        )
+
+        qc_barriers = QuantumCircuit(4, 1)
+        qc_barriers.x(0)
+        qc_barriers.x(1)
+        qc_barriers.x(2)
+        qc_barriers.x(3)
+        qc_barriers.barrier()
+        qc_barriers.measure(0, 0)
+        qc_barriers.delay(14, 0)
+        qc_barriers.x(1)
+        qc_barriers.x(2)
+        qc_barriers.x(3)
+        qc_barriers.barrier()
+
+        qc_dd = pm.run(qc_barriers)
+
+        expected = QuantumCircuit(4, 1)
+        expected.x(0)
+        expected.x(1)
+        expected.x(2)
+        expected.x(3)
+        expected.barrier()
+        expected.x(1)
+        expected.delay(208, 1)
+        expected.x(1)
+        expected.delay(448, 1)
+        expected.x(1)
+        expected.delay(208, 1)
+        expected.x(2)
+        expected.delay(80, 2)  # q1-q2 are coupled, staggered delays
+        expected.x(2)
+        expected.delay(704, 2)
+        expected.x(2)
+        expected.delay(80, 2)  # q2-q3 are uncoupled, same delays
+        expected.x(3)
+        expected.delay(208, 3)
+        expected.x(3)
+        expected.delay(448, 3)
+        expected.x(3)
+        expected.delay(208, 3)
+        expected.measure(0, 0)
+        expected.delay(14, 0)
+        expected.barrier()
+
+        self.assertEqual(qc_dd, expected)
+
+    def test_insert_dd_bad_spacings(self):
+        """Test DD raises when spacings don't add up to 1."""
+        dd_sequence = [XGate(), XGate()]
+        pm = PassManager(
+            [
+                ASAPScheduleAnalysis(self.durations),
+                PadDynamicalDecoupling(
+                    self.durations,
+                    dd_sequence,
+                    spacings=[0.1, 0.9, 0.1],
+                    coupling_map=self.coupling_map,
+                ),
+            ]
+        )
+
+        with self.assertRaises(TranspilerError):
+            pm.run(self.ghz4)
+
+    def test_insert_dd_bad_alt_spacings(self):
+        """Test DD raises when alt_spacings don't add up to 1."""
+        dd_sequence = [XGate(), XGate()]
+        pm = PassManager(
+            [
+                ASAPScheduleAnalysis(self.durations),
+                PadDynamicalDecoupling(
+                    self.durations,
+                    dd_sequence,
+                    alt_spacings=[0.1, 0.9, 0.1],
+                    coupling_map=self.coupling_map,
+                ),
+            ]
+        )
+
+        with self.assertRaises(TranspilerError):
+            pm.run(self.ghz4)
+
+    def test_unsupported_coupling_map(self):
+        """Test DD raises if coupling map is not supported."""
+        dd_sequence = [XGate(), XGate()]
+        pm = PassManager(
+            [
+                ASAPScheduleAnalysis(self.durations),
+                PadDynamicalDecoupling(
+                    self.durations,
+                    dd_sequence,
+                    coupling_map=CouplingMap([[0, 1], [0, 2], [1, 2], [2, 3]]),
+                ),
+            ]
+        )
+
+        with self.assertRaises(TranspilerError):
+            pm.run(self.ghz4)