diff --git a/qiskit/dagcircuit/collect_blocks.py b/qiskit/dagcircuit/collect_blocks.py index b15cfee71087..ea574536f45a 100644 --- a/qiskit/dagcircuit/collect_blocks.py +++ b/qiskit/dagcircuit/collect_blocks.py @@ -13,8 +13,13 @@ """Various ways to divide a DAG into blocks of nodes, to split blocks of nodes into smaller sub-blocks, and to consolidate blocks.""" +from __future__ import annotations -from qiskit.circuit import QuantumCircuit, CircuitInstruction, ClassicalRegister +from collections.abc import Iterable, Callable + +from qiskit.dagcircuit import DAGDepNode + +from qiskit.circuit import QuantumCircuit, CircuitInstruction, ClassicalRegister, Bit from qiskit.circuit.controlflow import condition_resources from . import DAGOpNode, DAGCircuit, DAGDependency from .exceptions import DAGCircuitError @@ -38,7 +43,7 @@ class BlockCollector: see https://github.com/Qiskit/qiskit-terra/issues/5775. """ - def __init__(self, dag): + def __init__(self, dag: DAGCircuit | DAGDependency): """ Args: dag (Union[DAGCircuit, DAGDependency]): The input DAG. @@ -48,8 +53,8 @@ def __init__(self, dag): """ self.dag = dag - self._pending_nodes = None - self._in_degree = None + self._pending_nodes: list[DAGOpNode | DAGDepNode] | None = None + self._in_degree: dict[DAGOpNode | DAGDepNode, int] | None = None self._collect_from_back = False if isinstance(dag, DAGCircuit): @@ -79,7 +84,7 @@ def _setup_in_degrees(self): if deg == 0: self._pending_nodes.append(node) - def _op_nodes(self): + def _op_nodes(self) -> Iterable[DAGOpNode | DAGDepNode]: """Returns DAG nodes.""" if not self.is_dag_dependency: return self.dag.op_nodes() @@ -134,7 +139,7 @@ def _have_uncollected_nodes(self): """Returns whether there are uncollected (pending) nodes""" return len(self._pending_nodes) > 0 - def collect_matching_block(self, filter_fn): + def collect_matching_block(self, filter_fn: Callable) -> list[DAGOpNode | DAGDepNode]: """Iteratively collects the largest block of input nodes (that is, nodes with ``_in_degree`` equal to 0) that match a given filtering function. Examples of this include collecting blocks of swap gates, @@ -205,7 +210,7 @@ def not_filter_fn(node): self._setup_in_degrees() # Iteratively collect non-matching and matching blocks. - matching_blocks = [] + matching_blocks: list[list[DAGOpNode | DAGDepNode]] = [] while self._have_uncollected_nodes(): self.collect_matching_block(not_filter_fn) matching_block = self.collect_matching_block(filter_fn) @@ -290,12 +295,12 @@ def run(self, block): return blocks -def split_block_into_layers(block): +def split_block_into_layers(block: list[DAGOpNode | DAGDepNode]): """Splits a block of nodes into sub-blocks of non-overlapping instructions (or, in other words, into depth-1 sub-blocks). """ - bit_depths = {} - layers = [] + bit_depths: dict[Bit, int] = {} + layers: list[list[DAGOpNode | DAGDepNode]] = [] for node in block: cur_bits = set(node.qargs) diff --git a/qiskit/dagcircuit/dagcircuit.py b/qiskit/dagcircuit/dagcircuit.py index c4da14078ffb..8c1332a8e604 100644 --- a/qiskit/dagcircuit/dagcircuit.py +++ b/qiskit/dagcircuit/dagcircuit.py @@ -20,10 +20,13 @@ composed, and modified. Some natural properties like depth can be computed directly from the graph. """ +from __future__ import annotations + from collections import OrderedDict, defaultdict, deque, namedtuple +from collections.abc import Callable, Sequence, Generator, Iterable import copy import math -from typing import Dict, Generator, Any, List +from typing import Any import numpy as np import rustworkx as rx @@ -35,6 +38,7 @@ WhileLoopOp, SwitchCaseOp, _classical_resource_map, + Operation, ) from qiskit.circuit.controlflow import condition_resources, node_resources, CONTROL_FLOW_OP_NAMES from qiskit.circuit.quantumregister import QuantumRegister, Qubit @@ -45,7 +49,7 @@ from qiskit.dagcircuit.exceptions import DAGCircuitError from qiskit.dagcircuit.dagnode import DAGNode, DAGOpNode, DAGInNode, DAGOutNode from qiskit.circuit.bit import Bit - +from qiskit.pulse import Schedule BitLocations = namedtuple("BitLocations", ("index", "registers")) @@ -97,18 +101,18 @@ def __init__(self): self.cregs = OrderedDict() # List of Qubit/Clbit wires that the DAG acts on. - self.qubits: List[Qubit] = [] - self.clbits: List[Clbit] = [] + self.qubits: list[Qubit] = [] + self.clbits: list[Clbit] = [] # Dictionary mapping of Qubit and Clbit instances to a tuple comprised of # 0) corresponding index in dag.{qubits,clbits} and # 1) a list of Register-int pairs for each Register containing the Bit and # its index within that register. - self._qubit_indices: Dict[Qubit, BitLocations] = {} - self._clbit_indices: Dict[Clbit, BitLocations] = {} + self._qubit_indices: dict[Qubit, BitLocations] = {} + self._clbit_indices: dict[Clbit, BitLocations] = {} - self._global_phase = 0 - self._calibrations = defaultdict(dict) + self._global_phase: float | ParameterExpression = 0.0 + self._calibrations: dict[str, dict[tuple, Schedule]] = defaultdict(dict) self._op_names = {} @@ -133,7 +137,7 @@ def global_phase(self): return self._global_phase @global_phase.setter - def global_phase(self, angle): + def global_phase(self, angle: float | ParameterExpression): """Set the global phase of the circuit. Args: @@ -150,7 +154,7 @@ def global_phase(self, angle): self._global_phase = angle % (2 * math.pi) @property - def calibrations(self): + def calibrations(self) -> dict[str, dict[tuple, Schedule]]: """Return calibration dictionary. The custom pulse definition of a given gate is of the form @@ -159,7 +163,7 @@ def calibrations(self): return dict(self._calibrations) @calibrations.setter - def calibrations(self, calibrations): + def calibrations(self, calibrations: dict[str, dict[tuple, Schedule]]): """Set the circuit calibration data from a dictionary of calibration definition. Args: @@ -637,7 +641,14 @@ def copy_empty_like(self): return target_dag - def apply_operation_back(self, op, qargs=(), cargs=(), *, check=True): + def apply_operation_back( + self, + op: Operation, + qargs: Iterable[Qubit] = (), + cargs: Iterable[Clbit] = (), + *, + check: bool = True, + ) -> DAGOpNode: """Apply an operation to the output of the circuit. Args: @@ -683,7 +694,14 @@ def apply_operation_back(self, op, qargs=(), cargs=(), *, check=True): ) return node - def apply_operation_front(self, op, qargs=(), cargs=(), *, check=True): + def apply_operation_front( + self, + op: Operation, + qargs: Sequence[Qubit] = (), + cargs: Sequence[Clbit] = (), + *, + check: bool = True, + ) -> DAGOpNode: """Apply an operation to the input of the circuit. Args: @@ -1075,7 +1093,7 @@ def _key(x): return iter(rx.lexicographical_topological_sort(self._multi_graph, key=key)) - def topological_op_nodes(self, key=None) -> Generator[DAGOpNode, Any, Any]: + def topological_op_nodes(self, key: Callable | None = None) -> Generator[DAGOpNode, Any, Any]: """ Yield op nodes in topological order. @@ -1092,7 +1110,9 @@ def topological_op_nodes(self, key=None) -> Generator[DAGOpNode, Any, Any]: """ return (nd for nd in self.topological_nodes(key) if isinstance(nd, DAGOpNode)) - def replace_block_with_op(self, node_block, op, wire_pos_map, cycle_check=True): + def replace_block_with_op( + self, node_block: list[DAGOpNode], op: Operation, wire_pos_map, cycle_check=True + ): """Replace a block of nodes with a single node. This is used to consolidate a block of DAGOpNodes into a single @@ -1381,7 +1401,7 @@ def edge_weight_map(wire): return {k: self._multi_graph[v] for k, v in node_map.items()} - def substitute_node(self, node, op, inplace=False, propagate_condition=True): + def substitute_node(self, node: DAGOpNode, op, inplace: bool = False, propagate_condition=True): """Replace an DAGOpNode with a single operation. qargs, cargs and conditions for the new operation will be inferred from the node to be replaced. The new operation will be checked to match the shape of the @@ -1471,7 +1491,7 @@ def substitute_node(self, node, op, inplace=False, propagate_condition=True): self._decrement_op(node.op) return new_node - def separable_circuits(self, remove_idle_qubits=False) -> List["DAGCircuit"]: + def separable_circuits(self, remove_idle_qubits: bool = False) -> list["DAGCircuit"]: """Decompose the circuit into sets of qubits with no gates connecting them. Args: @@ -1892,7 +1912,7 @@ def filter_fn(node): group_list = rx.collect_runs(self._multi_graph, filter_fn) return {tuple(x) for x in group_list} - def collect_1q_runs(self): + def collect_1q_runs(self) -> list[list[DAGOpNode]]: """Return a set of non-conditional runs of 1q "op" nodes.""" def filter_fn(node): diff --git a/qiskit/dagcircuit/dagdependency.py b/qiskit/dagcircuit/dagdependency.py index a519f0b53f7f..4316c9471401 100644 --- a/qiskit/dagcircuit/dagdependency.py +++ b/qiskit/dagcircuit/dagdependency.py @@ -12,10 +12,13 @@ """DAGDependency class for representing non-commutativity in a circuit. """ +from __future__ import annotations import math import heapq +import typing from collections import OrderedDict, defaultdict +from collections.abc import Iterator import rustworkx as rx @@ -25,6 +28,10 @@ from qiskit.circuit.classicalregister import ClassicalRegister, Clbit from qiskit.dagcircuit.exceptions import DAGDependencyError from qiskit.dagcircuit.dagdepnode import DAGDepNode +from qiskit.pulse import Schedule + +if typing.TYPE_CHECKING: + from qiskit.circuit.parameterexpression import ParameterExpression # ToDo: DagDependency needs to be refactored: @@ -106,8 +113,8 @@ def __init__(self): self.qubits = [] self.clbits = [] - self._global_phase = 0 - self._calibrations = defaultdict(dict) + self._global_phase: float | ParameterExpression = 0.0 + self._calibrations: dict[str, dict[tuple, Schedule]] = defaultdict(dict) self.duration = None self.unit = "dt" @@ -120,7 +127,7 @@ def global_phase(self): return self._global_phase @global_phase.setter - def global_phase(self, angle): + def global_phase(self, angle: float | ParameterExpression): """Set the global phase of the circuit. Args: @@ -139,7 +146,7 @@ def global_phase(self, angle): self._global_phase = angle % (2 * math.pi) @property - def calibrations(self): + def calibrations(self) -> dict[str, dict[tuple, Schedule]]: """Return calibration dictionary. The custom pulse definition of a given gate is of the form @@ -148,7 +155,7 @@ def calibrations(self): return dict(self._calibrations) @calibrations.setter - def calibrations(self, calibrations): + def calibrations(self, calibrations: dict[str, dict[tuple, Schedule]]): """Set the circuit calibration data from a dictionary of calibration definition. Args: @@ -219,7 +226,7 @@ def add_creg(self, creg): if creg[j] not in existing_clbits: self.clbits.append(creg[j]) - def _add_multi_graph_node(self, node): + def _add_multi_graph_node(self, node: DAGDepNode) -> int: """ Args: node (DAGDepNode): considered node. @@ -231,14 +238,14 @@ def _add_multi_graph_node(self, node): node.node_id = node_id return node_id - def get_nodes(self): + def get_nodes(self) -> Iterator[DAGDepNode]: """ Returns: generator(dict): iterator over all the nodes. """ return iter(self._multi_graph.nodes()) - def get_node(self, node_id): + def get_node(self, node_id: int) -> DAGDepNode: """ Args: node_id (int): label of considered node. @@ -248,7 +255,7 @@ def get_node(self, node_id): """ return self._multi_graph.get_node_data(node_id) - def _add_multi_graph_edge(self, src_id, dest_id, data): + def _add_multi_graph_edge(self, src_id: int, dest_id: int, data): """ Function to add an edge from given data (dict) between two nodes. @@ -311,7 +318,7 @@ def get_out_edges(self, node_id): """ return self._multi_graph.out_edges(node_id) - def direct_successors(self, node_id): + def direct_successors(self, node_id: int) -> list[int]: """ Direct successors id of a given node as sorted list. @@ -335,7 +342,7 @@ def direct_predecessors(self, node_id): """ return sorted(self._multi_graph.adj_direction(node_id, True).keys()) - def successors(self, node_id): + def successors(self, node_id: int) -> list[int]: """ Successors id of a given node as sorted list. @@ -347,7 +354,7 @@ def successors(self, node_id): """ return self._multi_graph.get_node_data(node_id).successors - def predecessors(self, node_id): + def predecessors(self, node_id: int) -> list[int]: """ Predecessors id of a given node as sorted list. diff --git a/qiskit/dagcircuit/dagdepnode.py b/qiskit/dagcircuit/dagdepnode.py index 0ac7be65f103..fe63f57d3d4f 100644 --- a/qiskit/dagcircuit/dagdepnode.py +++ b/qiskit/dagcircuit/dagdepnode.py @@ -13,6 +13,7 @@ # pylint: disable=redefined-builtin """Object to represent the information at a node in the DAGCircuit.""" +from __future__ import annotations from qiskit.exceptions import QiskitError @@ -49,15 +50,15 @@ def __init__( name=None, qargs=(), cargs=(), - successors=None, - predecessors=None, + successors: list[int] | None = None, + predecessors: list[int] | None = None, reachable=None, - matchedwith=None, - successorstovisit=None, - isblocked=None, - qindices=None, - cindices=None, - nid=-1, + matchedwith: list[int] | None = None, + successorstovisit: list[int] | None = None, + isblocked: bool | None = None, + qindices: list[int] | None = None, + cindices: list[int] | None = None, + nid: int = -1, ): self.type = type @@ -66,15 +67,17 @@ def __init__( self._qargs = tuple(qargs) if qargs is not None else () self.cargs = tuple(cargs) if cargs is not None else () self.node_id = nid - self.sort_key = str(self._qargs) - self.successors = successors if successors is not None else [] - self.predecessors = predecessors if predecessors is not None else [] + self.sort_key: str = str(self._qargs) + self.successors: list[int] = successors if successors is not None else [] + self.predecessors: list[int] = predecessors if predecessors is not None else [] self.reachable = reachable - self.matchedwith = matchedwith if matchedwith is not None else [] - self.isblocked = isblocked - self.successorstovisit = successorstovisit if successorstovisit is not None else [] - self.qindices = qindices if qindices is not None else [] - self.cindices = cindices if cindices is not None else [] + self.matchedwith: list[int] = matchedwith if matchedwith is not None else [] + self.isblocked: bool = isblocked + self.successorstovisit: list[int] = ( + successorstovisit if successorstovisit is not None else [] + ) + self.qindices: list[int] = qindices if qindices is not None else [] + self.cindices: list[int] = cindices if cindices is not None else [] @property def op(self): diff --git a/qiskit/dagcircuit/dagnode.py b/qiskit/dagcircuit/dagnode.py index 3b3c8fb38c7a..97283c75b8fc 100644 --- a/qiskit/dagcircuit/dagnode.py +++ b/qiskit/dagcircuit/dagnode.py @@ -12,10 +12,13 @@ """Objects to represent the information at a node in the DAGCircuit.""" +from __future__ import annotations import itertools +import typing import uuid -from typing import Iterable +from collections.abc import Iterable + from qiskit.circuit import ( Qubit, @@ -27,11 +30,16 @@ SwitchCaseOp, ForLoopOp, Parameter, + Operation, + QuantumCircuit, ) from qiskit.circuit.classical import expr +if typing.TYPE_CHECKING: + from qiskit.dagcircuit import DAGCircuit + -def _legacy_condition_eq(cond1, cond2, bit_indices1, bit_indices2): +def _legacy_condition_eq(cond1, cond2, bit_indices1, bit_indices2) -> bool: if cond1 is cond2 is None: return True elif None in (cond1, cond2): @@ -49,7 +57,7 @@ def _legacy_condition_eq(cond1, cond2, bit_indices1, bit_indices2): return False -def _circuit_to_dag(circuit, node_qargs, node_cargs, bit_indices): +def _circuit_to_dag(circuit: QuantumCircuit, node_qargs, node_cargs, bit_indices) -> DAGCircuit: """Get a :class:`.DAGCircuit` of the given :class:`.QuantumCircuit`. The bits in the output will be ordered in a canonical order based on their indices in the outer DAG, as defined by the ``bit_indices`` mapping and the ``node_{q,c}args`` arguments.""" @@ -252,7 +260,9 @@ class DAGOpNode(DAGNode): __slots__ = ["op", "qargs", "cargs", "sort_key"] - def __init__(self, op, qargs: Iterable[Qubit] = (), cargs: Iterable[Clbit] = (), dag=None): + def __init__( + self, op: Operation, qargs: Iterable[Qubit] = (), cargs: Iterable[Clbit] = (), dag=None + ): """Create an Instruction node""" super().__init__() self.op = op