Loading quantify_scheduler/backends/qblox/compiler_abc.py +6 −3 Original line number Diff line number Diff line Loading @@ -1168,9 +1168,12 @@ class PulsarBase(ControlDeviceCompiler, ABC): def extract_mapping_item(acquisition: OpInfo) -> Tuple[Tuple[int, int], str]: return ( ( acquisition.data["acq_channel"], acquisition.data["acq_index"], ), acquisition.data["protocol"] ), acquisition.data["protocol"], ) acq_mapping = dict() for sequencer in self.sequencers.values(): Loading quantify_scheduler/helpers/waveforms.py +45 −0 Original line number Diff line number Diff line Loading @@ -458,3 +458,48 @@ def normalize_waveform_data(data: np.ndarray) -> Tuple[np.ndarray, float, float] norm_data_i = data.imag / amp_imag if amp_imag != 0.0 else np.zeros(data.imag.shape) rescaled_data = norm_data_r + 1.0j * norm_data_i return rescaled_data, amp_real, amp_imag def area_pulses(pulses: List[Dict[str, Any]], sampling_rate: int) -> float: """ Calculates the area of a set of pulses. Parameters ---------- pulses List of dictinary with information of the pulses sampling_rate Sampling rate for the pulse Returns ------- : The area formed by all the pulses """ area: float = 0.0 for pulse in pulses: area += area_pulse(pulse, sampling_rate) return area def area_pulse(pulse: Dict[str, Any], sampling_rate: int) -> float: """ Calculates the area of a set of pulses. Parameters ---------- pulse The dictionary with information of the pulse sampling_rate Sampling rate for the pulse Returns ------- : The area defined by the pulse """ assert sampling_rate > 0 waveform: np.ndarray = get_waveform(pulse, sampling_rate) # Nice to have: Give the user the option to choose integration algorithm return waveform.sum() / sampling_rate quantify_scheduler/pulse_library.py +109 −1 Original line number Diff line number Diff line Loading @@ -4,10 +4,11 @@ # pylint: disable= too-many-arguments, too-many-ancestors from __future__ import annotations from typing import Optional from typing import Any, Dict, List, Optional from qcodes import validators from quantify_scheduler.types import Operation from quantify_scheduler.resources import BasebandClockResource from quantify_scheduler.helpers.waveforms import area_pulses class IdlePulse(Operation): Loading Loading @@ -481,3 +482,110 @@ class DRAGPulse(Operation): def __str__(self) -> str: pulse_info = self.data["pulse_info"][0] return self._get_signature(pulse_info) class DCCompensationPulse(SquarePulse): """ Calculates a SquarePulse to counteract charging effects based on a list of pulses. Parameters ---------- pulses List of pulses to compensate amp Desired amplitude of the DCCompensationPulse. Leave to None to calculate the value for compensation, in this case you must assign a value to duration. The sign of the amplitude is ignored and ajusted automatically to perform the compensation. duration Desired pulse duration in seconds. Leave to None to calculate the value for compensation, in this case you must assign a value to amp. The sign of the value of amp given in the previous step is ajusted to perform the compensation. port Port to perform the compensation. Any pulse that does not belong to the specified port is ignored. clock Clock used to modulate the pulse. phase Phase of the pulse in degrees. t0 Time in seconds when to start the pulses relative to the start time of the Operation in the Schedule. data The operation's dictionary, by default None Note: if the data parameter is not None all other parameters are overwritten using the contents of data. """ def __init__( self, pulses: List[Operation], sampling_rate: int, port: str, t0: float = 0, amp: Optional[float] = None, duration: Optional[float] = None, data: Optional[Dict[str, Any]] = None, ) -> None: # Make sure that the list contains at least one element assert len(pulses) > 0 pulse_info_list: List[Dict[str, Any]] = DCCompensationPulse._extract_pulses( pulses, port ) # Calculate the area given by the list of pulses area: float = area_pulses(pulse_info_list, sampling_rate) # Calculate the compensation amplitude and duration based on area c_duration: float c_amp: float if amp is None and duration is not None: assert duration > 0 c_duration = duration c_amp = -area / c_duration elif amp is not None and duration is None: if area > 0: c_amp = -abs(amp) else: c_amp = abs(amp) c_duration = area / c_amp else: raise ValueError( "The `DCCompensationPulse` allows either amp or duration to " + "be specified, not both. Both amp and duration were passed." ) super().__init__( amp=c_amp, duration=c_duration, port=port, clock=BasebandClockResource.IDENTITY, phase=0, t0=t0, data=data, ) @staticmethod def _extract_pulses(pulses: List[Operation], port: str) -> List[Dict[str, Any]]: # Collect all pulses for the given port pulse_info_list: List[Dict[str, Any]] = list() for pulse in pulses: for pulse_info in pulse["pulse_info"]: if ( pulse_info["port"] == port and pulse_info["clock"] == BasebandClockResource.IDENTITY ): pulse_info_list.append(pulse_info) if len(pulse_info_list) == 0: raise ValueError( "`DCCompensationPulse` needs at least one pulse with" + "clock=BasebandClockResource.IDENTITY for the port {}".format(port) ) return pulse_info_list tests/scheduler/backends/test_qblox_backend.py +1 −3 Original line number Diff line number Diff line Loading @@ -48,9 +48,7 @@ from quantify_scheduler.backends.qblox.helpers import ( to_grid_time, ) from quantify_scheduler.backends import qblox_backend as qb from quantify_scheduler.backends.types.qblox import ( QASMRuntimeSettings, ) from quantify_scheduler.backends.types.qblox import QASMRuntimeSettings from quantify_scheduler.backends.qblox.instrument_compilers import ( Pulsar_QCM, Pulsar_QRM, Loading tests/scheduler/backends/test_zhinst_backend.py +4 −4 Original line number Diff line number Diff line Loading @@ -151,9 +151,9 @@ def create_uhfqa_mock(mocker): def create_uhfqa_awg(i: int) -> uhfqa.AWG: _sequence_params = { "sequence_parameters": { "clock_rate": 1.8e9, # GSa/s } "clock_rate": 1.8e9, } } # GSa/s def get_string(value: str): if value == "directory": Loading Loading @@ -223,9 +223,9 @@ def create_hdawg_mock(mocker): # Section: 4.14.3 Constansts and Variables (page 181) _sequence_params = { "sequence_parameters": { "clock_rate": 2.4e9, # GSa/s } "clock_rate": 2.4e9, } } # GSa/s def get_string(value: str): if value == "directory": Loading Loading
quantify_scheduler/backends/qblox/compiler_abc.py +6 −3 Original line number Diff line number Diff line Loading @@ -1168,9 +1168,12 @@ class PulsarBase(ControlDeviceCompiler, ABC): def extract_mapping_item(acquisition: OpInfo) -> Tuple[Tuple[int, int], str]: return ( ( acquisition.data["acq_channel"], acquisition.data["acq_index"], ), acquisition.data["protocol"] ), acquisition.data["protocol"], ) acq_mapping = dict() for sequencer in self.sequencers.values(): Loading
quantify_scheduler/helpers/waveforms.py +45 −0 Original line number Diff line number Diff line Loading @@ -458,3 +458,48 @@ def normalize_waveform_data(data: np.ndarray) -> Tuple[np.ndarray, float, float] norm_data_i = data.imag / amp_imag if amp_imag != 0.0 else np.zeros(data.imag.shape) rescaled_data = norm_data_r + 1.0j * norm_data_i return rescaled_data, amp_real, amp_imag def area_pulses(pulses: List[Dict[str, Any]], sampling_rate: int) -> float: """ Calculates the area of a set of pulses. Parameters ---------- pulses List of dictinary with information of the pulses sampling_rate Sampling rate for the pulse Returns ------- : The area formed by all the pulses """ area: float = 0.0 for pulse in pulses: area += area_pulse(pulse, sampling_rate) return area def area_pulse(pulse: Dict[str, Any], sampling_rate: int) -> float: """ Calculates the area of a set of pulses. Parameters ---------- pulse The dictionary with information of the pulse sampling_rate Sampling rate for the pulse Returns ------- : The area defined by the pulse """ assert sampling_rate > 0 waveform: np.ndarray = get_waveform(pulse, sampling_rate) # Nice to have: Give the user the option to choose integration algorithm return waveform.sum() / sampling_rate
quantify_scheduler/pulse_library.py +109 −1 Original line number Diff line number Diff line Loading @@ -4,10 +4,11 @@ # pylint: disable= too-many-arguments, too-many-ancestors from __future__ import annotations from typing import Optional from typing import Any, Dict, List, Optional from qcodes import validators from quantify_scheduler.types import Operation from quantify_scheduler.resources import BasebandClockResource from quantify_scheduler.helpers.waveforms import area_pulses class IdlePulse(Operation): Loading Loading @@ -481,3 +482,110 @@ class DRAGPulse(Operation): def __str__(self) -> str: pulse_info = self.data["pulse_info"][0] return self._get_signature(pulse_info) class DCCompensationPulse(SquarePulse): """ Calculates a SquarePulse to counteract charging effects based on a list of pulses. Parameters ---------- pulses List of pulses to compensate amp Desired amplitude of the DCCompensationPulse. Leave to None to calculate the value for compensation, in this case you must assign a value to duration. The sign of the amplitude is ignored and ajusted automatically to perform the compensation. duration Desired pulse duration in seconds. Leave to None to calculate the value for compensation, in this case you must assign a value to amp. The sign of the value of amp given in the previous step is ajusted to perform the compensation. port Port to perform the compensation. Any pulse that does not belong to the specified port is ignored. clock Clock used to modulate the pulse. phase Phase of the pulse in degrees. t0 Time in seconds when to start the pulses relative to the start time of the Operation in the Schedule. data The operation's dictionary, by default None Note: if the data parameter is not None all other parameters are overwritten using the contents of data. """ def __init__( self, pulses: List[Operation], sampling_rate: int, port: str, t0: float = 0, amp: Optional[float] = None, duration: Optional[float] = None, data: Optional[Dict[str, Any]] = None, ) -> None: # Make sure that the list contains at least one element assert len(pulses) > 0 pulse_info_list: List[Dict[str, Any]] = DCCompensationPulse._extract_pulses( pulses, port ) # Calculate the area given by the list of pulses area: float = area_pulses(pulse_info_list, sampling_rate) # Calculate the compensation amplitude and duration based on area c_duration: float c_amp: float if amp is None and duration is not None: assert duration > 0 c_duration = duration c_amp = -area / c_duration elif amp is not None and duration is None: if area > 0: c_amp = -abs(amp) else: c_amp = abs(amp) c_duration = area / c_amp else: raise ValueError( "The `DCCompensationPulse` allows either amp or duration to " + "be specified, not both. Both amp and duration were passed." ) super().__init__( amp=c_amp, duration=c_duration, port=port, clock=BasebandClockResource.IDENTITY, phase=0, t0=t0, data=data, ) @staticmethod def _extract_pulses(pulses: List[Operation], port: str) -> List[Dict[str, Any]]: # Collect all pulses for the given port pulse_info_list: List[Dict[str, Any]] = list() for pulse in pulses: for pulse_info in pulse["pulse_info"]: if ( pulse_info["port"] == port and pulse_info["clock"] == BasebandClockResource.IDENTITY ): pulse_info_list.append(pulse_info) if len(pulse_info_list) == 0: raise ValueError( "`DCCompensationPulse` needs at least one pulse with" + "clock=BasebandClockResource.IDENTITY for the port {}".format(port) ) return pulse_info_list
tests/scheduler/backends/test_qblox_backend.py +1 −3 Original line number Diff line number Diff line Loading @@ -48,9 +48,7 @@ from quantify_scheduler.backends.qblox.helpers import ( to_grid_time, ) from quantify_scheduler.backends import qblox_backend as qb from quantify_scheduler.backends.types.qblox import ( QASMRuntimeSettings, ) from quantify_scheduler.backends.types.qblox import QASMRuntimeSettings from quantify_scheduler.backends.qblox.instrument_compilers import ( Pulsar_QCM, Pulsar_QRM, Loading
tests/scheduler/backends/test_zhinst_backend.py +4 −4 Original line number Diff line number Diff line Loading @@ -151,9 +151,9 @@ def create_uhfqa_mock(mocker): def create_uhfqa_awg(i: int) -> uhfqa.AWG: _sequence_params = { "sequence_parameters": { "clock_rate": 1.8e9, # GSa/s } "clock_rate": 1.8e9, } } # GSa/s def get_string(value: str): if value == "directory": Loading Loading @@ -223,9 +223,9 @@ def create_hdawg_mock(mocker): # Section: 4.14.3 Constansts and Variables (page 181) _sequence_params = { "sequence_parameters": { "clock_rate": 2.4e9, # GSa/s } "clock_rate": 2.4e9, } } # GSa/s def get_string(value: str): if value == "directory": Loading