Source code for uqrng_direct.utils

from dataclasses import dataclass
from typing import TypedDict, List
import numpy as np
import grpc
[docs]
class resultNISTdetail(TypedDict):
    """
    Dictionary containing detailed results from NIST-STS returned within
    :class:`.resultNIST` generated by either
    :meth:`.UqrngClient.FetchHealthTest` or 
    :meth:`.UqrngClient.HealthTest`. 
    
    :param test_name: a list of names for randomness tests
    :param p_value: a list of p_values associated with the tests.
    :param proportion: a list of proportion of pass vs fail for test.
    :param passed: list of bools with passing tests indicated by True
        and failing tests returning False.
    :param elapsed_time_mins: returns time in fractional mins since 
        last health check was completed.
    """
    test_name: List[str]
    p_value: List[float]
    proportion: List[float]
    passed: List[bool]
    elapsed_time_mins: float
[docs]
class resultNIST(TypedDict):
    """
    Dictionary containing results summary from NIST-STS generated by
    :meth:`.UqrngClient.FetchHealthTest` or
    :meth:`.UqrngClient.HealthTest`. 
    
    :param all_pass: indicates whether all tests in health check
        passed if any test failed then returns False.
    :param tests_detail: detailed test results of form
        :class:`.resultNISTdetail`
    :param summary_table: a string formatted to print as a table which
       summarizes the randomness test detailed results.
    """
    all_pass: bool
    tests_detail: resultNISTdetail
    summary_table: str
[docs]
class SystemInfoDict(TypedDict):
    """
    Dictionary structure for :meth:`.client.UqrngClient.SystemInfo`
    :param device_name: the type of device 
    :param server_version: the current semantic version for the 
       device server
    :param test_interval_mins: current number of
        minutes between consecutive health test for server. For
        information on how this value is set see
        :meth:`.client.UqrngClient.ScheduleHealthTest`.
    """
    device_type: str
    server_version: str
    test_interval_mins: int
[docs]
@dataclass
class SysStatus:
    """
    Status codes for system paired with their descriptions.
    """
    IDLE = {"status_code": 0, "status_desc": "IDLE"}
    SAMPLING = {"status_code": 1, "status_desc": "SAMPLING"}
[docs]
class StatusDict(TypedDict):
    """
    Status message
    
    :param status_code: integer code for response 
    :param status_desc: description for status code
    """
    status_code: int
    status_desc: str
[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 create_summary_table(detail_result: resultNISTdetail) -> str:
    column_names = ["STATISTICAL_TEST", "P-VALUE", "PROPORTION", "PASS"]
    prop_pass_print = np.round(detail_result["proportion_pass"], 6)
    pval_print = np.round(detail_result["p_value"],6)
    prop_col_width = max(len("PROPORTION"),max([len(str(num)) for num in prop_pass_print]))
    test_col_width = max(len("STATISTICAL TEST"),
                         max([len(str(num)) for num in detail_result["test_name"]]))
    pval_col_width = max(len("P-VALUE"),max([len(str(num)) for num in pval_print]))
    pass_col_width = max(len("PASS"),max([len(str(num)) for num in detail_result["passed"]]))
    col_widths = [test_col_width, pval_col_width, prop_col_width, pass_col_width]
    # Calculate column widths
    for i in range(len(column_names)):
        header = " | ".join(f"{col:{col_widths[i]}}" for i, col in enumerate(column_names))
    str_table = header + "\n"
    str_table += ("-" * len(header)+"\n")
    # Print rows
    for t_name, p_val, p_pass, pass_val in zip(detail_result["test_name"],
                                         pval_print,
                                         prop_pass_print,
                                         detail_result["passed"]):
        entry = " | ".join(f"{str(val):{col_widths[i]}}" for i, val in enumerate([t_name, 
                                                                         p_val, 
                                                                         p_pass, 
                                                                         pass_val]))
        str_table += (entry+"\n")
    return str_table
[docs]
def check_qrng_busy_error(rpc_err: grpc.RpcError):
    """
    Utility function used in waiting loop to determine if error is from
    busy or another RPC error which can't be handled
    
    :param rpc_error: RpcError to check
    :return: bool whether error is from QRNG is in use
    """
    if rpc_err.code() == grpc.StatusCode.UNAVAILABLE:
        error_details = rpc_err.details()
        return True
    else:
        return False