Merge pull request #1026 from qiboteam/sequence-building

Sequence building

src/qibolab/native.py

@@ -22,6 +22,13 @@ class Native(ABC, PulseSequence):
     def create_sequence(self, *args, **kwargs) -> PulseSequence:
         """Create a sequence for single-qubit rotation."""
 
+    def __call__(self, *args, **kwargs) -> PulseSequence:
+        """Create a sequence for single-qubit rotation.
+
+        Alias to :meth:`create_sequence`.
+        """
+        return self.create_sequence(*args, **kwargs)
+
 
 class RxyFactory(Native):
     """Factory for pulse sequences that generate single-qubit rotations around

src/qibolab/sequence.py

@@ -93,17 +93,69 @@ def pulse_channels(self, pulse_id: int) -> 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]
 
-    def concatenate(self, other: "PulseSequence") -> None:
+    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
+
+    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.
         """
-        _synchronize(self, other.channels)
+        _synchronize(self, PulseSequence(other).channels | self.channels)
         self.extend(other)
 
+    def __ior__(self, other: Iterable[_Element]) -> "PulseSequence":
+        """Juxtapose two sequences.
+
+        Alias to :meth:`concatenate`.
+        """
+        self.juxtapose(other)
+        return self
+
+    def __or__(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
+
     def align(self, channels: list[ChannelId]) -> Align:
         """Introduce align commands to the sequence."""
         align = Align()

tests/integration/test_sequence.py

@@ -0,0 +1,39 @@
+from qibolab import create_platform
+from qibolab.execution_parameters import ExecutionParameters
+from qibolab.pulses import Delay
+
+
+def test_sequence_creation():
+    platform = create_platform("dummy")
+
+    single = platform.natives.single_qubit
+    two = platform.natives.two_qubit
+
+    # How a complex sequence is supposed to be constructed
+    # ----------------------------------------------------
+
+    p02 = two[(0, 2)]
+    p12 = two[(1, 2)]
+    q0 = single[0]
+    q1 = single[1]
+    q2 = single[2]
+    ch1 = platform.qubits[1]
+
+    seq = (
+        q1.RX()
+        | p12.CZ()
+        | [(ch1.drive, Delay(duration=6.5))]
+        | q2.RX()
+        | q0.RX12()
+        | p02.CZ()
+    )
+    for q in range(3):
+        seq |= single[q].MZ()
+
+    # ----------------------------------------------------
+
+    nshots = 17
+    res = platform.execute([seq], ExecutionParameters(nshots=nshots))
+
+    for r in res.values():
+        assert r.shape == (nshots,)

tests/test_native.py

@@ -48,6 +48,9 @@ def test_fixed_sequence_factory():
     assert np not in seq.channels
     assert np not in fseq2.channels
 
+    # test alias
+    assert factory() == seq
+
 
 @pytest.mark.parametrize(
     "args,amplitude,phase",

tests/test_sequence.py

@@ -1,3 +1,5 @@
+from copy import deepcopy
+
 from pydantic import TypeAdapter
 
 from qibolab.pulses import (
@@ -144,6 +146,79 @@ def test_concatenate():
     assert isinstance(s1[3][1], Pulse)
     assert s1[3][0] == "a"
 
+    # Check aliases
+    sa1 = deepcopy(s1)
+    sc1 = deepcopy(s1)
+    sa1 <<= s2
+    sc1.concatenate(s2)
+    assert sa1 == sc1
+    assert sc1 == s1 << s2
+
+
+def test_juxtapose():
+    p1 = Pulse(duration=40, amplitude=0.9, envelope=Drag(rel_sigma=0.2, beta=1))
+    sequence1 = PulseSequence([("ch1", p1)])
+    p2 = Pulse(duration=60, amplitude=0.9, envelope=Drag(rel_sigma=0.2, beta=1))
+    sequence2 = PulseSequence([("ch2", p2)])
+
+    sequence1.juxtapose(sequence2)
+    assert set(sequence1.channels) == {"ch1", "ch2"}
+    assert len(list(sequence1.channel("ch1"))) == 1
+    assert len(list(sequence1.channel("ch2"))) == 2
+    assert sequence1.duration == 40 + 60
+    channel, delay = sequence1[1]
+    assert channel == "ch2"
+    assert isinstance(delay, Delay)
+    assert delay.duration == 40
+
+    sequence3 = PulseSequence(
+        [
+            (
+                "ch2",
+                Pulse(duration=80, amplitude=0.9, envelope=Drag(rel_sigma=0.2, beta=1)),
+            ),
+            (
+                "ch3",
+                Pulse(
+                    duration=100, amplitude=0.9, envelope=Drag(rel_sigma=0.2, beta=1)
+                ),
+            ),
+        ]
+    )
+
+    sequence1.juxtapose(sequence3)
+    assert sequence1.channels == {"ch1", "ch2", "ch3"}
+    assert len(list(sequence1.channel("ch1"))) == 2
+    assert len(list(sequence1.channel("ch2"))) == 3
+    assert len(list(sequence1.channel("ch3"))) == 2
+    assert isinstance(next(iter(sequence1.channel("ch3"))), Delay)
+    assert sequence1.duration == 40 + 60 + 100
+    assert sequence1.channel_duration("ch1") == 40 + 60
+    assert sequence1.channel_duration("ch2") == 40 + 60 + 80
+    assert sequence1.channel_duration("ch3") == 40 + 60 + 100
+    delay = list(sequence1.channel("ch3"))[0]
+    assert isinstance(delay, Delay)
+    assert delay.duration == 100
+
+    # Check order preservation, even with various channels
+    vz = VirtualZ(phase=0.1)
+    s1 = PulseSequence([("a", p1), ("b", vz)])
+    s2 = PulseSequence([("a", vz), ("a", p2)])
+    s1.juxtapose(s2)
+    target_channels = ["a", "b", "b", "a", "a"]
+    target_pulse_types = [Pulse, VirtualZ, Delay, VirtualZ, Pulse]
+    for i, (channel, pulse) in enumerate(s1):
+        assert channel == target_channels[i]
+        assert isinstance(pulse, target_pulse_types[i])
+
+    # Check aliases
+    sa1 = deepcopy(s1)
+    sc1 = deepcopy(s1)
+    sa1 |= s2
+    sc1.juxtapose(s2)
+    assert sa1 == sc1
+    assert sc1 == s1 | s2
+
 
 def test_copy():
     sequence = PulseSequence(