nvflare.recipe.poc_env module

class PocEnv(*, num_clients: int | None = 2, clients: list[str] | None = None, gpu_ids: list[int] | None = None, use_he: bool = False, docker_image: str | None = None, project_conf_path: str = '', username: str = 'admin@nvidia.com', study: str = 'default', extra: dict | None = None)[source]

Bases: ExecEnv

Proof of Concept execution environment for local testing and development.

This environment sets up a POC deployment on a single machine with multiple processes representing the server, clients, and admin console.

Initialize POC execution environment.

Parameters:

  • num_clients (int, optional) – Number of clients to use in POC mode. Defaults to 2.

  • clients (list[str], optional) – List of client names. If None, will generate site-1, site-2, etc. Defaults to None. If specified, number_of_clients argument will be ignored.

  • gpu_ids (list[int], optional) – List of GPU IDs to assign to clients. If None, uses CPU only. Defaults to None.

  • use_he (bool, optional) – Whether to use HE. Defaults to False.

  • docker_image (str, optional) – Docker image to use for POC. Defaults to None.

  • project_conf_path (str, optional) – Path to the project configuration file. Defaults to “”. If specified, ‘number_of_clients’,’clients’ and ‘docker’ specific options will be ignored.

  • username (str, optional) – Admin user. Defaults to “admin@nvidia.com”.

  • study (str, optional) – Study name to tag submitted jobs. Defaults to “default”.

  • extra – extra env info.

abort_job(job_id: str) → None[source]

Abort a running job.

Parameters:

job_id – The job ID to abort.

deploy(job: FedJob) → str[source]

Deploy a FedJob to the POC environment.

Parameters:

job (FedJob) – The FedJob to deploy.

Returns:

Job ID.

Return type:

str

Raises:

ValueError – If scripts do not exist locally.

get_job_result(job_id: str, timeout: float = 0.0) → str | None[source]

Get the result workspace of a job.

Parameters:

  • job_id – The job ID to get results for.

  • timeout – The timeout for the job to complete. Defaults to 0.0 (no timeout).

Returns:

The result workspace path if job completed, None otherwise.

Return type:

Optional[str]

get_job_status(job_id: str) → str | None[source]

Get the status of a job.

Parameters:

job_id – The job ID to check status for.

Returns:

The status of the job, or None if not available.

Return type:

Optional[str]

stop(clean_up: bool = False) → None[source]

Try to stop and clean existing POC.

This method is idempotent - safe to call multiple times.

Parameters:

clean_up (bool, optional) – Whether to clean the POC workspace. Defaults to False.