nvflare.fuel.hci.proto module

class Buffer

Bases: object

Buffer to append to for nvflare.fuel.hci.conn.Connection.

append_command(cmd: str)

append_dict(data: dict, meta: dict | None = None)

append_error(data: str, meta: dict | None = None)

append_shutdown(msg: str)

append_string(data: str, meta: dict | None = None)

append_success(data: str, meta: dict | None = None)

append_table(headers: List[str], name=None) → Table

append_token(token: str)

encode()

reset()

update_meta(meta: dict)

class ConfirmMethod

Bases: object

AUTH = 'auth'

PASSWORD = 'pwd'

USER_NAME = 'username'

YESNO = 'yesno'

class CredentialType(value)

Bases: str, Enum

An enumeration.

CERT = 'cert'

LOCAL_CERT = 'local_cert'

PASSWORD = 'password'

class InternalCommands

Bases: object

CERT_LOGIN = '_cert_login'

CHECK_SESSION = '_check_session'

GET_CMD_LIST = '_commands'

LIST_SESSIONS = 'list_sessions'

LOGOUT = '_logout'

PWD_LOGIN = '_login'

commands = ['_login', '_cert_login', '_logout', '_commands', '_check_session', 'list_sessions']

classmethod contains_commmand(command: str)

class MetaKey

Bases: object

APP_NAME = 'app_name'

CLIENTS = 'clients'

CLIENT_LAST_CONNECT_TIME = 'client_last_conn_time'

CLIENT_NAME = 'client_name'

CLIENT_STATUS = 'client_status'

CMD_NAME = 'cmd_name'

CMD_TIMEOUT = 'cmd_timeout'

CUSTOM_PROPS = 'custom_props'

DATA_TYPE = 'data_type'

DURATION = 'duration'

FILES = 'files'

FOLDER_NAME = 'folder_name'

INFO = 'info'

JOBS = 'jobs'

JOB_DATA = 'job_data'

JOB_DOWNLOAD_URL = 'job_download_url'

JOB_ID = 'job_id'

JOB_META = 'job_meta'

JOB_NAME = 'job_name'

LOCATION = 'location'

SERVER_START_TIME = 'server_start_time'

SERVER_STATUS = 'server_status'

STATUS = 'status'

SUBMIT_TIME = 'submit_time'

TX_ID = 'tx_id'

WORKSPACE = 'workspace'

class MetaStatusValue

Bases: object

CLIENTS_RUNNING = 'clients_running'

ERROR = 'error'

INTERNAL_ERROR = 'internal_error'

INVALID_JOB_DEFINITION = 'invalid_job_def'

INVALID_JOB_ID = 'invalid_job_id'

INVALID_TARGET = 'invalid_target'

JOB_NOT_RUNNING = 'job_not_running'

JOB_RUNNING = 'job_running'

NOT_AUTHENTICATED = 'not_authenticated'

NOT_AUTHORIZED = 'not_authorized'

NO_CLIENTS = 'no_clients'

NO_JOBS = 'no_jobs'

NO_REPLY = 'no_reply'

OK = 'ok'

SYNTAX_ERROR = 'syntax_error'

class ProtoKey

Bases: object

COMMAND = 'command'

DATA = 'data'

DETAILS = 'details'

DICT = 'dict'

ERROR = 'error'

META = 'meta'

ROWS = 'rows'

SHUTDOWN = 'shutdown'

STATUS = 'status'

STRING = 'string'

SUCCESS = 'success'

TABLE = 'table'

TIME = 'time'

TOKEN = 'token'

TYPE = 'type'

make_error(data: str)

make_meta(status: str, info: str = '', extra: dict | None = None) → dict

validate_proto(line: str)

Validate that the line being received is of the expected format.

Parameters:

line – str containing a JSON document

Returns: deserialized JSON document