temp commit

Melodie/calibrator.py

@@ -1,5 +1,6 @@
 import copy
 import logging
+import os
 import time
 from typing import (
     Dict,
@@ -13,7 +14,9 @@
     Type,
     Iterator,
     cast,
+    Literal
 )
+import pandas as pd
 
 from MelodieInfra import create_db_conn, Config, MelodieExceptions
 from MelodieInfra.parallel.parallel_manager import ParallelManager
@@ -82,7 +85,8 @@ class CalibratorAlgorithmMeta:
     """
 
     def __init__(self):
-        self._freeze = False  # If new attributes can be added by attribute assignment.
+        # If new attributes can be added by attribute assignment.
+        self._freeze = False
         self.id_calibrator_scenario = 0
         self.id_calibrator_params_scenario = 1
         self.id_path = 0
@@ -192,7 +196,8 @@ def get_params(self, id_chromosome: int) -> Dict[str, Any]:
         :param id_chromosome:
         :return:
         """
-        chromosome_value = self.algorithm.chrom2x(self.algorithm.Chrom)[id_chromosome]
+        chromosome_value = self.algorithm.chrom2x(
+            self.algorithm.Chrom)[id_chromosome]
         env_parameters_dict = {}
         for i, param_name in enumerate(self.env_param_names):
             env_parameters_dict[param_name] = chromosome_value[i]
@@ -209,7 +214,8 @@ def target_function_to_cache(
 
         :return:
         """
-        self.cache[(generation, id_chromosome)] = env_data["target_function_value"]
+        self.cache[(generation, id_chromosome)
+                   ] = env_data["target_function_value"]
 
     def generate_target_function(self) -> Callable[[], float]:
         """
@@ -220,7 +226,8 @@ def generate_target_function(self) -> Callable[[], float]:
 
         def f(*args):
             self._chromosome_counter += 1
-            value = self.cache[(self._current_generation, self._chromosome_counter)]
+            value = self.cache[(self._current_generation,
+                                self._chromosome_counter)]
             return value
 
         return f
@@ -239,7 +246,6 @@ def record_agent_properties(
         :param meta:
         :return:
         """
-        import pandas as pd
 
         agent_records = {}
         environment_record = {}
@@ -254,20 +260,14 @@ def record_agent_properties(
                 d.update(agent_container_data)
 
                 agent_records[container_name].append(d)
-            create_db_conn(self.manager.config).write_dataframe(
-                f"{container_name}_calibrator_result",
-                pd.DataFrame(agent_records[container_name]),
-                if_exists="append",
-            )
+            self.manager._write_to_table("csv", f"{container_name}_calibrator_result", pd.DataFrame(
+                agent_records[container_name]))
         environment_record.update(meta_dict)
         environment_record.update(env_data)
         environment_record.pop("target_function_value")
 
-        create_db_conn(self.manager.config).write_dataframe(
-            "environment_calibrator_result",
-            pd.DataFrame([environment_record]),
-            if_exists="append",
-        )
+        self.manager._write_to_table(
+            "csv", "environment_calibrator_result", pd.DataFrame([environment_record]))
         return agent_records, environment_record
 
     def calc_cov_df(
@@ -284,7 +284,6 @@ def calc_cov_df(
         :param meta:
         :return:
         """
-        import pandas as pd
 
         pd.set_option("max_colwidth", 500)
         pd.set_option("display.max_columns", None)
@@ -309,24 +308,21 @@ def calc_cov_df(
                         {prop_name + "_mean": mean, prop_name + "_cov": cov}
                     )
                 container_agent_record_list.append(cov_records)
-            create_db_conn(self.manager.config).write_dataframe(
-                f"{container_name}_calibrator_result_cov",
-                pd.DataFrame(container_agent_record_list),
-                if_exists="append",
-            )
+
+            self.manager._write_to_table("csv", f"{container_name}_calibrator_result_cov",
+                                         pd.DataFrame(container_agent_record_list))
         env_record = {}
         env_record.update(meta_dict)
         for prop_name in (
             self.env_param_names + self.recorded_env_properties + ["distance"]
         ):
             mean = env_df[prop_name].mean()
             cov = env_df[prop_name].std() / env_df[prop_name].mean()
-            env_record.update({prop_name + "_mean": mean, prop_name + "_cov": cov})
-        create_db_conn(self.manager.config).write_dataframe(
-            "environment_calibrator_result_cov",
-            pd.DataFrame([env_record]),
-            if_exists="append",
-        )
+            env_record.update(
+                {prop_name + "_mean": mean, prop_name + "_cov": cov})
+
+        self.manager._write_to_table("csv", "environment_calibrator_result_cov",
+                                     pd.DataFrame([env_record]))
 
     def pre_check(self, meta):
         """
@@ -342,7 +338,6 @@ def pre_check(self, meta):
         )
 
     def run(self, scenario: Scenario, meta: Union[GACalibratorAlgorithmMeta]):
-        import pandas as pd
 
         self.pre_check(meta)
 
@@ -383,7 +378,8 @@ def run(self, scenario: Scenario, meta: Union[GACalibratorAlgorithmMeta]):
                 self.target_function_to_cache(env_data, i, chrom)
 
             self.calc_cov_df(
-                {k: pd.DataFrame(v) for k, v in agent_records_collector.items()},
+                {k: pd.DataFrame(v)
+                 for k, v in agent_records_collector.items()},
                 pd.DataFrame(env_records_list),
                 meta,
             )
@@ -466,9 +462,9 @@ def get_params_scenarios(self) -> List:
 
         :return: A list of dict, and each dict contains parameters.
         """
-        import pandas as pd
 
-        calibrator_scenarios_table = self.get_dataframe("calibrator_params_scenarios")
+        calibrator_scenarios_table = self.get_dataframe(
+            "calibrator_params_scenarios")
         assert isinstance(
             calibrator_scenarios_table, pd.DataFrame
         ), "No learning scenarios table specified!"

Melodie/scenario_manager.py

@@ -67,6 +67,7 @@ def _setup(self, data: dict = None):
         if data is not None:
             for col_name in data.keys():
                 setattr(self, col_name, data[col_name])
+        # self.load()
         self.load_data()
         self.setup_data()
 

Melodie/simulator.py

@@ -9,7 +9,9 @@
 import threading
 import time
 from multiprocessing import Pool
-from typing import Optional, List, Tuple, Type
+from typing import Literal, Optional, List, Tuple, Type
+
+import pandas as pd
 
 from MelodieInfra import (
     create_db_conn,
@@ -136,6 +138,26 @@ def generate_scenarios(self) -> List[Scenario]:
         """
         pass
 
+    def _write_to_table(self, kind: Literal["csv", "sql"], table_name: str, data: pd.DataFrame):
+        """
+        Write a pandas dataframe to a table in output directory or database
+        """
+        if kind == "sql":
+            create_db_conn(self.config).write_dataframe(
+                table_name,
+                data,
+                if_exists="append",
+            )
+        elif kind == "csv":
+            csv_file = os.path.join(self.config.output_tables_path(),
+                                    table_name+".csv")
+            if os.path.exists(csv_file):
+                data.to_csv(csv_file, mode="a", header=False)
+            else:
+                data.to_csv(csv_file)
+        else:
+            raise NotImplementedError
+
 
 class SimulatorMeta:
     """
@@ -419,7 +441,6 @@ def run_parallel(self, cores: int = 1):
         logger.info(
             f"Verification finished, now using {cores} cores for parallel computing!"
         )
-        print(parameters[0])
         pool.starmap(self.run_model, parameters)
 
         pool.close()
@@ -501,7 +522,6 @@ def new_parallel(self, cores: int = 1):
         try:
             logger.info(f"Melodie will run for {len(self.scenarios)} times!.")
             first_run = False
-            print(self.scenarios)
             tasks_count = 0
             for scenario in self.scenarios:
                 for id_run in range(scenario.run_num):
@@ -514,7 +534,6 @@ def new_parallel(self, cores: int = 1):
                         )
                         first_run = True
                     else:
-                        print("put task:", scenario.to_json())
                         parallel_manager.put_task(
                             (id_run, scenario.to_json(), None))
                         tasks_count += 1