Pam verify bug fixes

bqskit/compiler/compile.py

@@ -1259,6 +1259,7 @@ def build_seqpam_mapping_optimization_workflow(
                 PAMLayoutPass(num_layout_passes),
                 PAMRoutingPass(0.1),
                 post_pam_seq,
+                ApplyPlacement(),
                 UnfoldPass(),
             ],
         ),

bqskit/compiler/passdata.py

@@ -146,7 +146,12 @@ def placement(self, _val: Sequence[int]) -> None:
 
     @property
     def initial_mapping(self) -> list[int]:
-        """Return the initial mapping of logical to physical qudits."""
+        """
+        Return the initial mapping of logical to physical qudits.
+
+        This always maps how the logical qudits from the original circuit start
+        on the physical qudits of the current circuit.
+        """
         return self._initial_mapping
 
     @initial_mapping.setter
@@ -166,7 +171,12 @@ def initial_mapping(self, _val: Sequence[int]) -> None:
 
     @property
     def final_mapping(self) -> list[int]:
-        """Return the final mapping of logical to physical qudits."""
+        """
+        Return the final mapping of logical to physical qudits.
+
+        This always maps how the logical qudits from the original circuit end on
+        the physical qudits of the current circuit.
+        """
         return self._final_mapping
 
     @final_mapping.setter

bqskit/ir/circuit.py

@@ -1152,7 +1152,8 @@ def append_circuit(
                 (Default: False)
 
         Returns:
-            int: The starting cycle index of the appended circuit.
+            int: The starting cycle index of the appended circuit. If the
+                appended circuit is empty, then this will be -1.
 
         Raises:
             ValueError: If `circuit` is not the same size as `location`.
@@ -1178,16 +1179,14 @@ def append_circuit(
             op = Operation(CircuitGate(circuit, move), location, circuit.params)
             return self.append(op)
 
-        cycle_index: int | None = None
+        cycle_index = -1
+
         for op in circuit:
             mapped_location = [location[q] for q in op.location]
             ci = self.append(Operation(op.gate, mapped_location, op.params))
             if cycle_index is None:
                 cycle_index = ci
 
-        if cycle_index is None:
-            raise RuntimeError('Cannot append empty circuit.')
-
         return cycle_index
 
     def extend(self, ops: Iterable[Operation]) -> None:

bqskit/passes/mapping/apply.py

@@ -17,13 +17,13 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
         """Perform the pass's operation, see :class:`BasePass` for more."""
         model = data.model
         placement = data.placement
+
+        # Place circuit on the model according to the placement
         physical_circuit = Circuit(model.num_qudits, model.radixes)
         physical_circuit.append_circuit(circuit, placement)
         circuit.become(physical_circuit)
-        if 'final_mapping' in data:
-            pi = data['final_mapping']
-            data['final_mapping'] = [placement[p] for p in pi]
-        if 'initial_mapping' in data:
-            pi = data['initial_mapping']
-            data['initial_mapping'] = [placement[p] for p in pi]
+
+        # Update the relevant data variables
+        data.initial_mapping = [placement[p] for p in data.initial_mapping]
+        data.final_mapping = [placement[p] for p in data.final_mapping]
         data.placement = list(i for i in range(model.num_qudits))

bqskit/passes/mapping/embed.py

@@ -135,7 +135,10 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
 
         datas = []
         for graph in graphs:
-            model = MachineModel(circuit.num_qudits, graph, data.gate_set, data.model.radixes)
+            model = MachineModel(
+                circuit.num_qudits, graph,
+                data.gate_set, data.model.radixes,
+            )
             target_data = copy.deepcopy(data)
             target_data.model = model
             datas.append(target_data)

bqskit/passes/mapping/layout/pam.py

@@ -82,10 +82,11 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
         if not subgraph.is_fully_connected():
             raise RuntimeError('Cannot route circuit on disconnected qudits.')
 
-        pi = data.initial_mapping
+        pi = [i for i in range(circuit.num_qudits)]
+
         for _ in range(self.total_passes):
             self.forward_pass(circuit, pi, subgraph, perm_data)
             self.backward_pass(circuit, pi, subgraph)
 
-        data.initial_mapping = pi
-        _logger.info(f'Found layout: {str(pi)}')
+        self._apply_perm(pi, data.placement)
+        _logger.info(f'Found layout: {pi}, new placement: {data.placement}')

bqskit/passes/mapping/layout/sabre.py

@@ -69,15 +69,15 @@ def __init__(
 
     async def run(self, circuit: Circuit, data: PassData) -> None:
         """Perform the pass's operation, see :class:`BasePass` for more."""
-        subgraph = self.get_connectivity(circuit, data)
+        subgraph = data.connectivity
         if not subgraph.is_fully_connected():
             raise RuntimeError('Cannot layout circuit on disconnected qudits.')
 
-        pi = data.initial_mapping
+        pi = [i for i in range(circuit.num_qudits)]
+
         for _ in range(self.total_passes):
             self.forward_pass(circuit, pi, subgraph)
             self.backward_pass(circuit, pi, subgraph)
 
-        # select qubits
-        data.initial_mapping = pi
-        _logger.info(f'Found layout: {str(pi)}')
+        self._apply_perm(pi, data.placement)
+        _logger.info(f'Found layout: {pi}, new placement: {data.placement}')

bqskit/passes/mapping/pam.py

@@ -403,10 +403,3 @@ def _score_perm(
 
         pi[:] = pi_bkp[:]
         return front + extend
-
-    def _apply_perm(self, perm: Sequence[int], pi: list[int]) -> None:
-        """Apply the `perm` permutation to the current mapping `pi`."""
-        _logger.debug('applying permutation %s' % str(perm))
-        pi_c = {q: pi[perm[i]] for i, q in enumerate(sorted(perm))}
-        for q in perm:
-            pi[q] = pi_c[q]

bqskit/passes/mapping/routing/pam.py

@@ -1,7 +1,6 @@
 """This module implements the PAMRoutingPass."""
 from __future__ import annotations
 
-import copy
 import logging
 
 from bqskit.compiler.basepass import BasePass
@@ -21,26 +20,18 @@ class PAMRoutingPass(PermutationAwareMappingAlgorithm, BasePass):
 
     async def run(self, circuit: Circuit, data: PassData) -> None:
         """Perform the pass's operation, see :class:`BasePass` for more."""
-        model = self.get_model(circuit, data)
-        placement = self.get_placement(circuit, data)
-        subgraph = model.coupling_graph.get_subgraph(placement)
+        subgraph = data.connectivity
+        if not subgraph.is_fully_connected():
+            raise RuntimeError('Cannot route circuit on disconnected qudits.')
 
         perm_data: dict[CircuitPoint, PAMBlockTAPermData] = {}
         block_datas = data[ForEachBlockPass.key][-1]
         for block_data in block_datas:
             perm_data[block_data['point']] = block_data['permutation_data']
 
-        if not subgraph.is_fully_connected():
-            raise RuntimeError('Cannot route circuit on disconnected qudits.')
-
-        if 'initial_mapping' in data:
-            pi = copy.deepcopy(data['initial_mapping'])
-        else:
-            pi = [i for i in range(circuit.num_qudits)]
-
+        pi = [i for i in range(circuit.num_qudits)]
         out_data = self.forward_pass(circuit, pi, subgraph, perm_data, True)
-        if 'final_mapping' in data:
-            self._apply_perm(data['final_mapping'], pi)
-        data['final_mapping'] = pi
+        data.final_mapping = [pi[x] for x in data.final_mapping]
+
         _logger.info(f'Finished routing with layout: {str(pi)}')
         data[self.out_data_key] = out_data

bqskit/passes/mapping/routing/sabre.py

@@ -1,7 +1,6 @@
 """This module implements the GeneralizedSabreRoutingPass class."""
 from __future__ import annotations
 
-import copy
 import logging
 
 from bqskit.compiler.basepass import BasePass
@@ -22,16 +21,12 @@ class GeneralizedSabreRoutingPass(BasePass, GeneralizedSabreAlgorithm):
 
     async def run(self, circuit: Circuit, data: PassData) -> None:
         """Perform the pass's operation, see :class:`BasePass` for more."""
-        subgraph = self.get_connectivity(circuit, data)
+        subgraph = data.connectivity
         if not subgraph.is_fully_connected():
             raise RuntimeError('Cannot route circuit on disconnected qudits.')
 
-        if 'initial_mapping' in data:
-            pi = copy.deepcopy(data['initial_mapping'])
-        else:
-            pi = [i for i in range(circuit.num_qudits)]
-
+        pi = [i for i in range(circuit.num_qudits)]
         self.forward_pass(circuit, pi, subgraph, modify_circuit=True)
-        # TODO: if final_mapping is already in data, apply it first
-        data['final_mapping'] = pi
+        data.final_mapping = [pi[x] for x in data.final_mapping]
+
         _logger.info(f'Finished routing with layout: {str(pi)}')

bqskit/passes/mapping/sabre.py

@@ -9,6 +9,7 @@
 import numpy as np
 
 from bqskit.ir.circuit import Circuit
+from bqskit.ir.gates.circuitgate import CircuitGate
 from bqskit.ir.gates.constant.swap import SwapGate
 from bqskit.ir.operation import Operation
 from bqskit.ir.point import CircuitPoint
@@ -329,9 +330,13 @@ def backward_pass(
 
     def _can_exe(self, op: Operation, pi: list[int], cg: CouplingGraph) -> bool:
         """Return true if `op` is executable given the current mapping `pi`."""
-        # TODO: check if circuitgate of only 1-qubit gates
+        if isinstance(op.gate, CircuitGate):
+            if all(g.num_qudits == 1 for g in op.gate._circuit.gate_set):
+                return True
+
         if op.num_qudits == 1:
             return True
+
         physical_qudits = [pi[i] for i in op.location]
         return cg.get_subgraph(physical_qudits).is_fully_connected()
 
@@ -510,3 +515,10 @@ def _uphill_swaps(
                 if pi[center_qudit] == p1 or pi[center_qudit] == p2:
                     continue
                 yield (p1, p2)
+
+    def _apply_perm(self, perm: Sequence[int], pi: list[int]) -> None:
+        """Apply the `perm` permutation to the current mapping `pi`."""
+        _logger.debug('applying permutation %s' % str(perm))
+        pi_c = {q: pi[perm[i]] for i, q in enumerate(sorted(perm))}
+        for q in perm:
+            pi[q] = pi_c[q]

bqskit/passes/mapping/verify.py

@@ -49,18 +49,20 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
         # calculate exact panel unitary
         exact_circuit = Circuit(circuit.num_qudits, circuit.radixes)
         for op in circuit:
-            if not isinstance(op.gate, TaggedGate):
-                raise RuntimeError('Expected tagged gate.')
-
-            pi = op.gate.tag['pre_perm']
-            pf = op.gate.tag['post_perm']
-            in_utry = op.gate.tag['original_utry']
-            PI = PermutationMatrix.from_qubit_location(in_utry.num_qudits, pi)
-            PF = PermutationMatrix.from_qubit_location(in_utry.num_qudits, pf)
-            exact_circuit.append_gate(
-                ConstantUnitaryGate(PF @ in_utry @ PI.T),
-                op.location,
-            )
+            if isinstance(op.gate, TaggedGate):
+                pi = op.gate.tag['pre_perm']
+                pf = op.gate.tag['post_perm']
+                in_utry = op.gate.tag['original_utry']
+                n = in_utry.num_qudits
+                PI = PermutationMatrix.from_qubit_location(n, pi)
+                PF = PermutationMatrix.from_qubit_location(n, pf)
+                exact_circuit.append_gate(
+                    ConstantUnitaryGate(PF.T @ in_utry @ PI),
+                    op.location,
+                )
+
+            else:
+                exact_circuit.append_gate(op.gate, op.location, op.params)
 
         exact_unitary = exact_circuit.get_unitary()
 
@@ -74,15 +76,13 @@ class UnTagPAMBlockDataPass(BasePass):
     async def run(self, circuit: Circuit, data: PassData) -> None:
         """Perform the pass's operation, see :class:`BasePass` for more."""
         for cycle, op in circuit.operations_with_cycles():
-            if not isinstance(op.gate, TaggedGate):
-                raise RuntimeError('Expected tagged gate.')
-
-            circuit.replace_gate(
-                (cycle, op.location[0]),
-                op.gate.gate,
-                op.location,
-                op.params,
-            )
+            if isinstance(op.gate, TaggedGate):
+                circuit.replace_gate(
+                    (cycle, op.location[0]),
+                    op.gate.gate,
+                    op.location,
+                    op.params,
+                )
 
 
 class PAMVerificationSequence(PassAlias):

tests/compiler/compile/test_compile.py

@@ -1,26 +1,60 @@
 from __future__ import annotations
 
+from typing import Callable
+
+import pytest
+
 from bqskit.compiler.compile import compile
 from bqskit.compiler.compiler import Compiler
+from bqskit.compiler.machine import MachineModel
 from bqskit.ir.circuit import Circuit
+from bqskit.qis.graph import CouplingGraph
 from bqskit.qis.permutation import PermutationMatrix
 
 
+def default_model_gen(circuit: Circuit) -> MachineModel:
+    """Generate a default model for the given circuit."""
+    return MachineModel(circuit.num_qudits)
+
+
+def linear_model_gen(circuit: Circuit) -> MachineModel:
+    """Generate a linear model for the given circuit."""
+    return MachineModel(
+        circuit.num_qudits,
+        CouplingGraph.linear(circuit.num_qudits),
+    )
+
+
+@pytest.mark.parametrize(
+    'gen_model',
+    [
+        default_model_gen,
+        linear_model_gen,
+    ],
+    ids=[
+        'default',
+        'linear',
+    ],
+)
 def test_medium_circuit_compile(
     compiler: Compiler,
     optimization_level: int,
     medium_qasm_file: str,
+    gen_model: Callable[[Circuit], MachineModel],
 ) -> None:
     circuit = Circuit.from_file(medium_qasm_file)
+    model = gen_model(circuit)
     out_circuit, pi, pf = compile(
         circuit,
         optimization_level=optimization_level,
         with_mapping=True,
         compiler=compiler,
+        model=model,
     )
     in_utry = circuit.get_unitary()
     out_utry = out_circuit.get_unitary()
     PI = PermutationMatrix.from_qubit_location(out_circuit.num_qudits, pi)
     PF = PermutationMatrix.from_qubit_location(out_circuit.num_qudits, pf)
     error = out_utry.get_distance_from(PF.T @ in_utry @ PI, 1)
     assert error <= 1e-8
+    assert model.is_compatible(out_circuit)

tests/passes/mapping/test_sabre.py

@@ -9,6 +9,7 @@
 from bqskit.passes import GeneralizedSabreRoutingPass
 from bqskit.passes import GreedyPlacementPass
 from bqskit.passes import SetModelPass
+from bqskit.passes.mapping.apply import ApplyPlacement
 from bqskit.qis import PermutationMatrix
 from bqskit.qis.unitary.unitarymatrix import UnitaryMatrix
 
@@ -38,12 +39,17 @@ def test_simple(compiler: Compiler) -> None:
         GreedyPlacementPass(),
         GeneralizedSabreLayoutPass(),
         GeneralizedSabreRoutingPass(),
+        ApplyPlacement(),
     ]
 
     cc, data = compiler.compile(circuit, workflow, True)
     pi = data['initial_mapping']
     pf = data['final_mapping']
     PI = PermutationMatrix.from_qubit_location(5, pi)
     PF = PermutationMatrix.from_qubit_location(5, pf)
+    inactive = [i for i in range(cc.num_qudits) if i not in cc.active_qudits]
+    inactive.sort(reverse=True)
+    for i in inactive:
+        cc.pop_qudit(i)
     assert cc.get_unitary().get_distance_from(PF.T @ in_utry @ PI) < 1e-7
     assert all(e in cg for e in cc.coupling_graph)