"""
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
}