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