Bug fix macros.measure with backendv2

qiskit/pulse/macros.py

@@ -11,14 +11,19 @@
 # that they have been altered from the originals.
 
 """Module for common pulse programming macros."""
+from __future__ import annotations
 
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Optional, Union, TYPE_CHECKING
 
 from qiskit.pulse import channels, exceptions, instructions, utils
 from qiskit.pulse.instruction_schedule_map import InstructionScheduleMap
 from qiskit.pulse.schedule import Schedule
 
 
+if TYPE_CHECKING:
+    from qiskit.transpiler import Target
+
+
 def measure(
     qubits: List[int],
     backend=None,
@@ -30,6 +35,13 @@ def measure(
     """Return a schedule which measures the requested qubits according to the given
     instruction mapping and measure map, or by using the defaults provided by the backend.
 
+    .. note::
+        This function internally dispatches schedule generation logic depending on input backend model.
+        For the :class:`.BackendV1`, it considers conventional :class:`.InstructionScheduleMap`
+        and utilizes the backend calibration defined for a group of qubits in the `meas_map`.
+        For the :class:`.BackendV2`, it assembles calibrations of single qubit measurement
+        defined in the backend target to build a composite measurement schedule for `qubits`.
+
     By default, the measurement results for each qubit are trivially mapped to the qubit
     index. This behavior is overridden by qubit_mem_slots. For instance, to measure
     qubit 0 into MemorySlot(1), qubit_mem_slots can be provided as {0: 1}.
@@ -47,18 +59,67 @@ def measure(
 
     Returns:
         A measurement schedule corresponding to the inputs provided.
+    """
+
+    # backend is V2.
+    if hasattr(backend, "target"):
+        try:
+            meas_map = backend.configuration().meas_map
+        except AttributeError:
+            # TODO add meas_map to Target in 0.25
+            meas_map = [list(range(backend.num_qubits))]
+
+        return _measure_v2(
+            qubits=qubits,
+            target=backend.target,
+            meas_map=meas_map,
+            qubit_mem_slots=qubit_mem_slots or dict(zip(qubits, range(len(qubits)))),
+            measure_name=measure_name,
+        )
+    # backend is V1 or backend is None.
+    else:
+        try:
+            return _measure_v1(
+                qubits=qubits,
+                inst_map=inst_map or backend.defaults().instruction_schedule_map,
+                meas_map=meas_map or backend.configuration().meas_map,
+                qubit_mem_slots=qubit_mem_slots,
+                measure_name=measure_name,
+            )
+        except AttributeError as ex:
+            raise exceptions.PulseError(
+                "inst_map or meas_map, and backend cannot be None simultaneously"
+            ) from ex
+
 
+def _measure_v1(
+    qubits: List[int],
+    inst_map: InstructionScheduleMap,
+    meas_map: Union[List[List[int]], Dict[int, List[int]]],
+    qubit_mem_slots: Optional[Dict[int, int]] = None,
+    measure_name: str = "measure",
+) -> Schedule:
+    """Return a schedule which measures the requested qubits according to the given
+    instruction mapping and measure map, or by using the defaults provided by the backendV1.
+
+    Args:
+        qubits: List of qubits to be measured.
+        backend (Union[Backend, BaseBackend]): A backend instance, which contains
+            hardware-specific data required for scheduling.
+        inst_map: Mapping of circuit operations to pulse schedules. If None, defaults to the
+                  ``instruction_schedule_map`` of ``backend``.
+        meas_map: List of sets of qubits that must be measured together. If None, defaults to
+                  the ``meas_map`` of ``backend``.
+        qubit_mem_slots: Mapping of measured qubit index to classical bit index.
+        measure_name: Name of the measurement schedule.
+    Returns:
+        A measurement schedule corresponding to the inputs provided.
     Raises:
         PulseError: If both ``inst_map`` or ``meas_map``, and ``backend`` is None.
     """
+
     schedule = Schedule(name=f"Default measurement schedule for qubits {qubits}")
-    try:
-        inst_map = inst_map or backend.defaults().instruction_schedule_map
-        meas_map = meas_map or backend.configuration().meas_map
-    except AttributeError as ex:
-        raise exceptions.PulseError(
-            "inst_map or meas_map, and backend cannot be None simultaneously"
-        ) from ex
+
     if isinstance(meas_map, list):
         meas_map = utils.format_meas_map(meas_map)
 
@@ -92,6 +153,68 @@ def measure(
     return schedule
 
 
+def _measure_v2(
+    qubits: List[int],
+    target: Target,
+    meas_map: Union[List[List[int]], Dict[int, List[int]]],
+    qubit_mem_slots: Dict[int, int],
+    measure_name: str = "measure",
+) -> Schedule:
+    """Return a schedule which measures the requested qubits according to the given
+    target and measure map, or by using the defaults provided by the backendV2.
+
+    Args:
+        qubits: List of qubits to be measured.
+        target: The :class:`~.Target` representing the target backend.
+        meas_map: List of sets of qubits that must be measured together.
+        qubit_mem_slots: Mapping of measured qubit index to classical bit index.
+        measure_name: Name of the measurement schedule.
+
+    Returns:
+        A measurement schedule corresponding to the inputs provided.
+    """
+    schedule = Schedule(name=f"Default measurement schedule for qubits {qubits}")
+
+    if isinstance(meas_map, list):
+        meas_map = utils.format_meas_map(meas_map)
+    meas_group = set()
+    for qubit in qubits:
+        meas_group |= set(meas_map[qubit])
+    meas_group = sorted(list(meas_group))
+
+    meas_group_set = set(range(max(meas_group) + 1))
+    unassigned_qubit_indices = sorted(set(meas_group) - qubit_mem_slots.keys())
+    unassigned_reg_indices = sorted(meas_group_set - set(qubit_mem_slots.values()), reverse=True)
+    if set(qubit_mem_slots.values()).issubset(meas_group_set):
+        for qubit in unassigned_qubit_indices:
+            qubit_mem_slots[qubit] = unassigned_reg_indices.pop()
+
+    for measure_qubit in meas_group:
+        try:
+            if measure_qubit in qubits:
+                default_sched = target.get_calibration(measure_name, (measure_qubit,)).filter(
+                    channels=[
+                        channels.MeasureChannel(measure_qubit),
+                        channels.AcquireChannel(measure_qubit),
+                    ]
+                )
+            else:
+                default_sched = target.get_calibration(measure_name, (measure_qubit,)).filter(
+                    channels=[
+                        channels.AcquireChannel(measure_qubit),
+                    ]
+                )
+        except KeyError as ex:
+            raise exceptions.PulseError(
+                "We could not find a default measurement schedule called '{}'. "
+                "Please provide another name using the 'measure_name' keyword "
+                "argument. For assistance, the instructions which are defined are: "
+                "{}".format(measure_name, target.instructions)
+            ) from ex
+        schedule += _schedule_remapping_memory_slot(default_sched, qubit_mem_slots)
+    return schedule
+
+
 def measure_all(backend) -> Schedule:
     """
     Return a Schedule which measures all qubits of the given backend.
@@ -104,3 +227,35 @@ def measure_all(backend) -> Schedule:
         A schedule corresponding to the inputs provided.
     """
     return measure(qubits=list(range(backend.configuration().n_qubits)), backend=backend)
+
+
+def _schedule_remapping_memory_slot(
+    schedule: Schedule, qubit_mem_slots: Dict[int, int]
+) -> Schedule:
+    """
+    A helper function to overwrite MemorySlot index of :class:`.Acquire` instruction.
+
+    Args:
+        schedule: A measurement schedule.
+        qubit_mem_slots: Mapping of measured qubit index to classical bit index.
+
+    Returns:
+        A measurement schedule with new memory slot index.
+    """
+    new_schedule = Schedule()
+    for t0, inst in schedule.instructions:
+        if isinstance(inst, instructions.Acquire):
+            qubit_index = inst.channel.index
+            reg_index = qubit_mem_slots.get(qubit_index, qubit_index)
+            new_schedule.insert(
+                t0,
+                instructions.Acquire(
+                    inst.duration,
+                    channels.AcquireChannel(qubit_index),
+                    mem_slot=channels.MemorySlot(reg_index),
+                ),
+                inplace=True,
+            )
+        else:
+            new_schedule.insert(t0, inst, inplace=True)
+    return new_schedule

releasenotes/notes/fix-macros-measure-with-backendV2-4354f00ab4f1cd3e.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Fixed failure in using :func:`~qiskit.pulse.macros.measure` 
+    with :class:`.BackendV2` backends.

test/python/pulse/test_macros.py

@@ -24,7 +24,7 @@
 )
 from qiskit.pulse import macros
 from qiskit.pulse.exceptions import PulseError
-from qiskit.providers.fake_provider import FakeOpenPulse2Q
+from qiskit.providers.fake_provider import FakeOpenPulse2Q, FakeHanoiV2
 from qiskit.test import QiskitTestCase
 
 
@@ -34,6 +34,7 @@ class TestMeasure(QiskitTestCase):
     def setUp(self):
         super().setUp()
         self.backend = FakeOpenPulse2Q()
+        self.backend_v2 = FakeHanoiV2()
         self.inst_map = self.backend.defaults().instruction_schedule_map
 
     def test_measure(self):
@@ -43,7 +44,6 @@ def test_measure(self):
             self.inst_map.get("measure", [0, 1]).filter(channels=[MeasureChannel(0)]),
             Acquire(10, AcquireChannel(0), MemorySlot(0)),
         )
-
         self.assertEqual(sched.instructions, expected.instructions)
 
     def test_measure_sched_with_qubit_mem_slots(self):
@@ -91,6 +91,78 @@ def test_fail_measure(self):
         with self.assertRaises(PulseError):
             macros.measure(qubits=[0], inst_map=self.inst_map)
 
+    def test_measure_v2(self):
+        """Test macro - measure with backendV2."""
+        sched = macros.measure(qubits=[0], backend=self.backend_v2)
+        expected = self.backend_v2.target.get_calibration("measure", (0,)).filter(
+            channels=[
+                MeasureChannel(0),
+            ]
+        )
+        measure_duration = expected.filter(instruction_types=[Play]).duration
+        for qubit in range(self.backend_v2.num_qubits):
+            expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(qubit))
+        self.assertEqual(sched.instructions, expected.instructions)
+
+    def test_measure_v2_sched_with_qubit_mem_slots(self):
+        """Test measure with backendV2 and custom qubit_mem_slots."""
+        sched = macros.measure(qubits=[0], backend=self.backend_v2, qubit_mem_slots={0: 2})
+        expected = self.backend_v2.target.get_calibration("measure", (0,)).filter(
+            channels=[
+                MeasureChannel(0),
+            ]
+        )
+        measure_duration = expected.filter(instruction_types=[Play]).duration
+        for qubit in range(self.backend_v2.num_qubits):
+            if qubit == 0:
+                expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(2))
+            elif qubit == 1:
+                expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(0))
+            elif qubit == 2:
+                expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(1))
+            else:
+                expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(qubit))
+        self.assertEqual(sched.instructions, expected.instructions)
+
+    def test_measure_v2_sched_with_meas_map(self):
+        """Test measure with backendV2 custom meas_map as list and dict."""
+        sched_with_meas_map_list = macros.measure(
+            qubits=[0], backend=self.backend_v2, meas_map=[[0, 1]]
+        )
+        sched_with_meas_map_dict = macros.measure(
+            qubits=[0], backend=self.backend_v2, meas_map={0: [0, 1], 1: [0, 1]}
+        )
+        expected = self.backend_v2.target.get_calibration("measure", (0,)).filter(
+            channels=[
+                MeasureChannel(0),
+            ]
+        )
+        measure_duration = expected.filter(instruction_types=[Play]).duration
+        for qubit in range(self.backend_v2.num_qubits):
+            expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(qubit))
+        self.assertEqual(sched_with_meas_map_list.instructions, expected.instructions)
+        self.assertEqual(sched_with_meas_map_dict.instructions, expected.instructions)
+
+    def test_multiple_measure_v2(self):
+        """Test macro - multiple qubit measure with backendV2."""
+        sched = macros.measure(qubits=[0, 1], backend=self.backend_v2)
+        expected = self.backend_v2.target.get_calibration("measure", (0,)).filter(
+            channels=[
+                MeasureChannel(0),
+            ]
+        )
+        expected += self.backend_v2.target.get_calibration("measure", (1,)).filter(
+            channels=[
+                MeasureChannel(1),
+            ]
+        )
+        measure_duration = expected.filter(instruction_types=[Play]).duration
+        expected += Acquire(measure_duration, AcquireChannel(0), MemorySlot(0))
+        expected += Acquire(measure_duration, AcquireChannel(1), MemorySlot(1))
+        for qubit in range(2, self.backend_v2.num_qubits):
+            expected += Acquire(measure_duration, AcquireChannel(qubit), MemorySlot(qubit))
+        self.assertEqual(sched.instructions, expected.instructions)
+
 
 class TestMeasureAll(QiskitTestCase):
     """Pulse measure all macro."""