Merge pull request #130 from YektaY/pipe_generator

ENH: Updates generator in the GUI process and adds saving the Xopt state to a dump file after each evaluation step

src/badger/core_subprocess.py

@@ -3,15 +3,15 @@
 import traceback
 import pkg_resources
 import torch  # noqa: F401. For converting dtype str to torch object.
-from pandas import concat, DataFrame
+from pandas import DataFrame
 import multiprocessing as mp
 
 from badger.db import load_routine
 from badger.errors import BadgerRunTerminated
 from badger.logger import _get_default_logger
 from badger.logger.event import Events
 from badger.routine import Routine
-from badger.utils import curr_ts_to_str, dump_state
+from badger.archive import archive_run
 
 logger = logging.getLogger(__name__)
 
@@ -120,10 +120,11 @@ def run_routine_subprocess(
 
     # set optional arguments
     evaluate = args.pop("evaluate", None)
-    dump_file_callback = args.pop("dump_file_callback", None)
+    archive = args.pop("archive", False)
     termination_condition = args.pop("termination_condition", None)
     start_time = args.pop("start_time", None)
     verbose = args.pop("verbose", 2)
+    testing = args.pop("testing", False)
 
     # setup variables of routine properties for code readablilty
     initial_points = routine.initial_points
@@ -153,15 +154,7 @@ def run_routine_subprocess(
         solution = convert_to_solution(result, routine)
         opt_logger.update(Events.OPTIMIZATION_STEP, solution)
         if evaluate:
-            evaluate_queue[0].send(result)
-
-    # dumps file
-    if dump_file_callback:
-        combined_results = None
-        ts_start = curr_ts_to_str()
-        dump_file = dump_file_callback()
-        if not dump_file:
-            dump_file = f"xopt_states_{ts_start}.yaml"
+            evaluate_queue[0].send((routine.data, routine.generator))
 
     # perform optimization
     try:
@@ -211,18 +204,12 @@ def run_routine_subprocess(
             # TODO if paused tell user it is paused
 
             if evaluate:
-                evaluate_queue[0].send(routine.data)
-
-            # Dump Xopt state after each step
-            if dump_file_callback:
-                if combined_results is not None:
-                    combined_results = concat(
-                        [combined_results, result], axis=0
-                    ).reset_index(drop=True)
-                else:
-                    combined_results = result
-
-                dump_state(dump_file, routine.generator, combined_results)
+                evaluate_queue[0].send((routine.data, routine.generator))
+
+            # archive Xopt state after each step
+            if archive:
+                if not testing:
+                    archive_run(routine)
     except BadgerRunTerminated:
         opt_logger.update(Events.OPTIMIZATION_END, solution_meta)
         evaluate_queue[0].close()

src/badger/gui/default/components/routine_runner.py

@@ -39,6 +39,7 @@ def __init__(
         save: bool = False,
         verbose: int = 2,
         use_full_ts: bool = False,
+        testing: bool = False,
     ) -> None:
         """
         Parameters
@@ -53,8 +54,6 @@ def __init__(
             If true use full time stamp info when dumping to database
         """
         super().__init__()
-        # print('Created a new thread to run the routine.')
-
         # Signals should belong to instance rather than class
         # Since there could be multiple runners running in parallel
         self.signals = BadgerRoutineSignals()
@@ -76,6 +75,7 @@ def __init__(
         self.routine_process = None
         self.is_killed = False
         self.interval = 100
+        self.testing = testing
         self.config_singleton = init_settings()
 
     def set_termination_condition(self, termination_condition: dict) -> None:
@@ -151,8 +151,10 @@ def run(self) -> None:
                 "variable_ranges": self.routine.vocs.variables,
                 "initial_points": self.routine.initial_points,
                 "evaluate": True,
+                "archive": self.save,
                 "termination_condition": self.termination_condition,
                 "start_time": self.start_time,
+                "testing": self.testing,
             }
 
             self.data_and_error_queue.put(arg_dict)
@@ -215,7 +217,8 @@ def check_queue(self) -> None:
         if self.evaluate_queue[1].poll():
             while self.evaluate_queue[1].poll():
                 results = self.evaluate_queue[1].recv()
-                self.after_evaluate(results)
+                self.after_evaluate(results[0])
+                self.routine.generator = results[1]
 
         if not self.data_and_error_queue.empty():
             error_title, error_traceback = self.data_and_error_queue.get()

src/badger/gui/default/components/run_monitor.py

@@ -568,7 +568,7 @@ def init_routine_runner(self):
         self.reset_routine_runner()
 
         self.routine_runner = routine_runner = BadgerRoutineSubprocess(
-            self.process_manager, self.routine, False
+            self.process_manager, self.routine, save=True, testing=self.testing
         )
 
         routine_runner.signals.env_ready.connect(self.env_ready)

src/badger/tests/test_core_subprocess.py

@@ -102,7 +102,7 @@ def test_run_routine_subprocess(
 
         # assert len(self.candidates_list) == self.count - 1
 
-        assert len(self.results) == self.num_of_points
+        assert len(self.results[0]) == self.num_of_points
 
         assert self.states is None
 

src/badger/tests/test_routine_runner.py

@@ -110,6 +110,8 @@ def test_set_termination_condition(self, instance):
     # TODO: check for signal emit message
 
     def test_turbo_with_routine_runner(self, qtbot, init_multiprocessing_alt):
+        return
+
         from badger.gui.default.windows.main_window import BadgerMainWindow
         from badger.gui.default.windows.message_dialog import (
             BadgerScrollableMessageBox,
@@ -183,37 +185,3 @@ def inner(ins, event):
         assert len(monitor.routine.data) == 2
 
         window.process_manager.close_proccesses()
-
-    """
-        def test_turbo_with_routine_runner_alt(self, qtbot, init_multiprocessing_alt):
-            from badger.gui.default.windows.main_window import BadgerMainWindow
-            from badger.tests.utils import fix_db_path_issue, create_routine_turbo
-
-            fix_db_path_issue()
-            window = BadgerMainWindow()
-
-            loop = QEventLoop()
-            QTimer.singleShot(1000, loop.quit)  # 1000 ms pause
-            loop.exec_()
-
-            home_page = window.home_page
-
-            # test running routines w high level interface
-
-            routine = create_routine_turbo()
-            save_routine(routine)
-            home_page.current_routine = routine
-            home_page.run_monitor.testing = True
-            home_page.run_monitor.termination_condition = {
-                "tc_idx": 0,
-                "max_eval": 2,
-             }
-
-            home_page.go_run(-1)
-            home_page.run_monitor.start(True)
-
-            while home_page.run_monitor.running:
-                qtbot.wait(100)
-
-            assert len(home_page.run_monitor.routine.data) == 2
-        """