From 68bd8f3d1bd0b8af66fdcceb12a75f4d856cb209 Mon Sep 17 00:00:00 2001 From: Alessandro Candido Date: Fri, 7 Feb 2025 13:13:13 +0100 Subject: [PATCH] fix: Filter probe channels To restrict their usage in association with acquisition --- .../_core/instruments/qblox/cluster.py | 17 +++++++- src/qibolab/_core/instruments/qblox/config.py | 43 +++++++++++++------ .../_core/instruments/qblox/validate.py | 6 +++ 3 files changed, 52 insertions(+), 14 deletions(-) create mode 100644 src/qibolab/_core/instruments/qblox/validate.py diff --git a/src/qibolab/_core/instruments/qblox/cluster.py b/src/qibolab/_core/instruments/qblox/cluster.py index fabcaf97b..c2c7f8d7c 100644 --- a/src/qibolab/_core/instruments/qblox/cluster.py +++ b/src/qibolab/_core/instruments/qblox/cluster.py @@ -8,6 +8,7 @@ from qblox_instruments.qcodes_drivers.module import Module from qcodes.instrument import find_or_create_instrument +from qibolab._core.components.channels import AcquisitionChannel from qibolab._core.components.configs import Configs from qibolab._core.execution_parameters import AcquisitionType, ExecutionParameters from qibolab._core.identifier import ChannelId, Result @@ -21,6 +22,7 @@ from .log import Logger from .results import AcquiredData, extract, integration_lenghts from .sequence import Q1Sequence, compile +from .validate import _assert_no_probe __all__ = ["Cluster"] @@ -46,6 +48,14 @@ def cluster(self) -> qblox.Cluster: def _modules(self) -> dict[SlotId, Module]: return {mod.slot_idx: mod for mod in self.cluster.modules if mod.present()} + @cached_property + def _probes(self) -> set[ChannelId]: + return { + ch.probe + for ch in self.channels.values() + if isinstance(ch, AcquisitionChannel) and ch.probe is not None + } + @cached_property def _channels_by_module(self) -> dict[SlotId, list[tuple[ChannelId, PortAddress]]]: addresses = { @@ -55,7 +65,9 @@ def _channels_by_module(self) -> dict[SlotId, list[tuple[ChannelId, PortAddress] k: [el[1] for el in g] for k, g in groupby( sorted( - (address.slot, (ch, address)) for ch, address in addresses.items() + (address.slot, (ch, address)) + for ch, address in addresses.items() + if ch not in self._probes ), key=lambda el: el[0], ) @@ -97,6 +109,7 @@ def play( log = Logger(configs) for ps in sequences: + _assert_no_probe(ps, self._probes) sequences_ = compile(ps, sweepers, options, self.sampling_rate) log.sequences(sequences_) sequencers = self._configure(sequences_, configs, options.acquisition_type) @@ -120,7 +133,7 @@ def _configure( for slot, chs in self._channels_by_module.items(): module = self._modules[slot] assert len(module.sequencers) >= len(chs) - config.module(module, dict(chs), self.channels, configs) + config.module(module, {ch[0] for ch in chs}, self.channels, configs) for idx, ((ch, address), sequencer) in enumerate( zip(chs, module.sequencers) ): diff --git a/src/qibolab/_core/instruments/qblox/config.py b/src/qibolab/_core/instruments/qblox/config.py index 10366e09e..410bd0e0a 100644 --- a/src/qibolab/_core/instruments/qblox/config.py +++ b/src/qibolab/_core/instruments/qblox/config.py @@ -67,7 +67,13 @@ def local_address(self): return f"{direction}{channels}" -def _probe(id_: ChannelId, channel: Channel) -> Optional[ChannelId]: +def _iqout(id_: ChannelId, channel: Channel) -> Optional[ChannelId]: + """Extract associated IQ output channel. + + This is the identity for each IQ output channel identifier, while it retrieves the + associated probe channel for acquisition ones, and :obj:`None` for any other one + (essentially, non-RF channels). + """ return ( id_ if isinstance(channel, IqChannel) @@ -79,9 +85,29 @@ def _probe(id_: ChannelId, channel: Channel) -> Optional[ChannelId]: ) +def _los( + mod_channels: set[ChannelId], channels: dict[ChannelId, Channel] +) -> set[tuple[ChannelId, str]]: + """Extract all LOs of a certain module. + + The result contains the associated channel, since required to + address the LO through the API. While the LO identifier is used to + retrieve the configuration. + """ + return { + (iq, lo) + for iq, lo in { + (iq, cast(IqChannel, channels[iq]).lo) + for iq in (_iqout(ch, channels[ch]) for ch in mod_channels) + if iq is not None + } + if lo is not None + } + + def module( mod: Module, - mod_channels: dict[ChannelId, PortAddress], + mod_channels: set[ChannelId], channels: dict[ChannelId, Channel], configs: Configs, ): @@ -94,15 +120,8 @@ def module( mod.scope_acq_trigger_mode_path1("sequencer") # set lo frequencies - los = { - (probe, cast(IqChannel, channels[probe]).lo) - for probe in (_probe(ch, channels[ch]) for ch in mod_channels) - if probe is not None - } - for probe, lo in los: - if lo is None: - continue - n = mod_channels[probe].ports[0] # TODO: check it is the correct path + for iq, lo in _los(mod_channels, channels): + n = PortAddress.from_path(channels[iq].path).ports[0] getattr(mod, f"out{n}_lo_freq")(cast(OscillatorConfig, configs[lo]).frequency) @@ -141,7 +160,7 @@ def sequencer( # demodulation seq.demod_en_acq(acquisition is not AcquisitionType.RAW) - probe = _probe(channel_id, channels[channel_id]) + probe = _iqout(channel_id, channels[channel_id]) if probe is not None: freq = cast(IqConfig, configs[probe]).frequency lo = cast(IqChannel, channels[probe]).lo diff --git a/src/qibolab/_core/instruments/qblox/validate.py b/src/qibolab/_core/instruments/qblox/validate.py new file mode 100644 index 000000000..28ca7e382 --- /dev/null +++ b/src/qibolab/_core/instruments/qblox/validate.py @@ -0,0 +1,6 @@ +from qibolab._core.identifier import ChannelId +from qibolab._core.sequence import PulseSequence + + +def _assert_no_probe(ps: PulseSequence, probes: set[ChannelId]): + return