openfl.federated.plan

class openfl.federated.plan.Plan

A class used to represent a Federated Learning plan.

This class provides methods to manage and manipulate federated learning plans.

Class Attributes:

  • logger (Logger) – Logger instance for the class.

  • config (dict) – Dictionary containing patched plan definition.

  • authorized_cols (list) – Authorized collaborator list.

  • cols_data_paths (dict) – Collaborator data paths dictionary.

  • collaborator_ (Collaborator) – Collaborator object.

  • aggregator_ (Aggregator) – Aggregator object.

  • assigner_ (Assigner) – Assigner object.

  • loader_ (DataLoader) – Data loader object.

  • runner_ (TaskRunner) – Task runner object.

  • server_ (AggregatorGRPCServer) – gRPC server object.

  • client_ (AggregatorGRPCClient) – gRPC client object.

  • pipe_ (CompressionPipeline) – Compression pipeline object.

  • straggler_policy_ (StragglerHandlingPolicy) – Straggler handling policy.

  • hash_ (str) – Hash of the instance.

  • name_ (str) – Name of the instance.

  • serializer_ (SerializerPlugin) – Serializer plugin.

static build(template, settings, **override)

Create an instance of a openfl Component or Federated DataLoader/TaskRunner.

Parameters:

  • template (str) – Fully qualified class template path.

  • settings (dict) – Keyword arguments to class constructor.

  • override (dict) – Additional settings to override the default ones.

Returns:

object – A Python object.

deserialize_interface_objects()

Deserialize objects for TaskRunner.

Returns:

tuple – Tuple containing the deserialized objects.

static dump(yaml_path, config, freeze=False)

Dump the plan config to YAML file.

Parameters:

  • yaml_path (Path) – Path to the YAML file.

  • config (dict) – Plan configuration to be dumped.

  • freeze (bool, optional) – Flag to freeze the plan. Defaults to False.

get_aggregator(tensor_dict=None)

Get federation aggregator.

This method retrieves the federation aggregator. If the aggregator does not exist, it is built using the configuration settings and the provided tensor dictionary.

Parameters:

tensor_dict (dict, optional) – The initial tensor dictionary to use when building the aggregator. Defaults to None.

Returns:

self.aggregator_ (Aggregator) – The federation aggregator.

Raises:

TypeError – If the log_metric_callback is not a callable object or cannot be imported from code.

get_assigner()

Get the plan task assigner.

get_client(collaborator_name, aggregator_uuid, federation_uuid, root_certificate=None, private_key=None, certificate=None)

Get gRPC client for the specified collaborator.

Parameters:

  • collaborator_name (str) – Name of the collaborator.

  • aggregator_uuid (str) – UUID of the aggregator.

  • federation_uuid (str) – UUID of the federation.

  • root_certificate (str, optional) – Root certificate for the collaborator. Defaults to None.

  • private_key (str, optional) – Private key for the collaborator. Defaults to None.

  • certificate (str, optional) – Certificate for the collaborator. Defaults to None.

Returns:

AggregatorGRPCClient – gRPC client for the specified collaborator.

get_collaborator(collaborator_name, root_certificate=None, private_key=None, certificate=None, task_runner=None, client=None, shard_descriptor=None)

Get collaborator.

This method retrieves a collaborator. If the collaborator does not exist, it is built using the configuration settings and the provided parameters.

Parameters:

  • collaborator_name (str) – Name of the collaborator.

  • root_certificate (str, optional) – Root certificate for the collaborator. Defaults to None.

  • private_key (str, optional) – Private key for the collaborator. Defaults to None.

  • certificate (str, optional) – Certificate for the collaborator. Defaults to None.

  • task_runner (TaskRunner, optional) – Task runner for the collaborator. Defaults to None.

  • client (Client, optional) – Client for the collaborator. Defaults to None.

  • shard_descriptor (ShardDescriptor, optional) – Descriptor of the data shard. Defaults to None.

Returns:

self.collaborator_ (Collaborator) – The collaborator instance.

get_core_task_runner(data_loader=None, model_provider=None, task_keeper=None)

Get core task runner.

Parameters:

  • data_loader (DataLoader, optional) – Data loader for the tasks. Defaults to None.

  • model_provider (ModelProvider, optional) – Provider for the model. Defaults to None.

  • task_keeper (TaskKeeper, optional) – Keeper for the tasks. Defaults to None.

Returns:

CoreTaskRunner – Core task runner for the tasks.

get_data_loader(collaborator_name)

Get data loader for a specific collaborator.

Parameters:

collaborator_name (str) – Name of the collaborator.

Returns:

DataLoader – Data loader for the specified collaborator.

get_serializer_plugin(**kwargs)

Get serializer plugin.

This plugin is used for serialization of interfaces in new interactive API.

Parameters:

**kwargs – Additional keyword arguments.

Returns:

SerializerPlugin – Serializer plugin.

get_server(root_certificate=None, private_key=None, certificate=None, **kwargs)

Get gRPC server of the aggregator instance.

Parameters:

  • root_certificate (str, optional) – Root certificate for the server. Defaults to None.

  • private_key (str, optional) – Private key for the server. Defaults to None.

  • certificate (str, optional) – Certificate for the server. Defaults to None.

  • **kwargs – Additional keyword arguments.

Returns:

AggregatorGRPCServer – gRPC server of the aggregator instance.

get_straggler_handling_policy()

Get straggler handling policy.

get_task_runner(data_loader)

Get task runner.

Parameters:

data_loader (DataLoader) – Data loader for the tasks.

Returns:

TaskRunner – Task runner for the tasks.

get_tasks()

Get federation tasks.

get_tensor_pipe()

Get data tensor pipeline.

property hash

Generate hash for this instance.

static import_(template)

Import an instance of a openfl Component or Federated DataLoader/TaskRunner.

Parameters:

template (str) – Fully qualified object path.

Returns:

object – A Python object.

initialize_data_loader(data_loader, shard_descriptor)

Initialize data loader.

Parameters:

  • data_loader (DataLoader) – Data loader to initialize.

  • shard_descriptor (ShardDescriptor) – Descriptor of the data shard.

Returns:

DataLoader – Initialized data loader.

interactive_api_get_server(*, tensor_dict, root_certificate, certificate, private_key, tls)

Get gRPC server of the aggregator instance for interactive API.

Parameters:

  • tensor_dict (dict) – Dictionary of tensors.

  • root_certificate (str) – Root certificate for the server.

  • certificate (str) – Certificate for the server.

  • private_key (str) – Private key for the server.

  • tls (bool) – Whether to use Transport Layer Security.

Returns:

AggregatorGRPCServer – gRPC server of the aggregator instance.

static load(yaml_path: Path, default: dict | None = None)

Load the plan from YAML file.

Parameters:

  • yaml_path (Path) – Path to the YAML file.

  • default (dict, optional) – Default plan configuration. Defaults to {}.

Returns:

dict – Plan configuration loaded from the YAML file.

static parse(plan_config_path: Path, cols_config_path: Path | None = None, data_config_path: Path | None = None, gandlf_config_path=None, resolve=True)

Parse the Federated Learning plan.

Parameters:

  • plan_config_path (Path) – The filepath to the Federated Learning plan.

  • cols_config_path (Path, optional) – The filepath to the Federation collaborator list. Defaults to None.

  • data_config_path (Path, optional) – The filepath to the Federation collaborator data configuration. Defaults to None.

  • gandlf_config_path (Path, optional) – The filepath to a yaml file that overrides the configuration. Defaults to None.

  • resolve (bool, optional) – Flag to resolve the plan settings. Defaults to True.

Returns:

Plan – A Federated Learning plan object.

resolve()

Resolve the federation settings.

restore_object(filename)

Deserialize an object.

Parameters:

filename (str) – Name of the file.

Returns:

object – Deserialized object.

plan

Plan module.