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