Fix behaviour for blocking used channels and add type-check to channe…

…l equality in test cases.

qualang_tools/wirer/instruments/instrument_channels.py

@@ -1,3 +1,4 @@
+from dataclasses import asdict
 from typing import Type
 
 from qualang_tools.wirer.instruments.instrument_channel import InstrumentChannel, InstrumentChannelLfFemInput, \
@@ -98,5 +99,11 @@ def __getitem__(self, item):
     def __iter__(self):
         return iter(self.stack)
 
+    def __contains__(self, item: InstrumentChannel):
+        for channel_type in self.stack:
+            if asdict(item) in [asdict(channel) for channel in self.stack[channel_type]]:
+                return True
+        return False
+
     def items(self):
         return self.stack.items()

qualang_tools/wirer/wirer/wirer.py

@@ -3,6 +3,7 @@
 into potential channel specifications, then to allocate them to the first valid
 combination of instrument channels.
 """
+import copy
 from typing import List
 
 from .channel_specs import ChannelSpecLfFemSingle, ChannelSpecOpxPlusSingle, ChannelSpecMwFemSingle, \
@@ -28,6 +29,8 @@ def allocate_wiring(connectivity: Connectivity, instruments: Instruments,
 
     specs = connectivity.specs
 
+    used_channel_cache = copy.deepcopy(instruments.used_channels)
+
     specs_with_untyped_lines = set()
     for line_type in line_type_fill_order:
         for spec in specs:
@@ -45,8 +48,9 @@ def allocate_wiring(connectivity: Connectivity, instruments: Instruments,
     if not block_used_channels:
         for channel_type, used_channels in instruments.used_channels.items():
             for i, used_channel in enumerate(reversed(used_channels)):
-                instruments.available_channels.insert(0, used_channel)
-                instruments.used_channels.remove(used_channel)
+                if used_channel not in used_channel_cache:
+                    instruments.available_channels.insert(0, used_channel)
+                    instruments.used_channels.remove(used_channel)
 
 
 def _allocate_wiring(spec: WiringSpec, instruments: Instruments):

tests/wirer/conftest.py

@@ -1,9 +1,12 @@
+from dataclasses import asdict
+
 from qualang_tools.wirer.instruments import Instruments
 import pytest
 
 
 def pytest_configure():
     pytest.visualize_flag = False
+    pytest.channels_are_equal = lambda x, y: type(x) == type(y) and asdict(x) == asdict(y)
 
 @pytest.fixture(params=["lf-fem", "opx+"])
 def instruments_qw_soprano(request) -> Instruments:

tests/wirer/test_add_dummy_line.py

@@ -1,5 +1,3 @@
-from dataclasses import asdict
-
 import pytest
 from qualang_tools.wirer import Connectivity, allocate_wiring, visualize
 from qualang_tools.wirer.connectivity.element import Element, Reference
@@ -30,7 +28,7 @@ def test_add_dummy_line(instruments_2lf_2mw):
     # regression test
     test_element = connectivity.elements[Reference('test')]
     for i, channel in enumerate(test_element.channels['ch']):
-        assert asdict(channel) == asdict([
+        assert pytest.channels_are_equal(channel, [
             InstrumentChannelLfFemInput(con=1, port=1, slot=1),
             InstrumentChannelLfFemOutput(con=1, port=6, slot=1)
         ][i])

tests/wirer/test_wirer_channel_reuse.py

@@ -23,25 +23,57 @@ def test_5q_allocation_with_channel_reuse(instruments_2lf_2mw):
 
         allocate_wiring(connectivity, instruments_2lf_2mw, block_used_channels=False)
 
-    # if visualize_flag:
-    visualize(connectivity.elements, instruments_2lf_2mw.available_channels)
+    if visualize_flag:
+        visualize(connectivity.elements, instruments_2lf_2mw.available_channels)
 
     for qubit in [1, 3]:
         # resonator lines re-used for qubits 1 & 3
         for i, channel in enumerate(connectivity.elements[QubitReference(index=qubit)].channels[WiringLineType.RESONATOR]):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelMwFemInput(con=1, port=1, slot=3),
                 InstrumentChannelMwFemOutput(con=1, port=1, slot=3)
             ][i])
 
         # drive lines re-used for qubits 1 & 3
         for i, channel in enumerate(connectivity.elements[QubitReference(index=qubit)].channels[WiringLineType.DRIVE]):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelMwFemOutput(con=1, port=2, slot=3)
             ][i])
 
         # flux lines re-used for qubits 1 & 3
         for i, channel in enumerate(connectivity.elements[QubitReference(index=qubit)].channels[WiringLineType.FLUX]):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelLfFemOutput(con=1, port=1, slot=1),
             ][i])
+
+
+def test_alternating_blocking_of_used_channels(instruments_2lf_2mw):
+    connectivity = Connectivity()
+
+    connectivity.add_qubit_drive_lines(qubits=1)
+    allocate_wiring(connectivity, instruments_2lf_2mw, block_used_channels=False)
+
+    connectivity.add_qubit_drive_lines(qubits=2)
+    allocate_wiring(connectivity, instruments_2lf_2mw)
+
+    connectivity.add_qubit_drive_lines(qubits=3)
+    allocate_wiring(connectivity, instruments_2lf_2mw, block_used_channels=False)
+
+    connectivity.add_qubit_drive_lines(qubits=4)
+    allocate_wiring(connectivity, instruments_2lf_2mw)
+
+    if visualize_flag:
+        visualize(connectivity.elements, instruments_2lf_2mw.available_channels)
+
+    expected_ports = [
+        1,  # q1 allocated to 1, but channel isn't blocked
+        1,  # q2 allocated to 1, since it wasn't blocked
+        2,  # q3 allocated to 2, since it's the next available channel, but not blocked
+        2   # q4 allocated to 2, since it wasn't blocked
+    ]
+    for qubit_index in [1, 2, 3, 4]:
+        drive_channels = connectivity.elements[QubitReference(qubit_index)].channels[WiringLineType.DRIVE]
+        for i, channel in enumerate(drive_channels):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelMwFemOutput(con=1, slot=3, port=expected_ports[qubit_index-1])
+            ][i])

tests/wirer/test_wirer_constraining.py

@@ -32,7 +32,7 @@ def test_opx_plus_resonator_constraining():
 
     # resonator lines should be hard-coded to I=9, Q=10, rf_out=1
     for i, channel in enumerate(connectivity.elements[QubitReference(index=1)].channels[WiringLineType.RESONATOR]):
-        assert asdict(channel) == asdict([
+        assert pytest.channels_are_equal(channel, [
             InstrumentChannelOpxPlusInput(con=1, port=1, slot=None),
             InstrumentChannelOpxPlusInput(con=1, port=2, slot=None),
             InstrumentChannelOpxPlusOutput(con=1, port=9, slot=None),

tests/wirer/test_wirer_lf_and_mw.py

@@ -1,5 +1,3 @@
-from dataclasses import asdict
-
 import pytest
 
 from qualang_tools.wirer import *
@@ -27,30 +25,30 @@ def test_6q_allocation(instruments_2lf_2mw):
 
     for qubit in qubits:
         # flux channels should have some port as qubit index since they're allocated sequentially
-        flux_channels = connectivity.elements[QubitReference(qubit)].channels[WiringLineType.FLUX]
-        assert [asdict(ch) for ch in flux_channels] == [
-            asdict(InstrumentChannelLfFemOutput(con=1, port=qubit, slot=1))
-        ]
+        for i, channel in enumerate(connectivity.elements[QubitReference(qubit)].channels[WiringLineType.FLUX]):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelLfFemOutput(con=1, port=qubit, slot=1)
+            ][i])
 
         # resonators all on same feedline, so should be first available input + outputs channels on MW-FEM
-        resonator_channels = connectivity.elements[QubitReference(qubit)].channels[WiringLineType.RESONATOR]
-        assert [asdict(ch) for ch in resonator_channels] == [
-            asdict(InstrumentChannelMwFemInput(con=1, port=1, slot=3)),
-            asdict(InstrumentChannelMwFemOutput(con=1, port=1, slot=3))
-        ]
+        for i, channel in enumerate(connectivity.elements[QubitReference(qubit)].channels[WiringLineType.RESONATOR]):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelMwFemInput(con=1, port=1, slot=3),
+                InstrumentChannelMwFemOutput(con=1, port=1, slot=3)
+            ][i])
 
         # drive channels are on MW-FEM
-        drive_channels = connectivity.elements[QubitReference(qubit)].channels[WiringLineType.DRIVE]
-        assert [asdict(ch) for ch in drive_channels] == [
-            asdict(InstrumentChannelMwFemOutput(con=1, port=qubit+1, slot=3))
-        ]
+        for i, channel in enumerate(connectivity.elements[QubitReference(qubit)].channels[WiringLineType.DRIVE]):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelMwFemOutput(con=1, port=qubit+1, slot=3)
+            ][i])
 
     for i, pair in enumerate(qubit_pairs):
         # coupler channels should have some port as pair index since they're allocated sequentially, but on slot 2
-        coupler_channels = connectivity.elements[QubitPairReference(*pair)].channels[WiringLineType.COUPLER]
-        assert [asdict(ch) for ch in coupler_channels] == [
-            asdict(InstrumentChannelLfFemOutput(con=1, port=i+1, slot=2))
-        ]
+        for j, channel in enumerate(connectivity.elements[QubitPairReference(*pair)].channels[WiringLineType.COUPLER]):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelLfFemOutput(con=1, port=i+1, slot=2)
+            ][j])
 
 def test_4rr_allocation(instruments_2lf_2mw):
     connectivity = Connectivity()
@@ -67,9 +65,9 @@ def test_4rr_allocation(instruments_2lf_2mw):
 
     # resonators all on different feedlines, so should fill all 4 inputs of 2x MW-FEM
     for i, qubit in enumerate([1, 2, 3, 4]):
-        resonator_channels = connectivity.elements[QubitReference(qubit)].channels[WiringLineType.RESONATOR]
-        assert [asdict(ch) for ch in resonator_channels] == [
-            asdict(InstrumentChannelMwFemInput(con=1, port=[1, 2, 1, 2][i], slot=[3, 3, 7, 7][i])),
-            asdict(InstrumentChannelMwFemOutput(con=1, port=[1, 2, 1, 2][i], slot=[3, 3, 7, 7][i]))
-        ]
+        for j, channel in enumerate(connectivity.elements[QubitReference(qubit)].channels[WiringLineType.RESONATOR]):
+            assert pytest.channels_are_equal(channel, [
+                InstrumentChannelMwFemInput(con=1, port=[1, 2, 1, 2][i], slot=[3, 3, 7, 7][i]),
+                InstrumentChannelMwFemOutput(con=1, port=[1, 2, 1, 2][i], slot=[3, 3, 7, 7][i])
+            ][j])
 

tests/wirer/test_wirer_lf_and_octave.py

@@ -29,7 +29,7 @@ def test_rf_io_allocation(instruments_1OPX1Octave):
     for qubit in qubits:
         # resonator lines should be the same because only 1 feedline
         for i, channel in enumerate(connectivity.elements[QubitReference(index=qubit)].channels[WiringLineType.RESONATOR]):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelOpxPlusInput(con=1, port=1, slot=None),
                 InstrumentChannelOpxPlusInput(con=1, port=2, slot=None),
                 InstrumentChannelOpxPlusOutput(con=1, port=1, slot=None),
@@ -42,7 +42,7 @@ def test_rf_io_allocation(instruments_1OPX1Octave):
 
         # drive lines should be allocated sequentially
         for i, channel in enumerate(connectivity.elements[QubitReference(index=qubit)].channels[WiringLineType.DRIVE]):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelOpxPlusOutput(con=1, port=1+2*qubit, slot=None),
                 InstrumentChannelOpxPlusOutput(con=1, port=2+2*qubit, slot=None),
                 InstrumentChannelOctaveOutput(con=1, port=1+qubit, slot=None),

tests/wirer/test_wirer_lf_charge.py

@@ -25,6 +25,6 @@ def test_1q_allocation_flux_charge(instruments_2lf_2mw):
         assert len(charge_channels) == 1
 
         for i, channel in enumerate(charge_channels):
-            assert asdict(channel) == asdict([
+            assert pytest.channels_are_equal(channel, [
                 InstrumentChannelLfFemOutput(con=1, port=qubit_index, slot=1)
             ][i])