nvflare.fuel.f3.streaming.transfer_progress module

class StreamingProgressConfig(streaming_idle_timeout: float = 600.0, streaming_max_peer_silence: float = 900.0)[source]

Bases: object

streaming_idle_timeout: float = 600.0

streaming_max_peer_silence: float = 900.0

class TransferProgressRecord(job_id: str, task_id: str, transfer_id: str, direction: str, receiver_id: str | None, sequence: int, bytes_done: int, items_done: int | None, started_time: float, last_progress_time: float, state: str = 'active', transfer_id_kind: str | None = None)[source]

Bases: object

bytes_done: int

direction: str

items_done: int | None

job_id: str

property key: Tuple[str, str, str, str, str | None]

last_progress_time: float

receiver_id: str | None

sequence: int

started_time: float

state: str = 'active'

task_id: str

property terminal: bool

transfer_id: str

transfer_id_kind: str | None = None

class TransferProgressState[source]

Bases: object

ABORTED = 'aborted'

ACTIVE = 'active'

COMPLETED = 'completed'

FAILED = 'failed'

TERMINAL_STATES = {'aborted', 'completed', 'failed'}

VALID_STATES = {'aborted', 'active', 'completed', 'failed'}

class TransferProgressTracker(*, idle_timeout: float = 600.0, clock: Callable[[], float] | None = None)[source]

Bases: object

Direction-neutral monotonic progress tracker for streamed transfers.

clear()[source]

get_record(*, job_id: str, task_id: str, transfer_id: str, direction: str, receiver_id: str | None = None) → TransferProgressRecord | None[source]

is_stalled(*, job_id: str, task_id: str, transfer_id: str, direction: str, receiver_id: str | None = None, now: float | None = None) → bool[source]

mark_terminal(*, job_id: str, task_id: str, transfer_id: str, direction: str, state: str, sequence: int | None = None, receiver_id: str | None = None, timestamp: float | None = None) → TransferProgressUpdate[source]

prune(*, before_time: float | None = None, include_active: bool = False, direction: str | None = None) → int[source]

records(*, job_id: str | None = None, task_id: str | None = None, direction: str | None = None, receiver_id: object = <object object>) → Iterable[TransferProgressRecord][source]

remove(*, job_id: str, task_id: str, transfer_id: str, direction: str, receiver_id: str | None = None) → bool[source]

set_idle_timeout(idle_timeout: float)[source]

stalled_records(now: float | None = None, *, job_id: str | None = None, task_id: str | None = None, direction: str | None = None, receiver_id: object = <object object>) → Iterable[TransferProgressRecord][source]

update(*, job_id: str, task_id: str, transfer_id: str, direction: str, sequence: int, bytes_done: int, items_done: int | None = None, state: str = 'active', transfer_id_kind: str | None = None, receiver_id: str | None = None, timestamp: float | None = None) → TransferProgressUpdate[source]

class TransferProgressUpdate(accepted: bool, progressed: bool, record: nvflare.fuel.f3.streaming.transfer_progress.TransferProgressRecord | None, reason: str = '')[source]

Bases: object

accepted: bool

progressed: bool

reason: str = ''

record: TransferProgressRecord | None

check_positive_finite_number(name: str, value) → float[source]

resolve_streaming_progress_config(config: Mapping[str, object] | None = None, *, streaming_idle_timeout: float | None = None, streaming_max_peer_silence: float | None = None) → StreamingProgressConfig[source]

Resolve generic streaming progress timeouts.

Explicit keyword values take precedence over mapping values. When the idle timeout is raised above the default and max peer silence is not explicit, derive max silence from the idle timeout.