[Stretch] Support stretch and duration expressions for Delay instruction.

crates/circuit/src/operations.rs

@@ -297,13 +297,14 @@ pub enum DelayUnit {
     MS,
     S,
     DT,
+    EXPR,
 }
 
 unsafe impl ::bytemuck::CheckedBitPattern for DelayUnit {
     type Bits = u8;
 
     fn is_valid_bit_pattern(bits: &Self::Bits) -> bool {
-        *bits < 6
+        *bits < 7
     }
 }
 unsafe impl ::bytemuck::NoUninit for DelayUnit {}
@@ -320,6 +321,7 @@ impl fmt::Display for DelayUnit {
                 DelayUnit::MS => "ms",
                 DelayUnit::S => "s",
                 DelayUnit::DT => "dt",
+                DelayUnit::EXPR => "expr",
             }
         )
     }
@@ -335,6 +337,7 @@ impl<'py> FromPyObject<'py> for DelayUnit {
             "ms" => DelayUnit::MS,
             "s" => DelayUnit::S,
             "dt" => DelayUnit::DT,
+            "expr" => DelayUnit::EXPR,
             unknown_unit => {
                 return Err(PyValueError::new_err(format!(
                     "Unit '{}' is invalid.",

qiskit/circuit/controlflow/builder.py

@@ -540,11 +540,15 @@ def remove_stretch(self, stretch: expr.Stretch):
         self._stretches_local.pop(stretch.name)
 
     def get_var(self, name: str):
+        if name in self._stretches_local:
+            return None
         if (out := self._vars_local.get(name)) is not None:
             return out
         return self._parent.get_var(name)
 
     def get_stretch(self, name: str):
+        if name in self._vars_local:
+            return None
         if (out := self._stretches_local.get(name)) is not None:
             return out
         return self._parent.get_stretch(name)

qiskit/circuit/delay.py

@@ -14,6 +14,8 @@
 Delay instruction (for circuit module).
 """
 import numpy as np
+
+from qiskit.circuit.classical import expr, types
 from qiskit.circuit.exceptions import CircuitError
 from qiskit.circuit.instruction import Instruction
 from qiskit.circuit.gate import Gate
@@ -28,13 +30,35 @@ class Delay(Instruction):
 
     _standard_instruction_type = StandardInstructionType.Delay
 
-    def __init__(self, duration, unit="dt"):
+    def __init__(self, duration, unit=None):
         """
         Args:
-            duration: the length of time of the duration.  Given in units of ``unit``.
-            unit: the unit of the duration.  Must be ``"dt"`` or an SI-prefixed seconds unit.
+            duration: the length of time of the duration. If this is an
+                :class:`~.expr.Expr`, it must be a constant expression of type
+                :class:`~.types.Duration` and the ``unit`` parameter should be
+                omitted (or MUST be "expr" if it is specified).
+            unit: the unit of the duration, if ``duration`` is a numeric
+                value. Must be ``"dt"``, an SI-prefixed seconds unit, or "expr".
+
+        Raises:
+            CircuitError: A ``duration`` expression was specified with a resolved
+                type that is not timing-based, or the ``unit`` was improperly specified.
         """
-        if unit not in {"s", "ms", "us", "ns", "ps", "dt"}:
+        if isinstance(duration, expr.Expr):
+            if unit is not None and unit != "expr":
+                raise CircuitError(
+                    "Argument 'unit' must not be specified for a duration expression."
+                )
+            if duration.type.kind is not types.Duration:
+                raise CircuitError(
+                    f"Expression of type '{duration.type}' is not valid for 'duration'."
+                )
+            if not duration.const:
+                raise CircuitError("Duration expressions must be constant.")
+            unit = "expr"
+        elif unit is None:
+            unit = "dt"
+        elif unit not in {"s", "ms", "us", "ns", "ps", "dt"}:
             raise CircuitError(f"Unknown unit {unit} is specified.")
         # Double underscore to differentiate from the private attribute in
         # `Instruction`. This can be changed to `_unit` in 2.0 after we
@@ -88,8 +112,9 @@ def __repr__(self):
         """Return the official string representing the delay."""
         return f"{self.__class__.__name__}(duration={self.params[0]}[unit={self.unit}])"
 
+    # pylint: disable=too-many-return-statements
     def validate_parameter(self, parameter):
-        """Delay parameter (i.e. duration) must be int, float or ParameterExpression."""
+        """Delay parameter (i.e. duration) must be Expr, int, float or ParameterExpression."""
         if isinstance(parameter, int):
             if parameter < 0:
                 raise CircuitError(
@@ -107,6 +132,12 @@ def validate_parameter(self, parameter):
                     raise CircuitError("Integer duration is expected for 'dt' unit.")
                 return parameter_int
             return parameter
+        elif isinstance(parameter, expr.Expr):
+            if parameter.type.kind is not types.Duration:
+                raise CircuitError(f"Expression duration of type '{parameter.type}' is not valid.")
+            if not parameter.const:
+                raise CircuitError("Duration expressions must be constant.")
+            return parameter
         elif isinstance(parameter, ParameterExpression):
             if len(parameter.parameters) > 0:
                 return parameter  # expression has free parameters, we cannot validate it

qiskit/circuit/quantumcircuit.py

@@ -4553,17 +4553,20 @@ def barrier(self, *qargs: QubitSpecifier, label=None) -> InstructionSet:
 
     def delay(
         self,
-        duration: ParameterValueType,
+        duration: Union[ParameterValueType, expr.Expr],
         qarg: QubitSpecifier | None = None,
-        unit: str = "dt",
+        unit: str | None = None,
     ) -> InstructionSet:
         """Apply :class:`~.circuit.Delay`. If qarg is ``None``, applies to all qubits.
         When applying to multiple qubits, delays with the same duration will be created.
 
         Args:
-            duration (int or float or ParameterExpression): duration of the delay.
+            duration (Object):
+                duration of the delay. If this is an :class:`~.expr.Expr`, it must be
+                a constant expression of type :class:`~.types.Duration`.
             qarg (Object): qubit argument to apply this delay.
-            unit (str): unit of the duration. Supported units: ``'s'``, ``'ms'``, ``'us'``,
+            unit (str | None): unit of the duration, unless ``duration`` is an :class:`~.expr.Expr`
+                in which case it must not be specified. Supported units: ``'s'``, ``'ms'``, ``'us'``,
                 ``'ns'``, ``'ps'``, and ``'dt'``. Default is ``'dt'``, i.e. integer time unit
                 depending on the target backend.
 

qiskit/qasm3/exporter.py

@@ -1151,6 +1151,8 @@ def build_duration(self, duration, unit) -> ast.Expression | None:
         """Build the expression of a given duration (if not ``None``)."""
         if duration is None:
             return None
+        if unit == "expr":
+            return self.build_expression(duration)
         if unit == "ps":
             return ast.DurationLiteral(1000 * duration, ast.DurationUnit.NANOSECOND)
         unit_map = {

qiskit/transpiler/passes/scheduling/padding/base_padding.py

@@ -210,6 +210,9 @@ def _pre_runhook(self, dag: DAGCircuit):
                 f"The input circuit {dag.name} is not scheduled. Call one of scheduling passes "
                 f"before running the {self.__class__.__name__} pass."
             )
+        if self.property_set["time_unit"] == "stretch":
+            # This should have already been raised during scheduling, but just in case.
+            raise TranspilerError("Scheduling cannot run on circuits with stretch durations.")
         for qarg, _ in enumerate(dag.qubits):
             if not self.__delay_supported(qarg):
                 logger.debug(

qiskit/transpiler/passes/scheduling/scheduling/alap.py

@@ -39,6 +39,8 @@ def run(self, dag):
         """
         if len(dag.qregs) != 1 or dag.qregs.get("q", None) is None:
             raise TranspilerError("ALAP schedule runs on physical circuits only")
+        if self.property_set["time_unit"] == "stretch":
+            raise TranspilerError("Scheduling cannot run on circuits with stretch durations.")
 
         clbit_write_latency = self.property_set.get("clbit_write_latency", 0)
 

qiskit/transpiler/passes/scheduling/scheduling/asap.py

@@ -39,6 +39,8 @@ def run(self, dag):
         """
         if len(dag.qregs) != 1 or dag.qregs.get("q", None) is None:
             raise TranspilerError("ASAP schedule runs on physical circuits only")
+        if self.property_set["time_unit"] == "stretch":
+            raise TranspilerError("Scheduling cannot run on circuits with stretch durations.")
 
         clbit_write_latency = self.property_set.get("clbit_write_latency", 0)
 

qiskit/transpiler/passes/scheduling/time_unit_conversion.py

@@ -13,12 +13,15 @@
 """Unify time unit in circuit for scheduling and following passes."""
 from typing import Set
 
-from qiskit.circuit import Delay
+from qiskit.circuit import Delay, Duration
+from qiskit.circuit.classical import expr
+from qiskit.circuit.duration import duration_in_dt
 from qiskit.dagcircuit import DAGCircuit
 from qiskit.transpiler.basepasses import TransformationPass
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.instruction_durations import InstructionDurations
 from qiskit.transpiler.target import Target
+from qiskit.utils import apply_prefix
 
 
 class TimeUnitConversion(TransformationPass):
@@ -71,52 +74,83 @@ def run(self, dag: DAGCircuit):
         if self._durations_provided:
             inst_durations.update(self.inst_durations, getattr(self.inst_durations, "dt", None))
 
+        # The float-value converted units for delay expressions, either all in 'dt'
+        # or all in seconds.
+        expression_durations = {}
+
         # Choose unit
-        if inst_durations.dt is not None:
-            time_unit = "dt"
-        else:
-            # Check what units are used in delays and other instructions: dt or SI or mixed
-            units_delay = self._units_used_in_delays(dag)
-            if self._unified(units_delay) == "mixed":
+        has_dt = False
+        has_si = False
+
+        # We _always_ need to traverse duration expressions to convert them to
+        # a float. But we also use the opportunity to note if they intermix cycles
+        # and wall-time, in case we don't have a `dt` to use to unify all instruction
+        # durations.
+        for node in dag.op_nodes(op=Delay):
+            if isinstance(node.op.duration, expr.Expr):
+                if any(
+                    isinstance(x, expr.Stretch) for x in expr.iter_identifiers(node.op.duration)
+                ):
+                    # If any of the delays use a stretch expression, we can't run scheduling
+                    # passes anyway, so we bail out. In theory, we _could_ still traverse
+                    # through the stretch expression and replace any Duration value nodes it may
+                    # contain with ones of the same units, but it'd be complex and probably unuseful.
+                    self.property_set["time_unit"] = "stretch"
+                    return dag
+
+                visitor = _EvalDurationImpl(inst_durations.dt)
+                duration = node.op.duration.accept(visitor)
+                if visitor.in_cycles():
+                    has_dt = True
+                    # We need to round in case the expression evaluated to a non-integral 'dt'.
+                    duration = duration_in_dt(duration, 1.0)
+                else:
+                    has_si = True
+                if duration < 0:
+                    raise TranspilerError(
+                        f"Expression '{node.op.duration}' resolves to a negative duration!"
+                    )
+                expression_durations[node._node_id] = duration
+            else:
+                if node.op.unit == "dt":
+                    has_dt = True
+                else:
+                    has_si = True
+            if inst_durations.dt is None and has_dt and has_si:
                 raise TranspilerError(
                     "Fail to unify time units in delays. SI units "
                     "and dt unit must not be mixed when dt is not supplied."
                 )
-            units_other = inst_durations.units_used()
-            if self._unified(units_other) == "mixed":
-                raise TranspilerError(
-                    "Fail to unify time units in instruction_durations. SI units "
-                    "and dt unit must not be mixed when dt is not supplied."
-                )
 
-            unified_unit = self._unified(units_delay | units_other)
-            if unified_unit == "SI":
+        if inst_durations.dt is None:
+            # Check what units are used in other instructions: dt or SI or mixed
+            units_other = inst_durations.units_used()
+            unified_unit = self._unified(units_other)
+            if unified_unit == "SI" and not has_dt:
                 time_unit = "s"
-            elif unified_unit == "dt":
+            elif unified_unit == "dt" and not has_si:
                 time_unit = "dt"
             else:
                 raise TranspilerError(
                     "Fail to unify time units. SI units "
                     "and dt unit must not be mixed when dt is not supplied."
                 )
+        else:
+            time_unit = "dt"
 
         # Make instructions with local durations consistent.
         for node in dag.op_nodes(Delay):
             op = node.op.to_mutable()
-            op.duration = inst_durations._convert_unit(op.duration, op.unit, time_unit)
+            if node._node_id in expression_durations:
+                op.duration = expression_durations[node._node_id]
+            else:
+                op.duration = inst_durations._convert_unit(op.duration, op.unit, time_unit)
             op.unit = time_unit
             dag.substitute_node(node, op)
 
         self.property_set["time_unit"] = time_unit
         return dag
 
-    @staticmethod
-    def _units_used_in_delays(dag: DAGCircuit) -> Set[str]:
-        units_used = set()
-        for node in dag.op_nodes(op=Delay):
-            units_used.add(node.op.unit)
-        return units_used
-
     @staticmethod
     def _unified(unit_set: Set[str]) -> str:
         if not unit_set:
@@ -135,3 +169,65 @@ def _unified(unit_set: Set[str]) -> str:
             return "SI"
 
         return "mixed"
+
+
+class _EvalDurationImpl(expr.ExprVisitor[float]):
+    """Evaluates the expression to a single float result.
+
+    If ``dt`` is provided or all durations are already in ``dt``, the result is in ``dt``.
+    Otherwise, the result will be in seconds, and all durations MUST be in wall-time (SI).
+    """
+
+    __slots__ = ("dt", "has_dt", "has_si")
+
+    def __init__(self, dt=None):
+        self.dt = dt
+        self.has_dt = False
+        self.has_si = False
+
+    def in_cycles(self):
+        """Returns ``True`` if units are 'dt' after visit."""
+        return self.has_dt or self.dt is not None
+
+    def visit_value(self, node, /) -> float:
+        if isinstance(node.value, float):
+            return node.value
+        if isinstance(node.value, Duration.dt):
+            if self.has_si and self.dt is None:
+                raise TranspilerError(
+                    "Fail to unify time units in delays. SI units "
+                    "and dt unit must not be mixed when dt is not supplied."
+                )
+            self.has_dt = True
+            return node.value[0]
+        if isinstance(node.value, Duration):
+            if self.has_dt and self.dt is None:
+                raise TranspilerError(
+                    "Fail to unify time units in delays. SI units "
+                    "and dt unit must not be mixed when dt is not supplied."
+                )
+            self.has_si = True
+            # Setting 'divisor' to 1 when there's no 'dt' is just to simplify
+            # the logic (we don't need to divide).
+            divisor = self.dt if self.dt is not None else 1
+            if isinstance(node.value, Duration.s):
+                return node.value[0] / divisor
+            from_unit = node.value.unit()
+            return apply_prefix(node.value[0], from_unit) / divisor
+        raise TranspilerError(f"invalid duration expression: {node}")
+
+    def visit_binary(self, node, /) -> float:
+        left = node.left.accept(self)
+        right = node.right.accept(self)
+        if node.op == expr.Binary.Op.ADD:
+            return left + right
+        if node.op == expr.Binary.Op.SUB:
+            return left - right
+        if node.op == expr.Binary.Op.MUL:
+            return left * right
+        if node.op == expr.Binary.Op.DIV:
+            return left / right
+        raise TranspilerError(f"invalid duration expression: {node}")
+
+    def visit_cast(self, node, /) -> float:
+        return node.operand.accept(self)