Source code for quantify_scheduler.backends.corrections

# Repository:
# Licensed according to the LICENCE file on the main branch
"""Pulse and acquisition corrections for hardware compilation."""
import logging
import warnings
from typing import Any, Dict, Generator, Optional, Tuple

import numpy as np
from quantify_scheduler import Schedule
from quantify_scheduler.helpers.importers import import_python_object_from_string
from quantify_scheduler.helpers.waveforms import get_waveform
from quantify_scheduler.operations.operation import Operation
from quantify_scheduler.operations.pulse_library import NumericalPulse

[docs] logger = logging.getLogger(__name__)
# pylint: disable=too-few-public-methods
[docs] def determine_relative_latency_corrections( hardware_cfg: Dict[str, Any] ) -> Dict[str, float]: """ Generates the latency configuration dict for all port-clock combinations that are present in the hardware_cfg. This is done by first setting unspecified latency corrections to zero, and then subtracting the minimum latency from all latency corrections. """ def _extract_port_clocks(hardware_cfg: Dict[str, Any]) -> Generator: """ Extracts all port-clock combinations that are present in a hardware configuration. Based on: """ if hasattr(hardware_cfg, "items"): for k, v in hardware_cfg.items(): if k == "port": port_clock = f'{hardware_cfg["port"]}-{hardware_cfg["clock"]}' yield port_clock elif isinstance(v, dict): for port_clock in _extract_port_clocks(v): yield port_clock elif isinstance(v, list): for d in v: for port_clock in _extract_port_clocks(d): yield port_clock if (raw_latency_dict := hardware_cfg.get("latency_corrections")) is None: return {} port_clocks = _extract_port_clocks(hardware_cfg=hardware_cfg) latency_dict = {} for port_clock in port_clocks: # Set unspecified latency corrections to zero to avoid ending up with # negative latency corrections after subtracting minimum latency_dict[port_clock] = raw_latency_dict.get(port_clock, 0) # Subtract lowest value to ensure minimal latency is used and offset the latency # corrections to be relative to the minimum. Note that this supports negative delays # (which is useful for calibrating) minimum_of_latency_corrections = min(latency_dict.values(), default=0) for port_clock, latency_at_port_clock in latency_dict.items(): latency_dict[port_clock] = ( latency_at_port_clock - minimum_of_latency_corrections ) return latency_dict
[docs] def distortion_correct_pulse( # pylint: disable=too-many-arguments pulse_data: Dict[str, Any], sampling_rate: int, filter_func_name: str, input_var_name: str, kwargs_dict: Dict[str, Any], clipping_values: Optional[Tuple[float]] = None, ) -> NumericalPulse: """ Sample pulse and apply filter function to the sample to distortion correct it. Parameters ---------- pulse_data Definition of the pulse. sampling_rate The sampling rate used to generate the time axis values. filter_func_name The filter function path of the dynamically loaded filter function. Example: ``"scipy.signal.lfilter"``. input_var_name The input variable name of the dynamically loaded filter function, most likely: ``"x"``. kwargs_dict Dictionary containing kwargs for the dynamically loaded filter function. Example: ``{"b": [0.0, 0.5, 1.0], "a": 1}``. clipping_values Min and max value to which the corrected pulse will be clipped, depending on allowed output values for the instrument. Returns ------- : The sampled, distortion corrected pulse wrapped in a ``NumericalPulse``. """ waveform_data = get_waveform(pulse_info=pulse_data, sampling_rate=sampling_rate) filter_func = import_python_object_from_string(filter_func_name) kwargs = {input_var_name: waveform_data, **kwargs_dict} corrected_waveform_data = filter_func(**kwargs) if clipping_values is not None and len(clipping_values) == 2: corrected_waveform_data = np.clip( corrected_waveform_data, clipping_values[0], clipping_values[1] ) if corrected_waveform_data.size == 1: # Interpolation requires two sample points corrected_waveform_data = np.append( corrected_waveform_data, corrected_waveform_data[-1] ) corrected_pulse = NumericalPulse( samples=corrected_waveform_data, t_samples=np.linspace( start=0, stop=pulse_data["duration"], num=corrected_waveform_data.size ), port=pulse_data["port"], clock=pulse_data["clock"], t0=pulse_data["t0"], ) return corrected_pulse
[docs] def _is_distortion_correctable(operation: Operation) -> bool: """Checks whether distortion corrections can be applied to the given operation.""" return operation.valid_pulse and not operation.has_voltage_offset
[docs] def apply_distortion_corrections( schedule: Schedule, hardware_cfg: Dict[str, Any] ) -> Schedule: """ Apply distortion corrections to operations in the schedule. Defined via the hardware configuration file, example: .. code-block:: "distortion_corrections": { "q0:fl-cl0.baseband": { "filter_func": "scipy.signal.lfilter", "input_var_name": "x", "kwargs": { "b": [0.0, 0.5, 1.0], "a": [1] }, "clipping_values": [-2.5, 2.5] } } Clipping values are the boundaries to which the corrected pulses will be clipped, upon exceeding, these are optional to supply. For pulses in need of correcting (indicated by their port-clock combination) we are **only** replacing the dict in ``"pulse_info"`` associated to that specific pulse. This means that we can have a combination of corrected (i.e., pre-sampled) and uncorrected pulses in the same operation. Note that we are **not** updating the ``"operation_id"`` key, used to reference the operation from schedulables. Parameters ---------- schedule The schedule that contains operations that are to be distortion corrected. hardware_cfg The hardware configuration of the setup. Returns ------- : The schedule with distortion corrected operations. Warns ----- RuntimeWarning If distortion correction can not be applied to the type of Operation in the schedule. Raises ------ KeyError when elements are missing in distortion correction config for a port-clock combination. KeyError when clipping values are supplied but not two values exactly, min and max. """ distortion_corrections_key = "distortion_corrections" if hardware_cfg.get(distortion_corrections_key) is None: logging.debug(f'No key "{distortion_corrections_key}" supplied in hardware_cfg') return schedule for operation_id in schedule.operations.keys(): substitute_operation = None for pulse_info_idx, pulse_data in enumerate( schedule.operations[operation_id].data["pulse_info"] ): portclock_key = f"{pulse_data['port']}-{pulse_data['clock']}" if portclock_key in hardware_cfg[distortion_corrections_key]: if not _is_distortion_correctable(schedule.operations[operation_id]): warnings.warn( f"Schedule contains an operation, for which distortion " f"correction is not implemented. Please either replace the " f"operation, or omit the distortion correction setting for " f"this port in order to suppress this warning. Offending " f"operation: {schedule.operations[operation_id]}", RuntimeWarning, ) continue correction_cfg = hardware_cfg[distortion_corrections_key][portclock_key] filter_func_name = correction_cfg.get("filter_func", None) input_var_name = correction_cfg.get("input_var_name", None) kwargs_dict = correction_cfg.get("kwargs", None) clipping_values = correction_cfg.get("clipping_values", None) sampling_rate = correction_cfg.get("sampling_rate") if None in (filter_func_name, input_var_name, kwargs_dict): raise KeyError( f"One or more elements missing in distortion correction config " f'for "{portclock_key}"\n\n' f'"filter_func": {filter_func_name}\n' f'"input_var_name": {input_var_name}\n' f'"kwargs": {kwargs_dict}' ) if clipping_values and len(clipping_values) != 2: raise KeyError( f'Clipping values for "{portclock_key}" should contain two ' "values, min and max.\n" f'"clipping_values": {clipping_values}' ) corrected_pulse = distortion_correct_pulse( pulse_data=pulse_data, sampling_rate=sampling_rate, filter_func_name=filter_func_name, input_var_name=input_var_name, kwargs_dict=kwargs_dict, clipping_values=clipping_values, ) schedule.operations[operation_id].data["pulse_info"][pulse_info_idx] = (["pulse_info"][0] ) if pulse_info_idx == 0: substitute_operation = corrected_pulse # Convert to operation-type of first entry in pulse_info, # required as first entry in pulse_info is used to generate signature in __str__ if substitute_operation is not None:["pulse_info"] = schedule.operations[ operation_id ].data["pulse_info"] schedule.operations[operation_id] = substitute_operation return schedule