Add "MappingManager" class and port older routing solution (#95)

* Copy code from private repository

* Add binders for mapping module

* Adding mapping module to setup.py

* Add shared_ptr to Architecture subclasses in binder file

* Port python test for mapping module

pytket/CMakeLists.txt

@@ -60,3 +60,4 @@ build_module(pauli binders/pauli.cpp)
 build_module(logging binders/logging.cpp)
 build_module(utils_serialization binders/utils_serialization.cpp)
 build_module(tailoring binders/tailoring.cpp)
+build_module(mapping binders/mapping.cpp)

pytket/binders/mapping.cpp

@@ -0,0 +1,80 @@
+#include <pybind11/functional.h>
+#include <pybind11/pybind11.h>
+#include <pybind11/stl.h>
+
+#include "Mapping/LexiRoute.hpp"
+#include "Mapping/MappingManager.hpp"
+#include "Mapping/RoutingMethodCircuit.hpp"
+
+namespace py = pybind11;
+
+namespace tket {
+PYBIND11_MODULE(mapping, m) {
+  py::class_<RoutingMethod>(
+      m, "RoutingMethod",
+      "Parent class for RoutingMethod, for inheritance purposes only, not for "
+      "usage.")
+      .def(py::init<>());
+
+  py::class_<RoutingMethodCircuit, RoutingMethod>(
+      m, "RoutingMethodCircuit",
+      "The RoutingMethod class captures a method for partially mapping logical"
+      "subcircuits to physical operations as permitted by some architecture. "
+      "Ranked RoutingMethod objects are used by the MappingManager to route "
+      "whole circuits.")
+      .def(
+          py::init<
+              const std::function<std::tuple<Circuit, unit_map_t, unit_map_t>(
+                  const Circuit&, const ArchitecturePtr&)>&,
+              const std::function<bool(const Circuit&, const ArchitecturePtr&)>,
+              unsigned, unsigned>(),
+          "Constructor for a routing method defined by partially routing "
+          "subcircuits.\n\n:param route_subcircuit: A function declaration "
+          "that given a Circuit and Architecture object, returns a tuple "
+          "containing a new modified circuit, the initial logical to physical "
+          "qubit mapping of the modified circuit and the permutation of "
+          "'logical to physical qubit mapping given operations in the "
+          "modified circuit\n:param check_subcircuit: A function declaration "
+          "that given a Circuit and Architecture object, returns a bool "
+          "stating whether the given method can modify the "
+          "given circuit\n:param max_size: The maximum number of gates "
+          "permitted in a subcircuit\n:param max_depth: The maximum permitted "
+          "depth of a subcircuit.",
+          py::arg("route_subcircuit"), py::arg("check_subcircuit"),
+          py::arg("max_size"), py::arg("max_depth"));
+
+  py::class_<LexiRouteRoutingMethod, RoutingMethod>(
+      m, "LexiRouteRoutingMethod",
+      "Defines a RoutingMethod object for mapping circuits that uses the "
+      "Lexicographical Comparison approach outlined in arXiv:1902.08091.")
+      .def(
+          py::init<unsigned>(),
+          "LexiRoute constructor.\n\n:param lookahead: Maximum depth of "
+          "lookahead "
+          "employed when picking SWAP for purpose of logical to physical "
+          "mapping.");
+
+  py::class_<MappingManager>(
+      m, "MappingManager",
+      "Defined by a pytket Architecture object, maps Circuit logical Qubits "
+      "to Physically permitted Architecture qubits. Mapping is completed by "
+      "sequential routing (full or partial) of subcircuits. Custom method for "
+      "routing (full or partial) of subcircuits can be defined in python "
+      "layer.")
+      .def(
+          py::init<const ArchitecturePtr&>(),
+          "MappingManager constructor.\n\n:param architecture: pytket "
+          "Architecure object MappingManager object defined by.",
+          py::arg("architecture"))
+      .def(
+          "route_circuit", &MappingManager::route_circuit,
+          "Maps from given logical circuit to physical circuit. Modification "
+          "defined by route_subcircuit, but typically this proceeds by "
+          "insertion of SWAP gates that permute logical qubits on physical "
+          "qubits. \n\n:param circuit: pytket circuit to be mapped"
+          "\n:param routing_methods: Ranked methods to use for routing "
+          "subcircuits. In given order, each method is sequentially checked "
+          "for viability, with the first viable method being used.",
+          py::arg("circuit"), py::arg("routing_methods"));
+}
+}  // namespace tket

pytket/binders/routing.cpp

@@ -93,7 +93,7 @@ std::pair<Circuit, qubit_mapping_t> route(
 }
 
 PYBIND11_MODULE(routing, m) {
-  py::class_<Architecture>(
+  py::class_<Architecture, ArchitecturePtr>(
       m, "Architecture",
       "The base architecture class, describing the connectivity of "
       "qubits on a device.")
@@ -106,11 +106,26 @@ PYBIND11_MODULE(routing, m) {
           "operations",
           py::arg("connections"))
       .def(
-          py::init<const std::vector<std::pair<Node, Node>>>(),
+          py::init<const std::vector<std::pair<Node, Node>> &>(),
           "The constructor for an architecture with connectivity "
           "between qubits.\n\n:param connections: A list of pairs "
           "representing Nodes that can perform two-qubit operations",
           py::arg("connections"))
+      .def(
+          "__repr__",
+          [](const Architecture &arc) {
+            return "<tket::Architecture, nodes=" +
+                   std::to_string(arc.n_uids()) + ">";
+          })
+      .def(
+          "get_distance", &Architecture::get_distance,
+          "given two nodes in Architecture, "
+          "returns distance between them",
+          py::arg("node_0"), py::arg("node_1"))
+      .def(
+          "get_adjacent_nodes", &Architecture::get_neighbour_uids,
+          "given a node, returns adjacent nodes in Architecture.",
+          py::arg("node"))
       .def_property_readonly(
           "nodes", &Architecture::get_all_uids_vec,
           "Returns all nodes of architecture as UnitID objects. ")
@@ -147,7 +162,7 @@ PYBIND11_MODULE(routing, m) {
           "equal "
           "if they have the same set of nodes and the same connections between "
           "nodes.");
-  py::class_<SquareGrid, Architecture>(
+  py::class_<SquareGrid, std::shared_ptr<SquareGrid>, Architecture>(
       m, "SquareGrid",
       "Architecture class for qubits arranged in a square lattice of "
       "given number of rows and columns. Qubits are arranged with qubits "
@@ -196,7 +211,7 @@ PYBIND11_MODULE(routing, m) {
                ", columns=" + std::to_string(arc.get_columns()) +
                ", layers=" + std::to_string(arc.get_layers()) + ">";
       });
-  py::class_<FullyConnected, Architecture>(
+  py::class_<FullyConnected, std::shared_ptr<FullyConnected>, Architecture>(
       m, "FullyConnected",
       "Architecture class for number of qubits connected to every other "
       "qubit.")
@@ -210,7 +225,7 @@ PYBIND11_MODULE(routing, m) {
         return "<tket::FullyConnected, nodes=" + std::to_string(arc.n_uids()) +
                ">";
       });
-  py::class_<RingArch, Architecture>(
+  py::class_<RingArch, std::shared_ptr<RingArch>, Architecture>(
       m, "RingArch",
       "Architecture class for number of qubits arranged in a ring.")
       .def(

pytket/pytket/mapping/__init__.py

@@ -0,0 +1,14 @@
+# Copyright 2019-2021 Cambridge Quantum Computing
+#
+# You may not use this file except in compliance with the Licence.
+# You may obtain a copy of the Licence in the LICENCE file accompanying
+# these documents or at:
+#
+#     https://cqcl.github.io/pytket/build/html/licence.html
+"""The mapping module provides an API to interact with the
+ tket :py:class:`MappingManager` suite, with methods for
+ mapping logical circuits to physical circuits and for
+ defining custom routing solutions. This module is provided
+ in binary form during the PyPI installation."""
+
+from pytket._tket.mapping import *  # type: ignore

pytket/setup.py

@@ -195,6 +195,7 @@ def build_extension(self, ext):
     "routing",
     "transform",
     "tailoring",
+    "mapping",
 ]
 
 

pytket/tests/mapping_test.py

@@ -0,0 +1,195 @@
+# Copyright 2019-2021 Cambridge Quantum Computing
+#
+# You may not use this file except in compliance with the Licence.
+# You may obtain a copy of the Licence in the LICENCE file accompanying
+# these documents or at:
+#
+#     https://cqcl.github.io/pytket/build/html/licence.html
+
+from pytket.mapping import MappingManager, RoutingMethodCircuit, LexiRouteRoutingMethod  # type: ignore
+from pytket.routing import Architecture  # type: ignore
+from pytket import Circuit, OpType
+from pytket.circuit import Node  # type: ignore
+from typing import Tuple, Dict
+
+
+# simple deterministic heuristic used for testing purposes
+def route_subcircuit_func(
+    circuit: Circuit, architecture: Architecture
+) -> Tuple[Circuit, Dict[Node, Node], Dict[Node, Node]]:
+    #     make a replacement circuit with identical unitds
+    replacement_circuit = Circuit()
+    for qb in circuit.qubits:
+        replacement_circuit.add_qubit(qb)
+    for bit in circuit.bits:
+        replacement_circuit.add_bit(bit)
+
+    # "place" unassigned logical qubits to physical qubits
+    unused_nodes = list(architecture.nodes)
+    relabelling_map = dict()
+
+    for qb in circuit.qubits:
+        for n in unused_nodes:
+            if n == qb:
+                unused_nodes.remove(n)
+
+    for qb in circuit.qubits:
+        if qb not in set(architecture.nodes):
+            relabelling_map[qb] = unused_nodes.pop()
+        else:
+            #           this is so later architecture.get_distance works
+            #           yes this is obviously bad, buts its a simple test heuristic so who cares?!
+            relabelling_map[qb] = qb
+
+    replacement_circuit.rename_units(relabelling_map)
+    permutation_map = dict()
+    for qb in replacement_circuit.qubits:
+        permutation_map[qb] = qb
+
+    #   very simple heuristic -> the first time a physically invalid CX is encountered, add a SWAP
+    #   then add all remaining gates as is (using updated physical mapping)
+    #   note this is possible as routing accepts partially solved problems
+    max_swaps = 1
+    swaps_added = 0
+    for com in circuit.get_commands():
+        rp_qubits = [permutation_map[relabelling_map[q]] for q in com.qubits]
+        if len(com.qubits) > 2:
+            raise ValueError("Command must have maximum two qubits")
+        if len(com.qubits) == 1:
+            replacement_circuit.add_gate(com.op.type, rp_qubits)
+        if len(com.qubits) == 2:
+            if swaps_added < max_swaps:
+                #               get node references for some stupid reason...
+                #               theres some stupid casting issue
+                #               just passing qubits didnt work.. whatever
+                for n in architecture.nodes:
+                    if n == rp_qubits[0]:
+                        n0 = n
+                    if n == rp_qubits[1]:
+                        n1 = n
+                distance = architecture.get_distance(n0, n1)
+                if distance > 1:
+                    for node in architecture.get_adjacent_nodes(n0):
+                        if architecture.get_distance(
+                            node, n1
+                        ) < architecture.get_distance(n0, n1):
+                            replacement_circuit.add_gate(
+                                OpType.SWAP, [rp_qubits[0], node]
+                            )
+
+                            permutation_map[rp_qubits[0]] = node
+                            permutation_map[node] = rp_qubits[0]
+                            rp_qubits = [
+                                permutation_map[relabelling_map[q]] for q in com.qubits
+                            ]
+                            swaps_added = swaps_added + 1
+                            break
+
+            replacement_circuit.add_gate(com.op.type, rp_qubits)
+
+    return (replacement_circuit, relabelling_map, permutation_map)
+
+
+def check_subcircuit_func_true(circuit: Circuit, architecture: Architecture) -> bool:
+    return True
+
+
+def check_subcircuit_func_false(circuit: Circuit, architecture: Architecture) -> bool:
+    return False
+
+
+def test_LexiRouteRoutingMethod() -> None:
+    test_c = Circuit(3).CX(0, 1).CX(0, 2).CX(1, 2)
+    nodes = [Node("test", 0), Node("test", 1), Node("test", 2)]
+    test_a = Architecture([[nodes[0], nodes[1]], [nodes[1], nodes[2]]])
+    test_mm = MappingManager(test_a)
+    test_mm.route_circuit(test_c, [LexiRouteRoutingMethod(50)])
+    routed_commands = test_c.get_commands()
+
+    assert routed_commands[0].op.type == OpType.CX
+    assert routed_commands[0].qubits == [nodes[1], nodes[0]]
+    assert routed_commands[1].op.type == OpType.CX
+    assert routed_commands[1].qubits == [nodes[1], nodes[2]]
+    assert routed_commands[2].op.type == OpType.SWAP
+    assert routed_commands[2].qubits == [nodes[2], nodes[1]]
+    assert routed_commands[3].op.type == OpType.CX
+    assert routed_commands[3].qubits == [nodes[0], nodes[1]]
+
+
+def test_RoutingMethodCircuit_custom() -> None:
+    test_c = Circuit(3).CX(0, 1).CX(0, 2).CX(1, 2)
+    nodes = [Node("test", 0), Node("test", 1), Node("test", 2)]
+    test_a = Architecture([[nodes[0], nodes[1]], [nodes[1], nodes[2]]])
+
+    test_mm = MappingManager(test_a)
+    test_mm.route_circuit(
+        test_c,
+        [RoutingMethodCircuit(route_subcircuit_func, check_subcircuit_func_true, 5, 5)],
+    )
+    routed_commands = test_c.get_commands()
+
+    assert routed_commands[0].op.type == OpType.CX
+    assert routed_commands[0].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[1].op.type == OpType.SWAP
+    assert routed_commands[1].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[2].op.type == OpType.CX
+    assert routed_commands[2].qubits == [nodes[1], nodes[2]]
+    assert routed_commands[3].op.type == OpType.SWAP
+    assert routed_commands[3].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[4].op.type == OpType.CX
+    assert routed_commands[4].qubits == [nodes[1], nodes[2]]
+
+
+def test_RoutingMethodCircuit_custom_list() -> None:
+    test_c = Circuit(3).CX(0, 1).CX(0, 2).CX(1, 2)
+    nodes = [Node("test", 0), Node("test", 1), Node("test", 2)]
+    test_a = Architecture([[nodes[0], nodes[1]], [nodes[1], nodes[2]]])
+
+    test_mm = MappingManager(test_a)
+    test_mm.route_circuit(
+        test_c,
+        [
+            RoutingMethodCircuit(
+                route_subcircuit_func, check_subcircuit_func_false, 5, 5
+            ),
+            LexiRouteRoutingMethod(50),
+        ],
+    )
+    routed_commands = test_c.get_commands()
+
+    assert routed_commands[0].op.type == OpType.CX
+    assert routed_commands[0].qubits == [nodes[1], nodes[0]]
+    assert routed_commands[1].op.type == OpType.CX
+    assert routed_commands[1].qubits == [nodes[1], nodes[2]]
+    assert routed_commands[2].op.type == OpType.SWAP
+    assert routed_commands[2].qubits == [nodes[2], nodes[1]]
+    assert routed_commands[3].op.type == OpType.CX
+    assert routed_commands[3].qubits == [nodes[0], nodes[1]]
+
+    test_c = Circuit(3).CX(0, 1).CX(0, 2).CX(1, 2)
+    test_mm.route_circuit(
+        test_c,
+        [
+            RoutingMethodCircuit(
+                route_subcircuit_func, check_subcircuit_func_true, 5, 5
+            ),
+            LexiRouteRoutingMethod(50),
+        ],
+    )
+    routed_commands = test_c.get_commands()
+    assert routed_commands[0].op.type == OpType.CX
+    assert routed_commands[0].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[1].op.type == OpType.SWAP
+    assert routed_commands[1].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[2].op.type == OpType.CX
+    assert routed_commands[2].qubits == [nodes[1], nodes[2]]
+    assert routed_commands[3].op.type == OpType.SWAP
+    assert routed_commands[3].qubits == [nodes[0], nodes[1]]
+    assert routed_commands[4].op.type == OpType.CX
+    assert routed_commands[4].qubits == [nodes[1], nodes[2]]
+
+
+if __name__ == "__main__":
+    test_LexiRouteRoutingMethod()
+    test_RoutingMethodCircuit_custom()
+    test_RoutingMethodCircuit_custom_list()