use ScriptExecutor

examples/hello-world/python_jobs/pt/client_api_pt.py

@@ -16,7 +16,7 @@
 
 from nvflare import FedJob
 from nvflare.app_common.workflows.fedavg import FedAvg
-from nvflare.app_opt.pt.in_process_client_api_executor import PTInProcessClientAPIExecutor
+from nvflare import ScriptExecutor
 
 
 if __name__ == "__main__":
@@ -33,7 +33,7 @@
     job.to(controller, "server")
 
     for i in range(n_clients):
-        executor = PTInProcessClientAPIExecutor(  # TODO: ScriptExecutor()
+        executor = ScriptExecutor(
             task_script_path=train_script,
             task_script_args=""  # f"--batch_size 32 --data_path /tmp/data/site-{i}"
         )

examples/hello-world/python_jobs/pt/client_api_pt_dp_filter.py

@@ -16,7 +16,7 @@
 
 from nvflare import FedJob, FilterType
 from nvflare.app_common.workflows.fedavg import FedAvg
-from nvflare.app_opt.pt.in_process_client_api_executor import PTInProcessClientAPIExecutor
+from nvflare import ScriptExecutor
 from nvflare.app_common.filters.percentile_privacy import PercentilePrivacy
 
 
@@ -34,7 +34,7 @@
     job.to(controller, "server")
 
     for i in range(n_clients):
-        executor = PTInProcessClientAPIExecutor(
+        executor = ScriptExecutor(
             task_script_path=train_script,
             task_script_args=""
         )

examples/hello-world/python_jobs/pt/client_api_pt_swarm.py

@@ -16,7 +16,7 @@
 
 from nvflare import FedJob
 from nvflare.apis.dxo import DataKind
-from nvflare.app_opt.pt.in_process_client_api_executor import PTInProcessClientAPIExecutor
+from nvflare import ScriptExecutor
 from nvflare.app_common.ccwf import SwarmServerController, CrossSiteEvalServerController, SwarmClientController, CrossSiteEvalClientController
 from nvflare.app_opt.pt.file_model_persistor import PTFileModelPersistor
 from nvflare.app_common.aggregators.intime_accumulate_model_aggregator import InTimeAccumulateWeightedAggregator

nvflare/__init__.py

@@ -20,3 +20,4 @@
 
 from nvflare.private.fed.app.simulator.simulator_runner import SimulatorRunner as SimulatorRunner
 from nvflare.fed_job import FedJob
+from nvflare.app_common.executors.script_executor import ScriptExecutor

nvflare/app_common/executors/script_executor.py

@@ -0,0 +1,68 @@
+# Copyright (c) 2024, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import Optional
+
+from nvflare.app_common.app_constant import AppConstants
+from nvflare.app_common.executors.in_process_client_api_executor import InProcessClientAPIExecutor
+from nvflare.app_opt.pt.decomposers import TensorDecomposer
+from nvflare.app_opt.pt.params_converter import NumpyToPTParamsConverter, PTToNumpyParamsConverter
+from nvflare.client.config import ExchangeFormat, TransferType
+from nvflare.fuel.utils import fobs
+
+
+class ScriptExecutor(InProcessClientAPIExecutor):
+    def __init__(
+        self,
+        task_script_path: str,
+        task_script_args: str = "",
+        task_wait_time: Optional[float] = None,
+        result_pull_interval: float = 0.5,
+        log_pull_interval: Optional[float] = None,
+        params_transfer_type: TransferType = TransferType.FULL,
+        from_nvflare_converter_id: Optional[str] = None,
+        to_nvflare_converter_id: Optional[str] = None,
+        train_with_evaluation: bool = True,
+        train_task_name: str = "train",
+        evaluate_task_name: str = "evaluate",
+        submit_model_task_name: str = "submit_model",
+        params_exchange_format=ExchangeFormat.PYTORCH,
+    ):
+        if params_exchange_format == ExchangeFormat.PYTORCH:
+            super(ScriptExecutor, self).__init__(
+                task_script_path=task_script_path,
+                task_script_args=task_script_args,
+                task_wait_time=task_wait_time,
+                result_pull_interval=result_pull_interval,
+                train_with_evaluation=train_with_evaluation,
+                train_task_name=train_task_name,
+                evaluate_task_name=evaluate_task_name,
+                submit_model_task_name=submit_model_task_name,
+                from_nvflare_converter_id=from_nvflare_converter_id,
+                to_nvflare_converter_id=to_nvflare_converter_id,
+                params_exchange_format=params_exchange_format,
+                params_transfer_type=params_transfer_type,
+                log_pull_interval=log_pull_interval,
+            )
+            fobs.register(TensorDecomposer)
+
+            if self._from_nvflare_converter is None:
+                self._from_nvflare_converter = NumpyToPTParamsConverter(
+                    [AppConstants.TASK_TRAIN, AppConstants.TASK_VALIDATION]
+                )
+            if self._to_nvflare_converter is None:
+                self._to_nvflare_converter = PTToNumpyParamsConverter(
+                    [AppConstants.TASK_TRAIN, AppConstants.TASK_SUBMIT_MODEL]
+                )
+        else:
+            raise NotImplementedError(f"`params_exchange_format`={params_exchange_format} not supported!")