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 }