Release v0.6.5: add SDK 0.6.0 (#19)

* **New Features**
  * Added configurable log streaming level for better control over logging output.
  * Introduced optional output file support for saving results to specified file paths.
  * Enhanced support for handling and saving multiple results to individual files with automatic path generation.

* **Bug Fixes**
  * Improved logging consistency across result submission and node lifecycle events.

* **Chores**
  * Updated package version and dependencies.

---------

Co-authored-by: davidhieber <david.hieber@uni-tuebingen.de>
Co-authored-by: Nightknight3000 <alexander.roehl@uni-tuebingen.de>

Author: Nightknight3000

flame/star/star_localdp/star_localdp_model.py

@@ -1,6 +1,7 @@
 from typing import Optional, Type, Literal, Union, Any
 
 from flamesdk import FlameCoreSDK
+from flamesdk.resources.utils.constants import LogTypeLiteral
 from flame.star.aggregator_client import Aggregator
 from flame.star.analyzer_client import Analyzer
 from flame.star.star_model import StarModel, _ERROR_MESSAGES
@@ -24,6 +25,8 @@ def __init__(self,
                  simple_analysis: bool = True,
                  output_type: Union[Literal['str', 'bytes', 'pickle'], list] = 'str',
                  multiple_results: bool = False,
+                 filename: Optional[Union[str, list[str]]] = None,
+                 stream_log_level: int = 20,
                  analyzer_kwargs: Optional[dict] = None,
                  aggregator_kwargs: Optional[dict] = None,
                  epsilon: Optional[float] = None,
@@ -39,6 +42,8 @@ def __init__(self,
                          simple_analysis=simple_analysis,
                          output_type=output_type,
                          multiple_results=multiple_results,
+                         filename=filename,
+                         stream_log_level=stream_log_level,
                          analyzer_kwargs=analyzer_kwargs,
                          aggregator_kwargs=aggregator_kwargs,
                          test_mode=test_mode,
@@ -49,6 +54,7 @@ def _start_aggregator(self,
                           simple_analysis: bool = True,
                           output_type: Union[Literal['str', 'bytes', 'pickle'], list] = 'str',
                           multiple_results: bool = False,
+                          filename: Optional[Union[str, list[str]]] = None,
                           aggregator_kwargs: Optional[dict] = None) -> None:
         if issubclass(aggregator, Aggregator):
             # init custom aggregator subclass
@@ -69,24 +75,27 @@ def _start_aggregator(self,
 
                 # Aggregate results
                 agg_res, converged = aggregator.aggregate(list(result_dict.values()), simple_analysis)
-                self.flame.flame_log(f"Aggregated results: {str(agg_res)[:100]}")
 
                 if converged:
                     if not self.test_mode:
                         self.flame.flame_log("Submitting final results using differential privacy...",
-                                             log_type='info',
-                                             end='')
+                                             log_type=LogTypeLiteral.INFO.value,
+                                             halt_submission=True)
                     if aggregator.delta_criteria and (self.epsilon is not None) and (self.sensitivity is not None):
                         local_dp = {"epsilon": self.epsilon, "sensitivity": self.sensitivity}
                     else:
                         local_dp = None
                     if self.test_mode and (local_dp is not None):
                         self.flame.flame_log(f"\tTest mode: Would apply local DP with epsilon={local_dp['epsilon']} "
                                              f"and sensitivity={local_dp['sensitivity']}",
-                                             log_type='info')
-                    response = self.flame.submit_final_result(agg_res, output_type, multiple_results, local_dp=local_dp)
+                                             log_type=LogTypeLiteral.INFO.value)
+                    response = self.flame.submit_final_result(agg_res,
+                                                              output_type,
+                                                              multiple_results,
+                                                              local_dp=local_dp,
+                                                              filename=filename)
                     if not self.test_mode:
-                        self.flame.flame_log(f"success (response={response})", log_type='info')
+                        self.flame.flame_log(f"success (response={response})", log_type=LogTypeLiteral.INFO.value)
                     self.flame.analysis_finished()
                     aggregator.node_finished()  # LOOP BREAK
                 else:

flame/star/star_model.py

@@ -2,6 +2,7 @@
 from typing import Optional, Type, Literal, Union, Any
 
 from flamesdk import FlameCoreSDK
+from flamesdk.resources.utils.constants import LogTypeLiteral
 from flame.star.aggregator_client import Aggregator
 from flame.star.analyzer_client import Analyzer
 from flame.utils.mock_flame_core import MockFlameCoreSDK
@@ -28,6 +29,8 @@ def __init__(self,
                  simple_analysis: bool = True,
                  output_type: Union[Literal['str', 'bytes', 'pickle'], list] = 'str',
                  multiple_results: bool = False,
+                 filename: Optional[Union[str, list[str]]] = None,
+                 stream_log_level: int = 20,
                  analyzer_kwargs: Optional[dict] = None,
                  aggregator_kwargs: Optional[dict] = None,
                  test_mode: bool = False,
@@ -38,27 +41,28 @@ def __init__(self,
             self.flame = MockFlameCoreSDK(test_kwargs=test_kwargs)
         else:
             self.test_kwargs = None
-            self.flame = FlameCoreSDK()
+            self.flame = FlameCoreSDK(stream_log_level=stream_log_level)
 
         if self._is_analyzer():
             self.flame.flame_log(f"Analyzer {test_kwargs['node_id'] + ' ' if self.test_mode else ''}started",
-                                 log_type='info')
+                                 log_type=LogTypeLiteral.INFO.value)
             self._start_analyzer(analyzer,
                                  data_type=data_type,
                                  query=query,
                                  simple_analysis=simple_analysis,
                                  analyzer_kwargs=analyzer_kwargs)
         elif self._is_aggregator():
-            self.flame.flame_log("Aggregator started", log_type='info')
+            self.flame.flame_log("Aggregator started", log_type=LogTypeLiteral.INFO.value)
             self._start_aggregator(aggregator,
                                    simple_analysis=simple_analysis,
                                    output_type=output_type,
                                    multiple_results=multiple_results,
+                                   filename=filename,
                                    aggregator_kwargs=aggregator_kwargs)
         else:
             raise BrokenPipeError("Has to be either analyzer or aggregator")
         if not self.test_mode:
-            self.flame.flame_log("Analysis finished!", log_type='info')
+            self.flame.flame_log("Analysis finished!", log_type=LogTypeLiteral.INFO.value)
             while True:
                 pass  # keep the node alive to allow for orderly shutdown
 
@@ -73,6 +77,7 @@ def _start_aggregator(self,
                           simple_analysis: bool = True,
                           output_type: Union[Literal['str', 'bytes', 'pickle'], list] = 'str',
                           multiple_results: bool = False,
+                          filename: Optional[Union[str, list[str]]] = None,
                           aggregator_kwargs: Optional[dict] = None) -> None:
         if issubclass(aggregator, Aggregator):
             # init custom aggregator subclass
@@ -89,21 +94,26 @@ def _start_aggregator(self,
 
             while not aggregator.finished:  # (**)
                 # Await intermediate results
+                self.flame.flame_log(f"Awaiting intermediate results...", log_type=LogTypeLiteral.INFO.value)
                 result_dict = self.flame.await_intermediate_data(analyzers)
 
                 # Aggregate results
                 agg_res, converged = aggregator.aggregate(list(result_dict.values()), simple_analysis)
 
                 if converged:
                     if not self.test_mode:
-                        self.flame.flame_log("Submitting final results...", log_type='info', end='')
-                    response = self.flame.submit_final_result(agg_res, output_type, multiple_results)
+                        self.flame.flame_log("Submitting final results...",
+                                             log_type=LogTypeLiteral.INFO.value,
+                                             halt_submission=True)
+                    response = self.flame.submit_final_result(agg_res, output_type, multiple_results,
+                                                              filename=filename)
                     if not self.test_mode:
-                        self.flame.flame_log(f"success (response={response})", log_type='info')
+                        self.flame.flame_log(f"success (response={response})", log_type=LogTypeLiteral.INFO.value)
                     self.flame.analysis_finished()
                     aggregator.node_finished()      # LOOP BREAK
                 else:
                     # Send aggregated result to analyzers
+                    self.flame.flame_log(f"Sending aggregated results...", log_type=LogTypeLiteral.INFO.value)
                     self.flame.send_intermediate_data(analyzers, agg_res)
         else:
             raise BrokenPipeError(_ERROR_MESSAGES.IS_INCORRECT_CLASS.value)
@@ -128,7 +138,6 @@ def _start_analyzer(self,
 
             # Get data
             self._get_data(query=query, data_type=data_type)
-            self.flame.flame_log(f"\tData extracted: {str(self.data)[:100]}", log_type='info')
 
             # Check converged status on Hub
             while not analyzer.finished:  # (**)
@@ -151,23 +160,27 @@ def _wait_until_partners_ready(self) -> None:
         if self._is_analyzer():
             aggregator_id = self.flame.get_aggregator_id()
             if not self.test_mode:
-                self.flame.flame_log("Awaiting contact with aggregator node...", log_type='info')
+                self.flame.flame_log("Awaiting contact with aggregator node...",
+                                     log_type=LogTypeLiteral.INFO.value)
             ready_check_dict = self.flame.ready_check([aggregator_id])
 
             if not ready_check_dict[aggregator_id]:
                 raise BrokenPipeError("Could not contact aggregator")
 
             if not self.test_mode:
-                self.flame.flame_log("Awaiting contact with aggregator node...success", log_type='info')
+                self.flame.flame_log("Awaiting contact with aggregator node...success",
+                                     log_type=LogTypeLiteral.INFO.value)
         else:
             analyzer_ids = self.flame.get_participant_ids()
             if not self.test_mode:
-                self.flame.flame_log("Awaiting contact with analyzer nodes...", log_type='info')
+                self.flame.flame_log("Awaiting contact with analyzer nodes...",
+                                     log_type=LogTypeLiteral.INFO.value)
             ready_check_dict = self.flame.ready_check(analyzer_ids)
             if not all(ready_check_dict.values()):
                 raise BrokenPipeError("Could not contact all analyzers")
             if not self.test_mode:
-                self.flame.flame_log("Awaiting contact with analyzer nodes...success", log_type='info')
+                self.flame.flame_log("Awaiting contact with analyzer nodes...success",
+                                     log_type=LogTypeLiteral.INFO.value)
 
     def _get_data(self,
                   data_type: Literal['fhir', 's3'],

flame/star/star_model_tester.py

@@ -19,11 +19,12 @@ def __init__(self,
                  simple_analysis: bool = True,
                  output_type: Union[Literal['str', 'bytes', 'pickle'], list] = 'str',
                  multiple_results: bool = False,
+                 filename: Optional[Union[str, list[str]]] = None,
+                 stream_log_level: int = 20,
                  analyzer_kwargs: Optional[dict] = None,
                  aggregator_kwargs: Optional[dict] = None,
                  epsilon: Optional[float] = None,
-                 sensitivity: Optional[float] = None,
-                 result_filepath: Optional[Union[str, list[str]]] = None) -> None:
+                 sensitivity: Optional[float] = None) -> None:
         num_splits = len(data_splits)
         self.test_input(data_splits[0])
         participants = []
@@ -52,6 +53,7 @@ def __init__(self,
                 'simple_analysis': simple_analysis,
                 'output_type': output_type,
                 'multiple_results': multiple_results,
+                'stream_log_level': stream_log_level,
                 'analyzer_kwargs': analyzer_kwargs,
                 'aggregator_kwargs': aggregator_kwargs,
                 'test_mode': True,
@@ -75,7 +77,8 @@ def run_node(kwargs=test_kwargs, use_dp=use_local_dp):
                         flame = StarModel(**kwargs).flame
                     else:
                         flame = StarLocalDPModel(**kwargs).flame
-                    results_queue.append(flame.final_results_storage)
+                    if kwargs['test_kwargs']['role'] == 'aggregator':
+                        results_queue.append(flame.final_results_storage)
                 except Exception:
                     stop_event = MockFlameCoreSDK.stop_event
                     if not stop_event:
@@ -103,7 +106,7 @@ def run_node(kwargs=test_kwargs, use_dp=use_local_dp):
 
         # write final results
         if results_queue:
-            self.write_result(results_queue[0], output_type, result_filepath, multiple_results)
+            self.write_result(results_queue[0], output_type, filename, multiple_results)
         else:
             print("No results to write. All threads failed with errors:")
             for (role, node_id), error in thread_errors.items():
@@ -147,12 +150,12 @@ def test_input(data: Any) -> None:
     @staticmethod
     def write_result(result: Any,
                      output_type: Union[Literal['str', 'bytes', 'pickle'], list],
-                     result_filepath: Optional[Union[str, list[str]]] = None,
+                     filename: Optional[Union[str, list[str]]] = None,
                      multiple_results: bool = False) -> None:
         if multiple_results:
             if isinstance(result, list) or isinstance(result, tuple):
-                if isinstance(result_filepath, list) and (len(result_filepath) != len(result)):
-                    print(f"Warning! Inconsistent number of result_filepaths (len={result_filepath}) "
+                if isinstance(filename, list) and (len(filename) != len(result)):
+                    print(f"Warning! Inconsistent number of filenames (len={filename}) "
                           f"and results (len={len(result)}) -> multiple_results will be ignored.")
                     multi_iterable_results = False
                 else:
@@ -164,20 +167,21 @@ def write_result(result: Any,
         else:
             multi_iterable_results = False
 
-        if result_filepath is not None:
+        if filename is not None:
             if not multi_iterable_results:
                 result = [result]
-                result_filepath = [result_filepath]
+                if not isinstance(filename, list):
+                    filename = [filename]
 
             for i, res in enumerate(result):
-                if isinstance(result_filepath, list):
-                    current_path = result_filepath[i]
+                if isinstance(filename, list):
+                    current_path = filename[i]
                 else:
-                    if '.' in result_filepath:
-                        result_filename, result_extension = result_filepath.rsplit('.', 1)
+                    if '.' in filename:
+                        result_filename, result_extension = filename.rsplit('.', 1)
                         current_path = f"{result_filename}_{i + 1}.{result_extension}"
                     else:
-                        current_path = f"{result_filepath}_{i + 1}"
+                        current_path = f"{filename}_{i + 1}"
                 if isinstance(output_type, list) and (len(output_type) == len(result)):
                     out_type = output_type[i]
                 else: