Fix dagcircuit mypy errors

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)

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):

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.