nvflare.app_common.streamers.file_streamer module

class FileStreamer[source]

Bases: StreamerBase

static get_file_location(stream_ctx: dict)[source]

Get the file location property from stream context. This method is intended to be used by the stream_done_cb() function of the receiving side.

Parameters:

stream_ctx – the stream context

Returns: location (full file path) of the received file

static get_file_name(stream_ctx: dict)[source]

Get the file base name property from stream context. This method is intended to be used by the stream_done_cb() function of the receiving side.

Parameters:

stream_ctx – the stream context

Returns: file base name

static get_file_size(stream_ctx: dict)[source]

Get the file size property from stream context. This method is intended to be used by the stream_done_cb() function of the receiving side.

Parameters:

stream_ctx – the stream context

Returns: size (in bytes) of the received file

static register_stream_processing(fl_ctx: FLContext, channel: str, topic: str, dest_dir: str | None = None, stream_done_cb=None, chunk_consumed_cb=None, **cb_kwargs)[source]

Register for stream processing on the receiving side.

Parameters:

  • fl_ctx – the FLContext object

  • channel – the app channel

  • topic – the app topic

  • dest_dir – the destination dir for received file. If not specified, system temp dir is used

  • stream_done_cb – if specified, the callback to be called when the file is completely received

  • chunk_consumed_cb – if specified, the callback to be called when a chunk is processed

  • **cb_kwargs – the kwargs for the stream_done_cb

Returns: None

Notes: the stream_done_cb must follow stream_done_cb_signature as defined in apis.streaming.

static stream_file(channel: str, topic: str, stream_ctx: dict, targets: ~typing.List[str], file_name: str, fl_ctx: ~nvflare.apis.fl_context.FLContext, chunk_size=None, chunk_timeout=None, optional=False, secure=False) -> (<class 'str'>, <class 'bool'>)[source]

Stream a file to one or more targets.

Parameters:

  • channel – the app channel

  • topic – the app topic

  • stream_ctx – context data of the stream

  • targets – targets that the file will be sent to

  • file_name – full path to the file to be streamed

  • fl_ctx – a FLContext object

  • chunk_size – size of each chunk to be streamed. If not specified, default to 1M bytes.

  • chunk_timeout – timeout for each chunk of data sent to targets.

  • optional – whether the file is optional

  • secure – whether P2P security is required

Returns: a tuple of (RC, Result):

  • RC is ReturnCode.OK or ReturnCode.ERROR;

  • Result is whether the streaming completed successfully

Notes: this is a blocking call - only returns after the streaming is done.