-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdecipher_results.py
190 lines (153 loc) · 6.53 KB
/
decipher_results.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import pickle
import numpy as np
import pandas as pd
import dimod
import os
from AGV_quantum import QuadraticAGV
from AGV_quantum import LinearProg
from typing import Any
def get_objective(lp: LinearProg, sample: dict) -> float:
"""computes objective value for sample
:param lp: the integer program with the relevant objective function
:type lp: LinearProg
:param sample: analyzed sample
:type sample: dict
:return: value of the objective funtion
:rtype: float
"""
return sum(
sample[f"x_{i}"] * coef for i, coef in zip(range(lp.nvars), lp.c) if coef != 0
)
def get_results(sampleset: dimod.SampleSet, prob: LinearProg) -> list[dict[str, Any]]:
"""Check samples one by one, and computes it statistics.
Statistics includes energy (as provided by D'Wave), objective function
value, feasibility analysis, the samples itself. Samples are sorted
according to value of the objetive function
:param sampleset: analyzed samples
:type sampleset: dimod.SampleSet
:param prob: integer problem according to which samples are analyzed
:type prob: pulp.LpProblem
:return: analyzed samples, sorted according to objective
:rtype: list[Dict[str,Any]]
"""
dict_list = []
for data in sampleset.data():
rdict = {}
sample = data.sample
rdict["energy"] = data.energy
rdict["objective"] = round(get_objective(prob, sample), 2)
rdict["feasible"] = all(analyze_constraints(prob, sample)[0].values())
rdict["sample"] = sample
rdict["feas_constraints"] = analyze_constraints(prob, sample)
dict_list.append(rdict)
return sorted(dict_list, key=lambda d: d["energy"])
def store_result(input_name: str, file_name: str, sampleset: dimod.SampleSet):
"""Save samples to the file
:param input_name: name of the input
:type input_name: str
:param file_name: name of the file
:type file_name: str
:param sampleset: samples
:type sampleset: dimod.SampleSet
"""
if not os.path.exists("annealing_results"):
os.mkdir("annealing_results")
folder = os.path.join("annealing_results", input_name)
if not os.path.exists(folder):
os.mkdir(folder)
sdf = sampleset.to_serializable()
with open(file_name, "wb") as handle:
pickle.dump(sdf, handle)
def load_results(file_name: str) -> dimod.SampleSet:
"""Load samples from the file
:param file_name: name of the file
:type file_name: str
:return: loaded samples
:rtype: dimod.SampleSet
"""
file = pickle.load(open(file_name, "rb"))
return dimod.SampleSet.from_serializable(file)
def analyze_constraints(
lp: LinearProg, sample: dict[str, int]
) -> tuple[dict[str, bool], int]:
"""check which constraints were satisfied
:param lp: analyzed integer model
:type lp: LinearProg
:param sample: samples generated by the optimizer
:type sample: Dict[str,int]
:return: dictionary mapping constraint to whether they were satisfied, and
the number of satisfied constraints
:rtype: tuple[dict[str, bool], int]
"""
result = {}
num_eq = 0
if lp.A_eq is not None:
for i in range(len(lp.A_eq)):
expr = sum(lp.A_eq[i][j] * sample[lp.var_names[j]] for j in range(lp.nvars))
result[f"eq_{num_eq}"] = expr == lp.b_eq[i]
num_eq += 1
if lp.A_ub is not None:
for i in range(len(lp.A_ub)):
expr = sum(lp.A_ub[i][j] * sample[lp.var_names[j]] for j in range(lp.nvars))
result[f"eq_{num_eq}"] = expr <= lp.b_ub[i]
num_eq += 1
return result, sum(x == False for x in result.values())
def print_results(dict_list):
soln = next((l for l in dict_list if l["feasible"]), None)
if soln is not None:
print("obj:", soln["objective"], "x:", list(soln["sample"].values()))
print("First 10 solutions")
for d in dict_list[:10]:
print(d)
else:
print("No feasible solution")
for d in dict_list[:10]:
print(
"Energy:",
d["energy"],
"Objective:",
d["objective"],
"Feasible",
d["feasible"],
"Broken constraints:",
d["feas_constraints"][1],
)
if __name__ == '__main__':
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
#path_to_results = os.path.join(ROOT, "AGV_quantum", "ising", "sbm_results", "H100_results.csv")
path_to_results = os.path.join(ROOT, "AGV_quantum", "ising", "sbm_results", "avg_sbm.csv")
path_to_annealing = os.path.join(ROOT, "AGV_quantum", "annealing_results", "tiny_2_AGV", "new_bqm.pkl")
size = "largest_ever"
instance = f"{size}_ising"
path_to_renumeration = os.path.join(ROOT, "AGV_quantum", "ising", f"{instance}_renumeration.pkl")
path_to_lp = os.path.join(ROOT, "AGV_quantum", "lp_files", f"lp_{size}.pkl")
results = pd.read_csv(path_to_results, sep=";")
ising_solution = results[results["instance"] == f"{instance}.csv"]
print(ising_solution)
state_ising = ising_solution["state"].item()
state_ising = eval(state_ising)
state = [(k+1)//2 for k in state_ising]
solution = {i + 1: state[i] for i in range(len(state))}
solution_ising = {i + 1: state_ising[i] for i in range(len(state_ising))}
with open(path_to_renumeration, "rb") as f:
var_to_nums, nums_to_var = pickle.load(f)
solutions_vars = {nums_to_var[k]: val for k, val in solution.items()}
solutions_vars_ising = {nums_to_var[k]: val for k, val in solution_ising.items()}
with open(path_to_lp, "rb") as f2:
lp = pickle.load(f2)
model = QuadraticAGV(lp)
p = 5 # TODO this has to be added
model.to_bqm_qubo_ising(p)
energy_computed = dimod.utilities.ising_energy(solutions_vars_ising, model.ising[0], model.ising[1])
sampleset = dimod.SampleSet.from_samples(solutions_vars, vartype=dimod.BINARY, energy=ising_solution["energy"].item())
decrypted_sapleset = model.interpreter(sampleset, "BIN")
decrypted_results = get_results(decrypted_sapleset, lp)
print("Feasible", decrypted_results[0]["feasible"])
broken_constr = decrypted_results[0]["feas_constraints"][1]
constraints = len(decrypted_results[0]["feas_constraints"][0])
print("broken constraints", broken_constr)
print("constraints", constraints)
print("prec broken constr", broken_constr/constraints)
print("energy computed", energy_computed)
print("ising offset", model.ising[2])
print("energy + offset", energy_computed + model.ising[2])