add LinearAncillaComposite

dwave/system/composites/__init__.py

@@ -17,3 +17,4 @@
 from dwave.system.composites.tiling import *
 from dwave.system.composites.virtual_graph import *
 from dwave.system.composites.reversecomposite import *
+from dwave.system.composites.linear_ancilla import *

dwave/system/composites/linear_ancilla.py

@@ -0,0 +1,213 @@
+# coding: utf-8
+# Copyright 2018 D-Wave Systems Inc.
+#
+#    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
+#
+#        http://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.
+
+"""Embedding composite to implement linear fields as polarized ancilla qubits.
+"""
+
+import numbers
+
+from typing import Sequence, Mapping, Any
+
+from collections import defaultdict
+
+import numpy as np
+
+from dimod.decorators import nonblocking_sample_method
+import dimod
+
+
+__all__ = ["LinearAncillaComposite"]
+
+
+class LinearAncillaComposite(dimod.ComposedSampler, dimod.Structured):
+    """Implements linear fields as polarized ancilla qubits.
+
+    Linear field `h_i` of qubit `i` is implemented through a coupling `J_{ij}` between
+    the qubit and a neighbouring qubit `j` that is fully polarized with a large flux bias.
+
+    Args:
+    child_sampler (:class:`dimod.Sampler`):
+        A dimod sampler, such as a :obj:`DWaveSampler`, that has flux bias controls.
+
+    """
+
+    def __init__(
+        self,
+        child_sampler: dimod.Sampler,
+    ):
+        self.children = [child_sampler]
+        self.parameters = child_sampler.parameters.copy()
+        self.properties = dict(child_properties=child_sampler.properties.copy())
+        self.nodelist = child_sampler.nodelist
+        self.edgelist = child_sampler.edgelist
+
+    def nodelist(self):
+        pass  # overwritten by init
+
+    def edgelist(self):
+        pass  # overwritten by init
+
+    children = None  # overwritten by init
+    """list [child_sampler]: List containing the structured sampler."""
+
+    parameters = None  # overwritten by init
+    """dict[str, list]: Parameters in the form of a dict.
+
+    For an instantiated composed sampler, keys are the keyword parameters
+    accepted by the child sampler and parameters added by the composite.
+    """
+
+    properties = None  # overwritten by init
+    """dict: Properties in the form of a dict.
+
+    Contains the properties of the child sampler.
+    """
+
+    @nonblocking_sample_method
+    def sample(
+        self,
+        bqm: dimod.BinaryQuadraticModel,
+        *,
+        h_tolerance: numbers.Number = 0,
+        default_flux_bias_range: Sequence[float] = [-0.005, 0.005],
+        **parameters,
+    ):
+        """Sample from the provided binary quadratic model.
+
+
+        Args:
+            bqm (:obj:`~dimod.BinaryQuadraticModel`):
+                Binary quadratic model to be sampled from.
+
+            h_tolerance (:class:`numbers.Number`):
+                Magnitude of the linear bias can be left on the qubit. Assumed to be positive. Defaults to zero.
+
+            default_flux_bias_range (:class:`typing.Sequence`):
+                Flux bias range safely accepted by the QPU, the larger the better.
+
+            **parameters:
+                Parameters for the sampling method, specified by the child
+                sampler.
+
+        Returns:
+            :obj:`~dimod.SampleSet`
+
+        """
+        if h_tolerance < 0:
+            raise ValueError("h_tolerance needs to be positive or zero")
+
+        child = self.child
+        qpu_properties = innermost_child_properties(child)
+        g_target = child.to_networkx_graph()
+        g_source = dimod.to_networkx_graph(bqm)
+        j_range = qpu_properties["extended_j_range"]
+        flux_bias_range = qpu_properties.get("flux_bias_range", default_flux_bias_range)
+
+        # Positive couplings tend to have smaller control error,
+        # we default to them if they have the same magnitude than negative couplings
+        # https://docs.dwavesys.com/docs/latest/c_qpu_ice.html#overview-of-ice
+        largest_j = j_range[1] if abs(j_range[1]) >= abs(j_range[0]) else j_range[0]
+        largest_j_sign = np.sign(largest_j)
+
+        # To implement the bias sign through flux bias sign,
+        # we pick a range (magnitude) that we can sign-flip
+        fb_magnitude = min(abs(b) for b in flux_bias_range)
+        flux_biases = [0] * qpu_properties["num_qubits"]
+
+        _bqm = bqm.copy()
+        used_ancillas = defaultdict(list)
+        for variable, bias in bqm.iter_linear():
+            if abs(bias) <= h_tolerance:
+                continue
+            if abs(bias) - h_tolerance > abs(largest_j):
+                return NotImplementedError(
+                    "linear biases larger than the strongest coupling are not supported yet"
+                )  # TODO: implement larger biases through multiple ancillas
+
+            available_ancillas = set(g_target.adj[variable]) - set(g_source.nodes())
+            if not len(available_ancillas):
+                raise ValueError(f"variable {variable} has no ancillas available")
+            unused_ancillas = available_ancillas - set(used_ancillas)
+            if len(unused_ancillas):
+                ancilla = unused_ancillas.pop()
+                # bias sign is handled by the flux bias
+                flux_biases[ancilla] = np.sign(bias) * largest_j_sign * fb_magnitude
+                _bqm.add_interaction(
+                    variable, ancilla, (abs(bias) - h_tolerance) * largest_j_sign
+                )
+            else:
+                if qpu_properties["j_range"][0] <= bias <= qpu_properties["j_range"][1]:
+                    # If j can be sign-flipped, select the least used ancilla
+                    ancilla = sorted(
+                        list(available_ancillas), key=lambda x: len(used_ancillas[x])
+                    )[0]
+                    _bqm.add_interaction(
+                        variable,
+                        ancilla,
+                        (bias - h_tolerance * np.sign(bias))
+                        * np.sign([flux_biases[ancilla]]),
+                    )
+                else:
+                    # Ancilla sharing is limited to flux biases with a  sign
+                    signed_ancillas = [
+                        ancilla
+                        for ancilla in available_ancillas
+                        if largest_j_sign
+                        == np.sign(flux_biases[ancilla]) * np.sign(bias)
+                    ]
+                    if not len(signed_ancillas):
+                        return ValueError(
+                            f"variable {variable} has no ancillas available"
+                        )
+                    else:
+                        ancilla = sorted(
+                            list(signed_ancillas), key=lambda x: len(used_ancillas[x])
+                        )[0]
+                        _bqm.add_interaction(
+                            variable,
+                            ancilla,
+                            largest_j_sign * (abs(bias) - h_tolerance),
+                        )
+
+            used_ancillas[ancilla].append(variable)
+            _bqm.set_linear(variable, h_tolerance * np.sign(bias))
+
+        sampleset = self.child.sample(_bqm, flux_biases=flux_biases, **parameters)
+        yield
+        yield dimod.SampleSet.from_samples_bqm(
+            [
+                {k: v for k, v in sample.items() if k not in used_ancillas}
+                for sample in sampleset.samples()
+            ],
+            bqm=bqm,
+            info=sampleset.info.update(used_ancillas),
+        )
+
+
+def innermost_child_properties(sampler: dimod.Sampler) -> Mapping[str, Any]:
+    """Returns the properties of the inner-most child sampler in a composite.
+
+    Args:
+        sampler: A dimod sampler
+
+    Returns:
+        properties (dict): The properties of the inner-most sampler
+
+    """
+
+    try:
+        return innermost_child_properties(sampler.child)
+    except AttributeError:
+        return sampler.properties

releasenotes/notes/add-linear-ancilla-composite-3281ed6733b0f0c7.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Add `LinearAncillaComposite` for implementing linear biases through ancilla qubits 

tests/test_linear_ancilla_composite.py

@@ -0,0 +1,122 @@
+# Copyright 2018 D-Wave Systems Inc.
+#
+#    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
+#
+#        http://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 unittest
+import collections
+
+import numpy as np
+
+import networkx as nx
+
+import dimod
+from dimod import TrackingComposite, StructureComposite
+
+from dwave.system.testing import MockDWaveSampler
+from dwave.system import LinearAncillaComposite
+
+
+class TestLinearAncillaComposite(unittest.TestCase):
+    def setUp(self):
+        self.qpu = MockDWaveSampler(properties=dict(extended_j_range=[-2, 1]))
+        self.tracked_qpu = TrackingComposite(self.qpu)
+
+        self.sampler = LinearAncillaComposite(
+            StructureComposite(
+                self.tracked_qpu,
+                nodelist=self.qpu.nodelist,
+                edgelist=self.qpu.edgelist,
+            )
+        )
+
+        self.submask = nx.subgraph(
+            self.sampler.to_networkx_graph(),
+            list(self.sampler.nodelist)[::2],
+        )
+
+        # this problem should run
+        self.linear_problem = dimod.BinaryQuadraticModel.from_ising(
+            {i: (-1) ** i for i in self.submask.nodes()},
+            {},
+        )
+
+        # this problem shouldn't run
+        self.linear_problem_full_graph = dimod.BinaryQuadraticModel.from_ising(
+            {i: (-1) ** i for i in self.qpu.nodelist},
+            {},
+        )
+
+    def test_only_quadratic(self):
+        """if no linear biases, the bqm remains intact"""
+
+        bqm = dimod.generators.ran_r(1, self.submask, seed=1)
+        self.sampler.sample(bqm)
+        self.assertEqual(bqm, self.tracked_qpu.input["bqm"])
+
+    def test_h_tolerance_too_large(self):
+        """if h tolerance is larger than the linear biases,
+        the bqm remains intact
+        """
+
+        self.sampler.sample(self.linear_problem, h_tolerance=1.01)
+        self.assertEqual(self.linear_problem, self.tracked_qpu.input["bqm"])
+
+    def test_intermediate_h_tolerance(self):
+        """check the desired h-tolerance is left in the qubit bias"""
+
+        h_tolerance = 0.5
+        self.sampler.sample(self.linear_problem, h_tolerance=h_tolerance)
+        for variable, bias in self.tracked_qpu.input["bqm"].linear.items():
+            if variable in self.linear_problem.variables:  # skip the ancillas
+                self.assertEqual(
+                    bias,
+                    np.sign(self.linear_problem.get_linear(variable)) * h_tolerance,
+                )
+
+    def test_no_ancillas_available(self):
+        """send a problem that uses all the qubits, not leaving any ancillas available"""
+
+        with self.assertRaises(ValueError):
+            ss = self.sampler.sample(self.linear_problem_full_graph)
+
+    def test_ancillas_present(self):
+        """check the solver used ancillas"""
+
+        self.sampler.sample(self.linear_problem)
+        self.assertGreater(
+            len(self.tracked_qpu.input["bqm"].variables),
+            len(self.linear_problem.variables),
+        )
+
+    def test_ancilla_cleanup(self):
+        """check the problem returned has no additional variables"""
+
+        sampleset = self.sampler.sample(self.linear_problem)
+        self.assertEqual(
+            len(self.linear_problem.variables),
+            len(sampleset.variables),
+        )
+
+    def test_flux_biases_present(self):
+        """check flux biases are applied to non-data qubits"""
+
+        self.sampler.sample(self.linear_problem)
+        flux_biases = np.array(self.tracked_qpu.input["flux_biases"])
+
+        # flux biases are used
+        self.assertGreater(sum(flux_biases != 0), 0)
+
+        # the qubits with flux biases are not data qubits
+        for qubit, flux_bias in enumerate(flux_biases):
+            if flux_bias != 0:
+                self.assertNotIn(qubit, self.linear_problem.variables)