qubo_lab/test_sat_to_qubo_workflow.py

#!/usr/bin/env python3

from util.kSAT import kSAT
import util.randomSAT as rand_sat
import util.SAT2QUBO as SAT2QUBO
import util.script as script
import util.queries as queries

from dwave.system.composites import FixedEmbeddingComposite
import dimod
from neal import SimulatedAnnealingSampler
import minorminer
import networkx as nx
import dwave_networkx as dnx
import dwave.embedding

from dwave_qbsolv import QBSolv

import numpy as np
import random

from tqdm import tqdm

def main():
    #__wmis()
    #__pqubo()
    __wmis3()
    

def __qubo_to_nx_graph(qubo):
    graph = nx.Graph()
    
    for coupler, energy in qubo.items():
        if coupler[0] != coupler[1]:
            graph.add_edge(coupler[0], coupler[1], weight=energy)
    
    return graph

def __wmis2():
    db = script.connect_to_instance_pool("dbc")
    target_graph =  dnx.chimera_graph(16, 16, 4)
    target_graph_id = queries.get_id_of_solver_graph(db["solver_graphs"],
                                                     nx.node_link_data(target_graph))
    instance_id = "5c9ccb998c6fe61926458351"
    qubo, wmis_id = queries.load_WMIS_qubo_of_instance(db["wmis_qubos"], instance_id)
    
    emb = queries.load_embedding(db["embeddings"], wmis_id, target_graph_id)
    
    chimera_sampler = dimod.StructureComposite(SimulatedAnnealingSampler(),
                                               target_graph.nodes(),
                                               target_graph.edges())
    
    sampler = FixedEmbeddingComposite(chimera_sampler, emb)
    
    res = sampler.sample_qubo(__negate_qubo(qubo))
    
    print(res.first)
    
def __wmis3():    
    db = script.connect_to_instance_pool()
    target_graph =  dnx.chimera_graph(16, 16, 4)
    target_graph_id = queries.get_id_of_solver_graph(db["solver_graphs"],
                                                     nx.node_link_data(target_graph))
    
    solver_input_query = queries.WMIS_solver_input_scope_query(db)
    solver_input_query.query("c42_v[5-84]_1", target_graph_id)
    
    base_sampler = SimulatedAnnealingSampler()
    chimera_sampler = dimod.StructureComposite(base_sampler,
                                               target_graph.nodes(),
                                               target_graph.edges())
    
    for solver_input in tqdm(solver_input_query):
        sampler = FixedEmbeddingComposite(chimera_sampler, solver_input["embeddings"][0])
        
        res = sampler.sample_qubo(__negate_qubo(solver_input["qubo"]))
        
        script.save_simulated_annealing_result(db["wmis_siman_results"],
                                               res,
                                               solver_input,
                                               0) 
    
    
def __wmis():
    sat = rand_sat.generateRandomKSAT(25, 6, 3)
    qubo = __negate_qubo(SAT2QUBO.WMISdictQUBO(sat))
    #qubo = SAT2QUBO.WMISdictQUBO(sat)
    nx_qubo = __qubo_to_nx_graph(qubo)
    
    target_graph = dnx.chimera_graph(16, 16, 4)
    
    emb = minorminer.find_embedding(nx_qubo.edges(),
                                    target_graph.edges(),
                                    return_overlap=True)
    
    if emb[1] != 1:
        print("no embedding found")
        return
    
    print(emb[0])
    
    chimera_sampler = dimod.StructureComposite(SimulatedAnnealingSampler(),
                                               target_graph.nodes(),
                                               target_graph.edges())
    
    
    
    sampler = FixedEmbeddingComposite(chimera_sampler, emb[0])

    
    res = sampler.sample_qubo(qubo)
    #res = SimulatedAnnealingSampler().sample_qubo(qubo)
    
    #dwave.embedding.chain_breaks.majority_vote(res, emb[0])
    
    first = res.first
    
    print("chain_break_fraction={}".format(first.chain_break_fraction))
    for lit, spin in first.sample.items():
        print(lit, spin)
    print("true count: {}".format(np.count_nonzero(list(first.sample.values()))))
    
    assignments = {}
    
    for coupler, energy in first.sample.items():
        
        var = abs(coupler[1])
        
        if var not in assignments:
            assignments[var] = {"all": []}
        
        if energy == 1:
            assignments[var]["all"].append(1 if coupler[1] > 0 else 0)
    
    __majority_vote(assignments)
                    
    #for var, a in assignments.items():
        ##print(var, np.sort(a["all"]), __majority_percentage(a["all"]))
        #print(var, a)
    final = __extract_assignment(assignments)
    print(final)
    print("satisfies sat: {}".format(sat.checkAssignment(final)))

def __optimize_assignment(sat, assignment, consistencies):
    rnd = random.Random()
    
    max_steps = 4000
    steps = 0
    
    while not sat.checkAssignment(assignment) and steps < max_steps:
        steps += 1
        
        for i in range(len(assignment)):
            if 0.9 * rnd.random() > consistencies[i]:
                assignment[i] = (1 + assignment[i]) % 2
                consistencies[i] = 1 - consistencies[i]
    
    print("steps: {}".format(steps))
    
    return assignment

def __random_assignment(number_of_variables):
    rnd = random.Random()
    assignment = [rnd.choice([0, 1]) for i in range(number_of_variables)]
    consistencies = [0.5 for i in range(number_of_variables)]
    
    return assignment, consistencies
        
    
def __extract_assignment(assignments):
    assignment = [0 for i in range(len(assignments))]
    
    for var, a in assignments.items():
        assignment[var - 1] = a["major"]
        
    return assignment

def __extract_consistencies(assignments)    :
    consistencies = [0.0 for i in range(len(assignments))]
    
    for var, a in assignments.items():
        consistencies[var - 1] = a["consistency"]
        
    return consistencies
    
def __majority_vote(assignments):
    for var, a in assignments.items():
        assignments[var]["major"] = 1 if __true_percentage(a["all"]) >= 0.5 else 0
        assignments[var]["consistency"] = __majority_percentage(a["all"])

def __true_percentage(a):
    if len(a) == 0:
        return 0
    
    return np.count_nonzero(a) / len(a)

def __majority_percentage(a):
    true_perc = __true_percentage(a)
    
    return true_perc if true_perc >= 0.5 else 1 - true_perc

def __pqubo():
    sat = rand_sat.generateRandomKSAT(25, 6, 3)
    ising = SAT2QUBO.primitiveQUBO(sat)
    nx_qubo = __qubo_to_nx_graph(ising)
    
    target_graph = dnx.chimera_graph(16, 16, 4)
    
    emb = minorminer.find_embedding(nx_qubo.edges(),
                                    target_graph.edges(),
                                    return_overlap=True)
    
    if emb[1] != 1:
        print("no embedding found")
        return
    
    chimera_sampler = dimod.StructureComposite(SimulatedAnnealingSampler(),
                                               target_graph.nodes(),
                                               target_graph.edges())
    
    sampler = FixedEmbeddingComposite(chimera_sampler, emb[0])
    
    h, J = __split_ising(ising)
    
    res = sampler.sample_ising(h, J)
    #res = QBSolv().sample_qubo(qubo, find_max=True)
    
    print(res.first)
    
def __split_ising(ising):
    h = {}
    J = {}
    
    for coupler, energy in ising.items():
        if coupler[0] == coupler[1]:
            h[coupler[0]] = energy
        else:
            J[coupler] = energy
    
    return h, J


def __negate_qubo(qubo):
    negative_qubo = {}
    
    for coupler, energy in qubo.items():
        negative_qubo[coupler] = -1 * energy
        
    return  negative_qubo

if __name__ == "__main__":
    main()