Cleanup timeline drawer with schedule analysis pass. (#7935)

* Cleanup timeline and add schedule info to scheduled circuit.

* review comment

* fix bug

* review comments

Co-authored-by: Matthew Treinish <[email protected]>

* lint

* disable cyclic import lint check (likely a bug)

* fix import path

* Revert "fix import path"

This reverts commit 1fa5382.

* fix bug

* add deprecation for unscheduled circuit drawing

Co-authored-by: Matthew Treinish <[email protected]>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

qiskit/circuit/quantumcircuit.py

@@ -245,6 +245,7 @@ def __init__(
         # Data contains a list of instructions and their contexts,
         # in the order they were applied.
         self._data = []
+        self._op_start_times = None
 
         # A stack to hold the instruction sets that are being built up during for-, if- and
         # while-block construction.  These are stored as a stripped down sequence of instructions,
@@ -299,6 +300,27 @@ def data(self) -> QuantumCircuitData:
         """
         return QuantumCircuitData(self)
 
+    @property
+    def op_start_times(self) -> List[int]:
+        """Return a list of operation start times.
+
+        This attribute is enabled once one of scheduling analysis passes
+        runs on the quantum circuit.
+
+        Returns:
+            List of integers representing instruction start times.
+            The index corresponds to the index of instruction in :attr:`QuantumCircuit.data`.
+
+        Raises:
+            AttributeError: When circuit is not scheduled.
+        """
+        if self._op_start_times is None:
+            raise AttributeError(
+                "This circuit is not scheduled. "
+                "To schedule it run the circuit through one of the transpiler scheduling passes."
+            )
+        return self._op_start_times
+
     @data.setter
     def data(
         self, data_input: List[Tuple[Instruction, List[QubitSpecifier], List[ClbitSpecifier]]]

qiskit/transpiler/basepasses.py

@@ -136,6 +136,15 @@ def __call__(self, circuit, property_set=None):
             result_circuit._clbit_write_latency = self.property_set["clbit_write_latency"]
         if self.property_set["conditional_latency"] is not None:
             result_circuit._conditional_latency = self.property_set["conditional_latency"]
+        if self.property_set["node_start_time"]:
+            # This is dictionary keyed on the DAGOpNode, which is invalidated once
+            # dag is converted into circuit. So this schedule information is
+            # also converted into list with the same ordering with circuit.data.
+            topological_start_times = []
+            start_times = self.property_set["node_start_time"]
+            for dag_node in result.topological_op_nodes():
+                topological_start_times.append(start_times[dag_node])
+            result_circuit._op_start_times = topological_start_times
 
         return result_circuit
 

qiskit/transpiler/runningpassmanager.py

@@ -132,6 +132,16 @@ def run(self, circuit, output_name=None, callback=None):
         circuit._clbit_write_latency = self.property_set["clbit_write_latency"]
         circuit._conditional_latency = self.property_set["conditional_latency"]
 
+        if self.property_set["node_start_time"]:
+            # This is dictionary keyed on the DAGOpNode, which is invalidated once
+            # dag is converted into circuit. So this schedule information is
+            # also converted into list with the same ordering with circuit.data.
+            topological_start_times = []
+            start_times = self.property_set["node_start_time"]
+            for dag_node in dag.topological_op_nodes():
+                topological_start_times.append(start_times[dag_node])
+            circuit._op_start_times = topological_start_times
+
         return circuit
 
     def _do_pass(self, pass_, dag, options):

qiskit/visualization/timeline/core.py

@@ -48,18 +48,17 @@
 If a plotter provides object handler for plotted shapes, the plotter API can manage
 the lookup table of the handler and the drawings by using this data key.
 """
-
+import warnings
 from copy import deepcopy
 from functools import partial
-from itertools import chain
 from typing import Tuple, Iterator, Dict
 from enum import Enum
 
 import numpy as np
 
 from qiskit import circuit
 from qiskit.visualization.exceptions import VisualizationError
-from qiskit.visualization.timeline import drawings, events, types
+from qiskit.visualization.timeline import drawings, types
 from qiskit.visualization.timeline.stylesheet import QiskitTimelineStyle
 
 
@@ -137,49 +136,82 @@ def add_data(self, data: drawings.ElementaryData):
             data.bits = [b for b in data.bits if not isinstance(b, circuit.Clbit)]
         self._collections[data.data_key] = data
 
+    # pylint: disable=cyclic-import
     def load_program(self, program: circuit.QuantumCircuit):
         """Load quantum circuit and create drawing..
 
         Args:
             program: Scheduled circuit object to draw.
+
+        Raises:
+           VisualizationError: When circuit is not scheduled.
         """
-        self.bits = program.qubits + program.clbits
-        stop_time = 0
+        not_gate_like = (circuit.Barrier,)
+
+        if getattr(program, "_op_start_times") is None:
+            # Run scheduling for backward compatibility
+            from qiskit import transpile
+            from qiskit.transpiler import InstructionDurations, TranspilerError
+
+            warnings.warn(
+                "Visualizing un-scheduled circuit with timeline drawer has been deprecated. "
+                "This circuit should be transpiled with scheduler though it consists of "
+                "instructions with explicit durations.",
+                DeprecationWarning,
+            )
+
+            try:
+                program = transpile(
+                    program, scheduling_method="alap", instruction_durations=InstructionDurations()
+                )
+            except TranspilerError as ex:
+                raise VisualizationError(
+                    f"Input circuit {program.name} is not scheduled and it contains "
+                    "operations with unknown delays. This cannot be visualized."
+                ) from ex
+
+        for t0, (inst, qargs, cargs) in zip(program.op_start_times, program.data):
+            bits = qargs + cargs
+            for bit_pos, bit in enumerate(qargs + cargs):
+                if not isinstance(inst, not_gate_like):
+                    # Generate draw object for gates
+                    gate_source = types.ScheduledGate(
+                        t0=t0,
+                        operand=inst,
+                        duration=inst.duration,
+                        bits=bits,
+                        bit_position=bit_pos,
+                    )
+                    for gen in self.generator["gates"]:
+                        obj_generator = partial(gen, formatter=self.formatter)
+                        for datum in obj_generator(gate_source):
+                            self.add_data(datum)
+                    if len(bits) > 1 and bit_pos == 0:
+                        # Generate draw object for gate-gate link
+                        line_pos = t0 + 0.5 * inst.duration
+                        link_source = types.GateLink(t0=line_pos, opname=inst.name, bits=bits)
+                        for gen in self.generator["gate_links"]:
+                            obj_generator = partial(gen, formatter=self.formatter)
+                            for datum in obj_generator(link_source):
+                                self.add_data(datum)
+                if isinstance(inst, circuit.Barrier):
+                    # Generate draw object for barrier
+                    barrier_source = types.Barrier(t0=t0, bits=bits, bit_position=bit_pos)
+                    for gen in self.generator["barriers"]:
+                        obj_generator = partial(gen, formatter=self.formatter)
+                        for datum in obj_generator(barrier_source):
+                            self.add_data(datum)
 
+        self.bits = program.qubits + program.clbits
         for bit in self.bits:
-            bit_events = events.BitEvents.load_program(scheduled_circuit=program, bit=bit)
-
-            # create objects associated with gates
-            for gen in self.generator["gates"]:
-                obj_generator = partial(gen, formatter=self.formatter)
-                draw_targets = [obj_generator(gate) for gate in bit_events.get_gates()]
-                for data in list(chain.from_iterable(draw_targets)):
-                    self.add_data(data)
-
-            # create objects associated with gate links
-            for gen in self.generator["gate_links"]:
-                obj_generator = partial(gen, formatter=self.formatter)
-                draw_targets = [obj_generator(link) for link in bit_events.get_gate_links()]
-                for data in list(chain.from_iterable(draw_targets)):
-                    self.add_data(data)
-
-            # create objects associated with barrier
-            for gen in self.generator["barriers"]:
-                obj_generator = partial(gen, formatter=self.formatter)
-                draw_targets = [obj_generator(barrier) for barrier in bit_events.get_barriers()]
-                for data in list(chain.from_iterable(draw_targets)):
-                    self.add_data(data)
-
-            # create objects associated with bit
             for gen in self.generator["bits"]:
+                # Generate draw objects for bit
                 obj_generator = partial(gen, formatter=self.formatter)
-                for data in obj_generator(bit):
-                    self.add_data(data)
-
-            stop_time = max(stop_time, bit_events.stop_time)
+                for datum in obj_generator(bit):
+                    self.add_data(datum)
 
         # update time range
-        t_end = max(stop_time, self.formatter["margin.minimum_duration"])
+        t_end = max(program.duration, self.formatter["margin.minimum_duration"])
         self.set_time_range(t_start=0, t_end=t_end)
 
     def set_time_range(self, t_start: int, t_end: int):

releasenotes/notes/cleanup-timeline-drawer-a6287bdab4459e6e.yaml

@@ -0,0 +1,33 @@
+---
+features:
+  - |
+    New attribute :attr:`op_start_times` has been added to :class:`~QuantumCircuit`.
+    This information is populated when one of scheduling analysis passes is run on the circuit.
+    It can be used to obtain circuit instruction with instruction time, for example:
+
+    .. code-block:: python
+
+      from qiskit import QuantumCircuit, transpile
+      from qiskit.test.mock import FakeMontreal
+
+      backend = FakeMontreal()
+
+      qc = QuantumCircuit(2)
+      qc.h(0)
+      qc.cx(0, 1)
+
+      qct = transpile(
+          qc, backend, initial_layout=[0, 1], coupling_map=[[0, 1]], scheduling_method="alap"
+      )
+      scheduled_insts = list(zip(qct.op_start_times, qct.data))
+
+fixes:
+  - |
+    Time misalignment bug of drawing classical register with :func:`~timeline_drawer`
+    has been fixed. Now classical register slots are drawn at correct position.
+deprecations:
+  - |
+    Calling :func:`~timeline_drawer` with unscheduled circuit has been deprecated.
+    All circuits, e.g. even though one consisting only of delay instructions,
+    must be transpiled with ``scheduling_method`` option to generate
+    schedule information being stored in :attr:`QuantumCircuit.op_start_times`.

test/python/circuit/test_circuit_properties.py

@@ -1266,6 +1266,16 @@ def test_metadata_copy_does_not_share_state(self):
 
         self.assertEqual(qc1.metadata["a"], 0)
 
+    def test_scheduling(self):
+        """Test cannot return schedule information without scheduling."""
+        qc = QuantumCircuit(2)
+        qc.h(0)
+        qc.cx(0, 1)
+
+        with self.assertRaises(AttributeError):
+            # pylint: disable=pointless-statement
+            qc.op_start_times
+
 
 if __name__ == "__main__":
     unittest.main()

test/python/visualization/timeline/test_core.py

@@ -143,7 +143,8 @@ def test_non_transpiled_delay_circuit(self):
             "gate_links": [],
         }
 
-        canvas.load_program(circ)
+        with self.assertWarns(DeprecationWarning):
+            canvas.load_program(circ)
         self.assertEqual(len(canvas._collections), 1)
 
     def test_multi_measurement_with_clbit_not_shown(self):