From 962b31b3c744288c363aba9a53818e47ec91f2f3 Mon Sep 17 00:00:00 2001 From: Yoav Romach Date: Mon, 23 Sep 2024 19:54:29 +0300 Subject: [PATCH] black --- .../advanced-driver/advanced_driver.py | 106 ++++------ .../advanced-driver/hello_qcodes_advanced.py | 12 +- .../basic-driver/hello_qcodes.py | 36 +--- .../stability-diagram/raster_scan.py | 12 +- .../stability-diagram/spiral_scan.py | 67 ++---- examples/callable_from_qua/qua_print.py | 1 + .../update_other_instruments.py | 1 + .../callable_from_qua/_callable_from_qua.py | 9 +- tests/conftest.py | 5 +- tests/test_bakery.py | 69 ++---- tests/test_config_builder.py | 47 ++--- tests/test_config_helper_tools.py | 28 +-- tests/test_digital_filters.py | 197 +++++++++++------- tests/test_fittinig_class.py | 29 +-- tests/test_octave_tools.py | 134 ++++++------ tests/test_simulator_tools.py | 32 +-- tests/test_time_units.py | 2 + tests/test_waveform_tools.py | 37 ++-- .../test_basic_qua_callable.py | 6 +- tests_against_server/conftest.py | 9 +- tests_against_server/test_bakery_server.py | 25 +-- .../test_digital_filters_server.py | 21 +- tests_against_server/test_long_wait.py | 62 +++--- tests_against_server/test_loops.py | 155 +++++++------- tests_against_server/test_multi_user.py | 8 +- 25 files changed, 499 insertions(+), 611 deletions(-) diff --git a/examples/Qcodes_drivers/advanced-driver/advanced_driver.py b/examples/Qcodes_drivers/advanced-driver/advanced_driver.py index 635f557f..955e7471 100644 --- a/examples/Qcodes_drivers/advanced-driver/advanced_driver.py +++ b/examples/Qcodes_drivers/advanced-driver/advanced_driver.py @@ -134,12 +134,8 @@ def __init__( def update_readout_parameters(self): """Update the config with the readout length and amplitude set with self.readout_pulse_length() and self.readout_pulse_amplitude().""" if self.readout_pulse_length() % 4 != 0: - raise ValueError( - "The readout pulse length must be an integer multiple of 4ns." - ) - self.config["pulses"]["readout_pulse"]["length"] = int( - self.readout_pulse_length() - ) + raise ValueError("The readout pulse length must be an integer multiple of 4ns.") + self.config["pulses"]["readout_pulse"]["length"] = int(self.readout_pulse_length()) self.config["integration_weights"]["cosine_weights"]["cosine"] = [ (1.0, self.config["pulses"]["readout_pulse"]["length"]) ] @@ -159,9 +155,7 @@ def update_readout_parameters(self): (-1.0, self.config["pulses"]["readout_pulse"]["length"]) ] # update readout pulse amplitude - pulse = self.config["elements"][self.readout_element()]["operations"][ - self.readout_operation() - ] + pulse = self.config["elements"][self.readout_element()]["operations"][self.readout_operation()] wf = self.config["pulses"][pulse]["waveforms"]["I"] self.config["waveforms"][wf]["sample"] = self.readout_pulse_amplitude() # Open QM @@ -200,10 +194,7 @@ def measurement_declaration(self): Q1_arr, Q2_arr, ] - elif ( - self.acquisition_mode() == "full_integration" - or self.acquisition_mode() == "full_demodulation" - ): + elif self.acquisition_mode() == "full_integration" or self.acquisition_mode() == "full_demodulation": I = declare(fixed) Q = declare(fixed) I_st = declare_stream() @@ -288,76 +279,49 @@ def stream_results(self): """Create the stream processing based on the measurement mode (self.acquisition_mode()) and OPX scan 0d, 1d or 2d (self.opx_scan())""" variables = self.measurement_variables if self.opx_scan() == "0d": - if ( - self.acquisition_mode() == "full_integration" - or self.acquisition_mode() == "full_demodulation" - ): + if self.acquisition_mode() == "full_integration" or self.acquisition_mode() == "full_demodulation": variables[2].buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("I") variables[3].buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("Q") elif self.acquisition_mode() == "raw_adc": - variables.input1().buffer(self.n_avg()).map( - FUNCTIONS.average() - ).save_all("adc1") - variables.input2().buffer(self.n_avg()).map( - FUNCTIONS.average() - ).save_all("adc2") - elif ( - self.acquisition_mode() == "sliced_integration" - or self.acquisition_mode() == "sliced_demodulation" - ): - variables[2].buffer(variables[5]).buffer(self.n_avg()).map( - FUNCTIONS.average() - ).save_all("I") - variables[3].buffer(variables[5]).buffer(self.n_avg()).map( - FUNCTIONS.average() - ).save_all("Q") + variables.input1().buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("adc1") + variables.input2().buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("adc2") + elif self.acquisition_mode() == "sliced_integration" or self.acquisition_mode() == "sliced_demodulation": + variables[2].buffer(variables[5]).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("I") + variables[3].buffer(variables[5]).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("Q") elif self.opx_scan() == "1d": - if ( - self.acquisition_mode() == "full_integration" - or self.acquisition_mode() == "full_demodulation" - ): - variables[2].buffer(self.axis1_npoints()).buffer(self.n_avg()).map( + if self.acquisition_mode() == "full_integration" or self.acquisition_mode() == "full_demodulation": + variables[2].buffer(self.axis1_npoints()).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("I") + variables[3].buffer(self.axis1_npoints()).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("Q") + elif self.acquisition_mode() == "raw_adc": + variables.buffer(self.axis1_npoints()).input1().buffer(self.n_avg()).map(FUNCTIONS.average()).save_all( + "adc1" + ) + variables.buffer(self.axis1_npoints()).input2().buffer(self.n_avg()).map(FUNCTIONS.average()).save_all( + "adc2" + ) + elif self.acquisition_mode() == "sliced_integration" or self.acquisition_mode() == "sliced_demodulation": + variables[2].buffer(variables[5]).buffer(self.axis2_npoints()).buffer(self.n_avg()).map( FUNCTIONS.average() ).save_all("I") - variables[3].buffer(self.axis1_npoints()).buffer(self.n_avg()).map( + variables[3].buffer(variables[5]).buffer(self.axis2_npoints()).buffer(self.n_avg()).map( FUNCTIONS.average() ).save_all("Q") - elif self.acquisition_mode() == "raw_adc": - variables.buffer(self.axis1_npoints()).input1().buffer( - self.n_avg() - ).map(FUNCTIONS.average()).save_all("adc1") - variables.buffer(self.axis1_npoints()).input2().buffer( - self.n_avg() - ).map(FUNCTIONS.average()).save_all("adc2") - elif ( - self.acquisition_mode() == "sliced_integration" - or self.acquisition_mode() == "sliced_demodulation" - ): - variables[2].buffer(variables[5]).buffer(self.axis2_npoints()).buffer( - self.n_avg() - ).map(FUNCTIONS.average()).save_all("I") - variables[3].buffer(variables[5]).buffer(self.axis2_npoints()).buffer( - self.n_avg() - ).map(FUNCTIONS.average()).save_all("Q") elif self.opx_scan() == "2d": - if ( - self.acquisition_mode() == "full_integration" - or self.acquisition_mode() == "full_demodulation" - ): + if self.acquisition_mode() == "full_integration" or self.acquisition_mode() == "full_demodulation": if self.outer_averaging_loop(): - variables[2].buffer(self.axis1_npoints()).buffer( - self.axis2_npoints() - ).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("I") - variables[3].buffer(self.axis1_npoints()).buffer( - self.axis2_npoints() - ).buffer(self.n_avg()).map(FUNCTIONS.average()).save_all("Q") - else: - variables[2].buffer(self.axis1_npoints()).buffer(self.n_avg()).map( + variables[2].buffer(self.axis1_npoints()).buffer(self.axis2_npoints()).buffer(self.n_avg()).map( FUNCTIONS.average() - ).buffer(self.axis2_npoints()).save_all("I") - variables[3].buffer(self.axis1_npoints()).buffer(self.n_avg()).map( + ).save_all("I") + variables[3].buffer(self.axis1_npoints()).buffer(self.axis2_npoints()).buffer(self.n_avg()).map( FUNCTIONS.average() - ).buffer(self.axis2_npoints()).save_all("Q") + ).save_all("Q") + else: + variables[2].buffer(self.axis1_npoints()).buffer(self.n_avg()).map(FUNCTIONS.average()).buffer( + self.axis2_npoints() + ).save_all("I") + variables[3].buffer(self.axis1_npoints()).buffer(self.n_avg()).map(FUNCTIONS.average()).buffer( + self.axis2_npoints() + ).save_all("Q") else: raise Exception( "It is advises to use 'full_integration' or 'full_demodulation' only for performing 2d scans on the OPX to avoid memory overflow." diff --git a/examples/Qcodes_drivers/advanced-driver/hello_qcodes_advanced.py b/examples/Qcodes_drivers/advanced-driver/hello_qcodes_advanced.py index 22e7cc3d..10e32c14 100644 --- a/examples/Qcodes_drivers/advanced-driver/hello_qcodes_advanced.py +++ b/examples/Qcodes_drivers/advanced-driver/hello_qcodes_advanced.py @@ -22,15 +22,11 @@ qc.config.core.db_location = db_file_path initialise_or_create_database_at(db_file_path) # Initialize qcodes experiment -experiment = load_or_create_experiment( - experiment_name=exp_name, sample_name=sample_name -) +experiment = load_or_create_experiment(experiment_name=exp_name, sample_name=sample_name) # Initialize the qcodes station to which instruments will be added station = qc.Station() # Create the OPX instrument class -opx_instrument = OPXCustomSequence( - config, name="OPX_demo", host="127.0.0.1", cluster_name="my_cluster" -) +opx_instrument = OPXCustomSequence(config, name="OPX_demo", host="127.0.0.1", cluster_name="my_cluster") # Add the OPX instrument to the qcodes station station.add_component(opx_instrument) @@ -143,9 +139,7 @@ def custom_sequence(self): opx_instrument.outer_averaging_loop(False) opx_instrument.n_avg(n_avg) # Axis1 is the most inner loop - opx_instrument.set_sweep_parameters( - "axis1", pulse_lengths * 4, "ns", "pulse duration" - ) + opx_instrument.set_sweep_parameters("axis1", pulse_lengths * 4, "ns", "pulse duration") # Axis2 is the second loop opx_instrument.set_sweep_parameters( "axis2", diff --git a/examples/Qcodes_drivers/basic-driver/hello_qcodes.py b/examples/Qcodes_drivers/basic-driver/hello_qcodes.py index 7f13041d..21c9528b 100644 --- a/examples/Qcodes_drivers/basic-driver/hello_qcodes.py +++ b/examples/Qcodes_drivers/basic-driver/hello_qcodes.py @@ -21,9 +21,7 @@ qc.config.core.db_location = db_file_path initialise_or_create_database_at(db_file_path) # Initialize qcodes experiment -experiment = load_or_create_experiment( - experiment_name=exp_name, sample_name=sample_name -) +experiment = load_or_create_experiment(experiment_name=exp_name, sample_name=sample_name) # Initialize the qcodes station to which instruments will be added station = qc.Station() # Create the OPX instrument class @@ -205,12 +203,8 @@ def OPX_1d_scan(simulate=False): ramp_to_zero("gate_1") with stream_processing(): - I_st.buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all( - "I" - ) - Q_st.buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all( - "Q" - ) + I_st.buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all("I") + Q_st.buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all("Q") return prog @@ -235,9 +229,7 @@ def OPX_1d_scan(simulate=False): do0d( opx_instrument.run_exp, opx_instrument.resume, - opx_instrument.get_measurement_parameter( - scale_factor=[("I", 1235, "pA"), ("Q", 1235, "pA")] - ), + opx_instrument.get_measurement_parameter(scale_factor=[("I", 1235, "pA"), ("Q", 1235, "pA")]), opx_instrument.halt, do_plot=True, exp=experiment, @@ -369,12 +361,8 @@ def OPX_sliced_scan(simulate=False): ramp_to_zero("gate_1") with stream_processing(): - I_st.buffer(nb_of_slices).buffer(len(biases)).buffer(n_avg).map( - FUNCTIONS.average() - ).save_all("I") - Q_st.buffer(nb_of_slices).buffer(len(biases)).buffer(n_avg).map( - FUNCTIONS.average() - ).save_all("Q") + I_st.buffer(nb_of_slices).buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all("I") + Q_st.buffer(nb_of_slices).buffer(len(biases)).buffer(n_avg).map(FUNCTIONS.average()).save_all("Q") return prog @@ -444,12 +432,8 @@ def OPX_reflectometry(simulate=False): save(Q, Q_st) with stream_processing(): - I_st.buffer(len(frequencies)).buffer(n_avg).map( - FUNCTIONS.average() - ).save_all("I") - Q_st.buffer(len(frequencies)).buffer(n_avg).map( - FUNCTIONS.average() - ).save_all("Q") + I_st.buffer(len(frequencies)).buffer(n_avg).map(FUNCTIONS.average()).save_all("I") + Q_st.buffer(len(frequencies)).buffer(n_avg).map(FUNCTIONS.average()).save_all("Q") return prog @@ -485,9 +469,7 @@ def OPX_reflectometry(simulate=False): # Get the results from the OPX data = opx_instrument.get_res() # Store the results in the scodes database - datasaver.add_result( - (VP1, VP1()), (OPX_param, np.array(list(data.values()))) - ) + datasaver.add_result((VP1, VP1()), (OPX_param, np.array(list(data.values())))) # Halt the OPX program at the end opx_instrument.halt() # Get the full dataset diff --git a/examples/Qcodes_drivers/stability-diagram/raster_scan.py b/examples/Qcodes_drivers/stability-diagram/raster_scan.py index 21b69501..15ded630 100644 --- a/examples/Qcodes_drivers/stability-diagram/raster_scan.py +++ b/examples/Qcodes_drivers/stability-diagram/raster_scan.py @@ -21,9 +21,7 @@ qc.config.core.db_location = db_file_path initialise_or_create_database_at(db_file_path) # Initialize qcodes experiment -experiment = load_or_create_experiment( - experiment_name=exp_name, sample_name=sample_name -) +experiment = load_or_create_experiment(experiment_name=exp_name, sample_name=sample_name) # Initialize the qcodes station to which instruments will be added station = qc.Station() # Create the OPX instrument class @@ -130,12 +128,8 @@ def raster_scan(simulate=False): save(I, I_st) save(Q, Q_st) with stream_processing(): - I_st.buffer(N1).buffer(N2).buffer(n_avg).map(FUNCTIONS.average()).save_all( - "I" - ) - Q_st.buffer(N1).buffer(N2).buffer(n_avg).map(FUNCTIONS.average()).save_all( - "Q" - ) + I_st.buffer(N1).buffer(N2).buffer(n_avg).map(FUNCTIONS.average()).save_all("I") + Q_st.buffer(N1).buffer(N2).buffer(n_avg).map(FUNCTIONS.average()).save_all("Q") return prog diff --git a/examples/Qcodes_drivers/stability-diagram/spiral_scan.py b/examples/Qcodes_drivers/stability-diagram/spiral_scan.py index 721399e5..73d28418 100644 --- a/examples/Qcodes_drivers/stability-diagram/spiral_scan.py +++ b/examples/Qcodes_drivers/stability-diagram/spiral_scan.py @@ -96,41 +96,24 @@ def get_res(self): # Get data and convert to Volt out = None # demodulated or integrated data - self.result_handles.get(self.results["names"][i]).wait_for_values( - self.counter - ) + self.result_handles.get(self.results["names"][i]).wait_for_values(self.counter) if self.results["types"][i] == "IQ": out = ( - -( - self.result_handles.get(self.results["names"][i]).fetch( - self.counter - 1 - )["value"] - ) + -(self.result_handles.get(self.results["names"][i]).fetch(self.counter - 1)["value"]) * 4096 / self.readout_pulse_length() ) # raw adc traces elif self.results["types"][i] == "adc": - out = ( - -( - self.result_handles.get(self.results["names"][i]).fetch( - self.counter - 1 - )["value"] - ) - / 4096 - ) + out = -(self.result_handles.get(self.results["names"][i]).fetch(self.counter - 1)["value"]) / 4096 # Reshape data out = out.reshape(self.results["buffers"][i][0]) - output[self.results["names"][i]] = out[ - self.spiral(int(np.sqrt(self.results["buffers"][i][0]))) - ] + output[self.results["names"][i]] = out[self.spiral(int(np.sqrt(self.results["buffers"][i][0])))] # Add amplitude and phase if I and Q are in the SP if "I" in output.keys() and "Q" in output.keys(): output["R"] = np.sqrt(output["I"] ** 2 + output["Q"] ** 2) - output["Phi"] = ( - np.unwrap(np.angle(output["I"] + 1j * output["Q"])) * 180 / np.pi - ) + output["Phi"] = np.unwrap(np.angle(output["I"] + 1j * output["Q"])) * 180 / np.pi return output def get_measurement_parameter(self, scale_factor=((),)): @@ -164,9 +147,7 @@ def get_measurement_parameter(self, scale_factor=((),)): self.axis1_stop(int(self.readout_pulse_length())) self.axis1_step(1) self.axis1_npoints(int(self.readout_pulse_length())) - self.axis1_full_list( - np.arange(self.axis1_start(), self.axis1_stop(), self.axis1_step()) - ) + self.axis1_full_list(np.arange(self.axis1_start(), self.axis1_stop(), self.axis1_step())) self.axis1_axis.unit = "ns" self.axis1_axis.label = "Readout time" @@ -176,12 +157,8 @@ def get_measurement_parameter(self, scale_factor=((),)): if len(scale_factor[0]) == 3: for param in scale_factor: if param[0] in self.results["names"]: - self.results["units"][ - self.results["names"].index(param[0]) - ] = param[2] - self.results["scale_factor"][ - self.results["names"].index(param[0]) - ] = param[1] + self.results["units"][self.results["names"].index(param[0])] = param[2] + self.results["scale_factor"][self.results["names"].index(param[0])] = param[1] else: raise ValueError( "scale_factor must be a list of tuples with 3 elements (the result name, the scale factor and the new unit), as in [('I', 0.152, 'pA'), ]." @@ -192,14 +169,10 @@ def get_measurement_parameter(self, scale_factor=((),)): "OPX_results", names=self.results["names"], units=self.results["units"], - shapes=((self.results["buffers"][0][0], self.results["buffers"][0][0]),) - * len(self.results["names"]), - setpoints=((self.axis2_axis(), self.axis1_axis()),) - * len(self.results["names"]), - setpoint_units=((self.axis2_axis.unit, self.axis1_axis.unit),) - * len(self.results["names"]), - setpoint_labels=((self.axis2_axis.label, self.axis1_axis.label),) - * len(self.results["names"]), + shapes=((self.results["buffers"][0][0], self.results["buffers"][0][0]),) * len(self.results["names"]), + setpoints=((self.axis2_axis(), self.axis1_axis()),) * len(self.results["names"]), + setpoint_units=((self.axis2_axis.unit, self.axis1_axis.unit),) * len(self.results["names"]), + setpoint_labels=((self.axis2_axis.label, self.axis1_axis.label),) * len(self.results["names"]), ) @@ -234,15 +207,11 @@ def measurement_macro(measured_element, I, I_stream, Q, Q_stream): qc.config.core.db_location = db_file_path initialise_or_create_database_at(db_file_path) # Initialize qcodes experiment -experiment = load_or_create_experiment( - experiment_name=exp_name, sample_name=sample_name -) +experiment = load_or_create_experiment(experiment_name=exp_name, sample_name=sample_name) # Initialize the qcodes station to which instruments will be added station = qc.Station() # Create the OPX instrument class -opx_instrument = OPXCustomSequence( - config, name="OPX", host=qop_ip, cluster_name=cluster_name -) +opx_instrument = OPXCustomSequence(config, name="OPX", host=qop_ip, cluster_name=cluster_name) # Add the OPX instrument to the qcodes station station.add_component(opx_instrument) @@ -408,9 +377,7 @@ def spiral_scan(x_element, y_element, readout_element, simulate=False): # Axis2 is the second loop opx_instrument.set_sweep_parameters("axis2", Vx_setpoints, "V", "Gate 1 biases") # Add the custom sequence to the OPX -opx_instrument.qua_program = spiral_scan( - "gate_1", "gate_2", "readout_element", simulate=True -) +opx_instrument.qua_program = spiral_scan("gate_1", "gate_2", "readout_element", simulate=True) # Simulate program opx_instrument.sim_time(100_000) # opx_instrument.simulate() @@ -419,9 +386,7 @@ def spiral_scan(x_element, y_element, readout_element, simulate=False): # Set the dc offset to the center of the spiral (these must be set to the slow voltage source) # qm.set_output_dc_offset_by_element("gate_2", "single", Vy_center()) # qm.set_output_dc_offset_by_element("gate_1", "single", Vx_center()) -opx_instrument.qua_program = spiral_scan( - "gate_1", "gate_2", "readout_element", simulate=False -) +opx_instrument.qua_program = spiral_scan("gate_1", "gate_2", "readout_element", simulate=False) do0d( opx_instrument.run_exp, opx_instrument.resume, diff --git a/examples/callable_from_qua/qua_print.py b/examples/callable_from_qua/qua_print.py index 56a411eb..5b2b01bd 100644 --- a/examples/callable_from_qua/qua_print.py +++ b/examples/callable_from_qua/qua_print.py @@ -6,6 +6,7 @@ # Patch to add the callable from qua functions to the main SDK patch_qua_program_addons() + # Define your callable_from_qua functions @callable_from_qua def qua_print(*args): diff --git a/examples/callable_from_qua/update_other_instruments.py b/examples/callable_from_qua/update_other_instruments.py index ffff6220..cb335029 100644 --- a/examples/callable_from_qua/update_other_instruments.py +++ b/examples/callable_from_qua/update_other_instruments.py @@ -9,6 +9,7 @@ # Patch to add the callable from qua functions to the main SDK patch_qua_program_addons() + # Define your callable_from_qua functions # Note that this framework can be easily adapted to update parameters from other instruments using their dedicated API @callable_from_qua diff --git a/qualang_tools/callable_from_qua/_callable_from_qua.py b/qualang_tools/callable_from_qua/_callable_from_qua.py index 760366eb..69b3e2d4 100644 --- a/qualang_tools/callable_from_qua/_callable_from_qua.py +++ b/qualang_tools/callable_from_qua/_callable_from_qua.py @@ -79,16 +79,13 @@ def run(self, job: QmJob): class ProgramAddon(ABC): @abstractmethod - def enter_program(self, program: Program): - ... + def enter_program(self, program: Program): ... @abstractmethod - def exit_program(self, exc_type, exc_val, exc_tb): - ... + def exit_program(self, exc_type, exc_val, exc_tb): ... @abstractmethod - def execute_program(self, program: Program, quantum_machine: QuantumMachine): - ... + def execute_program(self, program: Program, quantum_machine: QuantumMachine): ... class QuaCallableEventManager(ProgramAddon): diff --git a/tests/conftest.py b/tests/conftest.py index 4071d4f2..de271287 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,6 +18,7 @@ def ignore_deprecation_warnings(): @pytest.fixture(autouse=True) def capability_container(): container = create_capabilities_container(QuaMachineInfo([], ImplementationInfo("", "", ""))) - container.capabilities.override(ServerCapabilities(True, True, True, True, True, True, True, True, True, True, True, - True, True, True, 1, True)) + container.capabilities.override( + ServerCapabilities(True, True, True, True, True, True, True, True, True, True, True, True, True, True, 1, True) + ) return container diff --git a/tests/test_bakery.py b/tests/test_bakery.py index 6747c290..2351de1e 100644 --- a/tests/test_bakery.py +++ b/tests/test_bakery.py @@ -27,9 +27,7 @@ def IQ_imbalance(g, phi): c = np.cos(phi) s = np.sin(phi) N = 1 / ((1 - g**2) * (2 * c**2 - 1)) - return [ - float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c] - ] + return [float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c]] return { "version": 1, @@ -169,7 +167,6 @@ def test_override_waveform(config): assert b_new.get_op_length("qe3") == ref_length2 - def test_out_boolean(config): cfg = deepcopy(config) with baking(cfg) as b: @@ -219,37 +216,13 @@ def test_indices_behavior(config): b1.play("gaussOp", "qe2") b1.play("gaussOp", "qe3") - assert all( - [ - cfg["waveforms"]["qe2_baked_wf_I_0"]["samples"][i] - == gauss(0.2, 0, 15, 80)[i] - for i in range(80) - ] - ) - assert all( - [ - cfg["waveforms"]["qe3_baked_wf_I_0"]["samples"][i] - == gauss(0.2, 0, 15, 80)[i] - for i in range(80) - ] - ) + assert all([cfg["waveforms"]["qe2_baked_wf_I_0"]["samples"][i] == gauss(0.2, 0, 15, 80)[i] for i in range(80)]) + assert all([cfg["waveforms"]["qe3_baked_wf_I_0"]["samples"][i] == gauss(0.2, 0, 15, 80)[i] for i in range(80)]) with b1: b1.play("gaussOp", "qe2", amp=2) b1.play("gaussOp", "qe3", amp=2) - assert all( - [ - cfg["waveforms"]["qe2_baked_wf_I_0"]["samples"][i] - == gauss(0.4, 0, 15, 80)[i] - for i in range(80) - ] - ) - assert all( - [ - cfg["waveforms"]["qe3_baked_wf_I_0"]["samples"][i] - == gauss(0.4, 0, 15, 80)[i] - for i in range(80) - ] - ) + assert all([cfg["waveforms"]["qe2_baked_wf_I_0"]["samples"][i] == gauss(0.4, 0, 15, 80)[i] for i in range(80)]) + assert all([cfg["waveforms"]["qe3_baked_wf_I_0"]["samples"][i] == gauss(0.4, 0, 15, 80)[i] for i in range(80)]) def test_play_at_negative_t(config): @@ -267,12 +240,8 @@ def test_play_at_negative_t(config): b.play("Op1", "qe3") # The baked waveform is at this point I: [0.3, 0.3, 0.3, 0.3, 0.3] # Q: [0.2, 0.2, 0.2, 0.3, 0.3] - b.play_at( - "Op2", "qe2", t=-2 - ) # t indicates the time index where these new samples should be added - b.play_at( - "Op2", "qe3", t=-2 - ) # t indicates the time index where these new samples should be added + b.play_at("Op2", "qe2", t=-2) # t indicates the time index where these new samples should be added + b.play_at("Op2", "qe3", t=-2) # t indicates the time index where these new samples should be added # The baked waveform is now I: [0.3, 0.3, 0.3, 0.4, 0.4, 0.1, 0.1] # Q: [0.2, 0.2, 0.2, 0.4, 0.4, 0.1, 0.1] assert np.array_equal( @@ -302,12 +271,8 @@ def test_negative_wait(config): # Q: [0.2, 0.2, 0.2, 0.3, 0.3] b.wait(-3, "qe2") b.wait(-3, "qe3") - b.play( - "Op2", "qe2" - ) # t indicates the time index where these new samples should be added - b.play( - "Op2", "qe3" - ) # t indicates the time index where these new samples should be added + b.play("Op2", "qe2") # t indicates the time index where these new samples should be added + b.play("Op2", "qe3") # t indicates the time index where these new samples should be added # The baked waveform is now I: [0.3, 0.3, 0.3, 0.4, 0.4, 0.1, 0.1] # Q: [0.2, 0.2, 0.2, 0.4, 0.4, 0.1, 0.1] assert np.array_equal( @@ -339,12 +304,8 @@ def test_play_at_negative_t_too_large(config): Exception, match="too large for current baked samples length", ): - b.play_at( - "Op2", "qe2", t=-6 - ) # t indicates the time index where these new samples should be added - b.play_at( - "Op2", "qe3", t=-6 - ) # t indicates the time index where these new samples should be added + b.play_at("Op2", "qe2", t=-6) # t indicates the time index where these new samples should be added + b.play_at("Op2", "qe3", t=-6) # t indicates the time index where these new samples should be added # The baked waveform is now I: [0.3, 0.3, 0.3, 0.4, 0.4, 0.1, 0.1] # Q: [0.2, 0.2, 0.2, 0.4, 0.4, 0.1, 0.1] @@ -370,12 +331,8 @@ def test_negative_wait_too_large(config): ): b.wait(-6, "qe2") b.wait(-6, "qe3") - b.play( - "Op2", "qe2" - ) # t indicates the time index where these new samples should be added - b.play( - "Op2", "qe3" - ) # t indicates the time index where these new samples should be added + b.play("Op2", "qe2") # t indicates the time index where these new samples should be added + b.play("Op2", "qe3") # t indicates the time index where these new samples should be added # The baked waveform is now I: [0.3, 0.3, 0.3, 0.4, 0.4, 0.1, 0.1] # Q: [0.2, 0.2, 0.2, 0.4, 0.4, 0.1, 0.1] diff --git a/tests/test_config_builder.py b/tests/test_config_builder.py index 6e3ba4fe..74d94e8b 100644 --- a/tests/test_config_builder.py +++ b/tests/test_config_builder.py @@ -24,20 +24,10 @@ def config_resonator(): ] ro_pulse = MeasurePulse("ro_pulse", wfs, 16) - ro_pulse.add( - Weights(ConstantIntegrationWeights("integ_w1_I", cosine=1, sine=0, duration=16)) - ) - ro_pulse.add( - Weights( - ConstantIntegrationWeights("integ_w1_Q", cosine=0, sine=-1, duration=16) - ) - ) - ro_pulse.add( - Weights(ConstantIntegrationWeights("integ_w2_I", cosine=0, sine=1, duration=16)) - ) - ro_pulse.add( - Weights(ConstantIntegrationWeights("integ_w2_Q", cosine=1, sine=0, duration=16)) - ) + ro_pulse.add(Weights(ConstantIntegrationWeights("integ_w1_I", cosine=1, sine=0, duration=16))) + ro_pulse.add(Weights(ConstantIntegrationWeights("integ_w1_Q", cosine=0, sine=-1, duration=16))) + ro_pulse.add(Weights(ConstantIntegrationWeights("integ_w2_I", cosine=0, sine=1, duration=16))) + ro_pulse.add(Weights(ConstantIntegrationWeights("integ_w2_Q", cosine=1, sine=0, duration=16))) res.add(Operation(ro_pulse)) @@ -295,17 +285,9 @@ def config_3qb_3res(): ro_pulse = MeasurePulse("ro_pulse", [ro_I, ro_Q], 100) - ro_pulse.add( - Weights(ConstantIntegrationWeights("cos", cosine=1, sine=0, duration=100)) - ) - ro_pulse.add( - Weights( - ConstantIntegrationWeights("minus_sin", cosine=0, sine=-1, duration=100) - ) - ) - ro_pulse.add( - Weights(ConstantIntegrationWeights("sin", cosine=0, sine=1, duration=100)) - ) + ro_pulse.add(Weights(ConstantIntegrationWeights("cos", cosine=1, sine=0, duration=100))) + ro_pulse.add(Weights(ConstantIntegrationWeights("minus_sin", cosine=0, sine=-1, duration=100))) + ro_pulse.add(Weights(ConstantIntegrationWeights("sin", cosine=0, sine=1, duration=100))) res1.add(Operation(ro_pulse)) res2.add(Operation(ro_pulse)) @@ -363,17 +345,17 @@ def test_parameter_algebra(): assert (d**c)() == 10000 assert (c**2)() == 16 - e = c_vars.parameter("e", setter=lambda:[1, 2, 3]) + e = c_vars.parameter("e", setter=lambda: [1, 2, 3]) assert type(e) == Parameter assert e.len() == 3 + def test_deprecated_warning(): with pytest.warns(None) as record: cb = ConfigBuilder() cont = Controller("con1") cb.add(cont) - elm = Element("elm", analog_input_ports=[cont.analog_output(1)] - ) + elm = Element("elm", analog_input_ports=[cont.analog_output(1)]) cb.add(elm) cb.build() @@ -383,21 +365,20 @@ def test_deprecated_warning(): cb = ConfigBuilder() cont = Controller("con1") cb.add(cont) - elm = Element("elm", element_analog_inputs=[cont.analog_output(1)] - ) + elm = Element("elm", element_analog_inputs=[cont.analog_output(1)]) cb.add(elm) cb.build() assert len(record.list) == 1 + def test_digital_input(): cont = Controller("con1") - elm = Element("elm", - analog_input_ports=[cont.analog_output(1)], - digital_input_ports=[cont.digital_output(2)]) + elm = Element("elm", analog_input_ports=[cont.analog_output(1)], digital_input_ports=[cont.digital_output(2)]) elm.set_digital_input_delay(2, 20) assert "in2" in elm.dict["digitalInputs"].keys() assert elm.dict["digitalInputs"]["in2"]["delay"] == 20 + def test_element_none_arguments(): cont = Controller("con1") elm = Element("elm", digital_input_ports=[cont.digital_output(2)]) diff --git a/tests/test_config_helper_tools.py b/tests/test_config_helper_tools.py index f60b68e2..dcf5b6da 100644 --- a/tests/test_config_helper_tools.py +++ b/tests/test_config_helper_tools.py @@ -11,9 +11,7 @@ def IQ_imbalance(g, phi): c = np.cos(phi) s = np.sin(phi) N = 1 / ((1 - g**2) * (2 * c**2 - 1)) - return [ - float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c] - ] + return [float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c]] return { "version": 1, @@ -151,18 +149,14 @@ def test_update_integration_weight(config0): conf = deepcopy(config0) config = QuaConfig(conf) # Test update with only one operation using these iw - config.update_integration_weight( - "resonator", "readout", "cos", [(1, 180)], [(0, 180)] - ) + config.update_integration_weight("resonator", "readout", "cos", [(1, 180)], [(0, 180)]) assert conf["integration_weights"]["cosine_weights"]["cosine"][0] == (1, 180) assert conf["integration_weights"]["cosine_weights"]["sine"][0] == (0, 180) # Add another operation using the same iw config.copy_operation("resonator", "readout", "short_readout") # Test update with two operations using these iw anf force_update=False try: - config.update_integration_weight( - "resonator", "readout", "cos", [(1, 80)], [(0, 80)] - ) + config.update_integration_weight("resonator", "readout", "cos", [(1, 80)], [(0, 80)]) except Exception as e: assert ( e.__str__() @@ -170,9 +164,7 @@ def test_update_integration_weight(config0): ) assert conf["integration_weights"]["cosine_weights"]["cosine"][0] == (1, 180) # Test update with two operations using these iw anf force_update=True - config.update_integration_weight( - "resonator", "readout", "cos", [(1, 80)], [(0, 80)], True - ) + config.update_integration_weight("resonator", "readout", "cos", [(1, 80)], [(0, 80)], True) assert conf["integration_weights"]["cosine_weights"]["cosine"][0] == (1, 80) assert conf["integration_weights"]["cosine_weights"]["sine"][0] == (0, 80) @@ -181,9 +173,7 @@ def test_add_control_operation_iq(config0): conf = deepcopy(config0) config = QuaConfig(conf) - config.add_control_operation_iq( - "qubit", "gate", list(gaussian(112, 20)), [0.0 for _ in range(112)] - ) + config.add_control_operation_iq("qubit", "gate", list(gaussian(112, 20)), [0.0 for _ in range(112)]) assert "gate" in config["elements"]["qubit"]["operations"] assert "qubit_gate_pulse" in conf["pulses"] assert "qubit_gate_wf_i" in conf["waveforms"] @@ -213,9 +203,7 @@ def test_update_update_waveforms(config0): assert config["waveforms"]["flux_line_bias_single_wf"]["sample"] == 1.1 assert config["pulses"]["flux_line_bias_pulse"]["length"] == 175 config.update_waveforms("flux_line", "bias", (list(gaussian(112, 20)),)) - assert config["waveforms"]["flux_line_bias_single_wf"]["samples"] == list( - gaussian(112, 20) - ) + assert config["waveforms"]["flux_line_bias_single_wf"]["samples"] == list(gaussian(112, 20)) def test_transform_negative_delays(negative_delay_config): @@ -246,9 +234,7 @@ def test_transform_negative_delays(negative_delay_config): "mixers": {}, } - test_config = transform_negative_delays( - negative_delay_config, create_new_config=True - ) + test_config = transform_negative_delays(negative_delay_config, create_new_config=True) # Test that initial config has not been changed assert initial_config == negative_delay_config # Test that the updated config is correct diff --git a/tests/test_digital_filters.py b/tests/test_digital_filters.py index 2a1f0d56..3f3e2fc5 100644 --- a/tests/test_digital_filters.py +++ b/tests/test_digital_filters.py @@ -3,33 +3,43 @@ from qualang_tools.digital_filters import * -@pytest.mark.parametrize("low_pass_exp", [ - [500, -0.25, 200], - [500, 0.25, 200], -]) +@pytest.mark.parametrize( + "low_pass_exp", + [ + [500, -0.25, 200], + [500, 0.25, 200], + ], +) def test_exponential_decay(low_pass_exp): t_max = low_pass_exp[0] A = low_pass_exp[1] tau = low_pass_exp[2] t = np.arange(0, t_max, 1) - assert np.all(exponential_decay(t, A, tau) == 1 + A * np.exp(-t/tau)) + assert np.all(exponential_decay(t, A, tau) == 1 + A * np.exp(-t / tau)) -@pytest.mark.parametrize("high_pass_exp", [ - [500, 200], -]) + +@pytest.mark.parametrize( + "high_pass_exp", + [ + [500, 200], + ], +) def test_high_pass_exponential(high_pass_exp): t_max = high_pass_exp[0] tau = high_pass_exp[1] t = np.arange(0, t_max, 1) - assert np.all(high_pass_exponential(t, tau) == np.exp(-t/tau)) - - -@pytest.mark.parametrize("single_exp_correction", [ - [-0.25, 200, 1], - [0.25, 200, 1], - [0.25, 200, 2], - [-0.01, 2000, 1], -]) + assert np.all(high_pass_exponential(t, tau) == np.exp(-t / tau)) + + +@pytest.mark.parametrize( + "single_exp_correction", + [ + [-0.25, 200, 1], + [0.25, 200, 1], + [0.25, 200, 2], + [-0.01, 2000, 1], + ], +) def test_single_exp_correction(single_exp_correction): A = single_exp_correction[0] tau = single_exp_correction[1] @@ -42,15 +52,18 @@ def test_single_exp_correction(single_exp_correction): assert 0 <= feedback[0] # Low pass correction must be > 0 assert np.all(-2 < np.array(feedforward)) # Must be within (-2, 2) assert np.all(np.array(feedforward) < 2) # Must be within (-2, 2) - assert np.all(feedforward == single_exponential_correction(A, tau*2, 2*ts)[0]) - assert feedback == single_exponential_correction(A, tau*2, 2*ts)[1] - - -@pytest.mark.parametrize("hp_correction", [ - [20000, 1], - [2000000, 2], - [10, 1], -]) + assert np.all(feedforward == single_exponential_correction(A, tau * 2, 2 * ts)[0]) + assert feedback == single_exponential_correction(A, tau * 2, 2 * ts)[1] + + +@pytest.mark.parametrize( + "hp_correction", + [ + [20000, 1], + [2000000, 2], + [10, 1], + ], +) def test_hp_correction(hp_correction): tau = hp_correction[0] ts = hp_correction[1] @@ -61,21 +74,27 @@ def test_hp_correction(hp_correction): assert 0 <= feedback[0] # High pass correction must be > 0 assert np.all(-2 < np.array(feedforward)) # Must be within (-2, 2) assert np.all(np.array(feedforward) < 2) # Must be within (-2, 2) - assert np.all(feedforward == highpass_correction(tau/2, 2*ts)[0]) - assert feedback == highpass_correction(tau/2, 2*ts)[1] - -@pytest.mark.parametrize("calc_correction", [ - [[(0.25, 200)], None, 1], - [[(-0.25, 200)], None, 1], - [[(-0.25, 200)], None, 2], - [None, [20_000], 1], - [None, [20_000], 2], -]) + assert np.all(feedforward == highpass_correction(tau / 2, 2 * ts)[0]) + assert feedback == highpass_correction(tau / 2, 2 * ts)[1] + + +@pytest.mark.parametrize( + "calc_correction", + [ + [[(0.25, 200)], None, 1], + [[(-0.25, 200)], None, 1], + [[(-0.25, 200)], None, 2], + [None, [20_000], 1], + [None, [20_000], 2], + ], +) def test_single_calc_filter_taps(calc_correction): low_pass = calc_correction[0] high_pass = calc_correction[1] ts = calc_correction[2] - feedforward, feedback = calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts + ) print(f"\nfeedback: {feedback}") print(f"feedforward: {feedforward}") if low_pass is None: @@ -86,27 +105,36 @@ def test_single_calc_filter_taps(calc_correction): assert feedback[0] == single_exponential_correction(low_pass[0][0], low_pass[0][1], ts)[1] -@pytest.mark.parametrize("static_calc_correction", [ - [[(-0.25, 200)], None, 1, [[1.3322259136212624, -1.325581395348837], [0.9933554817275746]]], - [None, [20_000], 1, [[1.000025, -0.999975], [0.9999990463256836]]], -]) +@pytest.mark.parametrize( + "static_calc_correction", + [ + [[(-0.25, 200)], None, 1, [[1.3322259136212624, -1.325581395348837], [0.9933554817275746]]], + [None, [20_000], 1, [[1.000025, -0.999975], [0.9999990463256836]]], + ], +) def test_single_calc_filter_taps_static(static_calc_correction): low_pass = static_calc_correction[0] high_pass = static_calc_correction[1] ts = static_calc_correction[2] static_values = static_calc_correction[3] - feedforward, feedback = calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts + ) print(f"\nfeedback: {feedback}") print(f"feedforward: {feedforward}") assert np.all(feedforward == static_values[0]) assert feedback[0] == static_values[1][0] -@pytest.mark.parametrize("calc_correction", [ - [[(0.1, 100)], [10_000], 1, False], - [[(0.1, 100)], [10_000], 2, False], - [[(-0.1, 100)], [10_000], 1, True], - [[(-0.1, 100)], [10_000], 2, True], -]) + +@pytest.mark.parametrize( + "calc_correction", + [ + [[(0.1, 100)], [10_000], 1, False], + [[(0.1, 100)], [10_000], 2, False], + [[(-0.1, 100)], [10_000], 1, True], + [[(-0.1, 100)], [10_000], 2, True], + ], +) def test_double_calc_filter_taps(calc_correction): low_pass = calc_correction[0] high_pass = calc_correction[1] @@ -114,11 +142,13 @@ def test_double_calc_filter_taps(calc_correction): warning = calc_correction[3] if warning: with pytest.warns(expected_warning=UserWarning): - feedforward, feedback = calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=None, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts + ) else: - feedforward, feedback = calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=None, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts + ) print(f"\nfeedback: {feedback}") print(f"feedforward: {feedforward}") assert feedback[1] == highpass_correction(high_pass[0], ts)[1] @@ -126,34 +156,45 @@ def test_double_calc_filter_taps(calc_correction): assert np.all(np.abs(feedforward) < 2) -@pytest.mark.parametrize("calc_correction", [ - [None, None, 1, 10], - [None, None, 2, 10], - [None, None, 0.5, 11], - [[(0.1, 100)], [10_000], 1, 10], - [[(0.1, 100)], [10_000], 0.5, 10], -]) +@pytest.mark.parametrize( + "calc_correction", + [ + [None, None, 1, 10], + [None, None, 2, 10], + [None, None, 0.5, 11], + [[(0.1, 100)], [10_000], 1, 10], + [[(0.1, 100)], [10_000], 0.5, 10], + ], +) def test_delay_calc_filter_taps(calc_correction): low_pass = calc_correction[0] high_pass = calc_correction[1] ts = calc_correction[2] delay = calc_correction[3] - feedforward, feedback = calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=delay, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=delay, Ts=ts + ) print(f"\nfeedback: {feedback}") print(f"feedforward: {feedforward}") - assert feedforward[:int(delay/ts)] == [0.0] * int(delay/ts) - assert feedforward[int(delay/ts):] == calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=0, Ts=ts)[0] - assert feedback == calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=0, Ts=ts)[1] - -@pytest.mark.parametrize("calc_correction", [ - [None, None, 1], - [[(0.1, 100)], [10_000], 1], -]) + assert feedforward[: int(delay / ts)] == [0.0] * int(delay / ts) + assert ( + feedforward[int(delay / ts) :] + == calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=0, Ts=ts)[0] + ) + assert ( + feedback == calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=0, Ts=ts)[1] + ) + + +@pytest.mark.parametrize( + "calc_correction", + [ + [None, None, 1], + [[(0.1, 100)], [10_000], 1], + ], +) def test_fir_calc_filter_taps(calc_correction): from scipy.signal.windows import hann @@ -161,11 +202,15 @@ def test_fir_calc_filter_taps(calc_correction): high_pass = calc_correction[1] ts = calc_correction[2] - feedforward, feedback = calc_filter_taps(fir=hann(20)/np.sum(hann(20)), exponential=low_pass, highpass=high_pass, bounce=None, - delay=None, Ts=ts) + feedforward, feedback = calc_filter_taps( + fir=hann(20) / np.sum(hann(20)), exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts + ) print(f"\nfeedback: {feedback}") print(f"feedforward: {feedforward}") - assert feedforward == list(np.convolve(calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, - delay=None, Ts=ts)[0], hann(20)/np.sum(hann(20)))) - + assert feedforward == list( + np.convolve( + calc_filter_taps(fir=None, exponential=low_pass, highpass=high_pass, bounce=None, delay=None, Ts=ts)[0], + hann(20) / np.sum(hann(20)), + ) + ) diff --git a/tests/test_fittinig_class.py b/tests/test_fittinig_class.py index b98b96c4..95a6b7aa 100644 --- a/tests/test_fittinig_class.py +++ b/tests/test_fittinig_class.py @@ -3,6 +3,7 @@ from qualang_tools.plot.fitting import Fit + @pytest.mark.parametrize("a", [5, 10, -1]) @pytest.mark.parametrize("b", [1, 11, 0]) def test_linear(a, b): @@ -12,7 +13,8 @@ def test_linear(a, b): y = a * x + b out = fit.linear(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) + @pytest.mark.parametrize("amp", [0.2, 0.3, 0.4]) @pytest.mark.parametrize("T1", [10, 50, 100]) @@ -20,10 +22,10 @@ def test_linear(a, b): def test_T1(amp, T1, final_offset): fit = Fit() x = np.linspace(0, 100, 100) - y = amp * np.exp(-x * (1/T1)) + final_offset + y = amp * np.exp(-x * (1 / T1)) + final_offset out = fit.T1(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) @pytest.mark.parametrize("final_offset", [0.01, 0.05]) @@ -38,13 +40,12 @@ def test_ramsey(final_offset, T2, amp, initial_offset, f, phase): tau_max = 800 d_tau = 16 x = np.arange(tau_min, tau_max + 0.1, d_tau) - y = final_offset * (1 - np.exp(-x * (1/T2))) + amp / 2 * ( - np.exp(-x * (1/T2)) - * (initial_offset * 2 + np.cos(2 * np.pi * f * x + phase)) - ) + y = final_offset * (1 - np.exp(-x * (1 / T2))) + amp / 2 * ( + np.exp(-x * (1 / T2)) * (initial_offset * 2 + np.cos(2 * np.pi * f * x + phase)) + ) out = fit.ramsey(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) @pytest.mark.parametrize("kc", [3, 5]) @@ -54,10 +55,10 @@ def test_ramsey(final_offset, T2, amp, initial_offset, f, phase): def test_transmission_resonator_spectroscopy(kc, k, f, offset): fit = Fit() x = np.linspace(0, 100, 100) - y = ((kc/k) / (1 + (4 * ((x - f) ** 2) / (k ** 2)))) + offset + y = ((kc / k) / (1 + (4 * ((x - f) ** 2) / (k**2)))) + offset out = fit.transmission_resonator_spectroscopy(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) @pytest.mark.parametrize("kc", [3, 5]) @@ -68,10 +69,10 @@ def test_transmission_resonator_spectroscopy(kc, k, f, offset): def test_reflection_resonator_spectroscopy(kc, k, f, offset, slope): fit = Fit() x = np.linspace(0, 100, 100) - y = offset-((kc/k) / (1 + (4 * ((x - f) ** 2) / (k ** 2)))) + slope * x + y = offset - ((kc / k) / (1 + (4 * ((x - f) ** 2) / (k**2)))) + slope * x out = fit.reflection_resonator_spectroscopy(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) @pytest.mark.parametrize("offset", [0.01, 0.05]) @@ -85,8 +86,8 @@ def test_rabi(offset, T, amp, f, phase): tau_max = 800 d_tau = 16 x = np.arange(tau_min, tau_max + 0.1, d_tau) - y = amp * (np.sin(0.5 * (2 * np.pi * f) * x + phase))**2 * np.exp(-x / T) + offset + y = amp * (np.sin(0.5 * (2 * np.pi * f) * x + phase)) ** 2 * np.exp(-x / T) + offset out = fit.rabi(x_data=x, y_data=y) - np.testing.assert_allclose(out['fit_func'](x), y, rtol=1e-5, atol=1e-5) \ No newline at end of file + np.testing.assert_allclose(out["fit_func"](x), y, rtol=1e-5, atol=1e-5) diff --git a/tests/test_octave_tools.py b/tests/test_octave_tools.py index c71b1d72..e06659f0 100644 --- a/tests/test_octave_tools.py +++ b/tests/test_octave_tools.py @@ -6,83 +6,84 @@ from qualang_tools.octave_tools import get_calibration_parameters_from_db + @pytest.fixture def config(): return { - "version": 1, - "controllers": { - "con1": { - "analog_outputs": { - 1: {"offset": 0.0}, # I resonator - 2: {"offset": 0.0}, # Q resonator - 3: {"offset": 0.0}, # I qubit - 4: {"offset": 0.0}, # Q qubit - }, - "digital_outputs": {}, - "analog_inputs": { - 1: {"offset": 0.0, "gain_db": 0}, # I from down-conversion - 2: {"offset": 0.0, "gain_db": 0}, # Q from down-conversion + "version": 1, + "controllers": { + "con1": { + "analog_outputs": { + 1: {"offset": 0.0}, # I resonator + 2: {"offset": 0.0}, # Q resonator + 3: {"offset": 0.0}, # I qubit + 4: {"offset": 0.0}, # Q qubit + }, + "digital_outputs": {}, + "analog_inputs": { + 1: {"offset": 0.0, "gain_db": 0}, # I from down-conversion + 2: {"offset": 0.0, "gain_db": 0}, # Q from down-conversion + }, }, }, - }, - "elements": { - "qubit": { - "RF_inputs": {"port": ("octave1", 2)}, - "intermediate_frequency": 50e6, - "operations": { - "cw": "const_pulse", + "elements": { + "qubit": { + "RF_inputs": {"port": ("octave1", 2)}, + "intermediate_frequency": 50e6, + "operations": { + "cw": "const_pulse", + }, }, - }, - "resonator": { - "RF_inputs": {"port": ("octave1", 1)}, - "RF_outputs": {"port": ("octave1", 1)}, - "intermediate_frequency": -60e6, - "operations": { - "cw": "const_pulse", + "resonator": { + "RF_inputs": {"port": ("octave1", 1)}, + "RF_outputs": {"port": ("octave1", 1)}, + "intermediate_frequency": -60e6, + "operations": { + "cw": "const_pulse", + }, + "time_of_flight": 24, + "smearing": 0, }, - "time_of_flight": 24, - "smearing": 0, }, - }, - "octaves": { - "octave1": { - "RF_outputs": { - 1: { - "LO_frequency": 6e9, - "LO_source": "internal", - "output_mode": "always_on", - "gain": 0, + "octaves": { + "octave1": { + "RF_outputs": { + 1: { + "LO_frequency": 6e9, + "LO_source": "internal", + "output_mode": "always_on", + "gain": 0, + }, + 2: { + "LO_frequency": 5e9, + "LO_source": "internal", + "output_mode": "always_on", + "gain": 0, + }, }, - 2: { - "LO_frequency": 5e9, - "LO_source": "internal", - "output_mode": "always_on", - "gain": 0, + "RF_inputs": { + 1: { + "LO_frequency": 6e9, + "LO_source": "internal", + }, }, - }, - "RF_inputs": { - 1: { - "LO_frequency": 6e9, - "LO_source": "internal", + "connectivity": "con1", + } + }, + "pulses": { + "const_pulse": { + "operation": "control", + "length": 100, + "waveforms": { + "I": "const_wf", + "Q": "zero_wf", }, }, - "connectivity": "con1", - } - }, - "pulses": { - "const_pulse": { - "operation": "control", - "length": 100, - "waveforms": { - "I": "const_wf", - "Q": "zero_wf", - }, }, - }, - "waveforms": { - "const_wf": {"type": "constant", "sample": 0.125}, - }, -} + "waveforms": { + "const_wf": {"type": "constant", "sample": 0.125}, + }, + } def convert_to_correction(gain: float, phase: float): @@ -98,11 +99,13 @@ def convert_to_correction(gain: float, phase: float): return c00, c01, c10, c11 + def abs_path_to(rel_path: str) -> str: source_path = Path(__file__).resolve() source_dir = source_path.parent return os.path.join(source_dir, rel_path) + # res: 6e9, -60e6, gain=0 and 10 # qubit: 5e9, 50e6, gain=0 and -10 def test_validity_correction_parameters(config): @@ -115,6 +118,7 @@ def test_validity_correction_parameters(config): assert param_qubit["correction_matrix"] == convert_to_correction(0.000670812605205233, 0.16976806502793335) assert param_res["correction_matrix"] == convert_to_correction(-0.00357108327961989, -0.0270416534916431) + def test_verbose(config): param_qubit = get_calibration_parameters_from_db(abs_path_to(""), config, "qubit", 5e9, 50e6, 1, verbose_level=0) assert param_qubit["correction_matrix"] == () @@ -122,5 +126,3 @@ def test_verbose(config): assert param_qubit["correction_matrix"] == () with pytest.raises(Warning): get_calibration_parameters_from_db(abs_path_to(""), config, "qubit", 5e9, 50e6, 1, verbose_level=2) - - diff --git a/tests/test_simulator_tools.py b/tests/test_simulator_tools.py index 3712ad21..c37c3316 100644 --- a/tests/test_simulator_tools.py +++ b/tests/test_simulator_tools.py @@ -9,9 +9,7 @@ def test_two_controllers_connection_schema(): second_controller_list = [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2] second_port_list = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11] controller_connections = [ - ControllerConnection( - InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l) - ) + ControllerConnection(InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l)) for i, j, k, l in list( zip( first_controller_list, @@ -33,9 +31,7 @@ def test_three_controllers_connection_schema(): second_controller_list = [2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3] second_port_list = [0, 0, 1, 2, 2, 3, 4, 4, 5, 6, 6, 7, 8, 8, 9, 10, 10, 11] controller_connections = [ - ControllerConnection( - InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l) - ) + ControllerConnection(InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l)) for i, j, k, l in list( zip( first_controller_list, @@ -56,12 +52,16 @@ def test_last_two_in_a_three_controllers_connection_schema(): first_port_list = [0, 1, 1, 2, 3, 3, 4, 5, 5, 6, 7, 7, 8, 9, 9, 10, 11, 11] second_controller_list = [2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3] second_port_list = [0, 0, 1, 2, 2, 3, 4, 4, 5, 6, 6, 7, 8, 8, 9, 10, 10, 11] - filtered_lists = [(item1, item2, item3, item4) for item1, item2, item3, item4 in zip(first_controller_list, first_port_list, second_controller_list, second_port_list) if item1 != 1 and item3 != 1] + filtered_lists = [ + (item1, item2, item3, item4) + for item1, item2, item3, item4 in zip( + first_controller_list, first_port_list, second_controller_list, second_port_list + ) + if item1 != 1 and item3 != 1 + ] first_controller_list, first_port_list, second_controller_list, second_port_list = zip(*filtered_lists) controller_connections = [ - ControllerConnection( - InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l) - ) + ControllerConnection(InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l)) for i, j, k, l in list( zip( first_controller_list, @@ -82,12 +82,16 @@ def test_two_in_a_three_controllers_connection_schema(): first_port_list = [0, 1, 1, 2, 3, 3, 4, 5, 5, 6, 7, 7, 8, 9, 9, 10, 11, 11] second_controller_list = [2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3, 2, 3, 3] second_port_list = [0, 0, 1, 2, 2, 3, 4, 4, 5, 6, 6, 7, 8, 8, 9, 10, 10, 11] - filtered_lists = [(item1, item2, item3, item4) for item1, item2, item3, item4 in zip(first_controller_list, first_port_list, second_controller_list, second_port_list) if item1 != 2 and item3 != 2] + filtered_lists = [ + (item1, item2, item3, item4) + for item1, item2, item3, item4 in zip( + first_controller_list, first_port_list, second_controller_list, second_port_list + ) + if item1 != 2 and item3 != 2 + ] first_controller_list, first_port_list, second_controller_list, second_port_list = zip(*filtered_lists) controller_connections = [ - ControllerConnection( - InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l) - ) + ControllerConnection(InterOpxChannel(f"con{i}", j), InterOpxChannel(f"con{k}", l)) for i, j, k, l in list( zip( first_controller_list, diff --git a/tests/test_time_units.py b/tests/test_time_units.py index b6f382a0..6bdf606f 100644 --- a/tests/test_time_units.py +++ b/tests/test_time_units.py @@ -174,10 +174,12 @@ def test_frequency_conversion(unit_verbose): assert 4.1 * unit_verbose.MHz == 4_100_000 assert 4.1 * unit_verbose.GHz == 4_100_000_000 + def test_dBm2volts(unit): assert unit.dBm2volts(1) == 0.3548133892335754 assert unit.dBm2volts(1, Z=100) == 0.5017819071656863 + def test_volts2dBm(unit): assert unit.volts2dBm(0.5) == 3.9794000867203754 assert unit.volts2dBm(0.5, Z=100) == 0.9691001300805634 diff --git a/tests/test_waveform_tools.py b/tests/test_waveform_tools.py index 5155ccd8..fd09ea14 100644 --- a/tests/test_waveform_tools.py +++ b/tests/test_waveform_tools.py @@ -27,7 +27,7 @@ def test_drag_no_drag_gaussian_to_scipy(length, sampling_rate): anharmonicity=0, detuning=0, subtracted=False, - sampling_rate=sampling_rate + sampling_rate=sampling_rate, ) I_sub_wf, Q_sub_wf = drag_gaussian_pulse_waveforms( amplitude=amp, @@ -37,9 +37,9 @@ def test_drag_no_drag_gaussian_to_scipy(length, sampling_rate): anharmonicity=0, detuning=0, subtracted=True, - sampling_rate=sampling_rate + sampling_rate=sampling_rate, ) - gauss = amp * gaussian(int(length * sampling_rate/1e9), sigma * sampling_rate / 1e9) + gauss = amp * gaussian(int(length * sampling_rate / 1e9), sigma * sampling_rate / 1e9) sub_gauss = gauss - gauss[0] assert (I_wf == gauss).all() assert (I_sub_wf == sub_gauss).all() @@ -57,7 +57,7 @@ def test_drag_no_detune_symmetric(length, sampling_rate): anharmonicity=10e6, detuning=0, subtracted=False, - sampling_rate=sampling_rate + sampling_rate=sampling_rate, ) I_cos_wf, Q_cos_wf = drag_cosine_pulse_waveforms( amplitude=amp, length=length, alpha=0.1, anharmonicity=10e6, detuning=0, sampling_rate=sampling_rate @@ -83,12 +83,8 @@ def test_drag_no_detune_symmetric(length, sampling_rate): assert (np.array(I_gauss_first_half) == np.flip(I_gauss_second_half)).all() assert (np.array(Q_gauss_first_half) == -np.flip(Q_gauss_second_half)).all() - np.testing.assert_allclose( - I_cos_first_half, np.flip(I_cos_second_half), rtol=1e-7, atol=1e-7 - ) - np.testing.assert_allclose( - Q_cos_first_half, -np.flip(Q_cos_second_half), rtol=1e-7, atol=1e-7 - ) + np.testing.assert_allclose(I_cos_first_half, np.flip(I_cos_second_half), rtol=1e-7, atol=1e-7) + np.testing.assert_allclose(Q_cos_first_half, -np.flip(Q_cos_second_half), rtol=1e-7, atol=1e-7) def test_drag_detune(): @@ -141,14 +137,18 @@ def test_drag_zero_delta(): Exception, match="Cannot create a DRAG pulse", ): - drag_cosine_pulse_waveforms( - amplitude=0.1, length=40, alpha=0.1, anharmonicity=0, detuning=0 - ) + drag_cosine_pulse_waveforms(amplitude=0.1, length=40, alpha=0.1, anharmonicity=0, detuning=0) @pytest.mark.parametrize( "flat_length, rise_fall_length, sampling_rate", - list(zip([0, 16, 16, 21, 21, 60, 60, 0, 16, 16, 21, 21, 60, 60], [8, 5, 10, 5, 10, 0, 10, 8, 5, 10, 5, 10, 0, 10], [1e9, 1e9, 1e9, 1e9, 1e9, 1e9, 1e9, 2e9, 2e9, 2e9, 2e9, 2e9, 2e9, 2e9])), + list( + zip( + [0, 16, 16, 21, 21, 60, 60, 0, 16, 16, 21, 21, 60, 60], + [8, 5, 10, 5, 10, 0, 10, 8, 5, 10, 5, 10, 0, 10], + [1e9, 1e9, 1e9, 1e9, 1e9, 1e9, 1e9, 2e9, 2e9, 2e9, 2e9, 2e9, 2e9, 2e9], + ) + ), ) def test_flattop_flat_length(flat_length, rise_fall_length, sampling_rate): amp = 0.1 @@ -205,7 +205,12 @@ def test_flattop_flat_length(flat_length, rise_fall_length, sampling_rate): assert np.allclose( flattop_gaussian_rise + flattop_gaussian_fall, - (amp * gaussian(int(np.round(2 * rise_fall_length * sampling_rate / 1e9)), rise_fall_length / 5 * sampling_rate / 1e9)).tolist(), + ( + amp + * gaussian( + int(np.round(2 * rise_fall_length * sampling_rate / 1e9)), rise_fall_length / 5 * sampling_rate / 1e9 + ) + ).tolist(), rtol=1e-10, ) cosine_rise_part = ( @@ -238,7 +243,7 @@ def test_flattop_flat_length(flat_length, rise_fall_length, sampling_rate): np.linspace(2, 31, 30).astype(int).tolist(), np.linspace(-0.5, 0.5, 30).tolist(), np.linspace(0.5, -0.5, 30).tolist(), - [1e9]*15 + [2e9]*15, + [1e9] * 15 + [2e9] * 15, ) ), ) diff --git a/tests_against_server/callable_from_qua/test_basic_qua_callable.py b/tests_against_server/callable_from_qua/test_basic_qua_callable.py index 98ac249f..259a1b93 100644 --- a/tests_against_server/callable_from_qua/test_basic_qua_callable.py +++ b/tests_against_server/callable_from_qua/test_basic_qua_callable.py @@ -5,10 +5,13 @@ registered_calls = [] + + @callable_from_qua def register_calls(*args, **kwargs): registered_calls.append({"args": args, "kwargs": kwargs}) + def test_qua_callable_no_args(qmm, config): registered_calls.clear() with program() as prog: @@ -19,5 +22,6 @@ def test_qua_callable_no_args(qmm, config): job = qm.execute(prog) from time import sleep + sleep(10) - assert registered_calls == [{"args": (), "kwargs": {}}] \ No newline at end of file + assert registered_calls == [{"args": (), "kwargs": {}}] diff --git a/tests_against_server/conftest.py b/tests_against_server/conftest.py index 2c29ac24..c6037ee6 100644 --- a/tests_against_server/conftest.py +++ b/tests_against_server/conftest.py @@ -9,9 +9,7 @@ def qmm(): config_file = Path.home() / ".qm_config.toml" if not config_file.exists(): - raise FileNotFoundError( - "Could not locate ~/.qm_config.toml, cannot extract IP and port to execute tests" - ) + raise FileNotFoundError("Could not locate ~/.qm_config.toml, cannot extract IP and port to execute tests") config = toml.load(config_file) if "qmm" not in config: @@ -95,7 +93,6 @@ def config(): "length": 80, "waveforms": {"I": "const_wf", "Q": "zero_wf"}, }, - }, "waveforms": { "zero_wf": {"type": "constant", "sample": 0.0}, @@ -110,8 +107,8 @@ def config(): { "intermediate_frequency": 0, "lo_frequency": 0, - "correction": (1,0,0,1), + "correction": (1, 0, 0, 1), } ], }, - } \ No newline at end of file + } diff --git a/tests_against_server/test_bakery_server.py b/tests_against_server/test_bakery_server.py index 3c334d39..52ac58c5 100644 --- a/tests_against_server/test_bakery_server.py +++ b/tests_against_server/test_bakery_server.py @@ -20,9 +20,7 @@ def IQ_imbalance(g, phi): c = np.cos(phi) s = np.sin(phi) N = 1 / ((1 - g**2) * (2 * c**2 - 1)) - return [ - float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c] - ] + return [float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c]] return { "version": 1, @@ -117,9 +115,7 @@ def IQ_imbalance(g, phi): def simulate_program_and_return(config, prog, duration=20000): qmm = QuantumMachinesManager() qmm.close_all_quantum_machines() - job = qmm.simulate( - config, prog, SimulationConfig(duration, include_analog_waveforms=True) - ) + job = qmm.simulate(config, prog, SimulationConfig(duration, include_analog_waveforms=True)) return job @@ -153,14 +149,9 @@ def play_twice(b): job = simulate_program_and_return(cfg, prog, 200) samples = job.get_simulated_samples() - tstamp = int( - job.simulated_analog_waveforms()["controllers"]["con1"]["ports"]["1"][0][ - "timestamp" - ] - ) + tstamp = int(job.simulated_analog_waveforms()["controllers"]["con1"]["ports"]["1"][0]["timestamp"]) assert all( - samples.con1.analog["1"][tstamp : tstamp + 200] - == [i / 200 for i in range(100)] + [i / 200 for i in range(100)] + samples.con1.analog["1"][tstamp : tstamp + 200] == [i / 200 for i in range(100)] + [i / 200 for i in range(100)] ) @@ -193,12 +184,8 @@ def test_amp_modulation_run(config): samples3_data = samples3.con1.analog["1"] assert len(samples1_data) == len(samples2_data) - assert all( - [samples1_data[i] == samples3_data[i] for i in range(len(samples1_data))] - ) - assert all( - [samples2_data[i] == samples3_data[i] for i in range(len(samples2_data))] - ) + assert all([samples1_data[i] == samples3_data[i] for i in range(len(samples1_data))]) + assert all([samples2_data[i] == samples3_data[i] for i in range(len(samples2_data))]) def test_play_baked_with_existing_digital_wf(config): diff --git a/tests_against_server/test_digital_filters_server.py b/tests_against_server/test_digital_filters_server.py index 7ab0388c..56569e3f 100644 --- a/tests_against_server/test_digital_filters_server.py +++ b/tests_against_server/test_digital_filters_server.py @@ -11,6 +11,7 @@ from qualang_tools.config.waveform_tools import drag_gaussian_pulse_waveforms from qualang_tools.units import unit import matplotlib.pyplot as plt + ####################### # AUXILIARY FUNCTIONS # ####################### @@ -478,6 +479,7 @@ def IQ_imbalance(g, phi): tau_lp = 100 # feedforward, feedback = single_exponential_correction(A, tau_lp) from scipy.signal.windows import hann + # feedforward, feedback = calc_filter_taps(delay=11, Ts=2) # feedforward, feedback = calc_filter_taps(exponential=[(A, tau_lp)], highpass=[t_hp], delay=5.1) feedforward, feedback = calc_filter_taps(delay=10.1, highpass=[t_hp]) @@ -489,7 +491,8 @@ def IQ_imbalance(g, phi): # feedback = [-feedback[0]] config["controllers"]["con1"]["analog_outputs"][5] = { "offset": 0.0, - "filter": {"feedforward": feedforward, "feedback": feedback}} + "filter": {"feedforward": feedforward, "feedback": feedback}, +} ################### # The QUA program # @@ -521,9 +524,19 @@ def IQ_imbalance(g, phi): t = np.arange(0, len(samples.analog["5"]), 1) dt = np.where(samples.analog["5"] != 0)[0][0] plt.figure() - plt.plot(t[dt:const_flux_len+dt], const_flux_amp * (np.exp(-t[:const_flux_len] / t_hp)), 'b',label="Pulse before correction") - plt.plot(t, samples.analog["5"], 'g--', label="OPX pulse with correction") - plt.plot(t[dt:const_flux_len+dt], samples.analog["5"][dt:const_flux_len+dt] * (np.exp(-t[:const_flux_len] / t_hp)), 'r',label="Pulse after correction") + plt.plot( + t[dt : const_flux_len + dt], + const_flux_amp * (np.exp(-t[:const_flux_len] / t_hp)), + "b", + label="Pulse before correction", + ) + plt.plot(t, samples.analog["5"], "g--", label="OPX pulse with correction") + plt.plot( + t[dt : const_flux_len + dt], + samples.analog["5"][dt : const_flux_len + dt] * (np.exp(-t[:const_flux_len] / t_hp)), + "r", + label="Pulse after correction", + ) plt.legend() # plt.figure() diff --git a/tests_against_server/test_long_wait.py b/tests_against_server/test_long_wait.py index e8065774..b0cc4f47 100644 --- a/tests_against_server/test_long_wait.py +++ b/tests_against_server/test_long_wait.py @@ -9,7 +9,7 @@ def test_long_wait_time_live(qmm, config): - """ Tests that the program runs and finishes without crashing. """ + """Tests that the program runs and finishes without crashing.""" op = "playOp" wait_time = u.to_clock_cycles(30e9) # 50s @@ -42,13 +42,12 @@ def test_long_wait_time_live(qmm, config): dummy_max_wait_time = 200 # clock cycles -@pytest.mark.parametrize("wait_time", [4, 16, 100, - dummy_max_wait_time-1, - dummy_max_wait_time, - dummy_max_wait_time+1, - 100*dummy_max_wait_time]) +@pytest.mark.parametrize( + "wait_time", + [4, 16, 100, dummy_max_wait_time - 1, dummy_max_wait_time, dummy_max_wait_time + 1, 100 * dummy_max_wait_time], +) def test_long_wait_time_simulation(qmm, config, wait_time): - """ Extracts the wait time from simulation and asserts it to be equal to the input wait time.""" + """Extracts the wait time from simulation and asserts it to be equal to the input wait time.""" op = "playOp" with program() as prog: @@ -56,22 +55,22 @@ def test_long_wait_time_simulation(qmm, config, wait_time): long_wait(wait_time, threshold_for_looping=dummy_max_wait_time) play(op, "qe1") - element = config['elements']['qe1'] - pulse = config['pulses'][element['operations'][op]] - pulse_length = pulse['length'] - pulse_amplitude = config['waveforms'][pulse['waveforms']['single']]['sample'] + element = config["elements"]["qe1"] + pulse = config["pulses"][element["operations"][op]] + pulse_length = pulse["length"] + pulse_amplitude = config["waveforms"][pulse["waveforms"]["single"]]["sample"] - job = execute_program(qmm, config, prog, simulate=True, simulation_duration=wait_time + 2*pulse_length + 100) + job = execute_program(qmm, config, prog, simulate=True, simulation_duration=wait_time + 2 * pulse_length + 100) - output = job.get_simulated_samples().con1.analog['1-1'] + output = job.get_simulated_samples().con1.analog["1-1"] th = pulse_amplitude / 2 # edge threshold - rising_edges = np.where((output[:-1] <= th/2) & (output[1:] > th/2))[0] - falling_edges = np.where((output[:-1] > th/2) & (output[1:] <= th/2))[0] + rising_edges = np.where((output[:-1] <= th / 2) & (output[1:] > th / 2))[0] + falling_edges = np.where((output[:-1] > th / 2) & (output[1:] <= th / 2))[0] # in a square -> wait -> square program, the wait time lies between the first falling edge # and the second rising edge of the entire program. - wait_samples = output[falling_edges[0]:rising_edges[1]] + wait_samples = output[falling_edges[0] : rising_edges[1]] # plt.plot(output) # plt.axvline(falling_edges[0], linestyle='--') @@ -81,13 +80,12 @@ def test_long_wait_time_simulation(qmm, config, wait_time): assert len(wait_samples) == wait_time * u.clock_cycle -@pytest.mark.parametrize("wait_time", [4, 16, 100, - dummy_max_wait_time-1, - dummy_max_wait_time, - dummy_max_wait_time+1, - 100*dummy_max_wait_time]) +@pytest.mark.parametrize( + "wait_time", + [4, 16, 100, dummy_max_wait_time - 1, dummy_max_wait_time, dummy_max_wait_time + 1, 100 * dummy_max_wait_time], +) def test_long_wait_time_simulation_multi_element(qmm, config, wait_time): - """ Same as above, but multi-element waiting. """ + """Same as above, but multi-element waiting.""" op = "playOp" with program() as prog: @@ -97,24 +95,24 @@ def test_long_wait_time_simulation_multi_element(qmm, config, wait_time): play(op, "qe1") play(op, "qe2") - element = config['elements']['qe1'] - pulse = config['pulses'][element['operations'][op]] - pulse_length = pulse['length'] - pulse_amplitude = config['waveforms'][pulse['waveforms']['single']]['sample'] + element = config["elements"]["qe1"] + pulse = config["pulses"][element["operations"][op]] + pulse_length = pulse["length"] + pulse_amplitude = config["waveforms"][pulse["waveforms"]["single"]]["sample"] - job = execute_program(qmm, config, prog, simulate=True, simulation_duration=wait_time + 2*pulse_length + 100) + job = execute_program(qmm, config, prog, simulate=True, simulation_duration=wait_time + 2 * pulse_length + 100) - output_1 = job.get_simulated_samples().con1.analog['1-1'] - output_2 = job.get_simulated_samples().con1.analog['1-2'] + output_1 = job.get_simulated_samples().con1.analog["1-1"] + output_2 = job.get_simulated_samples().con1.analog["1-2"] for output in [output_1, output_2]: th = pulse_amplitude / 2 # edge threshold - rising_edges = np.where((output[:-1] <= th/2) & (output[1:] > th/2))[0] - falling_edges = np.where((output[:-1] > th/2) & (output[1:] <= th/2))[0] + rising_edges = np.where((output[:-1] <= th / 2) & (output[1:] > th / 2))[0] + falling_edges = np.where((output[:-1] > th / 2) & (output[1:] <= th / 2))[0] # in a square -> wait -> square program, the wait time lies between the first falling edge # and the second rising edge of the entire program. - wait_samples = output[falling_edges[0]:rising_edges[1]] + wait_samples = output[falling_edges[0] : rising_edges[1]] assert len(wait_samples) == wait_time * u.clock_cycle diff --git a/tests_against_server/test_loops.py b/tests_against_server/test_loops.py index 70b88c8b..11f3b5d3 100644 --- a/tests_against_server/test_loops.py +++ b/tests_against_server/test_loops.py @@ -98,19 +98,23 @@ def simulate_program_and_return(config, prog): return job -@pytest.mark.parametrize(["start", "stop" ,"step"], - [ [-0.05, -1, -0.15], - [10, 20, 1], - [20, 71, 2], - [20, 71, 1], - [11, 0, -1], - [0.1, 1, 0.2], - [0, 1, 0.2], - [0, 2, 0.0001], - [-1, 2, 0.00011], - [-1, 1, 0.2], - [0.00015, 1, 0.0001], - [0, -2, -0.001],]) +@pytest.mark.parametrize( + ["start", "stop", "step"], + [ + [-0.05, -1, -0.15], + [10, 20, 1], + [20, 71, 2], + [20, 71, 1], + [11, 0, -1], + [0.1, 1, 0.2], + [0, 1, 0.2], + [0, 2, 0.0001], + [-1, 2, 0.00011], + [-1, 1, 0.2], + [0.00015, 1, 0.0001], + [0, -2, -0.001], + ], +) def test_qua_arange(config, start, stop, step): def prog_maker(_start, _stop, _step): if float(step).is_integer(): @@ -144,33 +148,35 @@ def prog_maker(_start, _stop, _step): assert np.allclose(a_list, a_qua, atol=1e-4) -@pytest.mark.parametrize(["vector", "qua_type"], - [ - [np.logspace(np.log10(10000), np.log10(4), 100), "int"], - [np.logspace(np.log10(4), np.log10(10000), 29), "int"], - [np.logspace(np.log10(50), np.log10(12500), 19), "int"], - [np.logspace(np.log10(50000), np.log10(33), 72), "int"], - [np.logspace(6, 4, 19), "int"], - [np.logspace(3, 6, 199), "int"], - [np.logspace(-3, 0, 99), "fixed"], - [np.logspace(-3.5, -1, 11), "fixed"], - [np.logspace(0.5, -0.5, 22), "fixed"], - [np.logspace(0.5, -3.5, 21), "fixed"], - [np.arange(-7.0547, -2.2141, 0.1015), "fixed"], - [np.arange(-0.05, -1, -0.15), "fixed"], - [np.arange(-1, 2, 0.0006), "fixed"], - [np.arange(-11, -100, -2), "int"], - [np.arange(20, 71, 2), "int"], - [np.linspace(20, 71, 52), "int"], - [np.linspace(0.1, 1, 6), "fixed"], - [np.arange(10, 20, 1), "int"], - [np.arange(20, 71, 1), "int"], - [np.arange(0, 71, 2), "int"], - [np.arange(0, 1, 0.1), "fixed"], - [np.arange(0, 1, 0.2), "fixed"], - [np.arange(0.1, 1, 0.2), "fixed"], - # [np.arange(0.00015, 1, 0.0001), "fixed"], - ]) +@pytest.mark.parametrize( + ["vector", "qua_type"], + [ + [np.logspace(np.log10(10000), np.log10(4), 100), "int"], + [np.logspace(np.log10(4), np.log10(10000), 29), "int"], + [np.logspace(np.log10(50), np.log10(12500), 19), "int"], + [np.logspace(np.log10(50000), np.log10(33), 72), "int"], + [np.logspace(6, 4, 19), "int"], + [np.logspace(3, 6, 199), "int"], + [np.logspace(-3, 0, 99), "fixed"], + [np.logspace(-3.5, -1, 11), "fixed"], + [np.logspace(0.5, -0.5, 22), "fixed"], + [np.logspace(0.5, -3.5, 21), "fixed"], + [np.arange(-7.0547, -2.2141, 0.1015), "fixed"], + [np.arange(-0.05, -1, -0.15), "fixed"], + [np.arange(-1, 2, 0.0006), "fixed"], + [np.arange(-11, -100, -2), "int"], + [np.arange(20, 71, 2), "int"], + [np.linspace(20, 71, 52), "int"], + [np.linspace(0.1, 1, 6), "fixed"], + [np.arange(10, 20, 1), "int"], + [np.arange(20, 71, 1), "int"], + [np.arange(0, 71, 2), "int"], + [np.arange(0, 1, 0.1), "fixed"], + [np.arange(0, 1, 0.2), "fixed"], + [np.arange(0.1, 1, 0.2), "fixed"], + # [np.arange(0.00015, 1, 0.0001), "fixed"], + ], +) def test_from_array(config, vector, qua_type): def prog_maker(_vector, _qua_type): if _qua_type == "int": @@ -207,10 +213,7 @@ def prog_maker(_vector, _qua_type): assert np.allclose(a_list, a_qua, atol=1e-4) -@pytest.mark.parametrize(["vector", "qua_type"], - [ - [np.logspace(np.log10(4), np.log10(10000), 60), "int"] - ]) +@pytest.mark.parametrize(["vector", "qua_type"], [[np.logspace(np.log10(4), np.log10(10000), 60), "int"]]) def test_from_array_log_error(config, vector, qua_type): def prog_maker(_vector, _qua_type): if _qua_type == "int": @@ -240,16 +243,18 @@ def prog_maker(_vector, _qua_type): simulate_program_and_return(cfg, prog_maker(vector, qua_type)) -@pytest.mark.parametrize(["start", "stop", "N"], - [ - [0.1, 1, 5], - [0.1, 0.95, 5], - [-1, 0, 2], - [-1, 1, 2], - [-8, 7, 51], - [-0.1, 0.1, 5000], - [-1, 2, 11], - ]) +@pytest.mark.parametrize( + ["start", "stop", "N"], + [ + [0.1, 1, 5], + [0.1, 0.95, 5], + [-1, 0, 2], + [-1, 1, 2], + [-8, 7, 51], + [-0.1, 0.1, 5000], + [-1, 2, 11], + ], +) def test_qua_linspace(config, start, stop, N): def prog_maker(_start, _stop, _N): with program() as prog: @@ -272,14 +277,16 @@ def prog_maker(_start, _stop, _N): assert np.allclose(a_list, a_qua, atol=1e-4) -@pytest.mark.parametrize(["start", "stop", "N"], - [ - [0.5, -2, 24], - [0, -3, 51], - [0, -3, 50], - [-4, 0.5, 11], - [-3.8, 0.2, 7], - ]) +@pytest.mark.parametrize( + ["start", "stop", "N"], + [ + [0.5, -2, 24], + [0, -3, 51], + [0, -3, 50], + [-4, 0.5, 11], + [-3.8, 0.2, 7], + ], +) def test_qua_logspace_fixed(config, start, stop, N): def prog_maker(_start, _stop, _N): with program() as prog: @@ -302,12 +309,14 @@ def prog_maker(_start, _stop, _N): assert np.allclose(a_list, a_qua, atol=1e-4) -@pytest.mark.parametrize(["start", "stop", "N"], - [ - [np.log10(500), np.log10(12500), 11], - [np.log10(5000), np.log10(125), 30], - [np.log10(40), np.log10(1001), 51], - ]) +@pytest.mark.parametrize( + ["start", "stop", "N"], + [ + [np.log10(500), np.log10(12500), 11], + [np.log10(5000), np.log10(125), 30], + [np.log10(40), np.log10(1001), 51], + ], +) def test_qua_logspace_int(config, start, stop, N): def prog_maker(_start, _stop, _N): with program() as prog: @@ -330,10 +339,12 @@ def prog_maker(_start, _stop, _N): assert np.allclose(a_list, a_qua, atol=1e-4) -@pytest.mark.parametrize(["start", "stop", "N"], - [ - [np.log10(4), np.log10(10000), 60], - ]) +@pytest.mark.parametrize( + ["start", "stop", "N"], + [ + [np.log10(4), np.log10(10000), 60], + ], +) def test_qua_logspace_error(config, start, stop, N): def prog_maker(_start, _stop, _N): with program() as prog: diff --git a/tests_against_server/test_multi_user.py b/tests_against_server/test_multi_user.py index afe3a0b5..2fb44296 100644 --- a/tests_against_server/test_multi_user.py +++ b/tests_against_server/test_multi_user.py @@ -12,9 +12,7 @@ def IQ_imbalance(g, phi): c = np.cos(phi) s = np.sin(phi) N = 1 / ((1 - g**2) * (2 * c**2 - 1)) - return [ - float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c] - ] + return [float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c]] return { "version": 1, @@ -123,9 +121,7 @@ def IQ_imbalance(g, phi): c = np.cos(phi) s = np.sin(phi) N = 1 / ((1 - g**2) * (2 * c**2 - 1)) - return [ - float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c] - ] + return [float(N * x) for x in [(1 - g) * c, (1 + g) * s, (1 - g) * s, (1 + g) * c]] return { "version": 1,