# Repository: https://gitlab.com/qblox/packages/software/qblox-scheduler
# Licensed according to the LICENSE file on the main branch
#
# Copyright 2020-2025, Quantify Consortium
# Copyright 2025, Qblox B.V.
"""Classes for handling operations that are neither pulses nor acquisitions."""
from __future__ import annotations
from contextlib import contextmanager
from typing import TYPE_CHECKING, Any
import numpy as np
from qblox_scheduler._check_unsupported_expression import check_unsupported_expression
from qblox_scheduler.backends.qblox import constants, helpers, q1asm_instructions
from qblox_scheduler.backends.qblox.operation_handling.base import IOperationStrategy
from qblox_scheduler.backends.qblox.type_casting import _SIGNED_INT_CASTING_FNS, twos_complement
from qblox_scheduler.operations.expressions import (
BinaryExpression,
DType,
Expression,
UnaryExpression,
)
from qblox_scheduler.operations.variables import Variable
if TYPE_CHECKING:
from collections.abc import Iterator
from qblox_scheduler.backends.qblox.conditional import FeedbackTriggerCondition
from qblox_scheduler.backends.qblox.qasm_program import (
QASMProgram,
)
from qblox_scheduler.backends.types import qblox as types
[docs]
class IdleStrategy(IOperationStrategy):
"""
Defines the behavior for an operation that does not produce any output.
Parameters
----------
operation_info : qblox_scheduler.backends.types.qblox.OpInfo
The operation info that corresponds to this operation.
"""
def __init__(self, operation_info: types.OpInfo) -> None:
[docs]
self._op_info = operation_info
@property
[docs]
def operation_info(self) -> types.OpInfo:
"""Property for retrieving the operation info."""
return self._op_info
@property
[docs]
def generates_path_Q_output(self) -> bool:
"""
If True, this strategy compiles to instructions that generate output on Q. If False, the
instructions either generate no output, or output only on path I.
"""
return False
[docs]
def generate_data(self, wf_dict: dict[str, Any]) -> None:
"""Returns None as no waveforms are generated in this strategy."""
pass
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Add the assembly instructions for the Q1 sequence processor that corresponds to
this operation.
Not an abstractmethod, since it is allowed to use the IdleStrategy directly
(e.g. for IdlePulses), but can be overridden in subclass to add some assembly
instructions despite not outputting any data.
Parameters
----------
qasm_program
The QASMProgram to add the assembly instructions to.
"""
[docs]
class NcoPhaseShiftStrategy(IdleStrategy):
"""
Strategy for operation that does not produce any output, but rather applies a
phase shift to the NCO. Implemented as ``set_ph_delta`` and an ``upd_param`` of 8 ns,
leading to a total duration of 8 ns before the next command can be issued.
"""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Inserts the instructions needed to shift the NCO phase by a specific amount.
Parameters
----------
qasm_program
The QASMProgram to add the assembly instructions to.
"""
phase_shift = self.operation_info.data["phase_shift"]
if phase_shift == 0:
return
if isinstance(phase_shift, Variable):
phase_shift_arg = qasm_program.register_manager.get_register_of_variable(phase_shift)
label = f"{phase_shift_arg}_phase_is_wrapped"
max_nco_phase_steps_int = constants.MAX_NCO_PHASE_STEPS
qasm_program.emit(
q1asm_instructions.JUMP_LESS_THAN,
phase_shift_arg,
max_nco_phase_steps_int,
f"@{label}",
)
qasm_program.emit(
q1asm_instructions.SUB,
phase_shift_arg,
max_nco_phase_steps_int,
phase_shift_arg,
)
qasm_program.emit(q1asm_instructions.NEW_LINE, label=label)
else:
phase_shift_arg = helpers.get_nco_phase_arguments(phase_shift)
qasm_program.emit(
q1asm_instructions.INCR_NCO_PHASE_OFFSET,
phase_shift_arg,
comment="increment nco phase",
)
[docs]
class NcoResetClockPhaseStrategy(IdleStrategy):
"""
Strategy for operation that does not produce any output, but rather resets
the phase of the NCO.
"""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Inserts the instructions needed to reset the NCO phase.
Parameters
----------
qasm_program
The QASMProgram to add the assembly instructions to.
"""
reset_clock_phase = self.operation_info.data.get("reset_clock_phase")
if reset_clock_phase is None:
raise KeyError(
"NcoResetClockPhaseStrategy called, "
"but reset_clock_phase not present in operation_info.data"
)
qasm_program.emit(q1asm_instructions.RESET_PHASE)
[docs]
class NcoSetClockFrequencyStrategy(IdleStrategy):
"""
Strategy for operation that does not produce any output, but rather sets
the frequency of the NCO. Implemented as ``set_freq`` and an ``upd_param`` of 4 ns,
leading to a total duration of 4 ns before the next command can be issued.
"""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Inserts the instructions needed to set the NCO frequency.
Parameters
----------
qasm_program
The QASMProgram to add the assembly instructions to.
"""
clock_freq_new = self.operation_info.data.get("clock_freq_new")
clock_freq_old = self.operation_info.data.get("clock_freq_old")
interm_freq_old = self.operation_info.data.get("interm_freq_old")
if clock_freq_old is None or np.isnan(clock_freq_old):
raise RuntimeError(
f"Clock '{self.operation_info.data.get('clock')}' has an undefined "
f"initial frequency ({clock_freq_old=}); "
f"ensure this resource has been added to the schedule or to the device "
f"config."
)
if interm_freq_old is None:
raise RuntimeError(
f"Clock '{self.operation_info.data.get('clock')}' has an undefined "
f"associated intermodulation frequency ({interm_freq_old=}); make "
f"sure an 'interm_freq' is supplied or that 'mix_lo' is set to true in "
f"the hardware config."
)
if not isinstance(clock_freq_new, Variable):
check_unsupported_expression(
clock_freq_new,
clock_freq_old,
interm_freq_old,
operation_name=self.operation_info.name,
)
interm_freq_new = (
(interm_freq_old + clock_freq_new - clock_freq_old)
if (clock_freq_new is not None)
else interm_freq_old
)
frequency_args = helpers.get_nco_set_frequency_arguments(interm_freq_new)
else:
frequency_args = qasm_program.register_manager.get_register_of_variable(clock_freq_new)
qasm_program.emit(
q1asm_instructions.SET_FREQUENCY,
frequency_args,
comment="Update NCO frequency",
)
[docs]
class AwgOffsetStrategy(IdleStrategy):
"""
Strategy for compiling a DC voltage offset instruction. The generated Q1ASM contains
only the ``set_awg_offs`` instruction and no ``upd_param`` instruction.
"""
@property
[docs]
def generates_path_Q_output(self) -> bool:
"""
If True, this strategy compiles to instructions that generate output on Q. If False, the
instructions either generate no output, or output only on path I.
"""
return (
self.operation_info.data["offset_path_Q"] is not None
and self.operation_info.data["offset_path_Q"] != 0
)
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Add the Q1ASM instruction for a DC voltage offset.
Parameters
----------
qasm_program : QASMProgram
The QASMProgram to add the assembly instructions to.
"""
qasm_program.set_offset_from_float_or_variable(
offset_path_I=self.operation_info.data["offset_path_I"],
offset_path_Q=self.operation_info.data["offset_path_Q"],
operation=self.operation_info,
)
[docs]
class UpdateParameterStrategy(IdleStrategy):
"""Strategy for compiling an "update parameters" real-time instruction."""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Add the ``upd_param`` assembly instruction for the Q1 sequence processor.
Parameters
----------
qasm_program : QASMProgram
The QASMProgram to add the assembly instructions to.
"""
qasm_program.emit(
q1asm_instructions.UPDATE_PARAMETERS,
constants.MIN_TIME_BETWEEN_OPERATIONS,
)
qasm_program.elapsed_time += constants.MIN_TIME_BETWEEN_OPERATIONS
[docs]
class LoopStrategy(IdleStrategy):
"""
Strategy for compiling a "Loop" control flow instruction.
Empty as it is used for isinstance.
"""
[docs]
class ConditionalStrategy(IdleStrategy):
"""Strategy for compiling a "Conditional" control flow instruction."""
def __init__(
self,
operation_info: types.OpInfo,
trigger_condition: FeedbackTriggerCondition,
) -> None:
super().__init__(operation_info=operation_info)
[docs]
self.trigger_condition = trigger_condition
[docs]
class ControlFlowReturnStrategy(IdleStrategy):
"""
Strategy for compiling "ControlFlowReturn" control flow instruction.
Empty as it is used for isinstance.
"""
[docs]
class TimestampStrategy(IdleStrategy):
"""
Strategy for compiling
:class:`~qblox_scheduler.operations.pulse_library.Timestamp`.
"""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Inserts the instructions needed insert a time reference.
Parameters
----------
qasm_program
The QASMProgram to add the assembly instructions to.
"""
assert "timestamp" in self.operation_info.data
qasm_program.emit(q1asm_instructions.SET_TIME_REF)
[docs]
_BINARY_OP_TO_Q1ASM: dict[str, str] = {
"+": q1asm_instructions.ADD,
"-": q1asm_instructions.SUB,
"&": q1asm_instructions.AND,
"|": q1asm_instructions.OR,
"^": q1asm_instructions.XOR,
"<<": q1asm_instructions.ARITHMETIC_SHIFT_LEFT,
">>": q1asm_instructions.ARITHMETIC_SHIFT_RIGHT,
}
[docs]
class AssignStrategy(IdleStrategy):
"""
Strategy for compiling an :class:`~qblox_scheduler.operations.statements.Assign`
statement into backend register instructions.
Complex expressions are compiled recursively: each sub-expression is evaluated
into a temporary register that is freed as soon as it is consumed, keeping
register pressure at O(expression depth) rather than O(expression size).
``*`` (constant times runtime value only) requires a target with
``QASMISACapabilities.multiplication`` (Q1 ISA version >= 2.0, cluster
firmware >= 1.1) and raises :exc:`NotImplementedError` otherwise; see
:meth:`._compile_multiplication` for the fixed-point details. ``/`` and
``//`` are always unsupported, since Q1ASM has no divide instruction.
"""
[docs]
def insert_qasm(self, qasm_program: QASMProgram) -> None:
"""
Add register related assembly instruction for the Q1 sequence processor.
Parameters
----------
qasm_program : QASMProgram
The QASMProgram to add the assembly instructions to.
"""
variable = self.operation_info.data["variable"]
value = self.operation_info.data["value"]
if isinstance(value, Expression):
value = value.reduce()
if isinstance(value, complex):
raise NotImplementedError(
f"Cannot assign complex value {value!r} to a Q1ASM register; complex "
"variables must be split into separate real/imaginary registers before "
"reaching Q1ASM compilation."
)
register_manager = qasm_program.register_manager
if not register_manager.has_variable(variable):
register_manager.allocate_register_for_variable(variable)
destination_register = register_manager.get_register_of_variable(variable)
self._compile_into(qasm_program, value, variable.dtype, destination_register)
[docs]
def _compile_into(
self,
qasm_program: QASMProgram,
expr: Expression | int | float,
dtype: DType,
destination_register: str,
) -> None:
"""Compile *expr* and store the result directly into *destination_register*."""
match expr:
case Variable():
source_register = qasm_program.register_manager.get_register_of_variable(expr)
if source_register != destination_register:
qasm_program.emit(
q1asm_instructions.MOVE,
source_register,
destination_register,
comment="assign",
)
case int() | float():
cast_fn = _SIGNED_INT_CASTING_FNS.get(dtype, int)
qasm_program.emit(
q1asm_instructions.MOVE,
twos_complement(cast_fn(expr)),
destination_register,
comment="assign",
)
case BinaryExpression(operator="*"):
self._compile_multiplication(qasm_program, expr, dtype, destination_register)
case BinaryExpression():
op = expr.operator
if op not in _BINARY_OP_TO_Q1ASM:
raise NotImplementedError(
f"Operator '{op}' is not supported in Q1ASM arithmetic expressions. "
"Q1ASM has no divide instruction."
)
# _get_operand doesn't accept complex expressions but at this moment
# complex expressions should be lowered to multiple real expressions
# by now.
assert not isinstance(expr.rhs, complex)
assert not isinstance(expr.lhs, complex)
with (
self._get_operand(qasm_program, expr.lhs, dtype) as lhs,
self._get_operand(qasm_program, expr.rhs, dtype) as rhs,
):
qasm_program.emit(
_BINARY_OP_TO_Q1ASM[op],
lhs,
rhs,
destination_register,
comment=f"expr {op}",
)
case UnaryExpression(operator="+"):
self._compile_into(qasm_program, expr.operand, dtype, destination_register)
case UnaryExpression():
with self._get_operand(qasm_program, expr.operand, dtype) as operand:
if isinstance(operand, int):
evaluator = UnaryExpression.EVALUATORS[expr.operator]
qasm_program.emit(
q1asm_instructions.MOVE,
twos_complement(int(evaluator(operand))),
destination_register,
comment="assign",
)
else:
match expr.operator:
case "-":
# Two's complement negation: 'sub' does not accept an
# immediate as its first operand, so 0 - x is not valid.
qasm_program.emit(
q1asm_instructions.NOT,
operand,
destination_register,
comment="negate (two's complement)",
)
qasm_program.emit(
q1asm_instructions.ADD,
destination_register,
1,
destination_register,
)
case "~":
qasm_program.emit(
q1asm_instructions.NOT,
operand,
destination_register,
comment="bitwise not",
)
case _:
raise NotImplementedError(
f"Unary operator '{expr.operator}' is not supported."
)
case _:
raise NotImplementedError(
f"Cannot compile expression of type {type(expr).__name__} to Q1ASM."
)
[docs]
def _compile_multiplication(
self,
qasm_program: QASMProgram,
expr: BinaryExpression,
dtype: DType,
dst_reg: str,
) -> None:
"""
Compile ``constant * expr`` (or ``expr * constant``) into *dst_reg*.
The runtime operand stays in its dtype's fixed-point representation; the
``constant`` is represented as ``c_fixed = round(constant * 2**k)``
which is a dimensionless scale factor, encoded here as ``k``-bit integer
with ``k`` as large as possible (at most 31). The full 64-bit product
``operand * c_fixed`` is assembled from ``muls32h`` (higher 32 bits) and
``muls32l`` (lower 32 bits) and then shifted right by ``k`` to restore
the constant's fixed-point scale (``2**k``).
"""
match (expr.lhs, expr.rhs):
case ((int() | float()), (int() | float())):
self._compile_into(qasm_program, expr.lhs * expr.rhs, dtype, dst_reg)
return
case ((int() | float()) as constant, (Expression() as operand_expr)) | (
(Expression() as operand_expr),
(int() | float()) as constant,
):
pass
case _:
raise NotImplementedError(
"Multiplication of two runtime values is not supported in Q1ASM "
"arithmetic expressions; one operand must be a constant."
)
# Fold unary minus into the constant: (-x) * c == x * (-c). This avoids a
# runtime negation of the operand.
while isinstance(operand_expr, UnaryExpression) and operand_expr.operator == "-":
constant = -constant
operand_expr = operand_expr.operand
if not qasm_program.isa_capabilities.multiplication:
raise NotImplementedError(
"Operator '*' is not supported in Q1ASM arithmetic expressions on "
"this target. Sweeping over an expression that requires a runtime "
"multiplication (e.g. a swept amplitude in a stitched "
"long_ramp_pulse or long_chirp_pulse) requires a cluster with "
"firmware >= 1.1 (Q1 ISA version >= 2.0). Configure "
"`SequencerOptions.qasm_isa_version = (2, 0)` in the hardware "
"configuration if your cluster's firmware supports it."
)
if constant == 0:
self._compile_into(qasm_program, 0, dtype, dst_reg)
return
for k in range(31, 0, -1):
c_fixed = round(constant * 2**k)
if -(1 << 31) <= c_fixed < (1 << 31):
break
else:
raise ValueError(
f"Constant factor {constant} is too large to encode in a Q1ASM multiplication."
)
high_reg = qasm_program.register_manager.allocate_register()
try:
with self._get_operand(qasm_program, operand_expr, dtype) as operand:
# Unlike 'move'/'add', the multiply instructions take a *signed*
# immediate, so no two's complement encoding here.
qasm_program.emit(
q1asm_instructions.MULTIPLY_SIGNED_HIGH,
operand,
c_fixed,
high_reg,
comment=f"expr * ({constant} as Q{k}), high word",
)
qasm_program.emit(
q1asm_instructions.MULTIPLY_SIGNED_LOW,
operand,
c_fixed,
dst_reg,
comment=f"expr * ({constant} as Q{k}), low word",
)
# The high and low words land on disjoint bit ranges of the result
# (bits [31:32-k] and [31-k:0] respectively), so combining them here
# 'add'.
qasm_program.emit(
q1asm_instructions.ARITHMETIC_SHIFT_LEFT,
high_reg,
32 - k,
high_reg,
comment="restore fixed-point scale (high word)",
)
qasm_program.emit(
q1asm_instructions.LOGICAL_SHIFT_RIGHT,
dst_reg,
k,
dst_reg,
comment="restore fixed-point scale (low word)",
)
qasm_program.emit(
q1asm_instructions.ADD,
dst_reg,
high_reg,
dst_reg,
comment="combine high/low words",
)
finally:
qasm_program.register_manager.free_register(high_reg)
@contextmanager
[docs]
def _get_operand(
self,
qasm_program: QASMProgram,
expr: Expression | int | float,
dtype: DType,
) -> Iterator[str | int]:
"""Yield a register or immediate for *expr*, freeing any temp register on exit."""
match expr:
case Variable():
yield qasm_program.register_manager.get_register_of_variable(expr)
case int() | float():
cast_fn = _SIGNED_INT_CASTING_FNS.get(dtype, int) # type: ignore[arg-type]
yield twos_complement(cast_fn(expr))
case BinaryExpression() | UnaryExpression():
reg = qasm_program.register_manager.allocate_register()
try:
self._compile_into(qasm_program, expr, dtype, reg)
yield reg
finally:
qasm_program.register_manager.free_register(reg)
case _:
raise NotImplementedError(
f"Cannot compile expression of type {type(expr).__name__} to Q1ASM."
)
