style format

examples/advanced/monitoring/jobs/setup-1/fedavg_script_runner_pt.py

@@ -12,75 +12,68 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import argparse
 import os
+
 from src.simple_network import SimpleNetwork
 
 from nvflare.app_opt.pt.job_config.fed_avg import FedAvgJob
 from nvflare.fuel_opt.statsd.statsd_reporter import StatsDReporter
 from nvflare.job_config.script_runner import ScriptRunner
 from nvflare.metrics.job_metrics_collector import JobMetricsCollector
-import argparse
+
 
 def define_parser(parser):
 
     parser.add_argument(
-        "-j", 
-        "--job_configs_dir", 
-        type=str, 
-        default="/tmp/nvflare/jobs/job_config/", 
-        help="job configure folder"
+        "-j", "--job_configs_dir", type=str, default="/tmp/nvflare/jobs/job_config/", help="job configure folder"
     )
     return parser
 
 
-
 def main(job_configs_dir):
-        n_clients = 2
-        num_rounds = 2
-        train_script = "src/hello-pt_cifar10_fl.py"
-        job_name = "hello-pt"
-
-        job = FedAvgJob(
-            name=job_name,
-            n_clients=n_clients,
-            num_rounds=num_rounds,
-            initial_model=SimpleNetwork()
-        )
-
-        # add server side monitoring components
-
-        server_tags = {"site": "server", "env": "dev"}
-
-        metrics_reporter = StatsDReporter(site="server", host="localhost", port=9125)
-        metrics_collector = JobMetricsCollector(tags=server_tags, streaming_to_server=False)
-
-        job.to_server(metrics_collector, id="server_job_metrics_collector")
-        job.to_server(metrics_reporter, id="statsd_reporter")
-
-        # Add clients
-        for i in range(n_clients):
-            executor = ScriptRunner(script=train_script, script_args="")
-            client_site = f"site-{i + 1}"
-            job.to(executor, client_site)
-
-            # add client side monitoring components
-            tags = {"site": client_site, "env": "dev"}
-
-            metrics_collector = JobMetricsCollector(tags=tags)
-
-            job.to(metrics_collector, target=client_site, id=f"{client_site}_job_metrics_collector")
-            job.to(metrics_reporter, target=client_site, id=f"statsd_reporter")
-
-        job_config_path = os.path.join(job_configs_dir, job_name)
-        print(f"job config folder = {job_config_path}")
-
-        job.export_job(job_configs_dir)
-        # job.simulator_run("/tmp/nvflare/jobs/workdir", gpu="0")
+    n_clients = 2
+    num_rounds = 2
+    train_script = "src/hello-pt_cifar10_fl.py"
+    job_name = "hello-pt"
+
+    job = FedAvgJob(name=job_name, n_clients=n_clients, num_rounds=num_rounds, initial_model=SimpleNetwork())
+
+    # add server side monitoring components
+
+    server_tags = {"site": "server", "env": "dev"}
+
+    metrics_reporter = StatsDReporter(site="server", host="localhost", port=9125)
+    metrics_collector = JobMetricsCollector(tags=server_tags, streaming_to_server=False)
+
+    job.to_server(metrics_collector, id="server_job_metrics_collector")
+    job.to_server(metrics_reporter, id="statsd_reporter")
+
+    # Add clients
+    for i in range(n_clients):
+        executor = ScriptRunner(script=train_script, script_args="")
+        client_site = f"site-{i + 1}"
+        job.to(executor, client_site)
+
+        # add client side monitoring components
+        tags = {"site": client_site, "env": "dev"}
+
+        metrics_collector = JobMetricsCollector(tags=tags)
+
+        job.to(metrics_collector, target=client_site, id=f"{client_site}_job_metrics_collector")
+        job.to(metrics_reporter, target=client_site, id="statsd_reporter")
+
+    job_config_path = os.path.join(job_configs_dir, job_name)
+    print(f"job config folder = {job_config_path}")
+
+    job.export_job(job_configs_dir)
+    # job.simulator_run("/tmp/nvflare/jobs/workdir", gpu="0")
+
 
 if __name__ == "__main__":
 
     parser = argparse.ArgumentParser(description="FedAvg Job Script")
     parser = define_parser(parser)
     args = parser.parse_args()
-    
+
     main(args.job_configs_dir)

examples/advanced/monitoring/jobs/setup-2/fedavg_script_runner_pt.py

@@ -12,28 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from nvflare.app_common.widgets.convert_to_fed_event import ConvertToFedEvent
-from nvflare.metrics.metrics_keys import METRICS_EVENT_TYPE
-from nvflare.metrics.remote_metrics_receiver import RemoteMetricsReceiver
 from src.simple_network import SimpleNetwork
 
+from nvflare.app_common.widgets.convert_to_fed_event import ConvertToFedEvent
 from nvflare.app_opt.pt.job_config.fed_avg import FedAvgJob
 from nvflare.fuel_opt.statsd.statsd_reporter import StatsDReporter
 from nvflare.job_config.script_runner import ScriptRunner
 from nvflare.metrics.job_metrics_collector import JobMetricsCollector
+from nvflare.metrics.metrics_keys import METRICS_EVENT_TYPE
+from nvflare.metrics.remote_metrics_receiver import RemoteMetricsReceiver
 
 if __name__ == "__main__":
     n_clients = 2
     num_rounds = 2
     train_script = "src/hello-pt_cifar10_fl.py"
 
     job = FedAvgJob(
-        name="hello-pt_cifar10_fedavg",
-        n_clients=n_clients,
-        num_rounds=num_rounds,
-        initial_model=SimpleNetwork()
+        name="hello-pt_cifar10_fedavg", n_clients=n_clients, num_rounds=num_rounds, initial_model=SimpleNetwork()
     )
-    metrics_reporter = StatsDReporter(site = "server", host="localhost", port=9125)
+    metrics_reporter = StatsDReporter(site="server", host="localhost", port=9125)
     remote_metrics_collector = RemoteMetricsReceiver()
 
     server_tags = {"site": "server", "env": "dev"}

nvflare/fuel_opt/statsd/statsd_reporter.py

@@ -25,19 +25,19 @@
 
 class StatsDReporter:
 
-    def __init__(self,site: str="", host="localhost", port=9125):
+    def __init__(self, site: str = "", host="localhost", port=9125):
 
         # Initialize the DataDog StatsD client
         initialize(statsd_host=host, statsd_port=port)
         self.metrics = {}
         self.data_bus = DataBus()
- 
+
         self.data_bus.subscribe([ReservedTopic.APP_METRICS], self.process_metrics)
         self.logger = logging.getLogger(self.__class__.__name__)
         self.site = site
- 
+
     def process_metrics(self, topic, metrics, data_bus):
-        
+
         if topic == ReservedTopic.APP_METRICS:
             try:
                 for metric in metrics:

nvflare/metrics/job_metrics_collector.py

@@ -209,5 +209,3 @@ def handle_event(self, event: str, fl_ctx: FLContext):
 
     def publish_metrics(self, metrics: dict, metric_name: str, tags: dict, fl_ctx: FLContext):
         collect_metrics(self, self.streaming_to_server, metrics, metric_name, tags, self.data_bus, fl_ctx)
-
-

nvflare/metrics/metrics_keys.py

@@ -14,6 +14,7 @@
 
 METRICS_EVENT_TYPE = "metrics_event"
 
+
 class MetricTypes:
 
     COUNTER = "counter"

nvflare/metrics/metrics_publisher.py

@@ -40,29 +40,33 @@ def publish_app_metrics(metrics: dict, metric_name: str, tags: dict, data_bus: D
 
 
 def convert_metrics_to_event(
-        comp: FLComponent,
-        metrics: dict,
-        metric_name: str,
-        tags: dict,
-        fl_ctx: FLContext,
-    ) -> None:
-        metrics_data = {MetricKeys.metric_name: metric_name, MetricKeys.value: metrics, MetricKeys.tags: tags}
-        shareable = Shareable(data={"METRICS": metrics_data})
+    comp: FLComponent,
+    metrics: dict,
+    metric_name: str,
+    tags: dict,
+    fl_ctx: FLContext,
+) -> None:
+    metrics_data = {MetricKeys.metric_name: metric_name, MetricKeys.value: metrics, MetricKeys.tags: tags}
+    shareable = Shareable(data={"METRICS": metrics_data})
 
-        print("\nfire local event ==> ", METRICS_EVENT_TYPE)
-        with fl_ctx.get_engine().new_context() as fl_ctx2:
-            fl_ctx2.set_prop(key=FLContextKey.EVENT_DATA, value=shareable, private=True, sticky=False)
-            comp.fire_event(event_type=METRICS_EVENT_TYPE, fl_ctx=fl_ctx2)
-
+    print("\nfire local event ==> ", METRICS_EVENT_TYPE)
+    with fl_ctx.get_engine().new_context() as fl_ctx2:
+        fl_ctx2.set_prop(key=FLContextKey.EVENT_DATA, value=shareable, private=True, sticky=False)
+        comp.fire_event(event_type=METRICS_EVENT_TYPE, fl_ctx=fl_ctx2)
 
 
+def collect_metrics(
+    comp: FLComponent,
+    streaming_to_server: bool,
+    metrics: dict,
+    metric_name: str,
+    tags: dict,
+    data_bus: DataBus,
+    fl_ctx: FLContext,
+):
 
-def collect_metrics(comp: FLComponent, streaming_to_server: bool, metrics: dict, metric_name: str, tags: dict, data_bus: DataBus, fl_ctx: FLContext):
-
     if not streaming_to_server:
         publish_app_metrics(metrics, metric_name, tags, data_bus)
     else:
         print("converting metrics to event", metrics, metric_name, tags)
         convert_metrics_to_event(comp, metrics, metric_name, tags, fl_ctx)
-
-

nvflare/metrics/remote_metrics_receiver.py

@@ -1,18 +1,31 @@
+# Copyright (c) 2025, 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 List, Optional
 
 from nvflare.apis.fl_component import FLComponent
-from nvflare.apis.fl_constant import FLContextKey, ReservedKey, EventScope
+from nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey
 from nvflare.apis.fl_context import FLContext
+from nvflare.apis.shareable import Shareable
 from nvflare.fuel.data_event.data_bus import DataBus
 from nvflare.metrics.metrics_keys import METRICS_EVENT_TYPE, MetricKeys
-from nvflare.apis.shareable import Shareable
 from nvflare.metrics.metrics_publisher import publish_app_metrics
 
 
 class RemoteMetricsReceiver(FLComponent):
     def __init__(self, events: Optional[List[str]] = None):
-
         """Receives metrics data from client sites and publishes it to the local data bus.
         Args:
             events (optional, List[str]): A list of event that this receiver will handle.
@@ -47,7 +60,7 @@ def handle_event(self, event_type: str, fl_ctx: FLContext):
             if record_origin is None:
                 self.log_error(fl_ctx, "record_origin can't be None.", fire_event=False)
                 return
-            
+
             metrics_data = data.get("METRICS")
             if metrics_data is None:
                 self.log_error(fl_ctx, "Missing metrics data.", fire_event=False)
@@ -58,6 +71,3 @@ def handle_event(self, event_type: str, fl_ctx: FLContext):
             tags = metrics_data.get(MetricKeys.tags)
 
             publish_app_metrics(metrics=metrics, metric_name=metric_name, tags=tags, data_bus=self.data_bus)
-
-
-

nvflare/metrics/system_metrics_collector.py

@@ -34,7 +34,7 @@ def __init__(self, tags: dict, streaming_to_server: bool = False):
 
         self.tags = tags
         self.data_bus = DataBus()
-        self.streaming_to_server = streaming_to_server        
+        self.streaming_to_server = streaming_to_server
 
         # system events
         self.system_start = 0
@@ -51,7 +51,6 @@ def __init__(self, tags: dict, streaming_to_server: bool = False):
         self.after_check_client_resources = 0
         self.start = True
 
-
     def handle_event(self, event: str, fl_ctx: FLContext):
 
         current_time = time.time()
@@ -136,4 +135,3 @@ def handle_event(self, event: str, fl_ctx: FLContext):
     def publish_metrics(self, metrics: dict, metric_name: str, tags: dict, fl_ctx: FLContext):
 
         collect_metrics(self, self.streaming_to_server, metrics, metric_name, tags, self.data_bus, fl_ctx)
-