Porting qiskit-ibm-provider/787: Fix `DynamicCircuitInstructionDurati…

…ons.from_backend` for both `Backend versions` (#1383)

* porting qiskit-ibm-provider/pull/787

* porting qiskit-ibm-provider/pull/787

* black

* oops

* monkey patch Qiskit/qiskit#11727

* black lynt

* mypy

---------

Co-authored-by: Kevin Tian <[email protected]>

.pylintrc

@@ -379,6 +379,7 @@ function-naming-style=snake_case
 good-names=i,
            j,
            k,
+           dt,
            ex,
            Run,
            _

qiskit_ibm_runtime/transpiler/passes/scheduling/dynamical_decoupling.py

@@ -325,7 +325,11 @@ def _pre_runhook(self, dag: DAGCircuit) -> None:
                     self._dd_sequence_lengths[qubit] = []
 
                 physical_index = dag.qubits.index(qubit)
-                if self._qubits and physical_index not in self._qubits:
+                if (
+                    self._qubits
+                    and physical_index not in self._qubits
+                    or qubit in self._idle_qubits
+                ):
                     continue
 
                 for index, gate in enumerate(seq):

qiskit_ibm_runtime/transpiler/passes/scheduling/utils.py

@@ -21,7 +21,9 @@
     InstructionDurations,
     InstructionDurationsType,
 )
+from qiskit.transpiler.target import Target
 from qiskit.transpiler.exceptions import TranspilerError
+from qiskit.providers import Backend, BackendV1
 
 
 def block_order_op_nodes(dag: DAGCircuit) -> Generator[DAGOpNode, None, None]:
@@ -150,6 +152,75 @@ def __init__(
         self._enable_patching = enable_patching
         super().__init__(instruction_durations=instruction_durations, dt=dt)
 
+    @classmethod
+    def from_backend(cls, backend: Backend) -> "DynamicCircuitInstructionDurations":
+        """Construct a :class:`DynamicInstructionDurations` object from the backend.
+        Args:
+            backend: backend from which durations (gate lengths) and dt are extracted.
+        Returns:
+            DynamicInstructionDurations: The InstructionDurations constructed from backend.
+        """
+        if isinstance(backend, BackendV1):
+            # TODO Remove once https://github.com/Qiskit/qiskit/pull/11727 gets released in qiskit 0.46.1
+            # From here ---------------------------------------
+            def patch_from_backend(cls, backend: Backend):  # type: ignore
+                """
+                REMOVE me once https://github.com/Qiskit/qiskit/pull/11727 gets released in qiskit 0.46.1
+                """
+                instruction_durations = []
+                backend_properties = backend.properties()
+                if hasattr(backend_properties, "_gates"):
+                    for gate, insts in backend_properties._gates.items():
+                        for qubits, props in insts.items():
+                            if "gate_length" in props:
+                                gate_length = props["gate_length"][
+                                    0
+                                ]  # Throw away datetime at index 1
+                                instruction_durations.append((gate, qubits, gate_length, "s"))
+                    for (
+                        q,  # pylint: disable=invalid-name
+                        props,
+                    ) in backend.properties()._qubits.items():
+                        if "readout_length" in props:
+                            readout_length = props["readout_length"][
+                                0
+                            ]  # Throw away datetime at index 1
+                            instruction_durations.append(("measure", [q], readout_length, "s"))
+                try:
+                    dt = backend.configuration().dt
+                except AttributeError:
+                    dt = None
+
+                return cls(instruction_durations, dt=dt)
+
+            return patch_from_backend(DynamicCircuitInstructionDurations, backend)
+            # To here --------------------------------------- (remove comment ignore annotations too)
+            return super(  # type: ignore  # pylint: disable=unreachable
+                DynamicCircuitInstructionDurations, cls
+            ).from_backend(backend)
+
+        # Get durations from target if BackendV2
+        return cls.from_target(backend.target)
+
+    @classmethod
+    def from_target(cls, target: Target) -> "DynamicCircuitInstructionDurations":
+        """Construct a :class:`DynamicInstructionDurations` object from the target.
+        Args:
+            target: target from which durations (gate lengths) and dt are extracted.
+        Returns:
+            DynamicInstructionDurations: The InstructionDurations constructed from backend.
+        """
+
+        instruction_durations_dict = target.durations().duration_by_name_qubits
+        instruction_durations = []
+        for instr_key, instr_value in instruction_durations_dict.items():
+            instruction_durations += [(*instr_key, *instr_value)]
+        try:
+            dt = target.dt
+        except AttributeError:
+            dt = None
+        return cls(instruction_durations, dt=dt)
+
     def update(
         self, inst_durations: Optional[InstructionDurationsType], dt: float = None
     ) -> "DynamicCircuitInstructionDurations":
@@ -206,15 +277,23 @@ def _patch_instruction(self, key: InstrKey) -> None:
         elif name == "reset":
             self._patch_reset(key)
 
+    def _convert_and_patch_key(self, key: InstrKey) -> None:
+        """Convert duration to dt and patch key"""
+        prev_duration, unit = self._get_duration(key)
+        if unit != "dt":
+            prev_duration = self._convert_unit(prev_duration, unit, "dt")
+            # raise TranspilerError('Can currently only patch durations of "dt".')
+        odd_cycle_correction = self._get_odd_cycle_correction()
+        new_duration = prev_duration + self.MEASURE_PATCH_CYCLES + odd_cycle_correction
+        if unit != "dt":  # convert back to original unit
+            new_duration = self._convert_unit(new_duration, "dt", unit)
+        self._patch_key(key, new_duration, unit)
+
     def _patch_measurement(self, key: InstrKey) -> None:
         """Patch measurement duration by extending duration by 160dt as temporarily
         required by the dynamic circuit backend.
         """
-        prev_duration, unit = self._get_duration_dt(key)
-        if unit != "dt":
-            raise TranspilerError('Can currently only patch durations of "dt".')
-        odd_cycle_correction = self._get_odd_cycle_correction()
-        self._patch_key(key, prev_duration + self.MEASURE_PATCH_CYCLES + odd_cycle_correction, unit)
+        self._convert_and_patch_key(key)
         # Enforce patching of reset on measurement update
         self._patch_reset(("reset", key[1], key[2]))
 
@@ -227,31 +306,24 @@ def _patch_reset(self, key: InstrKey) -> None:
         # triggers the end of scheduling after the measurement pulse
         measure_key = ("measure", key[1], key[2])
         try:
-            measure_duration, unit = self._get_duration_dt(measure_key)
+            measure_duration, unit = self._get_duration(measure_key)
             self._patch_key(key, measure_duration, unit)
         except KeyError:
             # Fall back to reset key if measure not available
-            prev_duration, unit = self._get_duration_dt(key)
-            if unit != "dt":
-                raise TranspilerError('Can currently only patch durations of "dt".')
-            odd_cycle_correction = self._get_odd_cycle_correction()
-            self._patch_key(
-                key,
-                prev_duration + self.MEASURE_PATCH_CYCLES + odd_cycle_correction,
-                unit,
-            )
+            self._convert_and_patch_key(key)
 
-    def _get_duration_dt(self, key: InstrKey) -> Tuple[int, str]:
+    def _get_duration(self, key: InstrKey) -> Tuple[int, str]:
         """Handling for the complicated structure of this class.
 
         TODO: This class implementation should be simplified in Qiskit. Too many edge cases.
         """
         if key[1] is None and key[2] is None:
-            return self.duration_by_name[key[0]]
+            duration = self.duration_by_name[key[0]]
         elif key[2] is None:
-            return self.duration_by_name_qubits[(key[0], key[1])]
-
-        return self.duration_by_name_qubits_params[key]
+            duration = self.duration_by_name_qubits[(key[0], key[1])]
+        else:
+            duration = self.duration_by_name_qubits_params[key]
+        return duration
 
     def _patch_key(self, key: InstrKey, duration: int, unit: str) -> None:
         """Handling for the complicated structure of this class.

releasenotes/notes/fix-duration-patching-b80d45d77481dfa6.yaml

@@ -0,0 +1,11 @@
+---
+fixes:
+  - |
+    Fix the patching of :class:`.DynamicCircuitInstructions` for instructions
+    with durations that are not in units of ``dt``. 
+upgrade:
+  - |
+    Extend :meth:`.DynamicCircuitInstructions.from_backend` to extract and 
+    patch durations from both :class:`.BackendV1` and :class:`.BackendV2`
+    objects. Also add :meth:`.DynamicCircuitInstructions.from_target` to use a
+    :class:`.Target` object instead.

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

@@ -1038,35 +1038,46 @@ def test_disjoint_coupling_map(self):
         self.assertEqual(delay_dict[0], delay_dict[2])
 
     def test_no_unused_qubits(self):
-        """Test DD with if_test circuit that unused qubits are untouched and not scheduled.
-
-        This ensures that programs don't have unnecessary information for unused qubits.
-        Which might hurt performance in later executon stages.
+        """Test DD with if_test circuit that unused qubits are untouched and
+        not scheduled. Unused qubits may also have missing durations when
+        not operational.
+        This ensures that programs don't have unnecessary information for
+        unused qubits.
+        Which might hurt performance in later execution stages.
         """
 
+        # Here "x" on qubit 3 is not defined
+        durations = DynamicCircuitInstructionDurations(
+            [
+                ("h", 0, 50),
+                ("x", 0, 50),
+                ("x", 1, 50),
+                ("x", 2, 50),
+                ("measure", 0, 840),
+                ("reset", 0, 1340),
+            ]
+        )
+
         dd_sequence = [XGate(), XGate()]
         pm = PassManager(
             [
                 ASAPScheduleAnalysis(self.durations),
                 PadDynamicalDecoupling(
-                    self.durations,
+                    durations,
                     dd_sequence,
                     pulse_alignment=1,
                     sequence_min_length_ratios=[0.0],
                 ),
             ]
         )
 
-        qc = QuantumCircuit(3, 1)
+        qc = QuantumCircuit(4, 1)
         qc.measure(0, 0)
         qc.x(1)
-        with qc.if_test((0, True)):
-            qc.x(1)
-        qc.measure(0, 0)
         with qc.if_test((0, True)):
             qc.x(0)
         qc.x(1)
         qc_dd = pm.run(qc)
-        dont_use = qc_dd.qubits[-1]
+        dont_use = qc_dd.qubits[-2:]
         for op in qc_dd.data:
             self.assertNotIn(dont_use, op.qubits)

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

@@ -1774,23 +1774,16 @@ def test_transpile_both_paths(self):
 
         qr = QuantumRegister(7, name="q")
         expected = QuantumCircuit(qr, cr)
-        expected.delay(24080, qr[1])
-        expected.delay(24080, qr[2])
-        expected.delay(24080, qr[3])
-        expected.delay(24080, qr[4])
-        expected.delay(24080, qr[5])
-        expected.delay(24080, qr[6])
+        for q_ind in range(1, 7):
+            expected.delay(24240, qr[q_ind])
         expected.measure(qr[0], cr[0])
         with expected.if_test((cr[0], 1)):
             expected.x(qr[0])
         with expected.if_test((cr[0], 1)):
-            expected.delay(160, qr[0])
             expected.x(qr[1])
-            expected.delay(160, qr[2])
-            expected.delay(160, qr[3])
-            expected.delay(160, qr[4])
-            expected.delay(160, qr[5])
-            expected.delay(160, qr[6])
+            for q_ind in range(7):
+                if q_ind != 1:
+                    expected.delay(160, qr[q_ind])
         self.assertEqual(expected, scheduled)
 
     def test_c_if_plugin_conversion_with_transpile(self):
@@ -1837,7 +1830,7 @@ def test_no_unused_qubits(self):
         """Test DD with if_test circuit that unused qubits are untouched and not scheduled.
 
         This ensures that programs don't have unnecessary information for unused qubits.
-        Which might hurt performance in later executon stages.
+        Which might hurt performance in later execution stages.
         """
 
         durations = DynamicCircuitInstructionDurations([("x", None, 200), ("measure", None, 840)])

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

@@ -15,6 +15,7 @@
 from qiskit_ibm_runtime.transpiler.passes.scheduling.utils import (
     DynamicCircuitInstructionDurations,
 )
+from qiskit_ibm_runtime.fake_provider import FakeKolkata, FakeKolkataV2
 from .....ibm_test_case import IBMTestCase
 
 
@@ -51,6 +52,33 @@ def test_patch_measure(self):
         self.assertEqual(short_odd_durations.get("measure", (0,)), 1224)
         self.assertEqual(short_odd_durations.get("reset", (0,)), 1224)
 
+    def test_durations_from_backend_v1(self):
+        """Test loading and patching durations from a V1 Backend"""
+
+        durations = DynamicCircuitInstructionDurations.from_backend(FakeKolkata())
+
+        self.assertEqual(durations.get("x", (0,)), 160)
+        self.assertEqual(durations.get("measure", (0,)), 3200)
+        self.assertEqual(durations.get("reset", (0,)), 3200)
+
+    def test_durations_from_backend_v2(self):
+        """Test loading and patching durations from a V2 Backend"""
+
+        durations = DynamicCircuitInstructionDurations.from_backend(FakeKolkataV2())
+
+        self.assertEqual(durations.get("x", (0,)), 160)
+        self.assertEqual(durations.get("measure", (0,)), 3200)
+        self.assertEqual(durations.get("reset", (0,)), 3200)
+
+    def test_durations_from_target(self):
+        """Test loading and patching durations from a target"""
+
+        durations = DynamicCircuitInstructionDurations.from_target(FakeKolkataV2().target)
+
+        self.assertEqual(durations.get("x", (0,)), 160)
+        self.assertEqual(durations.get("measure", (0,)), 3200)
+        self.assertEqual(durations.get("reset", (0,)), 3200)
+
     def test_patch_disable(self):
         """Test if schedules circuits with c_if after measure with a common clbit.
         See: https://github.com/Qiskit/qiskit-terra/issues/7654"""