Source code for eqc_models.solvers.eqcdirect

# (C) Quantum Computing Inc., 2024.
from typing import Dict
import logging
import numpy as np
from eqc_direct.client import EqcClient
from eqc_models.base.base import ModelSolver, EqcModel

log = logging.getLogger(name=__name__)

class EqcDirectMixin:

    ip_addr = None
    port = None

    def connect(self, ip_addr : str, port : str) -> str:
        """ Explicitly set device address; if environment is configured with the connection, this call is not required """
        self.ip_addr = ip_addr
        self.port = port
        client = self.client
        return client.system_status()["status_desc"]

    @property
    def client(self):
    
        params = {}
        if self.ip_addr is not None:
            params["ip_address"] = self.ip_addr
        if self.port is not None:
            params["port"] = self.port
        return EqcClient(**params)

[docs] class EqcDirectSolver(ModelSolver, EqcDirectMixin):
[docs] def solve(self, model:EqcModel, relaxation_schedule:int=2, precision : float = 1.0) -> Dict: poly_coefficients, poly_indices = model.sparse # print(poly_indices) if model.machine_slacks > 0: # add a single 0 coefficient entry as the next-highest index highest_idx = int(np.max(poly_indices)) # print("POLY HIGHEST", highest_idx) for i in range(model.machine_slacks): addtl_index = [0 for i in range(len(poly_indices[0]))] addtl_index[-1] = highest_idx + i + 1 poly_indices = poly_indices.tolist() + [addtl_index] poly_coefficients = poly_coefficients.tolist() + [0] # print(poly_indices) scval = model.sum_constraint client = self.client lock_id, start_ts, end_ts = client.wait_for_lock() log.debug("Got device lock id %s. Wait time %f", lock_id, end_ts - start_ts) resp = None try: log.debug("Calling device with parameters relaxation_schedule %d sum_constraint %s lock_id %s solution_precision %f", relaxation_schedule, scval, lock_id, precision) resp = client.process_job(poly_coefficients=poly_coefficients, poly_indices=poly_indices, relaxation_schedule=relaxation_schedule, sum_constraint = scval, lock_id = lock_id, solution_precision=precision) log.debug("Received response with status %s", resp["err_desc"]) log.debug("Runtime %f resulting in energy %f", resp["runtime"], resp["energy"]) log.debug("Distillation runtime %s resulting in energy %f", resp["distilled_runtime"], resp["distilled_energy"]) finally: client.release_lock(lock_id=lock_id) if resp is not None: solution = resp["solution"] energy = resp["energy"] runtime = resp["runtime"] dirac3_sol = np.array(solution) log.debug("Energy %f Runtime %f Solution Size %i Solution Sum %f", energy, runtime, len(dirac3_sol), sum(dirac3_sol)) else: raise RuntimeError("FAILED TO GET RESPONSE") return resp
[docs] class Dirac3DirectSolver(EqcDirectSolver): """ Naming this for when when other devices are available and have different requirements. For instance, Dirac-3 requires the summation constraint parameter, but others might not. The same could be true for relaxation schedule. """