Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions dwave/experimental/multicolor_anneal/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@


def get_properties(sampler: DWaveSampler | Solver | str | None = None
) -> list[dict[str, Any]]:
"""Return multicolor-annealing properties for each annealing line.
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
"""Return multicolor-annealing properties

Args:
sampler:
Expand All @@ -37,19 +37,21 @@ def get_properties(sampler: DWaveSampler | Solver | str | None = None
an MCA-enabled solver.

Returns:
Annealing-line properties for all available annealing lines, formatted
as list of dicts in ascending order of annealing-line index.
A tuple containing two elements:
1. A dictionary of annealing-line independent properties.
2. Annealing-line specific properties for all available annealing lines,
formatted as list of dicts in ascending order of annealing-line index.

Examples:
Retrieve MCA properties for the annealing lines of a default solver, and
print the number of lines and first qubits on line 0.

>>> from dwave.experimental import multicolor_anneal as mca
...
>>> annealing_lines = mca.get_properties() # doctest: +SKIP
>>> len(annealing_lines) # doctest: +SKIP
>>> exp_feature_info = mca.get_properties() # doctest: +SKIP
>>> len(exp_feature_info[1]) # doctest: +SKIP
6
>>> annealing_lines[0]['qubits'] # doctest: +SKIP
>>> exp_feature_info[1][0]['qubits'] # doctest: +SKIP
[2, 6, 9, 14, 17, 18, ...]
"""

Expand Down
Binary file not shown.
2 changes: 1 addition & 1 deletion examples/mca_embedding.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ def main(
# when available, use feature-based search to default the solver.
if use_client:
qpu = DWaveSampler(solver=solver)
annealing_lines = get_properties(qpu)
annealing_lines = get_properties(qpu)[1]
line_assignments = {
n: al_idx for al_idx, al in enumerate(annealing_lines) for n in al["qubits"]
}
Expand Down
57 changes: 28 additions & 29 deletions examples/mca_shim_AO_FB.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
get_properties,
SOLVER_FILTER,
make_tds_graph,
qubit_to_Advantage2_annealing_line,
)
from dwave.experimental.shimming import shim_flux_biases

Expand Down Expand Up @@ -178,27 +177,27 @@ def _calc_anneal_offsets(
def artificial_data(
delays: np.ndarray,
A: float,
decay_time: float,
T2: float = 10.1,
num_independent_samples: int = float("Inf"),
prng: np.random.Generator | int | None = None,
):
) -> np.ndarray:
"""Create an artificial data set

y(t) = np.exp(-delays / decay_time) * np.cos(2* np.pi * A * delays)
y(t) = np.exp(-delays / T2) * np.cos(2* np.pi * A * delays)
with variance of (1 - y(t)^2) in the measured state. Given independent
and identically distributed samples we can model noise as normally
distributed.

Args:
delays: time of measurement
A: frequency
decay_time: exponential envelope time scale
T2: exponential envelope time scale
num_independent_samples: number of samples to model
prng: pseudo random number generator or seed.
Returns:
A model signal:
"""
y = np.exp(-delays / decay_time) * np.cos(2 * np.pi * A * delays)
y = np.exp(-delays / T2) * np.cos(2 * np.pi * A * delays)
if num_independent_samples != float("Inf"):
prng = np.random.default_rng(prng)
return y + np.sqrt((1 - y**2) / num_independent_samples) * prng.normal(
Expand All @@ -217,7 +216,10 @@ def run_parallel_experiment(
) -> np.ndarray:
"""Collect detector magnetization for a set of independent embeddings

See documentation example, here we simply parallelize.
Runs a Target-Detector-Source quench experiment on many parallel
embeddings with the specified delays applied to detector lines.
Sample averaged magnetization are calculated on detected qubits in
each embedding and returned as a numpy array.

Args:
sampler: A parallel embedding composite sampler, wrapping the qpu sampler.
Expand Down Expand Up @@ -320,11 +322,11 @@ def main(
target_c: float = 0.37,
no_flux_biases: bool = False,
no_anneal_offsets: bool = False,
delay_min: float = 0.005,
delay_max: float = 0.015,
delay_min: float = 0.01,
delay_max: float = 0.025,
delay_min_fit: float | None = None,
delay_max_fit: float | None = None,
fn_schedule: str = "09-1317A-D_Advantage2_research1_4_annealing_schedule.xlsx",
fn_schedule: str = "09-1323A-D_Advantage2_system4_annealing_schedule.xlsx",
):
"""Demonstrate t-d-s variability and mitigation strategies

Expand Down Expand Up @@ -400,13 +402,13 @@ def main(
if delay_max_fit is None:
delay_max_fit = delay_max # Can be automated for SNR in principle.
elif delay_max_fit > delay_max:
raise ValueError("Fit window exceeds data window")
raise ValueError("The fit window is incompatible with the data window")
if delay_min_fit is None:
delay_min_fit = delay_min # Can be automated for SNR in principle.
elif delay_min_fit < delay_min:
raise ValueError("Fit window exceeds data window")
raise ValueError("The fit window is incompatible with the data window")
if delay_min_fit > delay_max_fit:
raise ValueError("Fit window is empty")
raise ValueError("The fit window is empty")
# Schedule based approximations, target_A and dA/dc are approximated.
qpu_anneal_schedule = pd.read_excel(
fn_schedule, sheet_name="Fast-Annealing Schedule"
Expand Down Expand Up @@ -469,14 +471,14 @@ def main(
zephyr_shape = qpu.properties["topology"]["shape"]
exp_feature_info = get_properties(qpu)
line_assignments = {
n: al_idx for al_idx, al in enumerate(exp_feature_info) for n in al["qubits"]
n: al_idx for al_idx, al in enumerate(exp_feature_info[1]) for n in al["qubits"]
}
num_lines = len(exp_feature_info)
num_lines = len(exp_feature_info[1])
cmap = plt.colormaps.get_cmap("plasma")
line_color = [cmap(i / (num_lines - 1)) for i in range(num_lines)]

x_anneal_schedules = _make_anneal_schedules(
exp_feature_info,
exp_feature_info[1],
line_source=line_source,
line_detector=line_detector,
target_c=target_c,
Expand Down Expand Up @@ -526,7 +528,7 @@ def _target_assignments(n: int):
embs_by_line = {i: [] for i in range(num_lines)}
for i, emb in enumerate(embs):
q = emb[0][0]
embs_by_line[qubit_to_Advantage2_annealing_line(q, zephyr_shape)].append(emb)
embs_by_line[line_assignments[q]].append(emb)
embs = [emb for i in range(num_lines) for emb in embs_by_line[i]]

sampler = ParallelEmbeddingComposite(qpu, embeddings=embs)
Expand All @@ -538,13 +540,11 @@ def _target_assignments(n: int):
delays_ns = 5 * np.random.random() + 1000 * delays
ld = len(delays_ns)
frequencies = np.arange(ld) / dt / 1000 / ld
decay_time_ns = 20
for idx, A in enumerate([target_Aminus, target_A, target_Aplus]):
for num_independent_samples in [100, float("Inf")]:
signal = artificial_data(
delays_ns,
A,
decay_time=decay_time_ns,
num_independent_samples=num_independent_samples,
)
if num_independent_samples == float("Inf") and idx == 1:
Expand Down Expand Up @@ -595,11 +595,10 @@ def _target_assignments(n: int):
for e, J in bqm.quadratic.items()
},
)
# shimmed_variables = {n for n in bqm_embedded.variables if qubit_to_Advantage2_annealing_line(n, zephyr_shape) == line_detector}
shimmed_variables = {
n
for n in bqm_embedded.variables
if qubit_to_Advantage2_annealing_line(n, zephyr_shape) == line_detector
if line_assignments[n] == line_detector
}
# assert set(bqm_embedded.variables).issubset(qpu.nodelist) # Paranoia
# assert all(T.has_edge(*e) for e in bqm_embedded.quadratic) # Paranoia
Expand Down Expand Up @@ -656,7 +655,7 @@ def _target_assignments(n: int):
line_targets = set()
for idx, emb in enumerate(embs):
q = emb[0][0]
line_target = qubit_to_Advantage2_annealing_line(q, zephyr_shape)
line_target = line_assignments[q]
if line_target not in line_targets:
plt.plot(
delays * 1000,
Expand Down Expand Up @@ -706,7 +705,7 @@ def _target_assignments(n: int):
lines_represented = set()
for i, emb in enumerate(embs):
q = emb[0][0]
line_target = qubit_to_Advantage2_annealing_line(q, zephyr_shape)
line_target = line_assignments[q]
if line_target in lines_represented:
label = None
else:
Expand Down Expand Up @@ -766,7 +765,7 @@ def _target_assignments(n: int):
line_targets = set()
for i, emb in enumerate(embs):
q = emb[0][0]
line_target = qubit_to_Advantage2_annealing_line(q, zephyr_shape)
line_target = line_assignments[q]
if line_target not in line_targets:
plt.plot(
delays * 1000,
Expand Down Expand Up @@ -805,7 +804,7 @@ def _target_assignments(n: int):
lines_represented = set()
for i, emb in enumerate(embs):
q = emb[0][0]
line_target = qubit_to_Advantage2_annealing_line(q, zephyr_shape)
line_target = line_assignments[q]
if line_target in lines_represented:
label = None
else:
Expand Down Expand Up @@ -837,7 +836,7 @@ def _target_assignments(n: int):
lines_represented = set()
for i, emb in enumerate(embs):
q = emb[0][0]
line_target = qubit_to_Advantage2_annealing_line(q, zephyr_shape)
line_target = line_assignments[q]
if line_target in lines_represented:
label = None
else:
Expand Down Expand Up @@ -893,19 +892,19 @@ def _target_assignments(n: int):
"--target_c",
type=float,
help="target_c",
default=0.37, # First horizontal qubit line under 6-line control
default=0.387, # 2GHz experiment on Advantage2_research2
)
parser.add_argument(
"--delay_min",
type=float,
help="Initial delay time (us) for data collection",
default=0.005, # Sufficient for decoupling from source
default=0.01, # Sufficient for decoupling from source
)
parser.add_argument(
"--delay_max",
type=float,
help="Final delay time (us) for data collection",
default=0.015, # Oscillations not completely decayed
default=0.025, # Oscillations not completely decayed
)
parser.add_argument(
"--delay_min_fit",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
---
fixes:
- |
Update examples, tests and utils to accommodate modified get_properties
return format.
Comment thread
jackraymond marked this conversation as resolved.
notes:
- |
Defaults for examples are modified to reflect Advantage2_research2
properties and published documentation.
51 changes: 32 additions & 19 deletions tests/test_multicolor_anneal.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,22 +28,30 @@

class PropertiesCheckMixin:

properties = [
'annealingLine', 'minAnnealingTimeStep', 'minPolarizingTimeStep',
'depolarizationAnnealScheduleRequiredDelay', 'holdOvershootFor',
polarizing_line_properties = ['minPolarizingTimeStep',
'depolarizationAnnealScheduleRequiredDelay']

annealing_line_properties = [
'annealingLine', 'minAnnealingTimeStep', 'holdOvershootFor',
'minCOvershoot', 'maxCOvershoot', 'maxC', 'minC',
'scheduleDelayStep', 'qubits'
]

def validate_annealing_lines_properties(self, data):
def validate_exp_feature_info(self, data):
self.assertIsInstance(data, list)
self.assertGreater(len(data), 0)
n_lines = len(data)
self.assertEqual(len(data), 2)
polarizing_line_info, annealing_line_info = data
self.assertIsInstance(polarizing_line_info, dict)
for p in self.polarizing_line_properties:
self.assertIn(p, polarizing_line_info)
self.assertIsInstance(annealing_line_info, list)
self.assertGreater(len(annealing_line_info), 0)
n_lines = len(annealing_line_info)
for i in range(n_lines):
for p in self.properties:
self.assertIn(p, data[i])
self.assertEqual(data[i]['annealingLine'], i)
self.assertGreater(len(data[i]['qubits']), 0)
for p in self.annealing_line_properties:
self.assertIn(p, annealing_line_info[i])
self.assertEqual(annealing_line_info[i]['annealingLine'], i)
self.assertGreater(len(annealing_line_info[i]['qubits']), 0)


class MCA(unittest.TestCase, PropertiesCheckMixin):
Expand All @@ -55,9 +63,10 @@ def tearDown(self):
def test_sampler_properties(self):
n_lines = 6
n_qubits = 100
info = [{'annealingLine': i,
polarizing_line_info = {'minPolarizingTimeStep': 0.02,
'depolarizationAnnealScheduleRequiredDelay': 2.0}
annealing_line_info = [{'annealingLine': i,
'minAnnealingTimeStep': 0.01,
'minPolarizingTimeStep': 0.02,
'depolarizationAnnealScheduleRequiredDelay': 2.0,
'holdOvershootFor': 0.02,
'minCOvershoot': -7.0,
Expand All @@ -66,19 +75,23 @@ def test_sampler_properties(self):
'minC': -2.0,
'scheduleDelayStep': 1e-06,
'qubits': list(range(i*100, (i+1)*100))} for i in range(n_lines)]
info = [polarizing_line_info, annealing_line_info]

with unittest.mock.MagicMock() as sampler:
sampler.solver.edges = [(0,1)]
sampler.solver.sample_qubo.return_value.result.return_value = \
dict(x_get_multicolor_annealing_exp_feature_info=info)

lines = get_properties(sampler)
exp_feature_info = get_properties(sampler)

self.assertEqual(len(exp_feature_info), 2)
lines = exp_feature_info[1]

self.assertEqual(len(lines), n_lines)
self.assertTrue(all(lines[i]['annealingLine'] == i for i in range(n_lines)))
self.assertTrue(all(len(lines[i]['qubits']) == n_qubits for i in range(n_lines)))

self.validate_annealing_lines_properties(lines)
self.validate_exp_feature_info(exp_feature_info)

@unittest.mock.patch('dwave.experimental.fast_reverse_anneal.api.Client')
def test_default_solver_name(self, client):
Expand Down Expand Up @@ -112,15 +125,15 @@ def tearDown(self):
get_solver_name.cache_clear()

def test_get_parameters_from_sampler(self):
lines = get_properties(self.sampler)
self.validate_annealing_lines_properties(lines)
exp_feature_info = get_properties(self.sampler)
self.validate_exp_feature_info(exp_feature_info)

def test_get_parameters_from_name(self):
lines = get_properties(get_solver_name())
self.validate_annealing_lines_properties(lines)
exp_feature_info = get_properties(get_solver_name())
self.validate_exp_feature_info(exp_feature_info)

def test_6_line_accuracy(self):
annealing_lines = get_properties(self.sampler)
annealing_lines = get_properties(self.sampler)[1]
topology_type = self.sampler.properties["topology"]["type"]
if len(annealing_lines) == 6 and topology_type == "zephyr":
shape = self.sampler.properties["topology"]["shape"]
Expand Down