"""
Utilities for running server sim and client
"""
from dataclasses import dataclass
from typing import List, Tuple
import grpc
import time
import numpy as np
# levels for precision calc
PREC_MIN_RECOMMENDED_LEVELS = 200
PREC_MAX_LEVELS = 10000
[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"}
CALIBRATION = {"status_code": 2, "status_desc": "CALIBRATION"}
HEALTH_CHECK = {"status_code": 3, "status_desc": "HEALTH_CHECK"}
HARDWARE_FAILURE = {"status_code": [4, 5, 6, 7], "status_desc": "HARDWARE_FAILURE"}
[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"}
INDEX_OUT_OF_RANGE = {
"err_code": 1,
"err_desc": "Index in submitted data is out of range for specified number of variables",
}
COEF_INDEX_MISMATCH = {
"err_code": 2,
"err_desc": "Polynomial indices do not match required length for specified coefficient length",
}
DEVICE_BUSY = {
"err_code": 3,
"err_desc": "Device currently processing other request",
}
LOCK_MISMATCH = {
"err_code": 4,
"err_desc": "lock_id doesn't match current device lock",
}
HARDWARE_FAILURE = {
"err_code": 5,
"err_desc": "Device failed during execution",
}
INVALID_SUM_CONSTRAINT = {
"err_code": 6,
"err_desc": "Sum constraint must be greater than or equal to 1 and less than or equal to 10000",
}
INVALID_RELAXATION_SCHEDULE = {
"err_code": 7,
"err_desc": "Parameter relaxation_schedule must be in set {1,2,3,4}",
}
USER_INTERRUPT = {
"err_code": 8,
"err_desc": "User sent stop signal before result was returned",
}
EXCEEDS_MAX_SIZE = {
"err_code": 9,
"err_desc": "Exceeds max problem size for device",
}
DECREASING_INDEX = {
"err_code": 10,
"err_desc": "One of specified polynomial indices is not specified in non-decreasing order",
}
INVALID_PRECISION = {
"err_code": 11,
"err_desc": "The input precision exceeds maximum allowed precision for device",
}
DUPLICATE_INDEX = {
"err_code": 12,
"err_desc": "A duplicate polynomial index set was specified for the input polynomial",
}
PRECISION_CONSTRAINT_MISMATCH = {
"err_code": 13,
"err_desc": "Sum constraint must be divisible by solution_precision",
}
PRECISION_NONNEGATIVE = {
"err_code": 14,
"err_desc": "Input solution precision cannot be negative",
}
DEGREE_POSITIVE = {
"err_code": 15,
"err_desc": "Input degree must be greater than 0"
}
NUM_VARIABLES_POSITIVE = {
"err_code": 16,
"err_desc": "Input num_variables must be greater than 0"
}
[docs]
def message_to_dict(grpc_message) -> dict:
"""Convert a gRPC message to a dictionary."""
result = {}
for descriptor in grpc_message.DESCRIPTOR.fields:
field = getattr(grpc_message, descriptor.name)
if descriptor.type == descriptor.TYPE_MESSAGE:
if descriptor.label == descriptor.LABEL_REPEATED:
if field:
result[descriptor.name] = [message_to_dict(item) for item in field]
else:
result[descriptor.name] = []
else:
if field:
result[descriptor.name] = message_to_dict(field)
else:
result[descriptor.name] = {}
else:
result[descriptor.name] = field
return result
[docs]
def get_decimal_places(float_num: float)->int:
try:
decimal_places = len(str(float_num).split('.')[1])
except IndexError:
decimal_places = 0
return decimal_places