[runtime] QubitPlacer 2 - RandomDevicePlacer

cirq-google/cirq_google/workflow/_device_shim.py

@@ -0,0 +1,32 @@
+# Copyright 2021 The Cirq Developers
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cirq
+import networkx as nx
+import itertools
+from typing import Iterable, cast
+
+
+def _gridqubits_to_graph_device(qubits: Iterable[cirq.GridQubit]):
+    # cirq contrib: routing.gridqubits_to_graph_device
+    def _manhattan_distance(qubit1: cirq.GridQubit, qubit2: cirq.GridQubit) -> int:
+        return abs(qubit1.row - qubit2.row) + abs(qubit1.col - qubit2.col)
+
+    return nx.Graph(
+        pair for pair in itertools.combinations(qubits, 2) if _manhattan_distance(*pair) == 1
+    )
+
+
+def _Device_dot_get_nx_graph(device: 'cirq.Device') -> nx.Graph:
+    return _gridqubits_to_graph_device(cast(Iterable[cirq.GridQubit], device.qubit_set()))

cirq-google/cirq_google/workflow/qubit_placement.py

@@ -16,17 +16,26 @@
 
 import abc
 import dataclasses
+from functools import lru_cache
 from typing import Dict, Any, Tuple, TYPE_CHECKING
+from typing import List, Callable
 
 import numpy as np
 
 import cirq
+import cirq_google as cg
 from cirq import _compat
+from cirq.devices.named_topologies import get_placements
+from cirq_google.workflow._device_shim import _Device_dot_get_nx_graph
 
 if TYPE_CHECKING:
     import cirq_google as cg
 
 
+class CouldNotPlaceError(RuntimeError):
+    """Raised if a problem topology could not be placed on a device graph."""
+
+
 class QubitPlacer(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def place_circuit(
@@ -73,3 +82,95 @@ def _json_dict_(self) -> Dict[str, Any]:
 
     def __repr__(self) -> str:
         return _compat.dataclass_repr(self, namespace='cirq_google')
+
+
+def default_topo_node_to_qubit(node: Any) -> cirq.Qid:
+    """The default mapping from `cirq.NamedTopology` nodes and `cirq.Qid`.
+
+    If nodes are tuples of integers, map to `cirq.GridQubit`. Otherwise try
+    to map to `cirq.LineQubit` and rely on its validation.
+    """
+
+    try:
+        return cirq.GridQubit(*node)
+    except TypeError:
+        return cirq.LineQubit(node)
+
+
+@lru_cache()
+def _cached_get_placements(
+    problem_topo: 'cirq.NamedTopology', device: 'cirq.Device'
+) -> List[Dict[Any, 'cirq.Qid']]:
+    """Cache `cirq.get_placements` onto the specific device."""
+    return get_placements(
+        big_graph=_Device_dot_get_nx_graph(device), small_graph=problem_topo.graph
+    )
+
+
+def _get_random_placement(
+    problem_topology: 'cirq.NamedTopology',
+    device: 'cirq.Device',
+    rs: np.random.RandomState,
+    topo_node_to_qubit_func: Callable[[Any], 'cirq.Qid'] = default_topo_node_to_qubit,
+) -> Dict['cirq.Qid', 'cirq.Qid']:
+    """Place `problem_topology` randomly onto a device.
+
+    Used by `RandomDevicePlacer.place_circuit`.
+    """
+    placements = _cached_get_placements(problem_topology, device)
+    if len(placements) == 0:
+        raise CouldNotPlaceError
+    random_i = int(rs.random_integers(0, len(placements) - 1, size=1))
+    placement = placements[random_i]
+    placement_gq = {topo_node_to_qubit_func(k): v for k, v in placement.items()}
+    return placement_gq
+
+
+class RandomDevicePlacer(QubitPlacer):
+    def __init__(
+        self,
+        topo_node_to_qubit_func: Callable[[Any], cirq.Qid] = default_topo_node_to_qubit,
+    ):
+        """Initialize a RandomDevicePlacer
+
+        Args:
+            topo_node_to_qubit_func: A function that maps from `cirq.NamedTopology` nodes
+                to `cirq.Qid`. There is a correspondence between nodes and the "abstract" Qids
+                used to construct the un-placed circuit. `cirq.get_placements` returns a dictionary
+                mapping from node to Qid. We use this function to transform it into a mapping
+                from "abstract" Qid to device Qid. By default: nodes which are tuples correspond
+                to `cirq.GridQubit`s; otherwise `cirq.LineQubit`.
+        """
+        self.topo_node_to_qubit_func = topo_node_to_qubit_func
+
+    def place_circuit(
+        self,
+        circuit: 'cirq.AbstractCircuit',
+        problem_topology: 'cirq.NamedTopology',
+        shared_rt_info: 'cg.SharedRuntimeInfo',
+        rs: np.random.RandomState,
+    ) -> Tuple['cirq.FrozenCircuit', Dict[Any, 'cirq.Qid']]:
+        """Place a circuit with a given topology onto a device via `cirq.get_placements` with
+        randomized selection of the placement each time.
+
+        This requires device information to be present in `shared_rt_info`.
+
+        Args:
+            circuit: The circuit.
+            problem_topology: The topologies (i.e. connectivity) of the circuit.
+            shared_rt_info: A `cg.SharedRuntimeInfo` object that contains a `device` attribute
+                of type `cirq.Device` to enable placement.
+            rs: A `RandomState` as a source of randomness for random placements.
+
+        Returns:
+            A tuple of a new frozen circuit with the qubits placed and a mapping from input
+            qubits or nodes to output qubits.
+        """
+        device = shared_rt_info.device
+        if device is None:
+            raise ValueError("RandomDevicePlacer requires shared_rt_info.device to be a `cirq.Device`. "
+                             "This should have been set during the initialization phase of `cg.execute`.")
+        placement = _get_random_placement(
+            problem_topology, device, rs=rs, topo_node_to_qubit_func=self.topo_node_to_qubit_func
+        )
+        return circuit.unfreeze().transform_qubits(placement).freeze(), placement