"""PulseSequence class."""
from collections import UserList, defaultdict
from collections.abc import Callable, Iterable
from functools import cache
from typing import Any, Union
import numpy as np
from pydantic import TypeAdapter
from pydantic_core import core_schema
from qibolab._core.pulses.pulse import PulseId, VirtualZ
from .identifier import ChannelId
from .pulses import Acquisition, Align, Delay, Pulse, PulseLike, Readout
__all__ = ["PulseSequence"]
_Element = tuple[ChannelId, PulseLike]
InputOps = Union[Readout, Acquisition]
_adapted_sequence = TypeAdapter(list[_Element])
def _synchronize(sequence: "PulseSequence", channels: Iterable[ChannelId]) -> None:
"""Helper for ``concatenate`` and ``align_to_delays``.
Modifies given ``sequence`` in-place!
"""
durations = {ch: sequence.channel_duration(ch) for ch in channels}
max_duration = max(durations.values(), default=0.0)
for ch, duration in durations.items():
delay = max_duration - duration
if delay > 0:
sequence.append((ch, Delay(duration=delay)))
[docs]
class PulseSequence(UserList[_Element]):
"""Synchronized sequence of control instructions across multiple channels.
The sequence is a linear stream of instructions, which may be
executed in parallel over multiple channels.
Each instruction is composed by the pulse-like object representing
the action, and the channel on which it should be performed.
"""
@classmethod
def __get_pydantic_core_schema__(
cls, source_type: Any, handler: Callable[[Any], core_schema.CoreSchema]
) -> core_schema.CoreSchema:
schema = handler(list[_Element])
return core_schema.no_info_after_validator_function(
cls._validate,
schema,
serialization=core_schema.plain_serializer_function_ser_schema(
cls._serialize, info_arg=False
),
)
@classmethod
def _validate(cls, value):
return cls(value)
@staticmethod
def _serialize(value):
return _adapted_sequence.dump_python(list(value))
[docs]
@classmethod
def load(cls, value: list[tuple[str, PulseLike]]):
return TypeAdapter(cls).validate_python(value)
@property
def duration(self) -> float:
"""Duration of the entire sequence."""
return max((self.channel_duration(ch) for ch in self.channels), default=0.0)
@property
def channels(self) -> set[ChannelId]:
"""Channels involved in the sequence."""
return {ch for (ch, _) in self}
[docs]
def channel(self, channel: ChannelId) -> Iterable[PulseLike]:
"""Isolate pulses on a given channel."""
return (pulse for (ch, pulse) in self if ch == channel)
[docs]
def channel_duration(self, channel: ChannelId) -> float:
"""Duration of the given channel."""
sequence = self.align_to_delays()
return sum(pulse.duration for pulse in sequence.channel(channel))
[docs]
def pulse_channels(self, pulse_id: PulseId) -> list[ChannelId]:
"""Find channels on which a pulse with a given id plays."""
return [channel for channel, pulse in self if pulse.id == pulse_id]
[docs]
def concatenate(self, other: Iterable[_Element]) -> None:
"""Concatenate two sequences.
Appends ``other`` in-place such that the result is:
- ``self``
- necessary delays to synchronize channels
- ``other``
Guarantees that the all the channels in the concatenated
sequence will start simultaneously
"""
_synchronize(self, PulseSequence(other).channels)
self.extend(other)
def __ilshift__(self, other: Iterable[_Element]) -> "PulseSequence":
"""Juxtapose two sequences.
Alias to :meth:`concatenate`.
"""
self.concatenate(other)
return self
def __lshift__(self, other: Iterable[_Element]) -> "PulseSequence":
"""Juxtapose two sequences.
A copy is made, and no input is altered.
Other than that, it is based on :meth:`concatenate`.
"""
copy = self.copy()
copy <<= other
return copy
[docs]
def juxtapose(self, other: Iterable[_Element]) -> None:
"""Juxtapose two sequences.
Appends ``other`` in-place such that the result is:
- ``self``
- necessary delays to synchronize channels
- ``other``
Guarantee simultaneous start and no overlap.
.. deprecated:: 0.3.0
This is deprecated since 0.2.14. Use align_to_delays instead.
"""
_synchronize(self, PulseSequence(other).channels | self.channels)
self.extend(other)
def __ior__(self, other: Iterable[_Element]) -> "PulseSequence":
"""Pipe two sequences. ``other'' starts after ``self`` ends."""
other_channels = PulseSequence(other).channels
self.align(self.channels | other_channels)
self.extend(other)
return self
def __or__(self, other: Iterable[_Element]) -> "PulseSequence":
"""Pipe two sequences.
A copy is made, and no input is altered.
"""
copy = self.copy()
copy |= other
return copy
[docs]
def align(self, channels: Iterable[ChannelId]) -> Align:
"""Introduce align commands to the sequence."""
align = Align()
for channel in channels:
self.append((channel, align))
return align
[docs]
def align_to_delays(self) -> "PulseSequence":
"""Compile align commands to delays."""
# keep track of ``Align`` command that were already played
# because the same ``Align`` will appear on multiple channels
# in the sequence
processed_aligns = set()
new = type(self)()
for channel, pulse in self:
if isinstance(pulse, Align):
if pulse.id not in processed_aligns:
channels = self.pulse_channels(pulse.id)
_synchronize(new, channels)
processed_aligns.add(pulse.id)
else:
new.append((channel, pulse))
return new
[docs]
def trim(self) -> "PulseSequence":
"""Drop final delays.
The operation is not in place, and does not modify the original
sequence.
"""
terminated = set()
new = []
for ch, pulse in reversed(self):
if ch not in terminated:
if isinstance(pulse, Delay):
continue
terminated.add(ch)
new.append((ch, pulse))
return type(self)(reversed(new))
@property
def acquisitions(self) -> list[tuple[ChannelId, InputOps]]:
"""Return list of the readout pulses in this sequence.
.. note::
This selects only the :class:`Acquisition` events, and not all the
instructions directed to an acquistion channel
"""
# pulse filter needed to exclude delays
return [(ch, p) for ch, p in self if isinstance(p, (Acquisition, Readout))]
@property
def split_readouts(self) -> "PulseSequence":
"""Split readout operations in its constituents.
This will also double the rest of the channels (mainly delays) on which the
readouts are placed, assuming the probe channels to be absent.
.. note::
Since :class:`Readout` is only placed on an acquisition channel, the name of
the associated probe channel is actually unknown.
This function assumes the convention that the relevant channels are named
``.../acquisition`` and ``.../probe``.
"""
def unwrap(pulse: PulseLike, double: bool) -> tuple[PulseLike, ...]:
return (
(pulse.acquisition, pulse.probe)
if isinstance(pulse, Readout)
else (pulse, pulse)
if double
else (pulse,)
)
@cache
def probe(channel: ChannelId) -> ChannelId:
return channel.split("/")[0] + "/probe"
readouts = {ch for ch, p in self if isinstance(p, Readout)}
return type(self)(
[
(ch_, p_)
for ch, p in self
for ch_, p_ in zip((ch, probe(ch)), unwrap(p, ch in readouts))
]
)
@property
def by_channel(self) -> dict[ChannelId, list[PulseLike]]:
"""Separate sequence into channels dictionary."""
seqs = defaultdict(list)
for ch, pulse in self:
seqs[ch].append(pulse)
return seqs
[docs]
def to_vzs(self) -> "PulseSequence":
"""Transform :class:`Pulse` relative phases to :class:`VirtualZ` elements.
The relative phase for a pulse can be applied to the pulse itself, or applying
the inverse transformation to the reference frame, right before and right after
the pulse itself.
This method realizes this mapping from :class:`Pulse` relative phases to
:class:`VirtualZ`, leaving all relative phases to their default value of
``0.0``.
The method is compatible with the presence of further :class:`VirtualZ` in the
original sequence.
The basic formula just relies on the composition of Pauli matrices'
exponential. Cf. :cite:t:`Manenti:2023zzn`, Sec. 14.6.4 (or any other similar
reference).
"""
return PulseSequence(
[
el
for els in (
[(ch, ev)]
if not isinstance(ev, Pulse) or np.isclose(ev.relative_phase, 0)
else [
(ch, VirtualZ(phase=ev.relative_phase)),
(ch, ev.model_copy(update={"relative_phase": 0})),
(ch, VirtualZ(phase=-ev.relative_phase)),
]
for ch, ev in self
)
for el in els
]
)
[docs]
def to_relative_phases(self) -> "PulseSequence":
"""Embed :class:`VirtualZ` transformations into :class:`Pulse` relative phases.
This method implements a transformation which is the conceptual opposite of the
one realized by :meth:`to_vzs`.
The two functions are not exactly inverse to each other, since there could be
multiple :class:`VirtualZ` in the original sequence (which would be collapsed by
a round trip), and pulses' relative phases and frame transformations can
coexist.
"""
seq = PulseSequence()
phases = defaultdict(float)
for ch, ev in self:
if isinstance(ev, VirtualZ):
phases[ch] += ev.phase
elif isinstance(ev, Pulse):
seq.append(
(
ch,
ev.model_copy(
update={"relative_phase": ev.relative_phase + phases[ch]}
),
)
)
else:
seq.append((ch, ev))
return seq
[docs]
def collect_vzs(self) -> "PulseSequence":
"""Collect subsequent :class:`VirtualZ` rotations.
For each channel, it divides :class:`VirtualZ` in groups delimited by pulses.
Each group is collected and transformed in a single :class:`VirtualZ`, just summing
the angles.
"""
seq = PulseSequence()
phases = defaultdict(float)
def collect(ch: ChannelId):
if not np.isclose(phases[ch], 0.0):
seq.append((ch, VirtualZ(phase=phases[ch])))
phases[ch] = 0.0
for ch, ev in self:
if isinstance(ev, VirtualZ):
phases[ch] += ev.phase
continue
if isinstance(ev, Pulse):
collect(ch)
seq.append((ch, ev))
for ch in self.channels:
collect(ch)
return seq