Skip to content

Commit

Permalink
refactor: Scope grouping helper as property
Browse files Browse the repository at this point in the history
Since it has limited utility outside the class anyhow
  • Loading branch information
alecandido committed Dec 13, 2024
1 parent 6ffaec1 commit 9b7bae7
Showing 1 changed file with 16 additions and 15 deletions.
31 changes: 16 additions & 15 deletions src/qibolab/_core/instruments/qblox/cluster.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
import qblox_instruments as qblox
from qblox_instruments.qcodes_drivers.module import Module

from qibolab._core.components.channels import Channel
from qibolab._core.components.configs import Config, LogConfig
from qibolab._core.execution_parameters import ExecutionParameters
from qibolab._core.identifier import ChannelId, Result
Expand Down Expand Up @@ -91,6 +90,21 @@ def _modules(self) -> dict[SlotId, Module]:
assert self._cluster is not None
return {mod.slot_idx: mod for mod in self._cluster.modules if mod.present()}

@cached_property
def _channels_by_module(self) -> dict[SlotId, list[tuple[ChannelId, PortAddress]]]:
addresses = {
name: PortAddress.from_path(ch.path) for name, ch in self.channels.items()
}
return {
k: [el[1] for el in g]
for k, g in groupby(
sorted(
(address.slot, (ch, address)) for ch, address in addresses.items()
),
key=lambda el: el[0],
)
}

@property
def sampling_rate(self) -> int:
return SAMPLING_RATE
Expand Down Expand Up @@ -137,7 +151,7 @@ def _upload(
self, sequences: dict[ChannelId, Sequence]
) -> dict[SlotId, dict[ChannelId, SequencerId]]:
sequencers = defaultdict(dict)
for mod, chs in _channels_by_module(self.channels).items():
for mod, chs in self._channels_by_module.items():
module = self._modules[mod]
assert len(module.sequencers) > len(chs)
# Map sequencers to specific outputs (but first disable all sequencer connections)
Expand Down Expand Up @@ -166,19 +180,6 @@ def _execute(
return {}


def _channels_by_module(
channels: dict[ChannelId, Channel],
) -> dict[SlotId, list[tuple[ChannelId, PortAddress]]]:
addresses = {name: PortAddress.from_path(ch.path) for name, ch in channels.items()}
return {
k: [el[1] for el in g]
for k, g in groupby(
sorted((address.slot, (ch, address)) for ch, address in addresses.items()),
key=lambda el: el[0],
)
}


def _split_channels(sequence: PulseSequence) -> dict[ChannelId, PulseSequence]:
def unwrap(pulse: PulseLike, output: bool) -> PulseLike:
return (
Expand Down

0 comments on commit 9b7bae7

Please sign in to comment.