Merge branch 'zi_reacquire_acquisition' into 'develop' (2ce55aab) · Commits · Quantify / quantify-scheduler

quantify_scheduler/instrument_coordinator/init.py

+2 −2

Original line number	Diff line number	Diff line
		from .instrument_coordinator import InstrumentCoordinator
		from .instrument_coordinator import InstrumentCoordinator, ZIInstrumentCoordinator

		__all__ = ["InstrumentCoordinator"]
		__all__ = ["InstrumentCoordinator", "ZIInstrumentCoordinator"]

quantify_scheduler/instrument_coordinator/instrument_coordinator.py

+104 −0

Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@
		from __future__ import annotations

		from typing import Any, Dict, List, Tuple
		import numpy as np

		from qcodes.utils import validators
		from qcodes.instrument import parameter
		@@ -284,3 +285,106 @@ class InstrumentCoordinator(qcodes_base.Instrument):
		for instr_name in self.components():
		instrument = self.find_instrument(instr_name)
		self.get_component(instrument.name).wait_done(timeout_sec)


		def _convert_acquisition_data_format(raw_results):
		acquisition_dict = dict()
		for (channel, i) in raw_results.keys():
		if channel not in acquisition_dict.keys():
		acquisition_dict[channel] = []
		acquisition_dict[channel].append(raw_results.get((channel, i)))
		acquisitions_list = []
		for channel in acquisition_dict:
		acquisitions_list.append(np.array(acquisition_dict.get(channel)))
		return acquisitions_list


		class ZIInstrumentCoordinator(InstrumentCoordinator):
		"""
		This class is a hack and extension to the InstrumentCoordinator, which is
		introduced to support the quirks when using the ZI backend
		during the acquisition of results.
		"""

		__module__ = "quantify_scheduler.instrument_coordinator"

		def __init__(self, name: str) -> None:
		super().__init__(name)

		self.add_parameter(
		"timeout_reacquire",
		unit="",
		initial_value=True,
		vals=validators.Bool(),
		parameter_class=parameter.ManualParameter,
		docstring="Turns on reacquisition in case " "of timeouts.",
		)

		self.add_parameter(
		"max_num_reacquisitions",
		unit="",
		initial_value=5,
		vals=validators.Numbers(min_value=0, max_value=50),
		parameter_class=parameter.ManualParameter,
		docstring="The number of retries to retrieve acquisitions in case "
		"of timeouts.",
		)

		self._last_acquisition = None
		self._num_reacquisitions = 0

		def _compare_reacquire(self, raw_results):
		reacquire = True
		results_list = _convert_acquisition_data_format(raw_results)

		if self._last_acquisition is not None:
		last_acquisition_list = _convert_acquisition_data_format(
		self._last_acquisition
		)
		difference_np_array = np.linalg.norm(
		np.array(results_list[0]) - np.array(last_acquisition_list[0])
		)
		if difference_np_array > 0.0:
		reacquire = False
		self._num_reacquisitions = 0
		return reacquire

		def retrieve_acquisition(self):
		"""
		Retrieves the latest acquisition results of the components
		with acquisition capabilities.

		Returns
		-------
		:
		The acquisition data per component.
		"""
		# FIXME: update the description of the return type of the instrument
		# coordinator # pylint: disable=fixme

		raw_acq_results = super().retrieve_acquisition()
		if self.timeout_reacquire():
		reacquire = self._compare_reacquire(raw_acq_results)
		self._last_acquisition = raw_acq_results
		self._num_reacquisitions = 0

		while reacquire:
		self._num_reacquisitions += 1
		print(
		"Re-running acquisition "
		+ f"{self._num_reacquisitions}/{self.max_num_reacquisitions()}"
		)
		if self._num_reacquisitions >= self.max_num_reacquisitions():
		raise RuntimeError(
		"Exceeded number of reacquisitions:"
		+ f"{self._num_reacquisitions}/"
		+ f"{self.max_num_reacquisitions()}"
		)
		self.start()
		self.wait_done()
		# Acquire results
		raw_acq_results = super().retrieve_acquisition()
		reacquire = self._compare_reacquire(raw_acq_results)
		self._last_acquisition = raw_acq_results

		return raw_acq_results

tests/scheduler/instrument_coordinator/test_instrument_coordinator.py

+60 −1

Original line number	Diff line number	Diff line
		@@ -15,7 +15,10 @@ from unittest.mock import call
		import pytest
		from qcodes import Instrument
		from quantify_scheduler import Schedule, CompiledSchedule
		from quantify_scheduler.instrument_coordinator import InstrumentCoordinator
		from quantify_scheduler.instrument_coordinator import (
		InstrumentCoordinator,
		ZIInstrumentCoordinator,
		)
		from quantify_scheduler.instrument_coordinator.components import base as base_component


		@@ -83,6 +86,19 @@ def fixture_instrument_coordinator(request) -> InstrumentCoordinator:
		return instrument_coordinator


		@pytest.fixture(scope="function", name="zi_instrument_coordinator")
		def fixture_zi_instrument_coordinator(request) -> ZIInstrumentCoordinator:
		zi_instrument_coordinator = ZIInstrumentCoordinator("ic_zi_0000")

		def cleanup_tmp():
		# This should prevent the garbage collector from colleting the qcodes instrument
		zi_instrument_coordinator.close()

		request.addfinalizer(cleanup_tmp)

		return zi_instrument_coordinator


		def test_constructor(close_all_instruments, instrument_coordinator):

		# Assert
		@@ -280,6 +296,49 @@ def test_retrieve_acquisition(
		assert {(0, 0): [1, 2, 3, 4]} == data


		def test_reacquire_acquisition_successful(
		close_all_instruments, zi_instrument_coordinator, dummy_components
		):
		# Arrange
		component1 = dummy_components.pop(0)
		component2 = dummy_components.pop(0)
		zi_instrument_coordinator.add_component(component1)
		zi_instrument_coordinator.add_component(component2)

		component1.retrieve_acquisition.return_value = {(0, 0): [1, 2, 3, 4]}
		component2.retrieve_acquisition.return_value = None

		# Set the last cache of the ZIInstrumentCoordinator to something different
		zi_instrument_coordinator._last_acquisition = {(0, 0): [5, 6, 7, 8]}

		# Act
		data = zi_instrument_coordinator.retrieve_acquisition()

		# Assert
		component1.retrieve_acquisition.assert_called()
		component2.retrieve_acquisition.assert_called()
		assert zi_instrument_coordinator._last_acquisition is not None
		assert {(0, 0): [1, 2, 3, 4]} == data


		def test_reacquire_acquisition_failed(
		close_all_instruments, zi_instrument_coordinator, dummy_components
		):
		# Arrange
		component1 = dummy_components.pop(0)
		component2 = dummy_components.pop(0)
		zi_instrument_coordinator.add_component(component1)
		zi_instrument_coordinator.add_component(component2)

		component1.retrieve_acquisition.return_value = {(0, 0): [1, 2, 3, 4]}
		component2.retrieve_acquisition.return_value = None

		# Assert
		with pytest.raises(RuntimeError):
		# Act
		zi_instrument_coordinator.retrieve_acquisition()


		def test_wait_done(close_all_instruments, instrument_coordinator, dummy_components):
		# Arrange
		component1 = dummy_components.pop(0)