# Repository: https://gitlab.com/quantify-os/quantify-scheduler
# Licensed according to the LICENCE file on the main branch
"""Utilty classes for Qblox analog modules."""
from __future__ import annotations
import logging
import math
from abc import ABC, abstractmethod
from copy import deepcopy
from typing import TYPE_CHECKING, Any, Iterator
from quantify_scheduler.backends.qblox import (
constants,
helpers,
instrument_compilers,
q1asm_instructions,
)
from quantify_scheduler.backends.qblox.compiler_abc import (
ClusterModuleCompiler,
SequencerCompiler,
)
from quantify_scheduler.backends.qblox.enums import (
ChannelMode,
DistortionCorrectionLatencyEnum,
QbloxFilterConfig,
QbloxFilterMarkerDelay,
)
from quantify_scheduler.backends.qblox.exceptions import NcoOperationTimingError
from quantify_scheduler.backends.qblox.operation_handling.acquisitions import (
SquareAcquisitionStrategy,
)
from quantify_scheduler.backends.qblox.operation_handling.factory_analog import (
get_operation_strategy,
)
from quantify_scheduler.backends.qblox.operation_handling.virtual import (
NcoPhaseShiftStrategy,
NcoResetClockPhaseStrategy,
NcoSetClockFrequencyStrategy,
)
from quantify_scheduler.backends.qblox.qasm_program import get_marker_binary
from quantify_scheduler.backends.types.qblox import (
AnalogModuleSettings,
AnalogSequencerSettings,
BasebandModuleSettings,
ComplexChannelDescription,
ComplexInputGain,
DigitalChannelDescription,
InputAttenuation,
OpInfo,
OutputAttenuation,
RealInputGain,
RFModuleSettings,
StaticAnalogModuleProperties,
)
from quantify_scheduler.enums import BinMode
from quantify_scheduler.operations.pulse_library import SetClockFrequency
if TYPE_CHECKING:
from quantify_scheduler.backends.qblox.operation_handling.base import (
IOperationStrategy,
)
from quantify_scheduler.backends.qblox.qasm_program import QASMProgram
from quantify_scheduler.backends.qblox_backend import (
_ClusterModuleCompilationConfig,
_SequencerCompilationConfig,
)
from quantify_scheduler.schedules.schedule import AcquisitionMetadata
[docs]
logger = logging.getLogger(__name__)
logger.setLevel(logging.WARNING)
[docs]
class AnalogSequencerCompiler(SequencerCompiler):
"""
Class that performs the compilation steps on the sequencer level, for QCM and
QRM-type modules.
Parameters
----------
parent
A reference to the module compiler this sequencer belongs to.
index
Index of the sequencer.
static_hw_properties
The static properties of the hardware. This effectively gathers all the
differences between the different modules.
settings
The settings set to this sequencer.
sequencer_cfg
The instrument compiler config associated to this device.
"""
def __init__(
self,
parent: AnalogModuleCompiler,
index: int,
static_hw_properties: StaticAnalogModuleProperties,
settings: AnalogSequencerSettings,
sequencer_cfg: _SequencerCompilationConfig,
) -> None:
[docs]
self.static_hw_properties: StaticAnalogModuleProperties # help type checker
[docs]
self._settings: AnalogSequencerSettings # help type checker
super().__init__(
parent=parent,
index=index,
static_hw_properties=static_hw_properties,
settings=settings,
sequencer_cfg=sequencer_cfg,
)
[docs]
self.associated_ext_lo = sequencer_cfg.lo_name
[docs]
self.downconverter_freq = (
sequencer_cfg.hardware_description.downconverter_freq
if isinstance(sequencer_cfg.hardware_description, ComplexChannelDescription)
else None
)
[docs]
self.mix_lo = (
sequencer_cfg.hardware_description.mix_lo
if not isinstance(
sequencer_cfg.hardware_description, DigitalChannelDescription
)
else None
)
[docs]
self._marker_debug_mode_enable = (
(sequencer_cfg.hardware_description.marker_debug_mode_enable)
if not isinstance(
sequencer_cfg.hardware_description, DigitalChannelDescription
)
else None
)
[docs]
self._default_marker = (
self.static_hw_properties.channel_name_to_digital_marker.get(
self._settings.channel_name, self.static_hw_properties.default_marker
)
)
@property
[docs]
def frequency(self) -> float | None:
"""
The frequency used for modulation of the pulses.
Returns
-------
:
The frequency.
"""
return self._settings.modulation_freq
@frequency.setter
def frequency(self, freq: float) -> None:
"""
Assigns a modulation frequency to the sequencer.
Parameters
----------
freq
The frequency to be used for modulation.
Raises
------
ValueError
Attempting to set the modulation frequency to a new value even though a
different value has been previously assigned.
"""
if (
self._settings.modulation_freq is not None
and not math.isnan(self._settings.modulation_freq)
and not math.isclose(self._settings.modulation_freq, freq)
):
raise ValueError(
f"Attempting to set the modulation frequency of '{self.name}' of "
f"'{self.parent.name}' to {freq:e}, while it has previously been set "
f"to {self._settings.modulation_freq:e}."
)
self._settings.modulation_freq = freq
self._settings.nco_en = freq is not None
[docs]
def get_operation_strategy(
self,
operation_info: OpInfo,
) -> IOperationStrategy:
"""
Determines and instantiates the correct strategy object.
Parameters
----------
operation_info
The operation we are building the strategy for.
channel_name
Specifies the channel identifier of the hardware config (e.g. `complex_output_0`).
Returns
-------
:
The instantiated strategy object.
"""
return get_operation_strategy(operation_info, self.settings.channel_name)
[docs]
def add_operation_strategy(self, op_strategy: IOperationStrategy) -> None:
"""
Adds the operation strategy to the sequencer compiler.
Parameters
----------
op_strategy
The operation strategy.
"""
if op_strategy.operation_info.data.get("marker_pulse", False):
# A digital pulse always uses one output.
op_strategy.operation_info.data["output"] = self.connected_output_indices[0]
op_strategy.operation_info.data["default_marker"] = (
self.static_hw_properties.default_marker
)
super().add_operation_strategy(op_strategy)
[docs]
def _prepare_acq_settings(
self,
acquisitions: list[IOperationStrategy],
acq_metadata: AcquisitionMetadata,
) -> None:
"""
Sets sequencer settings that are specific to certain acquisitions.
For example for a TTL acquisition strategy.
Parameters
----------
acquisitions
List of the acquisitions assigned to this sequencer.
acq_metadata
Acquisition metadata.
"""
def _verify_param_range(
param_name: str,
val: float | None,
min_value: float,
max_value: float,
portclock: tuple[str, str],
) -> None:
if val is None:
return
if val < min_value or val > max_value:
raise ValueError(
f"Attempting to configure {param_name} to {val} for the sequencer "
f"specified with portclock '{portclock}' and while the "
f"hardware requires it to be between {min_value} and {max_value}."
)
acquisition_infos: list[OpInfo] = list(
map(lambda acq: acq.operation_info, acquisitions)
)
if acq_metadata.acq_protocol == "TriggerCount":
self._settings.ttl_acq_auto_bin_incr_en = (
acq_metadata.bin_mode == BinMode.AVERAGE
)
if len(self.connected_input_indices) == 1:
self._settings.ttl_acq_input_select = self.connected_input_indices[0]
elif len(self.connected_input_indices) > 1:
raise ValueError(
f"Please make sure you use a single real input for this "
f"portclock combination. "
f"Found: {len(self.connected_input_indices)} connected. "
f"TTL acquisition does not support multiple inputs."
f"Problem occurred for port {self.port} with"
f"clock {self.clock}, which corresponds to {self.name} of "
f"{self.parent.name}."
)
elif acq_metadata.acq_protocol == "ThresholdedAcquisition":
acq_rotation = acquisition_infos[0].data.get("acq_rotation")
_verify_param_range(
param_name="acq_rotation",
val=acq_rotation,
min_value=constants.MIN_PHASE_ROTATION_ACQ,
max_value=constants.MAX_PHASE_ROTATION_ACQ,
portclock=self.portclock,
)
self._settings.thresholded_acq_rotation = acq_rotation
acq_threshold = acquisition_infos[0].data.get("acq_threshold", 0.0)
_verify_param_range(
param_name="acq_threshold",
val=acq_threshold,
min_value=constants.MIN_DISCRETIZATION_THRESHOLD_ACQ,
max_value=constants.MAX_DISCRETIZATION_THRESHOLD_ACQ,
portclock=self.portclock,
)
integration_length = acquisition_infos[0].data.get("duration", 0.0) * 1e9
self._settings.thresholded_acq_threshold = (
acq_threshold * integration_length
)
for info in acquisition_infos:
if (address := info.data.get("feedback_trigger_address")) is not None:
self._settings.thresholded_acq_trigger_en = True
self._settings.thresholded_acq_trigger_address = address
self._settings.integration_length_acq = (
self._get_integration_length_from_acquisitions()
)
[docs]
def _get_integration_length_from_acquisitions(self) -> int | None:
"""
Get the (validated) integration_length sequencer setting.
Get the duration of all SSB integration acquisitions assigned to this sequencer
and validate that they are all the same.
"""
integration_length = None
for op_strat in self.op_strategies:
if not isinstance(op_strat, SquareAcquisitionStrategy):
continue
acq_duration_ns = round(op_strat.operation_info.duration * 1e9)
if acq_duration_ns % constants.GRID_TIME != 0:
raise ValueError(
"Attempting to perform square acquisition with a duration of "
f"{acq_duration_ns} ns. Please ensure the duration is a multiple "
f"of {constants.GRID_TIME} ns.\n\nException caused by "
f"{repr(op_strat)}."
)
if integration_length is None:
integration_length = acq_duration_ns
elif integration_length != acq_duration_ns:
raise ValueError(
f"Attempting to set an integration_length of {acq_duration_ns} "
f"ns, while this was previously determined to be "
f"{integration_length}. Please check whether all square "
"acquisitions in the schedule have the same duration."
)
return integration_length
[docs]
def prepare(self) -> None:
"""
Perform necessary operations on this sequencer's data before
:meth:`~quantify_scheduler.backends.qblox.compiler_abc.SequencerCompiler.compile`
is called.
"""
self._update_set_clock_frequency_operations()
if not self._settings.allow_off_grid_nco_ops:
self._assert_nco_operation_timing_on_grid(self._get_ordered_operations())
super().prepare()
[docs]
def _update_set_clock_frequency_operations(self) -> None:
for op_strat in self.op_strategies:
if op_strat.operation_info.name == SetClockFrequency.__name__:
op_strat.operation_info.data.update(
{
"interm_freq_old": self.frequency,
}
)
[docs]
def _get_latency_correction_ns(self, latency_correction: float) -> int:
# NCO grid alignment for NCO operations, _without_ latency corrections, is
# already checked in `_assert_nco_operation_timing_on_grid`. Therefore here we
# only check the latency corrections. Method overridden from superclass because
# only QRM/QCM modules have NCO operations.
latency_correction = super()._get_latency_correction_ns(latency_correction)
if not self._settings.allow_off_grid_nco_ops:
try:
helpers.to_grid_time(latency_correction * 1e-9, constants.NCO_TIME_GRID)
except ValueError as e:
raise NcoOperationTimingError(
f"The latency correction value of {latency_correction} ns for "
f"{self.port}-{self.clock} does not align with the grid time of "
f"{constants.NCO_TIME_GRID} ns for NCO operations. The latency "
"corrections must adhere to this grid time to ensure proper "
"alignment of all later operations in the schedule."
) from e
return latency_correction
@staticmethod
[docs]
def _assert_nco_operation_timing_on_grid(
ordered_op_strategies: list[IOperationStrategy],
) -> None:
"""Check whether this sequencer's operation adhere to NCO timing restrictions."""
last_freq_upd_time = -constants.NCO_SET_FREQ_WAIT
last_phase_upd_time = -constants.NCO_SET_PH_DELTA_WAIT
for op in ordered_op_strategies:
timing = round(op.operation_info.timing * 1e9)
if isinstance(op, NcoSetClockFrequencyStrategy):
if (diff := timing - last_freq_upd_time) < constants.NCO_SET_FREQ_WAIT:
raise NcoOperationTimingError(
f"Operation {op.operation_info} occurred {diff} ns after the "
"previous frequency update. The minimum time between frequency "
f"updates must be {constants.NCO_SET_FREQ_WAIT} ns."
)
else:
last_freq_upd_time = timing
if isinstance(op, (NcoPhaseShiftStrategy, NcoResetClockPhaseStrategy)):
timing = round(op.operation_info.timing * 1e9)
if (
diff := timing - last_phase_upd_time
) < constants.NCO_SET_PH_DELTA_WAIT:
raise NcoOperationTimingError(
f"Operation {op.operation_info} occurred {diff} ns after the "
"previous phase update. The minimum time between phase "
f"updates must be {constants.NCO_SET_PH_DELTA_WAIT} ns."
)
else:
last_phase_upd_time = timing
if isinstance(
op,
(
NcoSetClockFrequencyStrategy,
NcoPhaseShiftStrategy,
NcoResetClockPhaseStrategy,
),
):
try:
helpers.to_grid_time(timing * 1e-9, constants.NCO_TIME_GRID)
except ValueError as e:
raise NcoOperationTimingError(
f"NCO related operation {op.operation_info} must be on "
f"{constants.NCO_TIME_GRID} ns time grid"
) from e
[docs]
def _assert_total_play_time_on_nco_grid(self) -> None:
if self._settings.allow_off_grid_nco_ops:
return
try:
helpers.to_grid_time(self.parent.total_play_time, constants.NCO_TIME_GRID)
except ValueError as e:
raise NcoOperationTimingError(
"The schedule is repeated with a duration of "
f"{round(self.parent.total_play_time * 1e9)} ns per iteration, "
"which does not align with the grid time of "
f"{constants.NCO_TIME_GRID} ns for NCO operations. The duration "
"must adhere to this grid time to ensure proper alignment of NCO "
"operations for each iteration."
) from e
[docs]
def _write_pre_wait_sync_instructions(self, qasm: QASMProgram) -> None:
"""
Write instructions to the QASM program that must come before the first wait_sync.
The duration must be equal for all module types.
"""
qasm.emit(
q1asm_instructions.SET_MARKER,
get_marker_binary(self._default_marker),
comment=f"set markers to {self._default_marker}",
)
[docs]
def _insert_qasm(
self, op_strategy: IOperationStrategy, qasm_program: QASMProgram
) -> None:
"""
Get Q1ASM instruction(s) from ``op_strategy`` and insert them into ``qasm_program``.
Optionally wrap pulses and acquisitions in marker pulses depending on the
``marker_debug_mode_enable`` setting.
"""
if self._marker_debug_mode_enable and (
op_strategy.operation_info.is_acquisition
or op_strategy.operation_info.data.get("wf_func") is not None
):
marker = self._decide_markers(op_strategy)
qasm_program.emit(
q1asm_instructions.SET_MARKER,
get_marker_binary(marker),
comment=f"set markers to {marker}",
)
op_strategy.insert_qasm(qasm_program)
qasm_program.emit(
q1asm_instructions.SET_MARKER,
get_marker_binary(self._default_marker),
comment=f"set markers to {self._default_marker}",
)
qasm_program.emit(
q1asm_instructions.UPDATE_PARAMETERS,
constants.MIN_TIME_BETWEEN_OPERATIONS,
)
qasm_program.elapsed_time += constants.MIN_TIME_BETWEEN_OPERATIONS
else:
op_strategy.insert_qasm(qasm_program)
[docs]
def _decide_markers(self, operation: IOperationStrategy) -> int:
"""
Helper method to decide what markers should be pulled high when enable_marker is
set to True. Checks what module and operation are being processed, then builds
a bit string accordingly.
Note that with the current quantify structure a sequencer cannot have connected
inputs and outputs simultaneously. Therefore, the QRM baseband module pulls
both input or output markers high when doing an operation, as it is impossible
during compile time to find out what physical port is being used.
Parameters
----------
operation
The operation currently being processed by the sequence.
Returns
-------
A bit string passed on to the set_mrk function of the Q1ASM object.
"""
marker_bit_string = 0
instrument_type = self.static_hw_properties.instrument_type
if instrument_type == "QCM":
for output in self.connected_output_indices:
marker_bit_string |= 1 << output
elif instrument_type == "QRM":
marker_bit_string = (
0b1100 if operation.operation_info.is_acquisition else 0b0011
)
# For RF modules, the first two indices correspond to path enable/disable.
# Therefore, the index of the output is shifted by 2.
elif instrument_type == "QCM_RF":
# connected outputs are either 0,1 or 2,3 corresponding to
# marker bitstrings 0b0100 and 0b1000 respectively
output = self.connected_output_indices[0] // 2
marker_bit_string |= 1 << (output + 2)
marker_bit_string |= self._default_marker
elif instrument_type == "QRM_RF":
marker_bit_string = (
0b1011 if operation.operation_info.is_acquisition else 0b0111
)
return marker_bit_string
[docs]
class AnalogModuleCompiler(ClusterModuleCompiler, ABC):
"""
Base class for QCM and QRM-type modules.
This class is defined as an abstract base class since the distinctions between the
different devices are defined in subclasses.
Effectively, this base class contains the functionality shared by all Qblox
devices and serves to avoid repeated code between them.
Parameters
----------
parent: :class:`quantify_scheduler.backends.qblox.instrument_compilers.ClusterCompiler`
Reference to the parent object.
name
Name of the `QCoDeS` instrument this compiler object corresponds to.
total_play_time
Total time execution of the schedule should go on for. This parameter is
used to ensure that the different devices, potentially with different clock
rates, can work in a synchronized way when performing multiple executions of
the schedule.
instrument_cfg
The instrument compiler config referring to this device.
"""
[docs]
_settings: AnalogModuleSettings # type: ignore
def __init__(
self,
parent: instrument_compilers.ClusterCompiler,
name: str,
total_play_time: float,
instrument_cfg: _ClusterModuleCompilationConfig,
) -> None:
super().__init__(
parent=parent,
name=name,
total_play_time=total_play_time,
instrument_cfg=instrument_cfg,
)
[docs]
self.sequencers: dict[str, AnalogSequencerCompiler] = {}
@property
@abstractmethod
[docs]
def static_hw_properties(self) -> StaticAnalogModuleProperties:
"""
The static properties of the hardware. This effectively gathers all the
differences between the different modules.
"""
[docs]
def _construct_sequencer_compiler(
self, index: int, sequencer_cfg: _SequencerCompilationConfig
) -> AnalogSequencerCompiler:
"""Create an instance of :class:`AnalogSequencerCompiler`."""
settings = AnalogSequencerSettings.initialize_from_config_dict(
sequencer_cfg=sequencer_cfg,
channel_name=sequencer_cfg.channel_name,
connected_output_indices=self.static_hw_properties._get_connected_output_indices(
sequencer_cfg.channel_name
),
connected_input_indices=self.static_hw_properties._get_connected_input_indices(
sequencer_cfg.channel_name
),
)
return AnalogSequencerCompiler(
parent=self,
index=index,
static_hw_properties=self.static_hw_properties,
settings=settings,
sequencer_cfg=sequencer_cfg,
)
@abstractmethod
[docs]
def assign_frequencies(self, sequencer: AnalogSequencerCompiler) -> None:
"""
An abstract method that should be overridden. Meant to assign an IF frequency
to each sequencer, and an LO frequency to each output (if applicable).
"""
[docs]
def _set_lo_interm_freqs(
self,
freqs: helpers.Frequencies,
sequencer: AnalogSequencerCompiler,
compiler_lo_baseband: (
instrument_compilers.LocalOscillatorCompiler | None
) = None,
lo_freq_setting_rf: str | None = None,
) -> None:
"""
Sets the LO/IF frequencies, for baseband and RF modules.
Parameters
----------
freqs
LO, IF, and clock frequencies, supplied via an :class:`.helpers.Frequencies`
object.
sequencer
The sequencer for which frequences are to be set.
compiler_lo_baseband
For baseband modules, supply the
:class:`.instrument_compilers.LocalOscillatorCompiler` instrument compiler
of which the frequency is to be set.
lo_freq_setting_rf
For RF modules, supply the name of the LO frequency param from the
:class:`.RFModuleSettings` that is to be set.
Raises
------
ValueError
In case neither LO frequency nor IF has been supplied.
ValueError
In case both LO frequency and IF have been supplied and do not adhere to
:math:`f_{RF} = f_{LO} + f_{IF}`.
ValueError
In case of RF, when the LO frequency was already set to a different value.
"""
if freqs.LO is not None:
if compiler_lo_baseband is not None:
compiler_lo_baseband.frequency = freqs.LO
elif lo_freq_setting_rf is not None:
previous_lo_freq = getattr(self._settings, lo_freq_setting_rf)
if (
previous_lo_freq is not None
and not math.isnan(previous_lo_freq)
and not math.isclose(freqs.LO, previous_lo_freq)
):
raise ValueError(
f"Attempting to set '{lo_freq_setting_rf}' to frequency "
f"'{freqs.LO:e}', while it has previously already been set to "
f"'{previous_lo_freq:e}'!"
)
setattr(self._settings, lo_freq_setting_rf, freqs.LO)
if freqs.IF is not None:
sequencer.frequency = freqs.IF
@abstractmethod
[docs]
def assign_attenuation(self) -> None:
"""
An abstract method that should be overridden. Meant to assign
attenuation settings from the hardware configuration if there is any.
"""
[docs]
def prepare(self) -> None:
"""
Performs the logic needed before being able to start the compilation. In effect,
this means assigning the pulses and acquisitions to the sequencers and
calculating the relevant frequencies in case an external local oscillator is
used.
"""
self._configure_input_gains()
self._configure_mixer_offsets()
self._configure_hardware_distortion_corrections()
self._construct_all_sequencer_compilers()
for seq in self.sequencers.values():
self.assign_frequencies(seq)
self.distribute_data()
for seq in self.sequencers.values():
seq.prepare()
self._ensure_single_scope_mode_acquisition_sequencer()
self.assign_attenuation()
[docs]
def _ensure_single_scope_mode_acquisition_sequencer(self) -> None:
"""
Raises an error if multiple sequencers use scope mode acquisition,
because that's not supported by the hardware.
Also, see
:func:`~quantify_scheduler.instrument_coordinator.components.qblox._QRMComponent._determine_scope_mode_acquisition_sequencer_and_qblox_acq_index`
which also ensures the program that gets uploaded to the hardware satisfies this
requirement.
Raises
------
ValueError
Multiple sequencers have to perform trace acquisition. This is not
supported by the hardware.
""" # noqa: E501 line too long
def is_scope_acquisition(acquisition: OpInfo) -> bool:
return acquisition.data["protocol"] == "Trace"
scope_acq_seq = None
for seq in self.sequencers.values():
op_infos = [
op.operation_info
for op in seq.op_strategies
if op.operation_info.is_acquisition
]
has_scope = any(map(is_scope_acquisition, op_infos))
if has_scope:
if scope_acq_seq is not None:
helpers.single_scope_mode_acquisition_raise(
sequencer_0=scope_acq_seq,
sequencer_1=seq.index,
module_name=self.name,
)
scope_acq_seq = seq.index
[docs]
class BasebandModuleCompiler(AnalogModuleCompiler):
"""
Abstract class with all the shared functionality between the QRM and QCM baseband
modules.
"""
def __init__(
self,
parent: instrument_compilers.ClusterCompiler,
name: str,
total_play_time: float,
instrument_cfg: _ClusterModuleCompilationConfig,
) -> None:
super().__init__(
parent=parent,
name=name,
total_play_time=total_play_time,
instrument_cfg=instrument_cfg,
)
[docs]
self._settings: BasebandModuleSettings = ( # type: ignore
BasebandModuleSettings.extract_settings_from_mapping(instrument_cfg)
)
[docs]
def assign_frequencies(self, sequencer: AnalogSequencerCompiler) -> None:
"""
Determines LO/IF frequencies and assigns them, for baseband modules.
In case of **no** external local oscillator, the NCO is given the same
frequency as the clock -- unless NCO was permanently disabled via
`"interm_freq": 0` in the hardware config.
In case of **an** external local oscillator and `sequencer.mix_lo` is
``False``, the LO is given the same frequency as the clock
(via :func:`.helpers.determine_clock_lo_interm_freqs`).
"""
compiler_container = self.parent.parent
if (
sequencer.clock not in compiler_container.resources
or ChannelMode.DIGITAL in sequencer.settings.channel_name
):
return
clock_freq = compiler_container.resources[sequencer.clock]["freq"]
if sequencer.associated_ext_lo is None:
# Set NCO frequency to the clock frequency, unless NCO was permanently
# disabled via `"interm_freq": 0` in the hardware config
if sequencer.frequency != 0:
sequencer.frequency = clock_freq
else:
# In using external local oscillator, determine clock and LO/IF freqs,
# and then set LO/IF freqs, and enable NCO (via setter)
compiler_lo = compiler_container.instrument_compilers.get(
sequencer.associated_ext_lo
)
try:
freqs = helpers.determine_clock_lo_interm_freqs(
freqs=helpers.Frequencies(
clock=clock_freq,
LO=compiler_lo.frequency,
IF=sequencer.frequency,
),
downconverter_freq=sequencer.downconverter_freq,
mix_lo=sequencer.mix_lo,
)
except Exception as error: # Adding sequencer info to exception message
raise error.__class__(
f"{error} (for '{sequencer.name}' of '{self.name}' "
f"with port '{sequencer.port}' and clock '{sequencer.clock}')"
)
self._set_lo_interm_freqs(
freqs=freqs, sequencer=sequencer, compiler_lo_baseband=compiler_lo
)
[docs]
def assign_attenuation(self) -> None:
"""
Meant to assign attenuation settings from the hardware configuration, if there
is any. For baseband modules there is no attenuation parameters currently.
"""
[docs]
class RFModuleCompiler(AnalogModuleCompiler):
"""
Abstract class with all the shared functionality between the QRM-RF and QCM-RF
modules.
"""
def __init__(
self,
parent: instrument_compilers.ClusterCompiler,
name: str,
total_play_time: float,
instrument_cfg: _ClusterModuleCompilationConfig,
) -> None:
super().__init__(
parent=parent,
name=name,
total_play_time=total_play_time,
instrument_cfg=instrument_cfg,
)
[docs]
self._settings: RFModuleSettings = ( # type: ignore
RFModuleSettings.extract_settings_from_mapping(instrument_cfg)
)
[docs]
def assign_frequencies(self, sequencer: AnalogSequencerCompiler) -> None:
"""Determines LO/IF frequencies and assigns them for RF modules."""
compiler_container = self.parent.parent
if (
not sequencer.connected_output_indices
or sequencer.clock not in compiler_container.resources
or ChannelMode.DIGITAL in sequencer.settings.channel_name
):
return
for lo_idx in RFModuleCompiler._get_connected_lo_indices(sequencer):
lo_freq_setting_name = f"lo{lo_idx}_freq"
try:
freqs = helpers.determine_clock_lo_interm_freqs(
freqs=helpers.Frequencies(
clock=compiler_container.resources[sequencer.clock]["freq"],
LO=getattr(self._settings, lo_freq_setting_name),
IF=sequencer.frequency,
),
downconverter_freq=sequencer.downconverter_freq,
mix_lo=True,
)
except Exception as error: # Adding sequencer info to exception message
raise error.__class__(
f"{error} (for '{sequencer.name}' of '{self.name}' "
f"with port '{sequencer.port}' and clock '{sequencer.clock}')"
)
self._set_lo_interm_freqs(
freqs=freqs,
sequencer=sequencer,
lo_freq_setting_rf=lo_freq_setting_name,
)
@staticmethod
[docs]
def _get_connected_lo_indices(sequencer: AnalogSequencerCompiler) -> Iterator[int]:
"""
Identify the LO the sequencer is outputting.
Use the sequencer output to module output correspondence, and then
use the fact that LOX is connected to module output X.
"""
for sequencer_output_index in sequencer.connected_output_indices:
if sequencer_output_index % 2 != 0:
# We will only use real output 0 and 2, as they are part of the same
# complex outputs as real output 1 and 3
continue
module_output_index = 0 if sequencer_output_index == 0 else 1
yield module_output_index
[docs]
def assign_attenuation(self) -> None:
"""
Assigns attenuation settings from the hardware configuration.
Floats that are a multiple of 1 are converted to ints.
This is needed because the :func:`quantify_core.measurement.control.grid_setpoints`
converts setpoints to floats when using an attenuation as settable.
"""
def _convert_to_int(
value: InputAttenuation | OutputAttenuation | None, label: str
) -> int | None:
if value is not None:
if not math.isclose(value % 1, 0):
raise ValueError(
f'Trying to set "{label}" to non-integer value {value}'
)
return int(value)
in0_att = out0_att = out1_att = None
input_att_cfg = self.instrument_cfg.hardware_options.input_att
output_att_cfg = self.instrument_cfg.hardware_options.output_att
for (
portclock,
channel_name_path,
) in self.instrument_cfg.portclock_to_path.items():
channel_name = channel_name_path.split(".")[-1]
if channel_name in "complex_input_0" and input_att_cfg is not None:
in0_att = input_att_cfg.get(portclock)
for (
portclock,
channel_name_path,
) in self.instrument_cfg.portclock_to_path.items():
channel_name = channel_name_path.split(".")[-1]
if channel_name == "complex_output_0":
if input_att_cfg is not None:
in0_att_from_output = input_att_cfg.get(portclock)
if in0_att_from_output is not None:
if in0_att is not None:
raise ValueError(
f"'input_att' is defined for both 'complex_input_0' and "
f"'complex_output_0' on module '{self.name}', which is prohibited. "
f"Make sure you define it at a single place."
)
in0_att = in0_att_from_output
if output_att_cfg is not None:
out0_att = output_att_cfg.get(portclock)
if channel_name == "complex_output_1" and output_att_cfg is not None:
out1_att = output_att_cfg.get(portclock)
self._settings.in0_att = _convert_to_int(in0_att, label="in0_att")
self._settings.out0_att = _convert_to_int(out0_att, label="out0_att")
self._settings.out1_att = _convert_to_int(out1_att, label="out1_att")
