Source code for eqc_direct.utils

"""
Utilities for running server sim and client
"""
from dataclasses import dataclass
import grpc
import time
[docs]
@dataclass
class SysStatus:
    """
    Status codes for system paired with their descriptions.
    """
    IDLE = {"status_code": 0, "status_desc": "IDLE"}
    JOB_RUNNING = {"status_code": 1, "status_desc": "JOB_RUNNING"}
    HEALTH_CHECK = {"status_code": 2, "status_desc": "HEALTH_CHECK"}
    FAILED_ENTROPY = {"status_code": 3, "status_desc": "FAILED_ENTROPY"}
[docs]
@dataclass
class LockCheckStatus:
    """
    Statuses codes for checking lock status paired with their descriptions
    """
    AVAILABLE = {"status_code": 0, "status_desc": "Lock available"}
    USER_LOCKED = {
        "status_code": 1,
        "status_desc": "lock_id matches current server lock_id",
    }
    UNAVAILABLE = {
        "status_code": 2,
        "status_desc": "Execution lock is in use by another user",
    }
[docs]
@dataclass
class LockManageStatus:
    """
    Statuses and descriptions for acquiring and releasing lock
    """
    SUCCESS = {"status_code": 0, "status_desc": "Success"}
    MISMATCH = {
        "status_code": 1,
        "status_desc": "lock_id does not match current device lock_id",
    }
    BUSY = {
        "status_code": 2,
        "status_desc": "Lock currently in use unable to perform operation",
    }
[docs]
@dataclass
class JobCodes:
    """
    Job codes for errors paired with their descriptions
    """
    NORMAL = {
        "err_code": 0,
        "err_desc": "Success"
    }
    BAD_INPUT = {
        "err_code": 1,
        "err_desc": "Incorrectly formatted matrix"
    }
    DEVICE_BUSY = {
        "err_code": 2,
        "err_desc": "Device currently processing other request",
    }
    LOCK_MISMATCH = {
        "err_code": 3,
        "err_desc": "lock_id doesn't match current device lock",
    }
    NO_ENTROPY = {
        "err_code": 4,
        "err_desc": "Device failed to capture entropy during sampling",
    }
    INVALID_SUM_CONSTRAINT = {
        "err_code": 5,
        "err_desc": "Sum constraint must be greater than 0"
    }
    CONSTRAINT_SOLN_TYPE_MISMATCH = {
        "err_code": 6,
        "err_desc": "If `continuous_soln`=False then `sum_constraint` must be an integer"
    }
[docs]
def message_to_dict(grpc_message) -> dict:
    """
    Converts a grpc message to a dictionary
    :param grpc_message: original grpc message
    :return: original message parsed as a dict
    """
    return {
        field.name: getattr(grpc_message, field.name)
        for field in grpc_message.DESCRIPTOR.fields
    }