From 7391168c8944a5d901e0fdf02fb8955e82bc2ca1 Mon Sep 17 00:00:00 2001 From: Naoki Kanazawa Date: Fri, 17 Jun 2022 02:00:14 +0900 Subject: [PATCH] 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 * lint * disable cyclic import lint check (likely a bug) * fix import path * Revert "fix import path" This reverts commit 1fa5382df47e6dd778d791b270f1cbf9639ce300. * fix bug * add deprecation for unscheduled circuit drawing Co-authored-by: Matthew Treinish Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com> --- qiskit/circuit/quantumcircuit.py | 22 ++ qiskit/transpiler/basepasses.py | 9 + qiskit/transpiler/runningpassmanager.py | 10 + qiskit/visualization/timeline/core.py | 100 +++++--- qiskit/visualization/timeline/events.py | 121 --------- ...anup-timeline-drawer-a6287bdab4459e6e.yaml | 33 +++ .../python/circuit/test_circuit_properties.py | 10 + .../visualization/timeline/test_core.py | 3 +- .../visualization/timeline/test_events.py | 234 ------------------ 9 files changed, 152 insertions(+), 390 deletions(-) delete mode 100644 qiskit/visualization/timeline/events.py create mode 100644 releasenotes/notes/cleanup-timeline-drawer-a6287bdab4459e6e.yaml delete mode 100644 test/python/visualization/timeline/test_events.py diff --git a/qiskit/circuit/quantumcircuit.py b/qiskit/circuit/quantumcircuit.py index 612a6abeb353..d5d19f13035b 100644 --- a/qiskit/circuit/quantumcircuit.py +++ b/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]]] diff --git a/qiskit/transpiler/basepasses.py b/qiskit/transpiler/basepasses.py index 832b79e148a5..5e006d16c8da 100644 --- a/qiskit/transpiler/basepasses.py +++ b/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 diff --git a/qiskit/transpiler/runningpassmanager.py b/qiskit/transpiler/runningpassmanager.py index 43af938d1879..6cff16cb2c03 100644 --- a/qiskit/transpiler/runningpassmanager.py +++ b/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): diff --git a/qiskit/visualization/timeline/core.py b/qiskit/visualization/timeline/core.py index 3fd6ae9e473b..1c095a03f313 100644 --- a/qiskit/visualization/timeline/core.py +++ b/qiskit/visualization/timeline/core.py @@ -48,10 +48,9 @@ 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 @@ -59,7 +58,7 @@ 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): diff --git a/qiskit/visualization/timeline/events.py b/qiskit/visualization/timeline/events.py deleted file mode 100644 index 51374ac4c938..000000000000 --- a/qiskit/visualization/timeline/events.py +++ /dev/null @@ -1,121 +0,0 @@ -# This code is part of Qiskit. -# -# (C) Copyright IBM 2020. -# -# This code is licensed under the Apache License, Version 2.0. You may -# obtain a copy of this license in the LICENSE.txt file in the root directory -# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. -# -# Any modifications or derivative works of this code must retain this -# copyright notice, and modified files need to carry a notice indicating -# that they have been altered from the originals. - -""" -Bit event manager for scheduled circuits. - -This module provides a :py:class:`BitEvents` class that manages a series of instructions for a -specific circuit bit. Bit-wise filtering of the circuit program makes the arrangement of bits -easier in the core drawer function. The `BitEvents` class is expected to be called -by other programs (not by end-users). - -The :py:class:`BitEvents` class instance is created with the class method ``load_program``: - ```python - event = BitEvents.load_program(sched_circuit, qregs[0]) - ``` - -Loaded circuit instructions are saved as ``ScheduledGate``, which is a collection of instruction, -associated time, and bits. All gate instructions are returned by the `.get_gates` method. -Instruction types specified in `BitEvents._non_gates` are not considered as gates. -If an instruction is associated with multiple bits and the target bit of the class instance is -the primary bit of the instruction, the instance also generates a ``GateLink`` object -that shows the relationship between bits during multi-bit gates. -""" -from typing import List, Iterator - -from qiskit import circuit -from qiskit.visualization.exceptions import VisualizationError -from qiskit.visualization.timeline import types - - -class BitEvents: - """Bit event table.""" - - _non_gates = (circuit.Barrier,) - - def __init__(self, bit: types.Bits, instructions: List[types.ScheduledGate], t_stop: int): - """Create new event for the specified bit. - - Args: - bit: Bit object associated with this event table. - instructions: List of scheduled gate object. - t_stop: Stop time of this bit. - """ - self.bit = bit - self.instructions = instructions - self.stop_time = t_stop - - @classmethod - def load_program(cls, scheduled_circuit: circuit.QuantumCircuit, bit: types.Bits): - """Build new BitEvents from scheduled circuit. - - Args: - scheduled_circuit: Scheduled circuit object to draw. - bit: Target bit object. - - Returns: - BitEvents: New `BitEvents` object. - - Raises: - VisualizationError: When the circuit is not transpiled with duration. - """ - t0 = 0 - tf = scheduled_circuit.qubit_stop_time(bit) - - instructions = [] - for inst, qargs, cargs in scheduled_circuit.data: - associated_bits = qargs + cargs - if bit not in associated_bits: - continue - - duration = inst.duration - if duration is None: - raise VisualizationError( - "Instruction {oper} has no duration. " - "You need to transpile the QuantumCircuit with " - "gate durations before drawing.".format(oper=inst) - ) - - instructions.append( - types.ScheduledGate( - t0=t0, - operand=inst, - duration=duration, - bits=associated_bits, - bit_position=associated_bits.index(bit), - ) - ) - t0 += duration - - return BitEvents(bit, instructions, tf) - - def get_gates(self) -> Iterator[types.ScheduledGate]: - """Return scheduled gates.""" - for inst in self.instructions: - if not isinstance(inst.operand, self._non_gates): - yield inst - - def get_barriers(self) -> Iterator[types.Barrier]: - """Return barriers.""" - for inst in self.instructions: - if isinstance(inst.operand, circuit.Barrier): - barrier = types.Barrier(t0=inst.t0, bits=inst.bits, bit_position=inst.bit_position) - yield barrier - - def get_gate_links(self) -> Iterator[types.GateLink]: - """Return link between multi-bit gates.""" - for inst in self.get_gates(): - # generate link iff this is the primary bit. - if len(inst.bits) > 1 and inst.bit_position == 0: - t0 = inst.t0 + 0.5 * inst.duration - link = types.GateLink(t0=t0, opname=inst.operand.name, bits=inst.bits) - yield link diff --git a/releasenotes/notes/cleanup-timeline-drawer-a6287bdab4459e6e.yaml b/releasenotes/notes/cleanup-timeline-drawer-a6287bdab4459e6e.yaml new file mode 100644 index 000000000000..8d7de2fec700 --- /dev/null +++ b/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`. diff --git a/test/python/circuit/test_circuit_properties.py b/test/python/circuit/test_circuit_properties.py index c4355f2eee05..83f83a7b0ab2 100644 --- a/test/python/circuit/test_circuit_properties.py +++ b/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() diff --git a/test/python/visualization/timeline/test_core.py b/test/python/visualization/timeline/test_core.py index 38bf60c39df7..447618acf0eb 100644 --- a/test/python/visualization/timeline/test_core.py +++ b/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): diff --git a/test/python/visualization/timeline/test_events.py b/test/python/visualization/timeline/test_events.py deleted file mode 100644 index 994c1d539fc5..000000000000 --- a/test/python/visualization/timeline/test_events.py +++ /dev/null @@ -1,234 +0,0 @@ -# This code is part of Qiskit. -# -# (C) Copyright IBM 2020. -# -# This code is licensed under the Apache License, Version 2.0. You may -# obtain a copy of this license in the LICENSE.txt file in the root directory -# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0. -# -# Any modifications or derivative works of this code must retain this -# copyright notice, and modified files need to carry a notice indicating -# that they have been altered from the originals. - -"""Tests for event of timeline drawer.""" - -import qiskit -from qiskit import QuantumCircuit, transpile -from qiskit.circuit import library -from qiskit.test import QiskitTestCase -from qiskit.visualization.timeline import events, types - - -class TestLoadScheduledCircuit(QiskitTestCase): - """Test for loading program.""" - - def setUp(self) -> None: - """Setup.""" - super().setUp() - - circ = QuantumCircuit(3) - circ.delay(100, 2) - circ.barrier(0, 1, 2) - circ.h(0) - circ.cx(0, 1) - - self.circ = transpile( - circ, - scheduling_method="alap", - basis_gates=["h", "cx"], - instruction_durations=[("h", 0, 200), ("cx", [0, 1], 1000)], - optimization_level=0, - ) - - def test_create_from_program(self): - """Test factory method.""" - bit_event_q0 = events.BitEvents.load_program(self.circ, self.circ.qregs[0][0]) - bit_event_q1 = events.BitEvents.load_program(self.circ, self.circ.qregs[0][1]) - bit_event_q2 = events.BitEvents.load_program(self.circ, self.circ.qregs[0][2]) - - gates_q0 = list(bit_event_q0.get_gates()) - links_q0 = list(bit_event_q0.get_gate_links()) - barriers_q0 = list(bit_event_q0.get_barriers()) - - self.assertEqual(len(gates_q0), 3) - self.assertEqual(len(links_q0), 1) - self.assertEqual(len(barriers_q0), 1) - - # h gate - self.assertEqual(gates_q0[1].t0, 100) - - # cx gate - self.assertEqual(gates_q0[2].t0, 300) - - # link - self.assertEqual(links_q0[0].t0, 800) - - # barrier - self.assertEqual(barriers_q0[0].t0, 100) - - gates_q1 = list(bit_event_q1.get_gates()) - links_q1 = list(bit_event_q1.get_gate_links()) - barriers_q1 = list(bit_event_q1.get_barriers()) - - self.assertEqual(len(gates_q1), 3) - self.assertEqual(len(links_q1), 0) - self.assertEqual(len(barriers_q1), 1) - - # cx gate - self.assertEqual(gates_q0[2].t0, 300) - - # barrier - self.assertEqual(barriers_q1[0].t0, 100) - - gates_q2 = list(bit_event_q2.get_gates()) - links_q2 = list(bit_event_q2.get_gate_links()) - barriers_q2 = list(bit_event_q2.get_barriers()) - - self.assertEqual(len(gates_q2), 2) - self.assertEqual(len(links_q2), 0) - self.assertEqual(len(barriers_q2), 1) - - # barrier - self.assertEqual(barriers_q2[0].t0, 100) - - -class TestBitEvents(QiskitTestCase): - """Tests for bit events.""" - - def setUp(self) -> None: - """Setup.""" - super().setUp() - - self.qubits = list(qiskit.QuantumRegister(2)) - self.clbits = list(qiskit.ClassicalRegister(2)) - - self.instructions = [ - types.ScheduledGate( - t0=0, operand=library.U1Gate(0), duration=0, bits=[self.qubits[0]], bit_position=0 - ), - types.ScheduledGate( - t0=0, - operand=library.U2Gate(0, 0), - duration=10, - bits=[self.qubits[0]], - bit_position=0, - ), - types.ScheduledGate( - t0=10, - operand=library.CXGate(), - duration=50, - bits=[self.qubits[0], self.qubits[1]], - bit_position=0, - ), - types.ScheduledGate( - t0=100, - operand=library.U3Gate(0, 0, 0), - duration=20, - bits=[self.qubits[0]], - bit_position=0, - ), - types.ScheduledGate( - t0=120, - operand=library.Barrier(2), - duration=0, - bits=[self.qubits[0], self.qubits[1]], - bit_position=0, - ), - types.ScheduledGate( - t0=120, - operand=library.CXGate(), - duration=50, - bits=[self.qubits[1], self.qubits[0]], - bit_position=1, - ), - types.ScheduledGate( - t0=200, - operand=library.Barrier(1), - duration=0, - bits=[self.qubits[0]], - bit_position=0, - ), - types.ScheduledGate( - t0=200, - operand=library.Measure(), - duration=100, - bits=[self.qubits[0], self.clbits[0]], - bit_position=0, - ), - ] - - def test_gate_output(self): - """Test gate output.""" - bit_event = events.BitEvents(self.qubits[0], self.instructions, 300) - - gates = list(bit_event.get_gates()) - ref_list = [ - types.ScheduledGate( - t0=0, operand=library.U1Gate(0), duration=0, bits=[self.qubits[0]], bit_position=0 - ), - types.ScheduledGate( - t0=0, - operand=library.U2Gate(0, 0), - duration=10, - bits=[self.qubits[0]], - bit_position=0, - ), - types.ScheduledGate( - t0=10, - operand=library.CXGate(), - duration=50, - bits=[self.qubits[0], self.qubits[1]], - bit_position=0, - ), - types.ScheduledGate( - t0=100, - operand=library.U3Gate(0, 0, 0), - duration=20, - bits=[self.qubits[0]], - bit_position=0, - ), - types.ScheduledGate( - t0=120, - operand=library.CXGate(), - duration=50, - bits=[self.qubits[1], self.qubits[0]], - bit_position=1, - ), - types.ScheduledGate( - t0=200, - operand=library.Measure(), - duration=100, - bits=[self.qubits[0], self.clbits[0]], - bit_position=0, - ), - ] - - self.assertListEqual(gates, ref_list) - - def test_barrier_output(self): - """Test barrier output.""" - bit_event = events.BitEvents(self.qubits[0], self.instructions, 200) - - barriers = list(bit_event.get_barriers()) - ref_list = [ - types.Barrier(t0=120, bits=[self.qubits[0], self.qubits[1]], bit_position=0), - types.Barrier(t0=200, bits=[self.qubits[0]], bit_position=0), - ] - - self.assertListEqual(barriers, ref_list) - - def test_bit_link_output(self): - """Test link output.""" - bit_event = events.BitEvents(self.qubits[0], self.instructions, 250) - - links = list(bit_event.get_gate_links()) - ref_list = [ - types.GateLink( - t0=35.0, opname=library.CXGate().name, bits=[self.qubits[0], self.qubits[1]] - ), - types.GateLink( - t0=250.0, opname=library.Measure().name, bits=[self.qubits[0], self.clbits[0]] - ), - ] - - self.assertListEqual(links, ref_list)