Skip to content

Commit

Permalink
fix: Filter probe channels
Browse files Browse the repository at this point in the history
To restrict their usage in association with acquisition
  • Loading branch information
alecandido committed Feb 7, 2025
1 parent 4bd2a43 commit 68bd8f3
Show file tree
Hide file tree
Showing 3 changed files with 52 additions and 14 deletions.
17 changes: 15 additions & 2 deletions src/qibolab/_core/instruments/qblox/cluster.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"]

Expand All @@ -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 = {
Expand All @@ -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],
)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
):
Expand Down
43 changes: 31 additions & 12 deletions src/qibolab/_core/instruments/qblox/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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,
):
Expand All @@ -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)


Expand Down Expand Up @@ -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
Expand Down
6 changes: 6 additions & 0 deletions src/qibolab/_core/instruments/qblox/validate.py
Original file line number Diff line number Diff line change
@@ -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

0 comments on commit 68bd8f3

Please sign in to comment.