Scaleout Edge Clients

This section documents the client implementations available for Scaleout Edge. Clients run on edge nodes and communicate with the Scaleout Edge network to perform local training, evaluation, and model exchange. Multiple client implementations exist to support different environments and programming languages.

Python Client

The Python client provides a high-level API for integrating local training code with the Scaleout Edge network. It is suitable for servers, development machines, notebooks, and lightweight edge nodes.

class scaleout.EdgeClient(train_callback: Callable[[ScaleoutModel, Dict], Tuple[ScaleoutModel | None, Dict]] | None = None, validate_callback: Callable[[ScaleoutModel], Dict] | None = None, runtime: EdgeClientRuntime | None = None)

Bases: object

User-facing interface for an edge client.

Users instantiate this class directly. The runtime defaults to GrpcEdgeClientRuntime; passing runtime= at construction time swaps the implementation — useful for tests and alternative transports.

__init__(train_callback: Callable[[ScaleoutModel, Dict], Tuple[ScaleoutModel | None, Dict]] | None = None, validate_callback: Callable[[ScaleoutModel], Dict] | None = None, runtime: EdgeClientRuntime | None = None) → None

Initialize the EdgeClient.

add_priority_class(cls: PriorityClass) → None

Register a new priority class for use with log_* methods.

Parameters:

cls (PriorityClass) – The class to register. Its name must be unique.

Raises:

ValueError – If a class with that name already exists.

check_task_abort() → None

Check if the ongoing task has been aborted.

This function should be called periodically from the task callback to ensure that the task can be interrupted if needed. If called from a thread that do not run the task, this function is a no-op.

Raises:

StoppedException – If the task was aborted.

connect_to_api(url: str, json: dict | None = None, token: str | None = None, token_refresh_callback: Callable[[...], None] | None = None) → Tuple[ConnectToApiResult, Any]

Connect to the Scaleout API via the runtime.

property current_inference_context: InferenceContext | None

Get the current inference context for the running thread.

property current_logging_context: LoggingContext | None

Get the current logging context for the running thread.

get_access_token() → str | None

Return the current access token, if the runtime manages one.

inference_context(context: InferenceContext)

Set the inference context for the duration of the block.

init_grpchandler(config: GrpcConnectionOptions, token: str | None = None, url: str | None = None, token_refresh_callback: Callable[[...], None] | None = None) → bool

Initialize the runtime’s transport handler.

property link_quality: LinkQuality

Current client-internal link quality derived from the runtime.

log_attributes(attributes: dict, priority: str | None = None, check_task_abort: bool = True) → bool

Log the attributes to the server.

Parameters:

• attributes (dict) – The attributes to log.

• priority (str, optional) – Priority class name (e.g. “artifact”). Defaults to “artifact”.

• check_task_abort (bool, optional) – Whether or not to check for task abort. Defaults to True.

Returns:

True if the attributes were logged successfully, False otherwise.

Return type:

bool

log_inference_result(key: str, payload: dict, context: InferenceContext | None = None, priority: str | None = None) → bool

Log an inference result (sighting/observation) to the server.

Parameters:

• key (str) – A label for this result type (e.g. “detection”, “classification”).

• payload (dict) – JSON-serialisable observation data.

• context (InferenceContext, optional) – Override the thread-local inference context.

• priority (str, optional) – Priority class name (e.g. “telemetry”). Defaults to “inference”.

Returns:

True if the result was accepted into the queue, False if no context is set.

Return type:

bool

log_metric(metrics: dict, step: int = None, commit: bool = True, check_task_abort: bool = True, context: LoggingContext = None) → bool

Log the metrics to the server.

Parameters:

• metrics (dict) – The metrics to log.

• step (int, optional) – The step number.

• value. (If provided the context step will be set to this)

• provided (If not)

• used. (the step from the context will be)

• commit (bool, optional) – Whether or not to increment the step. Defaults to True.

• check_task_abort (bool, optional) – Whether or not to check for task abort. Defaults to True.

• context (LoggingContext, optional) – The logging context to use. Defaults to None, which uses the current context.

Returns:

True if the metrics were logged successfully, False otherwise.

Return type:

bool

log_telemetry(key: str | None = None, payload: dict | None = None, telemetry: dict | None = None, priority: str | None = None, check_task_abort: bool = True) → bool

Log a telemetry observation to the server.

Intended usage: log_telemetry(key="loss", payload={"value": 0.5, "step": 12}).

Parameters:

• key (str) – Telemetry key (e.g. metric name).

• payload (dict, optional) – JSON payload — typically {"value": <float>, ...} for plottable scalars, or arbitrary structured data for coupled observations (coordinates, bundles).

• telemetry (dict, optional) – Legacy — mapping of key to scalar value. In record mode this emits a warning and expands to one record per pair. In legacy mode (SCALEOUT_CLIENT_LEGACY_TELEMETRY=true) it is the accepted shape.

• priority (str, optional) – Priority class name (e.g. “alert”). Defaults to “telemetry”.

• check_task_abort (bool, optional) – Whether to check for task abort. Defaults to True.

Returns:

True if the telemetry was accepted into the queue.

Return type:

bool

Backward compat: existing positional callers passing a dict (log_telemetry({"loss": 0.5})) are accepted; the dict is routed through the legacy telemetry slot.

logging_context(context: LoggingContext)

Set the logging context for the duration of the block.

property priority_classes: list[PriorityClass]

Return the currently registered priority classes (excludes internal default).

remove_custom_callback(callback_name: str) → None

Remove a custom task callback.

remove_priority_class(name: str, force: bool = False) → int

Remove a registered priority class.

Parameters:

• name (str) – The class name to remove.

• force (bool) – If True, migrate queued records to the default class. If False, raises ValueError when the class is non-empty.

Returns:

Number of records migrated to the default class.

Return type:

int

replace_priority_class(cls: PriorityClass) → None

Update an existing priority class definition in-place.

Queued records are not moved — they pick up the new parameters on the next scheduling pass.

Parameters:

cls (PriorityClass) – Replacement definition. Its name must match an existing class.

Raises:

KeyError – If no class with that name exists.

run(with_heartbeat: bool = True, with_polling: bool = True) → None

Run the client’s event loop via the runtime.

run_inference(model: ScaleoutModel | str = None, params: Dict = None) → None

Run inference using the specified model.

Parameters:

• model – The ScaleoutModel or model ID string to use for inference.

• params – Additional parameters for inference.

set_client_id(client_id: str) → None

Set the client ID.

set_custom_callback(callback_name: str, callback: Callable[[scaleoututil.grpc.scaleout_pb2.TaskRequest], Dict]) → None

Set a custom task callback.

set_inference_callback(callback: Callable[[ScaleoutModel, Dict], Any]) → None

Set the inference callback.

set_name(name: str) → None

Set the client name.

set_stage_model_callback(callback: Callable[[ScaleoutModel], None]) → None

Set the stage-model callback, invoked after a model is staged for inference.

set_train_callback(callback: callable) → None

Register the callback invoked when a training task request arrives.

Called by the client with the current global model each time the combiner dispatches a training request. The callback should perform the local training update and return the new model.

Parameters:

callback (callable) –

(scaleout_model, settings) -> (model, metadata)

• scaleout_model (ScaleoutModel): The current model to train. Load parameters with scaleout_model.get_training_model(helper).

• settings (dict): Training settings for this round (e.g. epochs, batch size, learning rate).

• Returns a tuple of the updated model and a metadata dict. The metadata dict is used by the aggregator and for logging; for the default aggregators (fedavg, fedopt) it must at least contain {"training_metadata": {"num_examples": int}}.

Call self.check_task_abort() periodically during training (e.g. once few iterations) to allow the task to be stopped gracefully if the session is terminated from the server, and use self.log_metric(...) to report progress in real time.

set_validate_callback(callback: callable) → None

Register the callback invoked when a validation task request arrives.

Called by the client after a new global model has been produced, so the callback can validate that model and report metrics. Registering this callback is optional.

Parameters:

callback (callable) –

(scaleout_model) -> metrics

• scaleout_model (ScaleoutModel): The model to validate. Load parameters with scaleout_model.get_training_model(helper).

• Returns a dict of validation metrics. Scalar entries are captured and visualized in the Scaleout Edge UI; the entire dict is stored in the backend and accessible via the API/UI.

stage_model(model: ScaleoutModel | str) → ScaleoutModel

Stage a model for inference.

Parameters:

model – The ScaleoutModel or model id to stage.

start_autonomous_inference(model: ScaleoutModel | str, params: Dict = None) → Task

Start an autononmous inference task, i.e. inference initiated by the client. Starts the inference in a seperate thread as if it arrived from the server. Returns the Task; its correlation_id identifies the command on the server.

start_autonomous_task(callback: Callable[[], Any], command_type: str, parameters: dict | None = None) → Task

Start a task the client is initiating on its own accord as a stoppable command, without waiting for the server to dispatch it.

callback runs on a background thread; call check_task_abort() periodically inside it to respond to a server-issued stop.

stop_default_telemetry_loop() → None

Signal default_telemetry_loop to exit at its next sleep boundary.

class scaleout.ScaleoutModel

Bases: object

The ScaleoutModel class is the primary model representation in the Scaleout framework.

A ScaleoutModel is a self-describing, immutable container holding: - The serialized model weights (raw bytes, via a helper) - An optional full model representation (e.g. ONNX, .pt, …) - A metadata dict (eagerly loaded): model_id, helper, inference_model_format, and any user-supplied key-value pairs

Models are always immutable after construction. To create a modified copy, use to_builder() to obtain a pre-seeded ScaleoutModelBuilder:

new_model = model.to_builder().set_model_id("m-002").build()
new_model = model.to_builder().set_metadata("tag", "v2").build()

API

Reading metadata:

model.metadata          # shallow copy of the metadata dict
model.model_id          # convenience property

Model parameters:

with model.get_training_model_stream() as s:  # read directly from ZIP, no temp file
    data = s.read()
params = model.get_training_model(helper)     # decodes from ZIP

Full model representation:

model.has_inference_model                         # bool
model.get_inference_model_format()                # format string or None
with model.get_inference_model_stream() as s:    # stream inference_model.bin directly from ZIP
    data = s.read()

User-attached files:

model.list_files()              # list of attached file names
model.has_file("config.json")   # bool
with model.open_file("config.json") as f:
    data = f.read()
data = model.get_file("config.json")  # convenience, returns bytes

On-disk format

Files are ZIP archives (stdlib zipfile, DEFLATED) containing:

metadata.json – eagerly loaded on open training_model.bin – serialized weights inference_model.bin – optional full model representation files/{name} – zero or more user-attached files

Legacy raw-binary streams (no ZIP header, e.g. bare NPZ) are detected automatically and loaded with empty metadata for backward compatibility.

Storage

A single ZIP file (_zip_path) is the canonical representation from construction time onwards. The model always owns its ZIP and deletes it on garbage collection.

get_training_model_stream() and get_inference_model_stream() stream entries directly from the ZIP without extracting to a temp file.

__init__()

property checksum: str

close()

Explicitly release resources held by this model.

Closes the open file handle and deletes the owned temp ZIP file, if any. Deterministic alternative to relying on __del__/GC — callers that remove a model from storage while other references may still exist (e.g. caches, in-flight lambdas) should call this explicitly rather than dropping the reference and hoping for prompt collection.

Safe to call multiple times; the model should not be used afterwards.

static detect_format(stream: BytesIO) → str

Returns ‘zip’ if data is our ZIP container format, else ‘legacy’.

Legacy NPZ files (produced by NumpyHelper) are also ZIP-based and start with the PK magic bytes. We distinguish them from our container by checking for the presence of ‘metadata.json’ inside the archive.

static detect_format_file(file_path: str) → str

Returns ‘zip’ if the file is our ZIP container format, else ‘legacy’.

Reads only the central directory — does not load the entire file.

property format_version: int

The on-disk container format version this model was written with.

static from_file(file_path: str) → ScaleoutModel

Creates a ScaleoutModel from a file (ZIP or legacy raw binary).

Delegates to ScaleoutModelBuilder.

static from_filechunk_stream(filechunk_stream: Iterable[scaleoututil.grpc.scaleout_pb2.FileChunk]) → ScaleoutModel

Creates a ScaleoutModel from a gRPC FileChunk iterator.

Delegates to ScaleoutModelBuilder.

static from_stream(stream: BinaryIO) → ScaleoutModel

Creates a ScaleoutModel from a stream (ZIP or legacy raw binary).

Delegates to ScaleoutModelBuilder.

static from_training_model(model_params, helper=None, metadata: dict | None = None) → ScaleoutModel

Creates a ScaleoutModel from model parameters.

Delegates to ScaleoutModelBuilder. A fresh model_id is always generated; any model_id key present in metadata is ignored.

Parameters:

• model_params – Parameters accepted by the helper’s save() method.

• helper – Serialization helper. Falls back to NumpyHelper if not provided.

• metadata – Optional initial metadata dict.

get_file(name: str) → bytes

Read and return all bytes of an attached file.

get_file_stream()

Returns a new read handle to the packed ZIP file.

get_filechunk_stream(chunk_size=1048576)

Yields gRPC FileChunk messages of the raw model bytes.

get_inference_model_format() → str | None

Returns the format string of the full model representation, or None.

get_inference_model_stream()

Context manager yielding a read stream of the full model representation binary.

Reads directly from the ZIP entry — no temp-file extraction.

Usage:

if model.has_inference_model:
    with model.get_inference_model_stream() as stream:
        data = stream.read()

get_model_params(helper=None)

Alias for get_training_model() for backward compatibility.

get_training_model(helper=None)

Decodes and returns the model parameters via the helper.

get_training_model_stream()

Context manager yielding a read stream of the raw model binary.

Reads directly from the ZIP entry — no temp-file extraction.

Usage:

with model.get_training_model_stream() as stream:
    data = stream.read()

has_file(name: str) → bool

True if the model contains a user-attached file with this name.

property has_inference_model: bool

True if the model contains a full model representation.

property has_training_model: bool

True if the model contains serialized parameters.

property legacy_source: bool

True if this model was loaded from a legacy raw binary (no ZIP).

list_files() → list

Returns names of user-attached files (without the files/ prefix).

property metadata: dict

Returns a shallow copy of the metadata dict.

property model_id: str | None

open_file(name: str)

Context manager yielding a read stream for the named attached file.

Usage:

with model.open_file("config.json") as f:
    data = f.read()

save_to_file(file_path: str)

Saves the model to a file in ZIP format.

sign(private_key, signer_id: str | None = None) → dict

Compute and return a signature dict for this model.

Does not modify the ZIP. POST the returned dict to /api/v1/model-signatures/ to persist it.

Parameters:

• private_key – An Ed25519PrivateKey from the cryptography package.

• signer_id – Optional free-form string identifying the signer.

Returns:

A dict with keys model_id, algorithm, signature (base64-encoded), and optionally signer_id.

to_builder() → ScaleoutModelBuilder

Return a factory pre-loaded from this model’s current state.

Use this to create a modified copy without mutating the original:

new_model = model.to_builder().set_model_id("m-002").build()
new_model = model.to_builder().set_metadata("tag", "v2").build()

verify_checksum(checksum: str) → bool

verify_signature(public_key, signature: str) → bool

Return True if signature is valid for this model.

Parameters:

• public_key – An Ed25519PublicKey from the cryptography package.

• signature – Base64-encoded signature string — as returned by GET /api/v1/model-signatures/<id> signature field.

Additional Clients

Support for additional client implementations (e.g., C++ and Kotlin) will be included here in future versions of the documentation.