# 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.
"""Pulse compensation operations for use with the qblox_scheduler."""
from __future__ import annotations
import warnings
from typing import TYPE_CHECKING
import numpy as np
from qblox_scheduler.operations.operation import Operation
from qblox_scheduler.resources import BasebandClockResource
if TYPE_CHECKING:
from collections.abc import Iterable
from qblox_scheduler.schedule import Schedule
from qblox_scheduler.schedules.schedule import TimeableSchedule
"""Port on the hardware; this is an alias to str."""
[docs]
class PulseCompensation(Operation):
"""
Apply pulse compensation to an operation or schedule.
Inserts a pulse at the end of the operation or schedule set in ``body`` for each port.
The compensation pulses are calculated so that the integral of all pulses
(including the compensation pulses) are zero for each port.
Moreover, the compensating pulses are square pulses, and start just after the last
pulse on each port individually, and their maximum amplitude is the one
specified in the ``max_compensation_amp``. Their duration is divisible by ``duration_grid``.
The clock is assumed to be the baseband clock; any other clock is not allowed.
Parameters
----------
body
Operation to be pulse-compensated
qubits
For circuit-level operations, this is a list of device element names.
max_compensation_amp
Dictionary for each port the maximum allowed amplitude for the compensation pulse.
time_grid
Grid time of the duration of the compensation pulse.
sampling_rate
Sampling rate for pulse integration calculation.
min_duration
The minimum duration of the compensation pulse.
"""
def __init__(
self,
body: Operation | TimeableSchedule | Schedule,
qubits: str | Iterable[str] | None = None,
max_compensation_amp: dict[Port, float] | None = None,
time_grid: float | None = None,
sampling_rate: float | None = None,
min_duration: float | None = None,
) -> 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(
"PulseCompensation can not be defined over schedules "
"that contain non-realtime operations"
)
assert isinstance(timeable_schedule, TimeableSchedule)
body = timeable_schedule
device_elements = qubits
super().__init__(name="PulseCompensation")
if device_elements is not None:
if (
max_compensation_amp is not None
or time_grid is not None
or sampling_rate is not None
or min_duration is not None
):
raise ValueError(
"PulseCompensation can only be defined on gate-level or device-level, "
"but not both. If 'qubit' is defined, then 'max_compensation_amp', "
"'time_grid' and 'sampling_rate' must be 'None'."
)
if isinstance(device_elements, str):
device_elements = [device_elements]
self.data.update(
{
"pulse_compensation_info": {
"body": body,
"method": "software",
"device_elements": device_elements,
},
}
)
else:
self.data.update(
{
"pulse_compensation_info": {
"body": body,
"method": "software",
"max_compensation_amp": max_compensation_amp,
"time_grid": time_grid,
"sampling_rate": sampling_rate,
"min_duration": min_duration,
},
}
)
@property
[docs]
def body(self) -> Operation | TimeableSchedule:
"""Body of a pulse compensation."""
return self.data["pulse_compensation_info"]["body"]
@body.setter
def body(self, value: Operation | TimeableSchedule) -> None:
"""Body of a pulse compensation."""
self.data["pulse_compensation_info"]["body"] = value
@property
[docs]
def max_compensation_amp(self) -> dict[Port, float]:
"""For each port the maximum allowed amplitude for the compensation pulse."""
return self.data["pulse_compensation_info"]["max_compensation_amp"]
@property
[docs]
def time_grid(self) -> float:
"""Grid time of the duration of the compensation pulse."""
return self.data["pulse_compensation_info"]["time_grid"]
@property
[docs]
def sampling_rate(self) -> float:
"""Sampling rate for pulse integration calculation."""
return self.data["pulse_compensation_info"]["sampling_rate"]
@property
[docs]
def min_duration(self) -> float:
"""The minimum duration of the compensation pulse."""
min_duration = self.data["pulse_compensation_info"]["min_duration"]
# I don't know why, but this can be NaN sometimes instead of None.
if min_duration is None or np.isnan(min_duration):
ports = ", ".join(self.max_compensation_amp.keys())
warnings.warn(
"PulseCompensation has a new parameter, 'min_duration', which is undefined for "
f"the PulseCompensation scheduled on [{ports}]. Defaulting to 4e-9. "
"In the future this will result in an error",
DeprecationWarning,
)
min_duration = self.data["pulse_compensation_info"]["min_duration"] = 4e-9
return min_duration
[docs]
class NetZeroPulse(Operation):
"""
Play a pulse to remove any accumulated DC bias from the given ports.
As opposed to :class:`~PulseCompensation`, the bias correction is calculated by the hardware at
runtime, instead of by software at compile-time.
Parameters
----------
max_duration
The maximum duration of the pulse, in seconds.
ports
The ports to be compensated.
device_elements
For circuit-level operations, this is a list of device element names.
"""
def __init__(
self,
max_duration: float,
*,
ports: list[str] | None = None,
device_elements: str | Iterable[str] | None = None,
) -> None:
super().__init__(name="NetZeroPulse")
if ports is not None and isinstance(ports, str):
raise TypeError(f"Expected a sequence type but got str. {ports=}")
if device_elements is not None:
if ports is not None:
raise ValueError(
"NetZeroPulse can only be defined on gate-level or device-level, "
"but not both. If 'qubit' is defined, then 'ports' must be 'None'."
)
if isinstance(device_elements, str):
device_elements = [device_elements]
self.data.update(
{
"pulse_compensation_info": {
"method": "hardware",
"duration": max_duration,
"device_elements": device_elements,
},
}
)
else:
self.data.update(
{
"pulse_compensation_info": {
"method": "hardware",
"duration": max_duration,
"ports": ports,
},
}
)
[docs]
def get_used_port_clocks(self) -> set[tuple[str, str]]:
"""
Extracts which port-clock combinations are used in this operation.
Returns
-------
:
All (port, clock) combinations this operation uses.
"""
return {
(f"{port}", BasebandClockResource.IDENTITY)
for port in self["pulse_compensation_info"].get("ports", [])
}
@property
[docs]
def duration(self) -> float:
"""
Determine operation duration from pulse_info.
If the operation contains no pulse info, it is assumed to be ideal and
have zero duration.
"""
return self["pulse_compensation_info"]["duration"]
@property
[docs]
def valid_pulse(self) -> bool:
"""An operation is a valid pulse if it has pulse-level representation details."""
return len(self.data["pulse_compensation_info"]) > 0
@property
[docs]
def max_duration(self) -> float:
"""
The maximum duration of the pulse, in seconds.
The compiler treats this as the normal pulse duration.
"""
return self.data["pulse_compensation_info"]["duration"]
[docs]
class ResetNetZero(Operation):
"""
Reset the DC bias accumulator to zero without playing any compensation pulse.
Parameters
----------
ports
The ports to be compensated.
device_elements
For circuit-level operations, this is a list of device element names.
"""
def __init__(
self,
*,
ports: list[str] | None = None,
device_elements: str | Iterable[str] | None = None,
) -> None:
super().__init__(name="ResetNetZero")
if ports is not None and isinstance(ports, str):
raise TypeError(f"Expected a sequence type but got str. {ports=}")
if device_elements is not None:
if ports is not None:
raise ValueError(
"ResetNetZero can only be defined on gate-level or device-level, "
"but not both. If 'qubit' is defined, then 'ports' must be 'None'."
)
if isinstance(device_elements, str):
device_elements = [device_elements]
self.data.update(
{
"pulse_compensation_info": {
"method": "hardware",
"reset_netzero": True,
"device_elements": device_elements,
},
}
)
else:
self.data.update(
{
"pulse_compensation_info": {
"method": "hardware",
"reset_netzero": True,
"ports": ports,
},
}
)
[docs]
def get_used_port_clocks(self) -> set[tuple[str, str]]:
"""
Extracts which port-clock combinations are used in this operation.
Returns
-------
:
All (port, clock) combinations this operation uses.
"""
return {
(f"{port}", BasebandClockResource.IDENTITY)
for port in self["pulse_compensation_info"].get("ports", [])
}
@property
[docs]
def duration(self) -> float:
"""
Determine operation duration from pulse_info.
If the operation contains no pulse info, it is assumed to be ideal and
have zero duration.
"""
return 0
@property
[docs]
def valid_pulse(self) -> bool:
"""An operation is a valid pulse if it has pulse-level representation details."""
return False
