From be64f3c78dc5872571136e69c4650a458c138605 Mon Sep 17 00:00:00 2001 From: Alessandro Candido Date: Thu, 19 Dec 2024 14:25:55 +0100 Subject: [PATCH] fix: Move sequence compilation to its subpackage --- .../_core/instruments/qblox/cluster.py | 52 +------------------ src/qibolab/_core/instruments/qblox/config.py | 4 +- src/qibolab/_core/instruments/qblox/log.py | 2 +- .../instruments/qblox/sequence/sequence.py | 51 +++++++++++++++++- 4 files changed, 55 insertions(+), 54 deletions(-) diff --git a/src/qibolab/_core/instruments/qblox/cluster.py b/src/qibolab/_core/instruments/qblox/cluster.py index cf94293dc..7d1263d04 100644 --- a/src/qibolab/_core/instruments/qblox/cluster.py +++ b/src/qibolab/_core/instruments/qblox/cluster.py @@ -10,13 +10,14 @@ from qibolab._core.execution_parameters import ExecutionParameters from qibolab._core.identifier import ChannelId, Result from qibolab._core.instruments.abstract import Controller -from qibolab._core.pulses import PulseId, PulseLike, Readout +from qibolab._core.pulses import PulseId from qibolab._core.sequence import PulseSequence from qibolab._core.sweeper import ParallelSweepers from . import config from .config import PortAddress, SeqeuencerMap, SlotId from .log import Logger +from .sequence import Sequence, compile # from qcodes.instrument import find_or_create_instrument @@ -122,52 +123,3 @@ def _execute(self, sequencers: SeqeuencerMap) -> dict[PulseId, Result]: module.start_sequencer() return {} - - -def _split_channels(sequence: PulseSequence) -> dict[ChannelId, PulseSequence]: - def unwrap(pulse: PulseLike, output: bool) -> PulseLike: - return ( - pulse - if not isinstance(pulse, Readout) - else pulse.probe if output else pulse.acquisition - ) - - def unwrap_seq(seq: PulseSequence, output: bool) -> PulseSequence: - return PulseSequence((ch, unwrap(p, output)) for ch, p in seq) - - def ch_pulses(channel: ChannelId) -> PulseSequence: - return PulseSequence((ch, pulse) for ch, pulse in sequence if ch == channel) - - def probe(channel: ChannelId) -> ChannelId: - return channel.split("/")[0] + "/probe" - - def split(channel: ChannelId) -> dict[ChannelId, PulseSequence]: - seq = ch_pulses(channel) - readouts = any(isinstance(p, Readout) for _, p in seq) - assert not readouts or probe(channel) not in sequence.channels - return ( - {channel: seq} - if not readouts - else { - channel: unwrap_seq(seq, output=False), - probe(channel): unwrap_seq(seq, output=True), - } - ) - - return { - ch: seq for channel in sequence.channels for ch, seq in split(channel).items() - } - - -def _prepare( - sequence: PulseSequence, - sweepers: list[ParallelSweepers], - options: ExecutionParameters, - sampling_rate: float, -) -> dict[ChannelId, Sequence]: - return { - ch: Sequence.from_pulses(seq, sweepers, options, sampling_rate) - for ch, seq in _split_channels(sequence).items() - } - - diff --git a/src/qibolab/_core/instruments/qblox/config.py b/src/qibolab/_core/instruments/qblox/config.py index 5feed193a..0d96f7df0 100644 --- a/src/qibolab/_core/instruments/qblox/config.py +++ b/src/qibolab/_core/instruments/qblox/config.py @@ -1,10 +1,10 @@ from typing import Optional -from qblox_instruments.qcodes_drivers.sequencer import Sequencer from qblox_instruments.qcodes_drivers.module import Module +from qblox_instruments.qcodes_drivers.sequencer import Sequencer -from qibolab._core.serialize import Model from qibolab._core.identifier import ChannelId +from qibolab._core.serialize import Model from .sequence import Sequence diff --git a/src/qibolab/_core/instruments/qblox/log.py b/src/qibolab/_core/instruments/qblox/log.py index 6fe63b0da..372c468c9 100644 --- a/src/qibolab/_core/instruments/qblox/log.py +++ b/src/qibolab/_core/instruments/qblox/log.py @@ -6,8 +6,8 @@ from qibolab._core.components.configs import Configs, LogConfig from qibolab._core.identifier import ChannelId -from .sequence import Sequence from .config import SeqeuencerMap +from .sequence import Sequence __all__ = [] diff --git a/src/qibolab/_core/instruments/qblox/sequence/sequence.py b/src/qibolab/_core/instruments/qblox/sequence/sequence.py index de52b9948..825987e5f 100644 --- a/src/qibolab/_core/instruments/qblox/sequence/sequence.py +++ b/src/qibolab/_core/instruments/qblox/sequence/sequence.py @@ -4,6 +4,8 @@ from pydantic import PlainSerializer, PlainValidator from qibolab._core.execution_parameters import ExecutionParameters +from qibolab._core.identifier import ChannelId +from qibolab._core.pulses import PulseLike, Readout from qibolab._core.sequence import PulseSequence from qibolab._core.serialize import Model from qibolab._core.sweeper import ParallelSweepers @@ -14,7 +16,7 @@ from .program import program from .waveforms import Waveform, Waveforms, waveform_indices, waveforms -__all__ = ["Sequence"] +__all__ = ["Sequence", "compile"] Weight = Waveform @@ -55,3 +57,50 @@ def from_pulses( sampling_rate, ), ) + + +def _split_channels(sequence: PulseSequence) -> dict[ChannelId, PulseSequence]: + def unwrap(pulse: PulseLike, output: bool) -> PulseLike: + return ( + pulse + if not isinstance(pulse, Readout) + else pulse.probe if output else pulse.acquisition + ) + + def unwrap_seq(seq: PulseSequence, output: bool) -> PulseSequence: + return PulseSequence((ch, unwrap(p, output)) for ch, p in seq) + + def ch_pulses(channel: ChannelId) -> PulseSequence: + return PulseSequence((ch, pulse) for ch, pulse in sequence if ch == channel) + + def probe(channel: ChannelId) -> ChannelId: + return channel.split("/")[0] + "/probe" + + def split(channel: ChannelId) -> dict[ChannelId, PulseSequence]: + seq = ch_pulses(channel) + readouts = any(isinstance(p, Readout) for _, p in seq) + assert not readouts or probe(channel) not in sequence.channels + return ( + {channel: seq} + if not readouts + else { + channel: unwrap_seq(seq, output=False), + probe(channel): unwrap_seq(seq, output=True), + } + ) + + return { + ch: seq for channel in sequence.channels for ch, seq in split(channel).items() + } + + +def compile( + sequence: PulseSequence, + sweepers: list[ParallelSweepers], + options: ExecutionParameters, + sampling_rate: float, +) -> dict[ChannelId, Sequence]: + return { + ch: Sequence.from_pulses(seq, sweepers, options, sampling_rate) + for ch, seq in _split_channels(sequence).items() + }