We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add an event_type argument in AnalyticsSender
Right now AnalyticsSender is sending default event type "analytix_log_stats".

However, in the ConvertToFedEvent and TBAnalyticsReceiver JSON configuration, the user needs to config the "analytix_log_stats" in the events to work.
It's not natural for user to know where this event is coming from.
If we add an "event_type" argument  in the AnalyticsSender, we can make the event sending, converting and receiving in pairs.
That would be easier for people to understand.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvflare/app_common/widgets/streaming.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from abc import ABC, abstractmethod
16 from threading import Lock
17 from typing import List, Optional
18 
19 from nvflare.apis.analytix import AnalyticsData, AnalyticsDataType
20 from nvflare.apis.dxo import DXO
21 from nvflare.apis.event_type import EventType
22 from nvflare.apis.fl_component import FLComponent
23 from nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey
24 from nvflare.apis.fl_context import FLContext
25 from nvflare.apis.shareable import Shareable
26 from nvflare.app_common.app_event_type import AppEventType
27 from nvflare.widgets.widget import Widget
28 
29 
30 def send_analytic_dxo(
31     comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = AppEventType.ANALYTIC_EVENT_TYPE
32 ):
33     """Sends analytic dxo.
34 
35     Args:
36         comp (FLComponent): An FLComponent.
37         dxo (DXO): analytic data in dxo.
38         fl_ctx (FLContext): fl context info.
39         event_type (str): Event type.
40     """
41     if not isinstance(comp, FLComponent):
42         raise TypeError(f"expect comp to be an instance of FLComponent, but got {type(comp)}")
43     if not isinstance(dxo, DXO):
44         raise TypeError(f"expect dxo to be an instance of DXO, but got {type(dxo)}")
45     if not isinstance(fl_ctx, FLContext):
46         raise TypeError(f"expect fl_ctx to be an instance of FLContext, but got {type(fl_ctx)}")
47 
48     fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=dxo.to_shareable(), private=True, sticky=False)
49     comp.fire_event(event_type=event_type, fl_ctx=fl_ctx)
50 
51 
52 def create_analytic_dxo(tag: str, value, data_type: AnalyticsDataType, **kwargs) -> DXO:
53     """Creates the analytic DXO.
54 
55     Args:
56         tag (str): the tag associated with this value.
57         value: the analytic data.
58         data_type (AnalyticsDataType): analytic data type.
59         kwargs: additional arguments to be passed into the receiver side's function.
60 
61     Returns:
62         A DXO object that contains the analytic data.
63     """
64     data = AnalyticsData(tag=tag, value=value, data_type=data_type, kwargs=kwargs)
65     dxo = data.to_dxo()
66     return dxo
67 
68 
69 class AnalyticsSender(Widget):
70     def __init__(self):
71         """Sends analytics data.
72 
73         This class implements some common methods follows signatures from PyTorch SummaryWriter.
74         It provides a convenient way for Learner to use.
75         """
76         super().__init__()
77         self.engine = None
78 
79     def handle_event(self, event_type: str, fl_ctx: FLContext):
80         if event_type == EventType.START_RUN:
81             self.engine = fl_ctx.get_engine()
82 
83     def _add(
84         self,
85         tag: str,
86         value,
87         data_type: AnalyticsDataType,
88         global_step: Optional[int] = None,
89         kwargs: Optional[dict] = None,
90     ):
91         kwargs = kwargs if kwargs else {}
92         if global_step:
93             if not isinstance(global_step, int):
94                 raise TypeError(f"Expect global step to be an instance of int, but got {type(global_step)}")
95             kwargs["global_step"] = global_step
96         dxo = create_analytic_dxo(tag=tag, value=value, data_type=data_type, **kwargs)
97         with self.engine.new_context() as fl_ctx:
98             send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx)
99 
100     def add_scalar(self, tag: str, scalar: float, global_step: Optional[int] = None, **kwargs):
101         """Sends a scalar.
102 
103         Args:
104             tag (str): Data identifier.
105             scalar (float): Value to send.
106             global_step (optional, int): Global step value.
107             **kwargs: Additional arguments to pass to the receiver side.
108         """
109         self._add(tag=tag, value=scalar, data_type=AnalyticsDataType.SCALAR, global_step=global_step, kwargs=kwargs)
110 
111     def add_scalars(self, tag: str, scalars: dict, global_step: Optional[int] = None, **kwargs):
112         """Sends scalars.
113 
114         Args:
115             tag (str): The parent name for the tags.
116             scalars (dict): Key-value pair storing the tag and corresponding values.
117             global_step (optional, int): Global step value.
118             **kwargs: Additional arguments to pass to the receiver side.
119         """
120         self._add(tag=tag, value=scalars, data_type=AnalyticsDataType.SCALARS, global_step=global_step, kwargs=kwargs)
121 
122     def add_text(self, tag: str, text: str, global_step: Optional[int] = None, **kwargs):
123         """Sends a text.
124 
125         Args:
126             tag (str): Data identifier.
127             text (str): String to send.
128             global_step (optional, int): Global step value.
129             **kwargs: Additional arguments to pass to the receiver side.
130         """
131         self._add(tag=tag, value=text, data_type=AnalyticsDataType.TEXT, global_step=global_step, kwargs=kwargs)
132 
133     def add_image(self, tag: str, image, global_step: Optional[int] = None, **kwargs):
134         """Sends an image.
135 
136         Args:
137             tag (str): Data identifier.
138             image: Image to send.
139             global_step (optional, int): Global step value.
140             **kwargs: Additional arguments to pass to the receiver side.
141         """
142         self._add(tag=tag, value=image, data_type=AnalyticsDataType.IMAGE, global_step=global_step, kwargs=kwargs)
143 
144     def flush(self):
145         """Flushes out the message.
146 
147         This is doing nothing, it is defined for mimic the PyTorch SummaryWriter behavior.
148         """
149         pass
150 
151     def close(self):
152         """Close resources."""
153         if self.engine:
154             self.engine = None
155 
156 
157 class AnalyticsReceiver(Widget, ABC):
158     def __init__(self, events: Optional[List[str]] = None):
159         """Receives analytic data.
160 
161         Args:
162             events (optional, List[str]): A list of event that this receiver will handle.
163         """
164         super().__init__()
165         if events is None:
166             events = [AppEventType.ANALYTIC_EVENT_TYPE, f"fed.{AppEventType.ANALYTIC_EVENT_TYPE}"]
167         self.events = events
168         self._save_lock = Lock()
169         self._end = False
170 
171     @abstractmethod
172     def initialize(self, fl_ctx: FLContext):
173         """Initializes the receiver.
174 
175         Args:
176             fl_ctx (FLContext): fl context.
177         """
178         pass
179 
180     @abstractmethod
181     def save(self, fl_ctx: FLContext, shareable: Shareable, record_origin: str):
182         """Saves the received data.
183 
184         Args:
185             fl_ctx (FLContext): fl context.
186             shareable (Shareable): the received message.
187             record_origin (str): the sender of this message / record.
188         """
189         pass
190 
191     @abstractmethod
192     def finalize(self, fl_ctx: FLContext):
193         """Finalizes the receiver.
194 
195         Args:
196             fl_ctx (FLContext): fl context.
197         """
198         pass
199 
200     def handle_event(self, event_type: str, fl_ctx: FLContext):
201         if event_type == EventType.START_RUN:
202             self.initialize(fl_ctx)
203         elif event_type in self.events:
204             if self._end:
205                 self.log_debug(fl_ctx, f"Already received end run event, drop event {event_type}.", fire_event=False)
206                 return
207             data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)
208             if data is None:
209                 self.log_error(fl_ctx, "Missing event data.", fire_event=False)
210                 return
211             if not isinstance(data, Shareable):
212                 self.log_error(
213                     fl_ctx, f"Expect data to be an instance of Shareable but got {type(data)}", fire_event=False
214                 )
215                 return
216 
217             # if fed event use peer name to save
218             if fl_ctx.get_prop(FLContextKey.EVENT_SCOPE) == EventScope.FEDERATION:
219                 record_origin = data.get_peer_prop(ReservedKey.IDENTITY_NAME, None)
220             else:
221                 record_origin = fl_ctx.get_identity_name()
222 
223             if record_origin is None:
224                 self.log_error(fl_ctx, "record_origin can't be None.", fire_event=False)
225                 return
226             with self._save_lock:
227                 self.save(shareable=data, fl_ctx=fl_ctx, record_origin=record_origin)
228         elif event_type == EventType.END_RUN:
229             self._end = True
230             self.finalize(fl_ctx)
231
```

Path: `nvflare/app_common/app_event_type.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 class AppEventType(object):
17     """Defines application events."""
18 
19     START_ROUND = "_start_round"
20     END_ROUND = "_end_round"
21 
22     BEFORE_AGGREGATION = "_before_aggregation"
23     END_AGGREGATION = "_end_aggregation"
24 
25     SUBMIT_LOCAL_BEST_MODEL = "_submit_local_best_model"
26     SERVER_RECEIVE_BEST_MODEL = "_server_receive_best_model"
27     RECEIVE_VALIDATION_MODEL = "_receive_validation_model"
28     SEND_VALIDATION_RESULTS = "_send_validation_results"
29     RECEIVE_VALIDATION_RESULTS = "_receive_validation_results"
30 
31     BEFORE_INITIALIZE = "_before_initialize"
32     AFTER_INITIALIZE = "_after_initialize"
33     BEFORE_TRAIN = "_before_train"
34     AFTER_TRAIN = "_after_train"
35 
36     BEFORE_SHAREABLE_TO_LEARNABLE = "_before_model_update"
37     AFTER_SHAREABLE_TO_LEARNABLE = "_after_model_update"
38     BEFORE_LEARNABLE_PERSIST = "_before_save_model"
39     AFTER_LEARNABLE_PERSIST = "_after_save_model"
40     BEFORE_SEND_BEST_MODEL = "_before_send_best_model"
41     AFTER_SEND_BEST_MODEL = "_after_send_best_model"
42     LOCAL_BEST_MODEL_AVAILABLE = "_local_best_model_available"
43     GLOBAL_BEST_MODEL_AVAILABLE = "_global_best_model_available"
44     BEFORE_GET_VALIDATION_MODELS = "_before_get_validation_models"
45     AFTER_GET_VALIDATION_MODELS = "_after_get_validation_models"
46     SEND_MODEL_FOR_VALIDATION = "_send_model_for_validation"
47     BEFORE_VALIDATE_MODEL = "_before_validate_model"
48     AFTER_VALIDATE_MODEL = "_after_validate_model"
49     BEFORE_SUBMIT_VALIDATION_RESULTS = "_before_submit_validation_results"
50     AFTER_SUBMIT_VALIDATION_RESULTS = "_after_submit_validation_results"
51 
52     # Events
53     ROUND_STARTED = "_round_started"
54     ROUND_DONE = "_round_done"
55     INITIAL_MODEL_LOADED = "_initial_model_loaded"
56     BEFORE_TRAIN_TASK = "_before_train_task"
57     RECEIVE_CONTRIBUTION = "_receive_contribution"
58     AFTER_CONTRIBUTION_ACCEPT = "_after_contribution_accept"
59     AFTER_AGGREGATION = "_after_aggregation"
60     BEFORE_CONTRIBUTION_ACCEPT = "_before_contribution_accept"
61     GLOBAL_WEIGHTS_UPDATED = "_global_weights_updated"
62     TRAINING_STARTED = "_training_started"
63     TRAINING_FINISHED = "_training_finished"
64     TRAIN_DONE = "_train_done"
65 
66     CROSS_VAL_INIT = "_cross_val_init"
67     VALIDATION_RESULT_RECEIVED = "_validation_result_received"
68     RECEIVE_BEST_MODEL = "_receive_best_model"
69 
70     ANALYTIC_EVENT_TYPE = "analytix_log_stats"
71     LOGGING_EVENT_TYPE = "analytix_logging"
72
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvflare/app_common/app_event_type.py b/nvflare/app_common/app_event_type.py
--- a/nvflare/app_common/app_event_type.py
+++ b/nvflare/app_common/app_event_type.py
@@ -66,6 +66,3 @@
     CROSS_VAL_INIT = "_cross_val_init"
     VALIDATION_RESULT_RECEIVED = "_validation_result_received"
     RECEIVE_BEST_MODEL = "_receive_best_model"
-
-    ANALYTIC_EVENT_TYPE = "analytix_log_stats"
-    LOGGING_EVENT_TYPE = "analytix_logging"
diff --git a/nvflare/app_common/widgets/streaming.py b/nvflare/app_common/widgets/streaming.py
--- a/nvflare/app_common/widgets/streaming.py
+++ b/nvflare/app_common/widgets/streaming.py
@@ -23,13 +23,12 @@
 from nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey
 from nvflare.apis.fl_context import FLContext
 from nvflare.apis.shareable import Shareable
-from nvflare.app_common.app_event_type import AppEventType
 from nvflare.widgets.widget import Widget
 
+ANALYTIC_EVENT_TYPE = "analytix_log_stats"
 
-def send_analytic_dxo(
-    comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = AppEventType.ANALYTIC_EVENT_TYPE
-):
+
+def send_analytic_dxo(comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = ANALYTIC_EVENT_TYPE):
     """Sends analytic dxo.
 
     Args:
@@ -67,14 +66,19 @@
 
 
 class AnalyticsSender(Widget):
-    def __init__(self):
+    def __init__(self, event_type=ANALYTIC_EVENT_TYPE):
         """Sends analytics data.
 
-        This class implements some common methods follows signatures from PyTorch SummaryWriter.
-        It provides a convenient way for Learner to use.
+        Note::
+            This class implements some common methods follows signatures from PyTorch SummaryWriter.
+            It provides a convenient way for Learner to use.
+
+        Args:
+            event_type (str): event type to fire.
         """
         super().__init__()
         self.engine = None
+        self.event_type = event_type
 
     def handle_event(self, event_type: str, fl_ctx: FLContext):
         if event_type == EventType.START_RUN:
@@ -95,7 +99,7 @@
             kwargs["global_step"] = global_step
         dxo = create_analytic_dxo(tag=tag, value=value, data_type=data_type, **kwargs)
         with self.engine.new_context() as fl_ctx:
-            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx)
+            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx, event_type=self.event_type)
 
     def add_scalar(self, tag: str, scalar: float, global_step: Optional[int] = None, **kwargs):
         """Sends a scalar.
@@ -163,7 +167,7 @@
         """
         super().__init__()
         if events is None:
-            events = [AppEventType.ANALYTIC_EVENT_TYPE, f"fed.{AppEventType.ANALYTIC_EVENT_TYPE}"]
+            events = [ANALYTIC_EVENT_TYPE, f"fed.{ANALYTIC_EVENT_TYPE}"]
         self.events = events
         self._save_lock = Lock()
         self._end = False

{"golden_diff": "diff --git a/nvflare/app_common/app_event_type.py b/nvflare/app_common/app_event_type.py\n--- a/nvflare/app_common/app_event_type.py\n+++ b/nvflare/app_common/app_event_type.py\n@@ -66,6 +66,3 @@\n     CROSS_VAL_INIT = \"_cross_val_init\"\n     VALIDATION_RESULT_RECEIVED = \"_validation_result_received\"\n     RECEIVE_BEST_MODEL = \"_receive_best_model\"\n-\n-    ANALYTIC_EVENT_TYPE = \"analytix_log_stats\"\n-    LOGGING_EVENT_TYPE = \"analytix_logging\"\ndiff --git a/nvflare/app_common/widgets/streaming.py b/nvflare/app_common/widgets/streaming.py\n--- a/nvflare/app_common/widgets/streaming.py\n+++ b/nvflare/app_common/widgets/streaming.py\n@@ -23,13 +23,12 @@\n from nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey\n from nvflare.apis.fl_context import FLContext\n from nvflare.apis.shareable import Shareable\n-from nvflare.app_common.app_event_type import AppEventType\n from nvflare.widgets.widget import Widget\n \n+ANALYTIC_EVENT_TYPE = \"analytix_log_stats\"\n \n-def send_analytic_dxo(\n-    comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = AppEventType.ANALYTIC_EVENT_TYPE\n-):\n+\n+def send_analytic_dxo(comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = ANALYTIC_EVENT_TYPE):\n     \"\"\"Sends analytic dxo.\n \n     Args:\n@@ -67,14 +66,19 @@\n \n \n class AnalyticsSender(Widget):\n-    def __init__(self):\n+    def __init__(self, event_type=ANALYTIC_EVENT_TYPE):\n         \"\"\"Sends analytics data.\n \n-        This class implements some common methods follows signatures from PyTorch SummaryWriter.\n-        It provides a convenient way for Learner to use.\n+        Note::\n+            This class implements some common methods follows signatures from PyTorch SummaryWriter.\n+            It provides a convenient way for Learner to use.\n+\n+        Args:\n+            event_type (str): event type to fire.\n         \"\"\"\n         super().__init__()\n         self.engine = None\n+        self.event_type = event_type\n \n     def handle_event(self, event_type: str, fl_ctx: FLContext):\n         if event_type == EventType.START_RUN:\n@@ -95,7 +99,7 @@\n             kwargs[\"global_step\"] = global_step\n         dxo = create_analytic_dxo(tag=tag, value=value, data_type=data_type, **kwargs)\n         with self.engine.new_context() as fl_ctx:\n-            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx)\n+            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx, event_type=self.event_type)\n \n     def add_scalar(self, tag: str, scalar: float, global_step: Optional[int] = None, **kwargs):\n         \"\"\"Sends a scalar.\n@@ -163,7 +167,7 @@\n         \"\"\"\n         super().__init__()\n         if events is None:\n-            events = [AppEventType.ANALYTIC_EVENT_TYPE, f\"fed.{AppEventType.ANALYTIC_EVENT_TYPE}\"]\n+            events = [ANALYTIC_EVENT_TYPE, f\"fed.{ANALYTIC_EVENT_TYPE}\"]\n         self.events = events\n         self._save_lock = Lock()\n         self._end = False\n", "issue": "Add an event_type argument in AnalyticsSender\nRight now AnalyticsSender is sending default event type \"analytix_log_stats\".\r\n\r\nHowever, in the ConvertToFedEvent and TBAnalyticsReceiver JSON configuration, the user needs to config the \"analytix_log_stats\" in the events to work.\r\nIt's not natural for user to know where this event is coming from.\r\nIf we add an \"event_type\" argument  in the AnalyticsSender, we can make the event sending, converting and receiving in pairs.\r\nThat would be easier for people to understand.\n", "before_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom threading import Lock\nfrom typing import List, Optional\n\nfrom nvflare.apis.analytix import AnalyticsData, AnalyticsDataType\nfrom nvflare.apis.dxo import DXO\nfrom nvflare.apis.event_type import EventType\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.shareable import Shareable\nfrom nvflare.app_common.app_event_type import AppEventType\nfrom nvflare.widgets.widget import Widget\n\n\ndef send_analytic_dxo(\n    comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = AppEventType.ANALYTIC_EVENT_TYPE\n):\n    \"\"\"Sends analytic dxo.\n\n    Args:\n        comp (FLComponent): An FLComponent.\n        dxo (DXO): analytic data in dxo.\n        fl_ctx (FLContext): fl context info.\n        event_type (str): Event type.\n    \"\"\"\n    if not isinstance(comp, FLComponent):\n        raise TypeError(f\"expect comp to be an instance of FLComponent, but got {type(comp)}\")\n    if not isinstance(dxo, DXO):\n        raise TypeError(f\"expect dxo to be an instance of DXO, but got {type(dxo)}\")\n    if not isinstance(fl_ctx, FLContext):\n        raise TypeError(f\"expect fl_ctx to be an instance of FLContext, but got {type(fl_ctx)}\")\n\n    fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=dxo.to_shareable(), private=True, sticky=False)\n    comp.fire_event(event_type=event_type, fl_ctx=fl_ctx)\n\n\ndef create_analytic_dxo(tag: str, value, data_type: AnalyticsDataType, **kwargs) -> DXO:\n    \"\"\"Creates the analytic DXO.\n\n    Args:\n        tag (str): the tag associated with this value.\n        value: the analytic data.\n        data_type (AnalyticsDataType): analytic data type.\n        kwargs: additional arguments to be passed into the receiver side's function.\n\n    Returns:\n        A DXO object that contains the analytic data.\n    \"\"\"\n    data = AnalyticsData(tag=tag, value=value, data_type=data_type, kwargs=kwargs)\n    dxo = data.to_dxo()\n    return dxo\n\n\nclass AnalyticsSender(Widget):\n    def __init__(self):\n        \"\"\"Sends analytics data.\n\n        This class implements some common methods follows signatures from PyTorch SummaryWriter.\n        It provides a convenient way for Learner to use.\n        \"\"\"\n        super().__init__()\n        self.engine = None\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.engine = fl_ctx.get_engine()\n\n    def _add(\n        self,\n        tag: str,\n        value,\n        data_type: AnalyticsDataType,\n        global_step: Optional[int] = None,\n        kwargs: Optional[dict] = None,\n    ):\n        kwargs = kwargs if kwargs else {}\n        if global_step:\n            if not isinstance(global_step, int):\n                raise TypeError(f\"Expect global step to be an instance of int, but got {type(global_step)}\")\n            kwargs[\"global_step\"] = global_step\n        dxo = create_analytic_dxo(tag=tag, value=value, data_type=data_type, **kwargs)\n        with self.engine.new_context() as fl_ctx:\n            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx)\n\n    def add_scalar(self, tag: str, scalar: float, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends a scalar.\n\n        Args:\n            tag (str): Data identifier.\n            scalar (float): Value to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=scalar, data_type=AnalyticsDataType.SCALAR, global_step=global_step, kwargs=kwargs)\n\n    def add_scalars(self, tag: str, scalars: dict, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends scalars.\n\n        Args:\n            tag (str): The parent name for the tags.\n            scalars (dict): Key-value pair storing the tag and corresponding values.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=scalars, data_type=AnalyticsDataType.SCALARS, global_step=global_step, kwargs=kwargs)\n\n    def add_text(self, tag: str, text: str, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends a text.\n\n        Args:\n            tag (str): Data identifier.\n            text (str): String to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=text, data_type=AnalyticsDataType.TEXT, global_step=global_step, kwargs=kwargs)\n\n    def add_image(self, tag: str, image, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends an image.\n\n        Args:\n            tag (str): Data identifier.\n            image: Image to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=image, data_type=AnalyticsDataType.IMAGE, global_step=global_step, kwargs=kwargs)\n\n    def flush(self):\n        \"\"\"Flushes out the message.\n\n        This is doing nothing, it is defined for mimic the PyTorch SummaryWriter behavior.\n        \"\"\"\n        pass\n\n    def close(self):\n        \"\"\"Close resources.\"\"\"\n        if self.engine:\n            self.engine = None\n\n\nclass AnalyticsReceiver(Widget, ABC):\n    def __init__(self, events: Optional[List[str]] = None):\n        \"\"\"Receives analytic data.\n\n        Args:\n            events (optional, List[str]): A list of event that this receiver will handle.\n        \"\"\"\n        super().__init__()\n        if events is None:\n            events = [AppEventType.ANALYTIC_EVENT_TYPE, f\"fed.{AppEventType.ANALYTIC_EVENT_TYPE}\"]\n        self.events = events\n        self._save_lock = Lock()\n        self._end = False\n\n    @abstractmethod\n    def initialize(self, fl_ctx: FLContext):\n        \"\"\"Initializes the receiver.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def save(self, fl_ctx: FLContext, shareable: Shareable, record_origin: str):\n        \"\"\"Saves the received data.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n            shareable (Shareable): the received message.\n            record_origin (str): the sender of this message / record.\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def finalize(self, fl_ctx: FLContext):\n        \"\"\"Finalizes the receiver.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n        \"\"\"\n        pass\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.initialize(fl_ctx)\n        elif event_type in self.events:\n            if self._end:\n                self.log_debug(fl_ctx, f\"Already received end run event, drop event {event_type}.\", fire_event=False)\n                return\n            data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)\n            if data is None:\n                self.log_error(fl_ctx, \"Missing event data.\", fire_event=False)\n                return\n            if not isinstance(data, Shareable):\n                self.log_error(\n                    fl_ctx, f\"Expect data to be an instance of Shareable but got {type(data)}\", fire_event=False\n                )\n                return\n\n            # if fed event use peer name to save\n            if fl_ctx.get_prop(FLContextKey.EVENT_SCOPE) == EventScope.FEDERATION:\n                record_origin = data.get_peer_prop(ReservedKey.IDENTITY_NAME, None)\n            else:\n                record_origin = fl_ctx.get_identity_name()\n\n            if record_origin is None:\n                self.log_error(fl_ctx, \"record_origin can't be None.\", fire_event=False)\n                return\n            with self._save_lock:\n                self.save(shareable=data, fl_ctx=fl_ctx, record_origin=record_origin)\n        elif event_type == EventType.END_RUN:\n            self._end = True\n            self.finalize(fl_ctx)\n", "path": "nvflare/app_common/widgets/streaming.py"}, {"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nclass AppEventType(object):\n    \"\"\"Defines application events.\"\"\"\n\n    START_ROUND = \"_start_round\"\n    END_ROUND = \"_end_round\"\n\n    BEFORE_AGGREGATION = \"_before_aggregation\"\n    END_AGGREGATION = \"_end_aggregation\"\n\n    SUBMIT_LOCAL_BEST_MODEL = \"_submit_local_best_model\"\n    SERVER_RECEIVE_BEST_MODEL = \"_server_receive_best_model\"\n    RECEIVE_VALIDATION_MODEL = \"_receive_validation_model\"\n    SEND_VALIDATION_RESULTS = \"_send_validation_results\"\n    RECEIVE_VALIDATION_RESULTS = \"_receive_validation_results\"\n\n    BEFORE_INITIALIZE = \"_before_initialize\"\n    AFTER_INITIALIZE = \"_after_initialize\"\n    BEFORE_TRAIN = \"_before_train\"\n    AFTER_TRAIN = \"_after_train\"\n\n    BEFORE_SHAREABLE_TO_LEARNABLE = \"_before_model_update\"\n    AFTER_SHAREABLE_TO_LEARNABLE = \"_after_model_update\"\n    BEFORE_LEARNABLE_PERSIST = \"_before_save_model\"\n    AFTER_LEARNABLE_PERSIST = \"_after_save_model\"\n    BEFORE_SEND_BEST_MODEL = \"_before_send_best_model\"\n    AFTER_SEND_BEST_MODEL = \"_after_send_best_model\"\n    LOCAL_BEST_MODEL_AVAILABLE = \"_local_best_model_available\"\n    GLOBAL_BEST_MODEL_AVAILABLE = \"_global_best_model_available\"\n    BEFORE_GET_VALIDATION_MODELS = \"_before_get_validation_models\"\n    AFTER_GET_VALIDATION_MODELS = \"_after_get_validation_models\"\n    SEND_MODEL_FOR_VALIDATION = \"_send_model_for_validation\"\n    BEFORE_VALIDATE_MODEL = \"_before_validate_model\"\n    AFTER_VALIDATE_MODEL = \"_after_validate_model\"\n    BEFORE_SUBMIT_VALIDATION_RESULTS = \"_before_submit_validation_results\"\n    AFTER_SUBMIT_VALIDATION_RESULTS = \"_after_submit_validation_results\"\n\n    # Events\n    ROUND_STARTED = \"_round_started\"\n    ROUND_DONE = \"_round_done\"\n    INITIAL_MODEL_LOADED = \"_initial_model_loaded\"\n    BEFORE_TRAIN_TASK = \"_before_train_task\"\n    RECEIVE_CONTRIBUTION = \"_receive_contribution\"\n    AFTER_CONTRIBUTION_ACCEPT = \"_after_contribution_accept\"\n    AFTER_AGGREGATION = \"_after_aggregation\"\n    BEFORE_CONTRIBUTION_ACCEPT = \"_before_contribution_accept\"\n    GLOBAL_WEIGHTS_UPDATED = \"_global_weights_updated\"\n    TRAINING_STARTED = \"_training_started\"\n    TRAINING_FINISHED = \"_training_finished\"\n    TRAIN_DONE = \"_train_done\"\n\n    CROSS_VAL_INIT = \"_cross_val_init\"\n    VALIDATION_RESULT_RECEIVED = \"_validation_result_received\"\n    RECEIVE_BEST_MODEL = \"_receive_best_model\"\n\n    ANALYTIC_EVENT_TYPE = \"analytix_log_stats\"\n    LOGGING_EVENT_TYPE = \"analytix_logging\"\n", "path": "nvflare/app_common/app_event_type.py"}], "after_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom abc import ABC, abstractmethod\nfrom threading import Lock\nfrom typing import List, Optional\n\nfrom nvflare.apis.analytix import AnalyticsData, AnalyticsDataType\nfrom nvflare.apis.dxo import DXO\nfrom nvflare.apis.event_type import EventType\nfrom nvflare.apis.fl_component import FLComponent\nfrom nvflare.apis.fl_constant import EventScope, FLContextKey, ReservedKey\nfrom nvflare.apis.fl_context import FLContext\nfrom nvflare.apis.shareable import Shareable\nfrom nvflare.widgets.widget import Widget\n\nANALYTIC_EVENT_TYPE = \"analytix_log_stats\"\n\n\ndef send_analytic_dxo(comp: FLComponent, dxo: DXO, fl_ctx: FLContext, event_type: str = ANALYTIC_EVENT_TYPE):\n    \"\"\"Sends analytic dxo.\n\n    Args:\n        comp (FLComponent): An FLComponent.\n        dxo (DXO): analytic data in dxo.\n        fl_ctx (FLContext): fl context info.\n        event_type (str): Event type.\n    \"\"\"\n    if not isinstance(comp, FLComponent):\n        raise TypeError(f\"expect comp to be an instance of FLComponent, but got {type(comp)}\")\n    if not isinstance(dxo, DXO):\n        raise TypeError(f\"expect dxo to be an instance of DXO, but got {type(dxo)}\")\n    if not isinstance(fl_ctx, FLContext):\n        raise TypeError(f\"expect fl_ctx to be an instance of FLContext, but got {type(fl_ctx)}\")\n\n    fl_ctx.set_prop(key=FLContextKey.EVENT_DATA, value=dxo.to_shareable(), private=True, sticky=False)\n    comp.fire_event(event_type=event_type, fl_ctx=fl_ctx)\n\n\ndef create_analytic_dxo(tag: str, value, data_type: AnalyticsDataType, **kwargs) -> DXO:\n    \"\"\"Creates the analytic DXO.\n\n    Args:\n        tag (str): the tag associated with this value.\n        value: the analytic data.\n        data_type (AnalyticsDataType): analytic data type.\n        kwargs: additional arguments to be passed into the receiver side's function.\n\n    Returns:\n        A DXO object that contains the analytic data.\n    \"\"\"\n    data = AnalyticsData(tag=tag, value=value, data_type=data_type, kwargs=kwargs)\n    dxo = data.to_dxo()\n    return dxo\n\n\nclass AnalyticsSender(Widget):\n    def __init__(self, event_type=ANALYTIC_EVENT_TYPE):\n        \"\"\"Sends analytics data.\n\n        Note::\n            This class implements some common methods follows signatures from PyTorch SummaryWriter.\n            It provides a convenient way for Learner to use.\n\n        Args:\n            event_type (str): event type to fire.\n        \"\"\"\n        super().__init__()\n        self.engine = None\n        self.event_type = event_type\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.engine = fl_ctx.get_engine()\n\n    def _add(\n        self,\n        tag: str,\n        value,\n        data_type: AnalyticsDataType,\n        global_step: Optional[int] = None,\n        kwargs: Optional[dict] = None,\n    ):\n        kwargs = kwargs if kwargs else {}\n        if global_step:\n            if not isinstance(global_step, int):\n                raise TypeError(f\"Expect global step to be an instance of int, but got {type(global_step)}\")\n            kwargs[\"global_step\"] = global_step\n        dxo = create_analytic_dxo(tag=tag, value=value, data_type=data_type, **kwargs)\n        with self.engine.new_context() as fl_ctx:\n            send_analytic_dxo(self, dxo=dxo, fl_ctx=fl_ctx, event_type=self.event_type)\n\n    def add_scalar(self, tag: str, scalar: float, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends a scalar.\n\n        Args:\n            tag (str): Data identifier.\n            scalar (float): Value to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=scalar, data_type=AnalyticsDataType.SCALAR, global_step=global_step, kwargs=kwargs)\n\n    def add_scalars(self, tag: str, scalars: dict, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends scalars.\n\n        Args:\n            tag (str): The parent name for the tags.\n            scalars (dict): Key-value pair storing the tag and corresponding values.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=scalars, data_type=AnalyticsDataType.SCALARS, global_step=global_step, kwargs=kwargs)\n\n    def add_text(self, tag: str, text: str, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends a text.\n\n        Args:\n            tag (str): Data identifier.\n            text (str): String to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=text, data_type=AnalyticsDataType.TEXT, global_step=global_step, kwargs=kwargs)\n\n    def add_image(self, tag: str, image, global_step: Optional[int] = None, **kwargs):\n        \"\"\"Sends an image.\n\n        Args:\n            tag (str): Data identifier.\n            image: Image to send.\n            global_step (optional, int): Global step value.\n            **kwargs: Additional arguments to pass to the receiver side.\n        \"\"\"\n        self._add(tag=tag, value=image, data_type=AnalyticsDataType.IMAGE, global_step=global_step, kwargs=kwargs)\n\n    def flush(self):\n        \"\"\"Flushes out the message.\n\n        This is doing nothing, it is defined for mimic the PyTorch SummaryWriter behavior.\n        \"\"\"\n        pass\n\n    def close(self):\n        \"\"\"Close resources.\"\"\"\n        if self.engine:\n            self.engine = None\n\n\nclass AnalyticsReceiver(Widget, ABC):\n    def __init__(self, events: Optional[List[str]] = None):\n        \"\"\"Receives analytic data.\n\n        Args:\n            events (optional, List[str]): A list of event that this receiver will handle.\n        \"\"\"\n        super().__init__()\n        if events is None:\n            events = [ANALYTIC_EVENT_TYPE, f\"fed.{ANALYTIC_EVENT_TYPE}\"]\n        self.events = events\n        self._save_lock = Lock()\n        self._end = False\n\n    @abstractmethod\n    def initialize(self, fl_ctx: FLContext):\n        \"\"\"Initializes the receiver.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def save(self, fl_ctx: FLContext, shareable: Shareable, record_origin: str):\n        \"\"\"Saves the received data.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n            shareable (Shareable): the received message.\n            record_origin (str): the sender of this message / record.\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def finalize(self, fl_ctx: FLContext):\n        \"\"\"Finalizes the receiver.\n\n        Args:\n            fl_ctx (FLContext): fl context.\n        \"\"\"\n        pass\n\n    def handle_event(self, event_type: str, fl_ctx: FLContext):\n        if event_type == EventType.START_RUN:\n            self.initialize(fl_ctx)\n        elif event_type in self.events:\n            if self._end:\n                self.log_debug(fl_ctx, f\"Already received end run event, drop event {event_type}.\", fire_event=False)\n                return\n            data = fl_ctx.get_prop(FLContextKey.EVENT_DATA, None)\n            if data is None:\n                self.log_error(fl_ctx, \"Missing event data.\", fire_event=False)\n                return\n            if not isinstance(data, Shareable):\n                self.log_error(\n                    fl_ctx, f\"Expect data to be an instance of Shareable but got {type(data)}\", fire_event=False\n                )\n                return\n\n            # if fed event use peer name to save\n            if fl_ctx.get_prop(FLContextKey.EVENT_SCOPE) == EventScope.FEDERATION:\n                record_origin = data.get_peer_prop(ReservedKey.IDENTITY_NAME, None)\n            else:\n                record_origin = fl_ctx.get_identity_name()\n\n            if record_origin is None:\n                self.log_error(fl_ctx, \"record_origin can't be None.\", fire_event=False)\n                return\n            with self._save_lock:\n                self.save(shareable=data, fl_ctx=fl_ctx, record_origin=record_origin)\n        elif event_type == EventType.END_RUN:\n            self._end = True\n            self.finalize(fl_ctx)\n", "path": "nvflare/app_common/widgets/streaming.py"}, {"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nclass AppEventType(object):\n    \"\"\"Defines application events.\"\"\"\n\n    START_ROUND = \"_start_round\"\n    END_ROUND = \"_end_round\"\n\n    BEFORE_AGGREGATION = \"_before_aggregation\"\n    END_AGGREGATION = \"_end_aggregation\"\n\n    SUBMIT_LOCAL_BEST_MODEL = \"_submit_local_best_model\"\n    SERVER_RECEIVE_BEST_MODEL = \"_server_receive_best_model\"\n    RECEIVE_VALIDATION_MODEL = \"_receive_validation_model\"\n    SEND_VALIDATION_RESULTS = \"_send_validation_results\"\n    RECEIVE_VALIDATION_RESULTS = \"_receive_validation_results\"\n\n    BEFORE_INITIALIZE = \"_before_initialize\"\n    AFTER_INITIALIZE = \"_after_initialize\"\n    BEFORE_TRAIN = \"_before_train\"\n    AFTER_TRAIN = \"_after_train\"\n\n    BEFORE_SHAREABLE_TO_LEARNABLE = \"_before_model_update\"\n    AFTER_SHAREABLE_TO_LEARNABLE = \"_after_model_update\"\n    BEFORE_LEARNABLE_PERSIST = \"_before_save_model\"\n    AFTER_LEARNABLE_PERSIST = \"_after_save_model\"\n    BEFORE_SEND_BEST_MODEL = \"_before_send_best_model\"\n    AFTER_SEND_BEST_MODEL = \"_after_send_best_model\"\n    LOCAL_BEST_MODEL_AVAILABLE = \"_local_best_model_available\"\n    GLOBAL_BEST_MODEL_AVAILABLE = \"_global_best_model_available\"\n    BEFORE_GET_VALIDATION_MODELS = \"_before_get_validation_models\"\n    AFTER_GET_VALIDATION_MODELS = \"_after_get_validation_models\"\n    SEND_MODEL_FOR_VALIDATION = \"_send_model_for_validation\"\n    BEFORE_VALIDATE_MODEL = \"_before_validate_model\"\n    AFTER_VALIDATE_MODEL = \"_after_validate_model\"\n    BEFORE_SUBMIT_VALIDATION_RESULTS = \"_before_submit_validation_results\"\n    AFTER_SUBMIT_VALIDATION_RESULTS = \"_after_submit_validation_results\"\n\n    # Events\n    ROUND_STARTED = \"_round_started\"\n    ROUND_DONE = \"_round_done\"\n    INITIAL_MODEL_LOADED = \"_initial_model_loaded\"\n    BEFORE_TRAIN_TASK = \"_before_train_task\"\n    RECEIVE_CONTRIBUTION = \"_receive_contribution\"\n    AFTER_CONTRIBUTION_ACCEPT = \"_after_contribution_accept\"\n    AFTER_AGGREGATION = \"_after_aggregation\"\n    BEFORE_CONTRIBUTION_ACCEPT = \"_before_contribution_accept\"\n    GLOBAL_WEIGHTS_UPDATED = \"_global_weights_updated\"\n    TRAINING_STARTED = \"_training_started\"\n    TRAINING_FINISHED = \"_training_finished\"\n    TRAIN_DONE = \"_train_done\"\n\n    CROSS_VAL_INIT = \"_cross_val_init\"\n    VALIDATION_RESULT_RECEIVED = \"_validation_result_received\"\n    RECEIVE_BEST_MODEL = \"_receive_best_model\"\n", "path": "nvflare/app_common/app_event_type.py"}]}