# SPDX-FileCopyrightText: Copyright 2020-2025, Quantify Consortium
# SPDX-FileCopyrightText: © 2023 Qblox <https://qblox.com>
# SPDX-License-Identifier: BSD-3-Clause
"""Standard control flow operations for use with the qblox_scheduler."""
from __future__ import annotations
from abc import ABCMeta, abstractmethod
from copy import copy
from typing import TYPE_CHECKING
from qblox_scheduler.backends.qblox import constants
from qblox_scheduler.enums import StrEnum
from qblox_scheduler.operations.operation import Operation
if TYPE_CHECKING:
from qblox_scheduler.operations.loop_domains import LinearDomain
from qblox_scheduler.operations.variables import Variable
from qblox_scheduler.schedule import Schedule
from qblox_scheduler.schedules.schedule import TimeableSchedule
class ControlFlowOperation(Operation, metaclass=ABCMeta):
"""
Control flow operation that can be used as an ``Operation`` in ``.TimeableSchedule``.
This is an abstract class. Each concrete implementation
of the control flow operation decides how and when
their ``body`` operation is executed.
"""
@property
@abstractmethod
def body(self) -> Operation | TimeableSchedule:
"""Body of a control flow."""
pass
@body.setter
@abstractmethod
def body(self, value: Operation | TimeableSchedule) -> None:
"""Body of a control flow."""
pass
def __str__(self) -> str:
"""
Represent the Operation as a string.
Returns
-------
str
description
"""
return self._get_signature(self.data["control_flow_info"])
def get_used_port_clocks(self) -> set[tuple[str, str]]:
"""
Extracts which port-clock combinations are used in this control flow operation.
Returns
-------
:
All (port, clock) combinations that operations
in the body of this control flow operation uses.
"""
return self.body.get_used_port_clocks()
[docs]
class LoopStrategy(StrEnum):
"""
Strategy to use for implementing loops.
REALTIME: Use native loops.
UNROLLED: Unroll loop at compilation time into separate instructions.
"""
REALTIME = "realtime"
UNROLLED = "unrolled"
class LoopOperation(ControlFlowOperation):
"""
Loop over another operation predefined times.
Repeats the operation defined in ``body`` ``repetitions`` times.
The actual implementation depends on the backend.
One of ``domain`` or ``repetitions`` must be specified.
Parameters
----------
body
Operation to be repeated
repetitions
Number of repetitions, by default None
domain
Linear domain to loop over, by default None
strategy
Strategy to use for implementing this loop, by default None to make own decision
is_schedule_repetition
Internal flag indicating that this loop was created from a schedule's
top-level ``repetitions``, by
default False. Used by visualization code (e.g. ``timing_table`` and
``plot_pulse_diagram``) to limit how many repetitions are expanded by default.
"""
def __init__(
self,
body: Operation | TimeableSchedule | Schedule,
*,
repetitions: int | None = None,
domain: dict[Variable, LinearDomain] | None = None,
strategy: LoopStrategy | None = None,
is_schedule_repetition: bool = False,
) -> None:
# Delayed to prevent circular imports
from qblox_scheduler.schedules.schedule import TimeableSchedule
if not isinstance(body, (Operation, TimeableSchedule)):
timeable_schedule = body._timeable_schedule
if timeable_schedule is None:
raise ValueError(
"LoopOperation can not be defined over schedules "
"that contain non-realtime operations"
)
assert isinstance(timeable_schedule, TimeableSchedule)
body = timeable_schedule
if (repetitions is None and domain is None) or (
repetitions is not None and domain is not None
):
raise ValueError("One of `repetitions` or `domain` must be specified.")
if domain is not None:
for var, dom in domain.items():
if var.dtype != dom.dtype:
raise ValueError(f"dtypes of variable ({var}) and domain ({dom}) do not match.")
nums = {d.num_steps for d in domain.values()}
if len(nums) > 1:
raise ValueError("Domains have different amount of steps.")
repetitions = nums.pop()
else:
domain = {}
super().__init__(name="LoopOperation")
self.data.update(
{
"control_flow_info": {
"body": body,
"domain": domain,
"repetitions": repetitions,
"strategy": strategy,
"is_schedule_repetition": is_schedule_repetition,
},
}
)
@property
def body(self) -> Operation | TimeableSchedule:
"""Body of a control flow."""
return self.data["control_flow_info"]["body"]
@body.setter
def body(self, value: Operation | TimeableSchedule) -> None:
"""Body of a control flow."""
self.data["control_flow_info"]["body"] = value
@property
def duration(self) -> float:
"""Duration of a control flow."""
return self.repetitions * self.data["control_flow_info"]["body"].duration
@property
def domain(self) -> dict[Variable, LinearDomain]:
"""Linear domain to loop over."""
return self.data["control_flow_info"]["domain"]
@property
def repetitions(self) -> int:
"""Number of times the body will execute."""
return self.data["control_flow_info"]["repetitions"]
@property
def strategy(self) -> LoopStrategy | None:
"""What strategy to use for implementing this loop."""
return self.data["control_flow_info"].get("strategy", None)
@property
def is_schedule_repetition(self) -> bool:
"""Whether this loop was synthesized from a schedule's top-level ``repetitions``."""
return self.data["control_flow_info"].get("is_schedule_repetition", False)
def __getstate__(self) -> dict[str, object]:
# We need to customize how the "domain" is serialized, because
# the domain contains variables as keys, and domains as values, which is not a valid JSON,
# because the key is not a string, it's an object.
# Instead, we serialize them as dicts `{"key" variable, "value": domain}`.
state = super().__getstate__()
state = copy(state)
state["data"] = copy(state["data"])
state["data"]["control_flow_info"] = copy(state["data"]["control_flow_info"]) # type: ignore[reportIndexIssue]
state["data"]["control_flow_info"]["domain"] = copy( # type: ignore[reportIndexIssue]
state["data"]["control_flow_info"]["domain"] # type: ignore[reportIndexIssue]
)
if state["data"]["control_flow_info"]["domain"] is not None: # type: ignore[reportAttributeAccessIssue]
state["data"]["control_flow_info"]["domain"] = [ # type: ignore[reportAttributeAccessIssue]
{"key": variable, "value": domain}
for variable, domain in state["data"]["control_flow_info"]["domain"].items() # type: ignore[reportAttributeAccessIssue]
]
return state
def __setstate__(self, state: dict[str, dict]) -> None:
# See comment in serialization to understand why it's a custom deserialization.
super().__setstate__(state)
if state["data"]["control_flow_info"]["domain"] is not None:
self.data["control_flow_info"]["domain"] = {
domain["key"]: domain["value"]
for domain in state["data"]["control_flow_info"]["domain"]
}
self.update()
[docs]
class ConditionalOperation(ControlFlowOperation):
"""
Conditional over another operation.
If a preceding thresholded acquisition on ``feedback_trigger_label`` results in a
"1", the body will be executed, otherwise it will generate a wait time that is
equal to the time of the subschedule, to ensure the absolute timing of later
operations remains consistent.
Parameters
----------
body
Operation to be conditionally played
feedback_trigger_label
Name of the device element on which the body will be conditioned
hardware_buffer_time
Time buffer, by default the minimum time between operations on the hardware
Example
-------
A conditional reset can be implemented as follows:
.. jupyter-execute::
# relevant imports
from qblox_scheduler import Schedule
from qblox_scheduler import ConditionalOperation, Measure, X
# define conditional reset as a Schedule
conditional_reset = Schedule("conditional reset")
conditional_reset.add(Measure("q0", feedback_trigger_label="q0"))
conditional_reset.add(
ConditionalOperation(body=X("q0"), feedback_trigger_label="q0"),
rel_time=364e-9,
)
.. versionadded:: 0.22.0
For some hardware specific implementations, a ``hardware_buffer_time``
might be required to ensure the correct timing of the operations. This will
be added to the duration of the ``body`` to prevent overlap with other
operations.
"""
def __init__(
self,
body: Operation | TimeableSchedule | Schedule,
feedback_trigger_label: str | None = None,
hardware_buffer_time: float = constants.MIN_TIME_BETWEEN_OPERATIONS * 1e-9,
) -> None:
if feedback_trigger_label is None:
raise TypeError(
"ConditionalOperation.__init__() missing 1 required argument: "
"'feedback_trigger_label'"
)
# Delayed to prevent circular imports
from qblox_scheduler.schedules.schedule import TimeableSchedule
if not isinstance(body, (Operation, TimeableSchedule)):
timeable_schedule = body._timeable_schedule
if timeable_schedule is None:
raise ValueError(
"ConditionalOperation can not be defined over schedules "
"that contain non-realtime operations"
)
assert isinstance(timeable_schedule, TimeableSchedule)
body = timeable_schedule
device_element_name = feedback_trigger_label
super().__init__(name="ConditionalOperation")
self.data.update(
{
"control_flow_info": {
"body": body,
"feedback_trigger_label": device_element_name,
"feedback_trigger_address": None, # Filled in at compilation.
"hardware_buffer_time": hardware_buffer_time,
},
}
)
@property
def body(self) -> Operation | TimeableSchedule:
"""Body of a control flow."""
return self.data["control_flow_info"]["body"]
@body.setter
def body(self, value: Operation | TimeableSchedule) -> None:
"""Body of a control flow."""
self.data["control_flow_info"]["body"] = value
@property
def duration(self) -> float:
"""Duration of a control flow."""
return (
self.data["control_flow_info"]["body"].duration
+ self.data["control_flow_info"]["hardware_buffer_time"]
)
class ControlFlowSpec(metaclass=ABCMeta):
"""
Control flow specification to be used at ``Schedule.add``.
The users can specify any concrete control flow with
the ``control_flow`` argument to ``Schedule.add``.
The ``ControlFlowSpec`` is only a type which by itself
cannot be used for the ``control_flow`` argument,
use any concrete control flow derived from it.
"""
@abstractmethod
def create_operation(self, body: Operation | TimeableSchedule) -> Operation | TimeableSchedule:
"""Transform the control flow specification to an operation or schedule."""
pass
class Loop(ControlFlowSpec):
"""
Loop control flow specification to be used at ``Schedule.add``.
For more information, see ``LoopOperation``.
Parameters
----------
repetitions
Number of repetitions
strategy
Strategy to use for implementing the loop, by default None
"""
def __init__(self, repetitions: int, strategy: LoopStrategy | None = None) -> None:
self.repetitions = repetitions
self.strategy = strategy
def create_operation(self, body: Operation | TimeableSchedule) -> LoopOperation:
"""Transform the control flow specification to an operation or schedule."""
return LoopOperation(body, repetitions=self.repetitions, strategy=self.strategy)
class Conditional(ControlFlowSpec):
"""
Conditional control flow specification to be used at ``Schedule.add``.
For more information, see ``ConditionalOperation``.
Parameters
----------
feedback_trigger_label
Target device element.
"""
def __init__(
self,
feedback_trigger_label: str | None = None,
) -> None:
if feedback_trigger_label is None:
raise TypeError(
"Conditional.__init__() missing 1 required argument: 'feedback_trigger_label'"
)
self.device_element_name = feedback_trigger_label
def create_operation(self, body: Operation | TimeableSchedule) -> ConditionalOperation:
"""Transform the control flow specification to an operation or schedule."""
return ConditionalOperation(body, self.device_element_name)
