Merge branch '2-acquisition-representation' into 'develop' (be3e3f0a) · Commits · Quantify / quantify-scheduler

quantify/scheduler/acquisition_library.py

+159 −77

Original line number	Diff line number	Diff line
		# Repository: https://gitlab.com/quantify-os/quantify-scheduler
		# Licensed according to the LICENCE file on the master branch
		"""Standard acquisition protocols for use with the quantify.scheduler."""
		from typing import Any, Dict, List
		from typing import Any, Dict, List, Optional, Union

		import numpy as np
		from quantify.scheduler.enums import BinMode
		from quantify.scheduler.types import Operation


		class Trace(Operation):
		"""The Trace acquisition protocol measures a signal s(t)."""

		def __init__(
		self,
		duration: float,
		port: str,
		acq_channel: int = 0,
		acq_index: int = 0,
		bin_mode: BinMode = BinMode.APPEND,
		bin_mode: Union[BinMode, str] = BinMode.APPEND,
		t0: float = 0,
		data: Optional[dict] = None,
		):
		"""
		Creates a new instance of Trace.
		@@ -34,19 +38,33 @@ class Trace(Operation):
		duration :
		The acquisition duration in seconds.
		acq_channel :
		The data channel in which the acquisition is stored, by default 0. Describes the "where" information of the
		measurement, typically corresponds to a qubit idx.
		The data channel in which the acquisition is stored, is by default 0.
		Describes the "where" information of the measurement, which typically
		corresponds to a qubit idx.
		acq_index :
		The data register in which the acquisition is stored, by default 0. Describes the "when" information of the
		measurement, used to label/tag individual measurements in a large circuit. Typically corresponds to the
		setpoints of a schedule (e.g., tau in a T1 experiment).
		The data register in which the acquisition is stored, by default 0.
		Describes the "when" information of the measurement, used to label or
		tag individual measurements in a large circuit. Typically corresponds
		to the setpoints of a schedule (e.g., tau in a T1 experiment).
		bin_mode :
		Describes what is done when data is written to a register that already contains a value. Options are
		"append" which appends the result to the list or "average" which stores the weighted average value of the
		Describes what is done when data is written to a register that already
		contains a value. Options are "append" which appends the result to the
		list or "average" which stores the weighted average value of the
		new result and the old register value, by default BinMode.APPEND
		t0 :
		The acquisition start time in seconds, by default 0
		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.
		"""

		if data is None:
		if not isinstance(duration, float):
		duration = float(duration)
		if isinstance(bin_mode, str):
		bin_mode = BinMode(bin_mode)

		data = {
		"name": "Trace",
		"acquisition_info": [
		@@ -64,8 +82,17 @@ class Trace(Operation):
		}
		super().__init__(name=data["name"], data=data)

		def __str__(self) -> str:
		acq_info = self.data["acquisition_info"][0]
		return self._get_signature(acq_info)


		class WeightedIntegratedComplex(Operation):
		"""
		Weighted integration acquisition protocol on a
		complex signal in a custom complex window.
		"""

		def __init__(
		self,
		waveform_a: Dict[str, Any],
		@@ -75,9 +102,10 @@ class WeightedIntegratedComplex(Operation):
		duration: float,
		acq_channel: int = 0,
		acq_index: int = 0,
		bin_mode: BinMode = BinMode.APPEND,
		bin_mode: Union[BinMode, str] = BinMode.APPEND,
		phase: float = 0,
		t0: float = 0,
		data: Optional[dict] = None,
		):
		r"""
		Creates a new instance of WeightedIntegratedComplex.
		@@ -112,20 +140,27 @@ class WeightedIntegratedComplex(Operation):
		duration :
		The acquisition duration in seconds.
		acq_channel :
		The data channel in which the acquisition is stored, by default 0. Describes the "where" information of the
		measurement, typically corresponds to a qubit idx.
		The data channel in which the acquisition is stored, by default 0.
		Describes the "where" information of the measurement, which typically
		corresponds to a qubit idx.
		acq_index :
		The data register in which the acquisition is stored, by default 0. Describes the "when" information of the
		measurement, used to label/tag individual measurements in a large circuit. Typically corresponds to the
		setpoints of a schedule (e.g., tau in a T1 experiment).
		The data register in which the acquisition is stored, by default 0.
		Describes the "when" information of the measurement, used to label or
		tag individual measurements in a large circuit. Typically corresponds
		to the setpoints of a schedule (e.g., tau in a T1 experiment).
		bin_mode :
		Describes what is done when data is written to a register that already contains a value. Options are
		"append" which appends the result to the list or "average" which stores the weighted average value of the
		new result and the old register value, by default :code:`BinMode.APPEND`
		Describes what is done when data is written to a register that already
		contains a value. Options are "append" which appends the result to the
		list or "average" which stores the weighted average value of the
		new result and the old register value, by default BinMode.APPEND
		phase :
		The phase of the pulse and acquisition in degrees, by default 0
		t0 :
		The acquisition start time in seconds, by default 0
		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.

		Raises
		------
		@@ -136,6 +171,7 @@ class WeightedIntegratedComplex(Operation):
		# FIXME: need to be able to add phases to the waveform separate from the clock.
		raise NotImplementedError("Non-zero phase not yet implemented")

		if data is None:
		data = {
		"name": "WeightedIntegrationComplex",
		"acquisition_info": [
		@@ -155,6 +191,10 @@ class WeightedIntegratedComplex(Operation):
		}
		super().__init__(name=data["name"], data=data)

		def __str__(self) -> str:
		acq_info = self.data["acquisition_info"][0]
		return self._get_signature(acq_info)


		class SSBIntegrationComplex(WeightedIntegratedComplex):
		def __init__(
		@@ -164,21 +204,20 @@ class SSBIntegrationComplex(WeightedIntegratedComplex):
		duration: float,
		acq_channel: int = 0,
		acq_index: int = 0,
		bin_mode: BinMode = BinMode.APPEND,
		bin_mode: Union[BinMode, str] = BinMode.APPEND,
		phase: float = 0,
		t0: float = 0,
		data: Optional[dict] = None,
		):
		"""
		Creates a new instance of SSBIntegrationComplex.
		Single Sideband Integration acquisition protocol
		with complex results.
		Creates a new instance of SSBIntegrationComplex. Single Sideband
		Integration acquisition protocol with complex results.

		A weighted integrated acquisition on a complex
		signal using a square window for the acquisition
		weights.
		A weighted integrated acquisition on a complex signal using a
		square window for the acquisition weights.

		The signal is demodulated using the specified clock, and the square window then effectively specifies an
		integration window.
		The signal is demodulated using the specified clock, and the
		square window then effectively specifies an integration window.

		Parameters
		----------
		@@ -189,20 +228,27 @@ class SSBIntegrationComplex(WeightedIntegratedComplex):
		duration :
		The acquisition duration in seconds.
		acq_channel :
		The data channel in which the acquisition is stored, by default 0. Describes the "where" information of the
		measurement, typically corresponds to a qubit idx.
		The data channel in which the acquisition is stored, by default 0.
		Describes the "where" information of the measurement, which typically
		corresponds to a qubit idx.
		acq_index :
		The data register in which the acquisition is stored, by default 0. Describes the "when" information of the
		measurement, used to label/tag individual measurements in a large circuit. Typically corresponds to the
		setpoints of a schedule (e.g., tau in a T1 experiment).
		The data register in which the acquisition is stored, by default 0.
		Describes the "when" information of the measurement, used to label or
		tag individual measurements in a large circuit. Typically corresponds
		to the setpoints of a schedule (e.g., tau in a T1 experiment).
		bin_mode :
		Describes what is done when data is written to a register that already contains a value. Options are
		"append" which appends the result to the list or "average" which stores the weighted average value of the
		new result and the old register value, by default :code:`BinMode.APPEND`
		Describes what is done when data is written to a register that already
		contains a value. Options are "append" which appends the result to the
		list or "average" which stores the weighted average value of the
		new result and the old register value, by default BinMode.APPEND
		phase :
		The phase of the pulse and acquisition in degrees, by default 0
		t0 :
		The acquisition start time in seconds, by default 0
		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.
		"""
		waveform_i = {
		"port": port,
		@@ -233,6 +279,7 @@ class SSBIntegrationComplex(WeightedIntegratedComplex):
		bin_mode,
		phase,
		t0,
		data,
		)
		self.data["name"] = "SSBIntegrationComplex"

		@@ -248,9 +295,10 @@ class NumericalWeightedIntegrationComplex(WeightedIntegratedComplex):
		interpolation: str = "linear",
		acq_channel: int = 0,
		acq_index: int = 0,
		bin_mode: BinMode = BinMode.APPEND,
		bin_mode: Union[BinMode, str] = BinMode.APPEND,
		phase: float = 0,
		t0: float = 0,
		data: Optional[dict] = None,
		):
		r"""
		Creates a new instance of NumericalWeightedIntegrationComplex.
		@@ -285,22 +333,30 @@ class NumericalWeightedIntegrationComplex(WeightedIntegratedComplex):
		clock :
		The clock used to demodulate the acquisition.
		interpolation :
		The type of interpolation to use, by default "linear". This argument is passed to :obj:`~scipy.interpolate.interp1d`.
		The type of interpolation to use, by default "linear". This argument is
		passed to :obj:`~scipy.interpolate.interp1d`.
		acq_channel :
		The data channel in which the acquisition is stored, by default 0. Describes the "where" information of the
		measurement, typically corresponds to a qubit idx.
		The data channel in which the acquisition is stored, by default 0.
		Describes the "where" information of the measurement, which typically
		corresponds to a qubit idx.
		acq_index :
		The data register in which the acquisition is stored, by default 0. Describes the "when" information of the
		measurement, used to label/tag individual measurements in a large circuit. Typically corresponds to the
		setpoints of a schedule (e.g., tau in a T1 experiment).
		The data register in which the acquisition is stored, by default 0.
		Describes the "when" information of the measurement, used to label or
		tag individual measurements in a large circuit. Typically corresponds
		to the setpoints of a schedule (e.g., tau in a T1 experiment).
		bin_mode :
		Describes what is done when data is written to a register that already contains a value. Options are
		"append" which appends the result to the list or "average" which stores the weighted average value of the
		new result and the old register value, by default :code:`BinMode.APPEND`
		Describes what is done when data is written to a register that already
		contains a value. Options are "append" which appends the result to the
		list or "average" which stores the weighted average value of the
		new result and the old register value, by default BinMode.APPEND
		phase :
		The phase of the pulse and acquisition in degrees, by default 0
		t0 :
		The acquisition start time in seconds, by default 0
		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.
		"""
		waveforms_a = {
		"wf_func": "scipy.interpolate.interp1d",
		@@ -326,5 +382,31 @@ class NumericalWeightedIntegrationComplex(WeightedIntegratedComplex):
		bin_mode,
		phase,
		t0,
		data,
		)
		self.data["name"] = "NumericalWeightedIntegrationComplex"

		def __str__(self) -> str:
		acq_info = self.data["acquisition_info"][0]
		weights_a = np.array2string(
		acq_info["waveforms"][0]["weights"], separator=", ", precision=9
		)
		weights_b = np.array2string(
		acq_info["waveforms"][1]["weights"], separator=", ", precision=9
		)
		t = np.array2string(acq_info["waveforms"][0]["t"], separator=", ", precision=9)
		port = acq_info["port"]
		clock = acq_info["clock"]
		interpolation = acq_info["waveforms"][0]["interpolation"]
		acq_channel = acq_info["acq_channel"]
		acq_index = acq_info["acq_index"]
		bin_mode = acq_info["bin_mode"].value
		phase = acq_info["phase"]
		t0 = acq_info["t0"]

		return (
		f"{self.__class__.__name__}(weights_a={weights_a}, weights_b={weights_b}, "
		f"t={t}, port='{port}', clock='{clock}', interpolation='{interpolation}', "
		f"acq_channel={acq_channel}, acq_index={acq_index}, bin_mode='{bin_mode}', "
		f"phase={phase}, t0={t0})"
		)

quantify/scheduler/types.py

+30 −1

Original line number	Diff line number	Diff line
		@@ -4,10 +4,11 @@
		from __future__ import annotations

		import inspect
		import json
		from ast import literal_eval
		from collections import UserDict
		from copy import deepcopy
		import json
		from enum import Enum
		from typing import Any, Dict
		from uuid import uuid4

		@@ -15,6 +16,7 @@ import jsonschema
		import numpy as np
		from typing_extensions import Literal
		from quantify.scheduler import resources
		from quantify.scheduler.enums import BinMode
		from quantify.utilities import general


		@@ -174,6 +176,9 @@ class Operation(UserDict): # pylint: disable=too-many-ancestors
		:
		"""
		value = parameters[key]
		if isinstance(value, Enum):
		enum_value = value.value
		value = enum_value
		value = f"'{value}'" if isinstance(value, str) else value
		return f"{key}={value}"

		@@ -233,6 +238,20 @@ class Operation(UserDict): # pylint: disable=too-many-ancestors
		_data["gate_info"]["unitary"], separator=", ", precision=9
		)

		for acq_info in _data["acquisition_info"]:
		if "bin_mode" in acq_info and isinstance(acq_info["bin_mode"], BinMode):
		acq_info["bin_mode"] = acq_info["bin_mode"].value

		for waveform in acq_info["waveforms"]:
		if "t" in waveform:
		waveform["t"] = np.array2string(
		waveform["t"], separator=", ", precision=9
		)
		if "weights" in waveform:
		waveform["weights"] = np.array2string(
		waveform["weights"], separator=", ", precision=9
		)

		return _data

		def _deserialize(self) -> None:
		@@ -244,6 +263,16 @@ class Operation(UserDict): # pylint: disable=too-many-ancestors
		literal_eval(self.data["gate_info"]["unitary"])
		)

		for acq_info in self.data["acquisition_info"]:
		if "bin_mode" in acq_info and isinstance(acq_info["bin_mode"], str):
		acq_info["bin_mode"] = BinMode(acq_info["bin_mode"])

		for waveform in acq_info["waveforms"]:
		if "t" in waveform and isinstance(waveform["t"], str):
		waveform["t"] = np.array(literal_eval(waveform["t"]))
		if "weights" in waveform and isinstance(waveform["weights"], str):
		waveform["weights"] = np.array(literal_eval(waveform["weights"]))

		@classmethod
		def is_valid(cls, operation) -> bool:
		"""Checks if the operation is valid according to its schema."""

tests/scheduler/test_acquisition_library.py

+156 −6

Original line number	Diff line number	Diff line
		from quantify.scheduler.types import Operation
		from quantify.scheduler.enums import BinMode
		# Repository: https://gitlab.com/quantify-os/quantify-scheduler
		# Licensed according to the LICENCE file on the master branch
		"""Unit tests acquisition protocols for use with the quantify.scheduler."""

		# pylint: disable=missing-class-docstring
		# pylint: disable=missing-function-docstring
		# pylint: disable=eval-used
		from unittest import TestCase

		import pytest
		import numpy as np
		from quantify.scheduler.acquisition_library import (
		NumericalWeightedIntegrationComplex,
		SSBIntegrationComplex,
		Trace,
		)
		from quantify.scheduler.enums import BinMode
		from quantify.scheduler.gate_library import X90
		from quantify.scheduler.pulse_library import DRAGPulse
		from quantify.scheduler.types import Operation


		def test_ssb_integration_complex():
		@@ -66,7 +78,7 @@ def test_valid_acquisition():


		def test_trace():
		tr = Trace(
		trace = Trace(
		1234e-9,
		port="q0.res",
		acq_channel=4815162342,
		@@ -74,6 +86,144 @@ def test_trace():
		bin_mode=BinMode.AVERAGE,
		t0=12e-9,
		)
		assert Operation.is_valid(tr)
		assert tr.data["acquisition_info"][0]["acq_index"] == 4815162342
		assert tr.data["acquisition_info"][0]["acq_channel"] == 4815162342
		assert Operation.is_valid(trace)
		assert trace.data["acquisition_info"][0]["acq_index"] == 4815162342
		assert trace.data["acquisition_info"][0]["acq_channel"] == 4815162342


		@pytest.mark.parametrize(
		"operation",
		[
		Trace(
		duration=16e-9,
		port="q0.res",
		),
		SSBIntegrationComplex(
		port="q0.res",
		clock="q0.01",
		duration=100e-9,
		),
		NumericalWeightedIntegrationComplex(
		weights_a=np.zeros(3, dtype=complex),
		weights_b=np.ones(3, dtype=complex),
		t=np.linspace(0, 3, 1),
		port="q0.res",
		clock="q0.01",
		),
		],
		)
		def test__repr__(operation: Operation):
		assert eval(repr(operation)) == operation


		@pytest.mark.parametrize(
		"operation",
		[
		Trace(
		duration=16e-9,
		port="q0.res",
		),
		SSBIntegrationComplex(
		port="q0.res",
		clock="q0.01",
		duration=100e-9,
		),
		NumericalWeightedIntegrationComplex(
		weights_a=np.zeros(3, dtype=complex),
		weights_b=np.ones(3, dtype=complex),
		t=np.linspace(0, 3, 1),
		port="q0.res",
		clock="q0.01",
		),
		],
		)
		def test__str__(operation: Operation):
		assert isinstance(eval(str(operation)), type(operation))


		@pytest.mark.parametrize(
		"operation",
		[
		Trace(
		duration=16e-9,
		port="q0.res",
		),
		SSBIntegrationComplex(
		port="q0.res",
		clock="q0.01",
		duration=100e-9,
		),
		NumericalWeightedIntegrationComplex(
		weights_a=np.zeros(3, dtype=complex),
		weights_b=np.ones(3, dtype=complex),
		t=np.linspace(0, 3, 1),
		port="q0.res",
		clock="q0.01",
		),
		],
		)
		def test_deserialize(operation: Operation):
		# Arrange
		operation_repr: str = repr(operation)

		# Act
		obj = eval(operation_repr)

		# Assert
		if isinstance(operation, NumericalWeightedIntegrationComplex):
		waveforms = operation.data["acquisition_info"][0]["waveforms"]
		for i, waveform in enumerate(waveforms):
		assert isinstance(waveform["t"], (np.generic, np.ndarray))
		assert isinstance(waveform["weights"], (np.generic, np.ndarray))
		np.testing.assert_array_almost_equal(
		obj.data["acquisition_info"][0]["waveforms"][i]["t"],
		waveform["t"],
		decimal=9,
		)
		np.testing.assert_array_almost_equal(
		obj.data["acquisition_info"][0]["waveforms"][i]["weights"],
		waveform["weights"],
		decimal=9,
		)

		# TestCase().assertDictEqual cannot compare numpy arrays for equality
		# therefore "unitary" is removed
		del obj.data["acquisition_info"][0]["waveforms"][i]["t"]
		del waveform["t"]
		del obj.data["acquisition_info"][0]["waveforms"][i]["weights"]
		del waveform["weights"]

		TestCase().assertDictEqual(obj.data, operation.data)


		@pytest.mark.parametrize(
		"operation",
		[
		Trace(
		duration=16e-9,
		port="q0.res",
		),
		SSBIntegrationComplex(
		port="q0.res",
		clock="q0.01",
		duration=100e-9,
		),
		NumericalWeightedIntegrationComplex(
		weights_a=np.zeros(3, dtype=complex),
		weights_b=np.ones(3, dtype=complex),
		t=np.linspace(0, 3, 1),
		port="q0.res",
		clock="q0.01",
		),
		],
		)
		def test__repr__modify_not_equal(operation: Operation):
		# Arrange
		obj = eval(repr(operation))
		assert obj == operation

		# Act
		obj.data["acquisition_info"][0]["foo"] = "bar"

		# Assert
		assert obj != operation