Merge branch 'QAE-674-returning-xarray-acquisition-results' into 'main'

### Breaking changes

- Acquisition - `InstrumentCoordinator.retrieve_acquisition` returns an `xarray.Dataset` (!550, #362)

### Merged branches and closed issue

- Compilation - Add `determine_relative_latencies` that determines latencies for all port-clock combinations in the hardware config relative to the minimum latency (!566, #379)

    q1asm_instructions,

    register_manager,

)

from quantify_scheduler.backends.qblox.helpers import calc_from_units_volt

from quantify_scheduler.backends.qblox.helpers import (

    calc_from_units_volt,

    extract_acquisition_metadata_from_acquisitions,

    single_scope_mode_acquisition_raise,

)

from quantify_scheduler.backends.qblox.operation_handling.acquisitions import (

    AcquisitionStrategyPartial,

)

    MarkerConfiguration,

)

from quantify_scheduler.enums import BinMode

from quantify_scheduler.helpers.schedule import (

    _extract_acquisition_metadata_from_acquisitions,

)

logger = logging.getLogger(__name__)

logger.setLevel(logging.WARNING)

            acq.generate_data(wf_dict)

        return wf_dict

    def _prepare_acq_settings(self, acquisitions: List[IOperationStrategy]):

    def _prepare_acq_settings(

        self, acquisitions: List[IOperationStrategy], repetitions: int

    ):

        """

        Sets sequencer settings that are specific to certain acquisitions.

        For example for a TTL acquisition strategy.

        Parameters

        ----------

        acquisitions

            List of the acquisitions assigned to this sequencer.

        repetitions

            The number of times to repeat execution of the schedule.

        """

        acquisition_infos: List[OpInfo] = list(

            map(lambda acq: acq.operation_info, acquisitions)

        )

        acq_metadata = _extract_acquisition_metadata_from_acquisitions(

            acquisition_infos

        acq_metadata = extract_acquisition_metadata_from_acquisitions(

            acquisitions=acquisition_infos, repetitions=repetitions

        )

        if acq_metadata.acq_protocol == "trigger_count":

        if acq_metadata.acq_protocol == "TriggerCount":

            self._settings.ttl_acq_auto_bin_incr_en = (

                acq_metadata.bin_mode == BinMode.AVERAGE

            )

        )

        # acquisition metadata for acquisitions relevant to this sequencer only

        acq_metadata = _extract_acquisition_metadata_from_acquisitions(

            acquisition_infos

        acq_metadata = extract_acquisition_metadata_from_acquisitions(

            acquisitions=acquisition_infos, repetitions=repetitions

        )

        # initialize an empty dictionary for the format required by module

                    f"Please make sure the used bins increment by 1 starting from "

                    f"0. Found: {max(acq_indices)} as the highest bin out of "

                    f"{len(acq_indices)} for channel {acq_channel}, indicating "

                    f"an acquisition_index was skipped. "

                    f"an acquisition index was skipped. "

                    f"Problem occurred for port {self.port} with clock {self.clock},"

                    f"which corresponds to {self.name} of {self.parent.name}."

                )

            if acq_metadata.bin_mode == BinMode.APPEND:

                num_bins = repetitions * (max(acq_indices) + 1)

            elif acq_metadata.bin_mode == BinMode.AVERAGE:

                if acq_metadata.acq_protocol == "trigger_count":

                if acq_metadata.acq_protocol == "TriggerCount":

                    num_bins = constants.MAX_NUMBER_OF_BINS

                else:

                    num_bins = max(acq_indices) + 1

            weights_dict = {}

            acq_declaration_dict = {}

            if len(self.acquisitions) > 0:

                self._prepare_acq_settings(self.acquisitions)

                self._prepare_acq_settings(

                    acquisitions=self.acquisitions, repetitions=repetitions

                )

                weights_dict = self._generate_weights_dict()

                acq_declaration_dict = self._generate_acq_declaration_dict(

                    acquisitions=self.acquisitions, repetitions=repetitions

        self._configure_mixer_offsets()

        self._construct_sequencers()

        self.distribute_data()

        self._determine_scope_mode_acquisition_sequencer()

        self._ensure_single_scope_mode_acquisition_sequencer()

        for seq in self.sequencers.values():

            self.assign_frequencies(seq)

        self.assign_attenuation()

                    voltage_range, self.name, "dc_mixer_offset_Q", output_cfg

                )

    def _determine_scope_mode_acquisition_sequencer(self) -> None:

    def _ensure_single_scope_mode_acquisition_sequencer(self) -> None:

        """

        Finds which sequencer has to perform raw trace acquisitions and adds it to the

        `scope_mode_sequencer` of the settings.

        Raises an error if multiple sequencers use scope mode acquisition,

        because that's not supported by the hardware.

        Also, see

        :func:`~quantify_scheduler.instrument_coordinator.components.qblox.QRMComponent._determine_scope_mode_acquisition_sequencer_and_channel`

        which also ensures the program that gets uploaded to the hardware satisfies this requirement.

        Raises

        ------

        """

        def is_scope_acquisition(acquisition: OpInfo) -> bool:

            return acquisition.data["protocol"] == "trace"

            return acquisition.data["protocol"] == "Trace"

        scope_acq_seq = None

        for seq in self.sequencers.values():

            has_scope = any(map(is_scope_acquisition, op_infos))

            if has_scope:

                if scope_acq_seq is not None:

                    raise ValueError(

                        f'Both sequencer "{scope_acq_seq}" and "{seq.index}" of '

                        f'"{self.name}" need to perform scope mode '

                        "acquisitions. Only one sequencer per device can "

                        "trigger raw trace capture.\n\nPlease ensure that "

                        "only one port and clock combination has to "

                        "perform raw trace acquisition per instrument."

                    single_scope_mode_acquisition_raise(

                        sequencer_0=scope_acq_seq,

                        sequencer_1=seq.index,

                        module_name=self.name,

                    )

                scope_acq_seq = seq.index

        self._settings.scope_mode_sequencer = scope_acq_seq

    def compile(

        self,

        repetitions: int = 1,

    ) -> Optional[Dict[str, Any]]:

        """

        Performs the actual compilation steps for this module, by calling the sequencer

        level compilation functions and combining them into a single dictionary. The

        compiled program has a settings key, and keys for every sequencer.

        level compilation functions and combining them into a single dictionary.

        Parameters

        ----------

        Returns

        -------

        :

            The compiled program corresponding to this module. It contains an entry for

            every sequencer and general "settings". If the device is not actually used,

            The compiled program corresponding to this module.

            It contains an entry for every sequencer under the key `"sequencers"`,

            and acquisition metadata under the key `"acq_metadata"`,

            and the `"repetitions"` is an integer with

            the number of times the defined schedule is repeated.

            All the other generic settings are under the key `"settings"`.

            If the device is not actually used,

            and an empty program is compiled, None is returned instead.

        """

        program = {}

        if sequence_to_file is None:

            sequence_to_file = self.hw_mapping.get("sequence_to_file", True)

        program["sequencers"] = {}

        for seq_name, seq in self.sequencers.items():

            seq_program = seq.compile(

                repetitions=repetitions, sequence_to_file=sequence_to_file

            )

            if seq_program is not None:

                program[seq_name] = seq_program

                program["sequencers"][seq_name] = seq_program

        if len(program) == 0:

            return None

        self._settings.hardware_averages = repetitions

        program["settings"] = self._settings.to_dict()

        if self.supports_acquisition:

            # Add both acquisition metadata (a summary) and trace_acq_channel

            program["acq_metadata"] = {}

            for sequencer in self.sequencers.values():

                if not sequencer.acquisitions:

                    continue

                acq_metadata = _extract_acquisition_metadata_from_acquisitions(

                    [acq.operation_info for acq in sequencer.acquisitions]

                acq_metadata = extract_acquisition_metadata_from_acquisitions(

                    acquisitions=[acq.operation_info for acq in sequencer.acquisitions],

                    repetitions=repetitions,

                )

                program["acq_metadata"][sequencer.name] = acq_metadata

        program["repetitions"] = repetitions

            if (trace_acq_channel := self._get_trace_acq_channel()) is not None:

                program["trace_acq_channel"] = trace_acq_channel

        return program

    def _get_trace_acq_channel(self) -> Optional[int]:

        """

        Returns the acquisition channel set by the user on the only trace operation in

        the module.

        If there is no trace operation in the schedule, it returns None.

        If there are more than one, it raises an exception, this is not a valid

        schedule.

        Returns

        -------

        :

            If there is a trace operation, returns the channel for the trace operation.

            If there is no trace operation in the schedule, it returns `None`.

        Raises

        ------

        ValueError

            There is more than one trace acquisition operation.

        """

        def _is_trace_acquisition(acquisition):

            return acquisition.operation_info.data["protocol"] == "trace"

        def _extract_channel(acquisition):

            return acquisition.operation_info.data["acq_channel"]

        trace_acq_channels = []

        for sequencer in self.sequencers.values():

            for acquisition in sequencer.acquisitions:

                if _is_trace_acquisition(acquisition):

                    trace_acq_channels.append(_extract_channel(acquisition))

        if len(trace_acq_channels) == 0:

            return None

        elif len(trace_acq_channels) > 1:

            raise ValueError(

                f"Invalid schedule. "

                f"Only one trace acquisition is permitted per hardware component. "

                f"The following channels are attempting to use trace acquisition: "

                f"{trace_acq_channels}"

            )

        else:

            return trace_acq_channels[0]

class QbloxBasebandModule(QbloxBaseModule):

    """

from quantify_scheduler.backends.qblox import constants

from quantify_scheduler.helpers.waveforms import exec_waveform_function

from quantify_scheduler.schedules.schedule import AcquisitionMetadata

from quantify_scheduler.helpers.schedule import (

    extract_acquisition_metadata_from_acquisition_protocols,

)

from quantify_scheduler import Schedule

from quantify_scheduler.backends.types.qblox import OpInfo

        )

    return calculated_offset

def extract_acquisition_metadata_from_acquisitions(

    acquisitions: List[OpInfo], repetitions: int

) -> AcquisitionMetadata:

    """

    Variant of

    :func:`~quantify_scheduler.helpers.schedule.extract_acquisition_metadata_from_acquisition_protocols`

    for use with the Qblox backend.

    """

    return extract_acquisition_metadata_from_acquisition_protocols(

        acquisition_protocols=[acq.data for acq in acquisitions],

        repetitions=repetitions,

    )

def single_scope_mode_acquisition_raise(sequencer_0, sequencer_1, module_name):

    """

    Raises an error stating that only one scope mode acquisition can be used per module.

    Parameters

    ----------

    sequencer_0

        First sequencer which attempts to use the scope mode acquisition.

    sequencer_1

        Second sequencer which attempts to use the scope mode acquisition.

    module_name

        Name of the module.

    Raises

    ------

    ValueError

        Always raises the error message.

    """

    raise ValueError(

        f"Both sequencer '{sequencer_0}' and '{sequencer_1}' "

        f"of '{module_name}' attempts to perform scope mode acquisitions. "

        f"Only one sequencer per device can "

        f"trigger raw trace capture.\n\nPlease ensure that "

        f"only one port-clock combination performs "

        f"raw trace acquisition per instrument."

    )

    """Handles the logic for determining the correct acquisition type."""

    protocol = operation_info.data["protocol"]

    if protocol in ("trace", "ssb_integration_complex"):

        if protocol == "trace" and operation_info.data["bin_mode"] == BinMode.APPEND:

    if protocol in ("Trace", "SSBIntegrationComplex"):

        if protocol == "Trace" and operation_info.data["bin_mode"] == BinMode.APPEND:

            raise ValueError(

                f"Trace acquisition does not support APPEND bin mode.\n\n"

                f"{repr(operation_info)} caused this exception to occur."

            )

        return acquisitions.SquareAcquisitionStrategy(operation_info)

    if protocol == "weighted_integrated_complex":

    elif protocol == "WeightedIntegratedComplex":

        return acquisitions.WeightedAcquisitionStrategy(operation_info)

    if protocol == "trigger_count":

    elif protocol == "TriggerCount":

        return acquisitions.TriggerCountAcquisitionStrategy(operation_info)

    raise ValueError(

class BaseModuleSettings(DataClassJsonMixin):

    """Shared settings between all the Qblox modules."""

    scope_mode_sequencer: Optional[int] = None

    """The index of the sequencer that triggers scope mode Acquisitions. Only a single

    sequencer can perform trace acquisition. This setting gets set as a QCoDeS parameter

    on the driver as well as used for internal checks. Having multiple sequencers

    perform trace acquisition will result in an exception being raised."""

    offset_ch0_path0: Union[float, None] = None

    """The DC offset on the path 0 of channel 0."""

    offset_ch0_path1: Union[float, None] = None