Merge pull request #57 from cseptesting/53-correct-management-of-models-input-catalog-and-arg-files

Update management of input files

Author: pabloitu

floatcsep/__init__.py

@@ -2,7 +2,7 @@
 from floatcsep import experiment
 from floatcsep import model
 from floatcsep.infrastructure import engine, environments, registries, repositories, logger
-from floatcsep.utils import readers, accessors, helpers
+from floatcsep.utils import file_io, accessors, helpers
 from floatcsep.postprocess import reporting, plot_handler
 
 from importlib.metadata import version, PackageNotFoundError

floatcsep/evaluation.py

@@ -287,7 +287,7 @@ def plot_results(
                 # Regular consistency/comparative test plots (e.g., many models)
                 try:
                     for time_str in timewindow:
-                        fig_path = registry.get_figure(time_str, self.name)
+                        fig_path = registry.get_figure_key(time_str, self.name)
                         results = self.read_results(time_str, models)
                         ax = func(results, plot_args=fargs, **fkwargs)
                         if "code" in fargs:
@@ -307,7 +307,7 @@ def plot_results(
                                 registry.figures[time_str][fig_name] = os.path.join(
                                     time_str, "figures", fig_name
                                 )
-                                fig_path = registry.get_figure(time_str, fig_name)
+                                fig_path = registry.get_figure_key(time_str, fig_name)
                                 ax = func(result, plot_args=fargs, **fkwargs, show=False)
                                 if "code" in fargs:
                                     exec(fargs["code"])
@@ -318,7 +318,7 @@ def plot_results(
                                     pyplot.show()
 
             elif self.type in ["sequential", "sequential_comparative", "batch"]:
-                fig_path = registry.get_figure(timewindow[-1], self.name)
+                fig_path = registry.get_figure_key(timewindow[-1], self.name)
                 results = self.read_results(timewindow[-1], models)
                 ax = func(results, plot_args=fargs, **fkwargs)
 

floatcsep/experiment.py

@@ -25,6 +25,7 @@
     parse_nested_dicts,
 )
 from floatcsep.infrastructure.engine import Task, TaskGraph
+from floatcsep.infrastructure.logger import log_models_tree, log_results_tree
 
 log = logging.getLogger("floatLogger")
 
@@ -52,8 +53,8 @@ class Experiment:
             - growth (:class:`str`): `incremental` or `cumulative`
             - offset (:class:`float`): recurrence of forecast creation.
 
-            For further details, see :func:`~floatcsep.utils.timewindows_ti`
-            and :func:`~floatcsep.utils.timewindows_td`
+            For further details, see :func:`~floatcsep.utils.time_windows_ti`
+            and :func:`~floatcsep.utils.time_windows_td`
 
         region_config (dict): Contains all the spatial and magnitude
             specifications. It must contain the following keys:
@@ -75,6 +76,7 @@ class Experiment:
 
         model_config (str): Path to the models' configuration file
         test_config (str): Path to the evaluations' configuration file
+        run_mode (str): 'sequential' or 'parallel'
         default_test_kwargs (dict): Default values for the testing
          (seed, number of simulations, etc.)
         postprocess (dict): Contains the instruction for postprocessing
@@ -99,6 +101,7 @@ def __init__(
         postprocess: str = None,
         default_test_kwargs: dict = None,
         rundir: str = "results",
+        run_mode: str = "sequential",
         report_hook: dict = None,
         **kwargs,
     ) -> None:
@@ -118,14 +121,15 @@ def __init__(
         os.makedirs(os.path.join(workdir, rundir), exist_ok=True)
 
         self.name = name if name else "floatingExp"
-        self.registry = ExperimentRegistry(workdir, rundir)
+        self.registry = ExperimentRegistry.factory(workdir=workdir, run_dir=rundir)
         self.results_repo = ResultsRepository(self.registry)
         self.catalog_repo = CatalogRepository(self.registry)
 
         self.config_file = kwargs.get("config_file", None)
         self.original_config = kwargs.get("original_config", None)
         self.original_run_dir = kwargs.get("original_rundir", None)
         self.run_dir = rundir
+        self.run_mode = run_mode
         self.seed = kwargs.get("seed", None)
         self.time_config = read_time_cfg(time_config, **kwargs)
         self.region_config = read_region_cfg(region_config, **kwargs)
@@ -143,7 +147,7 @@ def __init__(
         log.info(f"Setting up experiment {self.name}:")
         log.info(f"\tStart: {self.start_date}")
         log.info(f"\tEnd: {self.end_date}")
-        log.info(f"\tTime windows: {len(self.timewindows)}")
+        log.info(f"\tTime windows: {len(self.time_windows)}")
         log.info(f"\tRegion: {self.region.name if self.region else None}")
         log.info(
             f"\tMagnitude range: [{numpy.min(self.magnitudes)},"
@@ -175,7 +179,7 @@ def __getattr__(self, item: str) -> object:
         Override built-in method to return the experiment attributes by also using the command
         ``experiment.{attr}``. Adds also to the experiment scope the keys of
         :attr:`region_config` or :attr:`time_config`. These are: ``start_date``, ``end_date``,
-        ``timewindows``, ``horizon``, ``offset``, ``region``, ``magnitudes``, ``mag_min``,
+        ``time_windows``, ``horizon``, ``offset``, ``region``, ``magnitudes``, ``mag_min``,
         `mag_max``, ``mag_bin``, ``depth_min`` depth_max .
         """
 
@@ -295,8 +299,8 @@ def stage_models(self) -> None:
         """
         log.info("Staging models")
         for i in self.models:
-            i.stage(self.timewindows)
-            self.registry.add_forecast_registry(i)
+            i.stage(self.time_windows, run_mode=self.run_mode, run_dir=self.run_dir)
+            self.registry.add_model_registry(i)
 
     def set_tests(self, test_config: Union[str, Dict, List]) -> list:
         """
@@ -376,17 +380,17 @@ def set_tasks(self) -> None:
         """
 
         # Set the file path structure
-        self.registry.build_tree(self.timewindows, self.models, self.tests)
+        self.registry.build_tree(self.time_windows, self.models, self.tests, self.run_mode)
 
         log.debug("Pre-run forecast summary")
-        self.registry.log_forecast_trees(self.timewindows)
+        log_models_tree(log, self.registry, self.time_windows)
         log.debug("Pre-run result summary")
-        self.registry.log_results_tree()
+        log_results_tree(log, self.registry)
 
         log.info("Setting up experiment's tasks")
 
         # Get the time windows strings
-        tw_strings = timewindow2str(self.timewindows)
+        tw_strings = timewindow2str(self.time_windows)
 
         # Prepare the testing catalogs
         task_graph = TaskGraph()
@@ -481,7 +485,7 @@ def set_tasks(self) -> None:
                         )
             # Set up the Sequential_Comparative Scores
             elif test_k.type == "sequential_comparative":
-                tw_strs = timewindow2str(self.timewindows)
+                tw_strs = timewindow2str(self.time_windows)
                 for model_j in self.models:
                     task_k = Task(
                         instance=test_k,
@@ -504,7 +508,7 @@ def set_tasks(self) -> None:
                         )
             # Set up the Batch comparative Scores
             elif test_k.type == "batch":
-                time_str = timewindow2str(self.timewindows[-1])
+                time_str = timewindow2str(self.time_windows[-1])
                 for model_j in self.models:
                     task_k = Task(
                         instance=test_k,
@@ -540,9 +544,9 @@ def run(self) -> None:
         self.task_graph.run()
         log.info("Calculation completed")
         log.debug("Post-run forecast registry")
-        self.registry.log_forecast_trees(self.timewindows)
+        log_models_tree(log, self.registry, self.time_windows)
         log.debug("Post-run result summary")
-        self.registry.log_results_tree()
+        log_results_tree(log, self.registry)
 
     def read_results(self, test: Evaluation, window: str) -> List:
         """
@@ -559,7 +563,7 @@ def make_repr(self) -> None:
 
         """
         log.info("Creating reproducibility config file")
-        repr_config = self.registry.get("repr_config")
+        repr_config = self.registry.get_attr("repr_config")
 
         # Dropping region to results folder if it is a file
         region_path = self.region_config.get("path", False)
@@ -604,7 +608,7 @@ def as_dict(self, extra: Sequence = (), extended=False) -> dict:
             "time_config": {
                 i: j
                 for i, j in self.time_config.items()
-                if (i not in ("timewindows",) or extended)
+                if (i not in ("time_windows",) or extended)
             },
             "region_config": {
                 i: j
@@ -731,7 +735,7 @@ def test_stat(test_orig, test_repr):
 
     def get_results(self):
 
-        win_orig = timewindow2str(self.original.timewindows)
+        win_orig = timewindow2str(self.original.time_windows)
 
         tests_orig = self.original.tests
 
@@ -787,7 +791,7 @@ def get_hash(filename):
 
     def get_filecomp(self):
 
-        win_orig = timewindow2str(self.original.timewindows)
+        win_orig = timewindow2str(self.original.time_windows)
 
         tests_orig = self.original.tests
 
@@ -801,8 +805,8 @@ def get_filecomp(self):
                 for tw in win_orig:
                     results[test.name][tw] = dict.fromkeys(models_orig)
                     for model in models_orig:
-                        orig_path = self.original.registry.get_result(tw, test, model)
-                        repr_path = self.reproduced.registry.get_result(tw, test, model)
+                        orig_path = self.original.registry.get_result_key(tw, test, model)
+                        repr_path = self.reproduced.registry.get_result_key(tw, test, model)
 
                         results[test.name][tw][model] = {
                             "hash": (self.get_hash(orig_path) == self.get_hash(repr_path)),
@@ -811,8 +815,8 @@ def get_filecomp(self):
             else:
                 results[test.name] = dict.fromkeys(models_orig)
                 for model in models_orig:
-                    orig_path = self.original.registry.get_result(win_orig[-1], test, model)
-                    repr_path = self.reproduced.registry.get_result(win_orig[-1], test, model)
+                    orig_path = self.original.registry.get_result_key(win_orig[-1], test, model)
+                    repr_path = self.reproduced.registry.get_result_key(win_orig[-1], test, model)
                     results[test.name][model] = {
                         "hash": (self.get_hash(orig_path) == self.get_hash(repr_path)),
                         "byte2byte": filecmp.cmp(orig_path, repr_path),

floatcsep/infrastructure/logger.py

@@ -60,3 +60,74 @@ def set_console_log_level(log_level):
     for handler in logger.handlers:
         if isinstance(handler, logging.StreamHandler):
             handler.setLevel(log_level)
+
+
+
+
+def log_models_tree(log, experiment_registry, time_windows):
+    """
+    Logs the forecasts for all models managed by this ExperimentFileRegistry.
+    """
+    log.debug("===================")
+    log.debug(f" Total Time Windows: {len(time_windows)}")
+    for model_name, registry in experiment_registry.model_registries.items():
+        log.debug(f"  Model: {model_name}")
+        exists_group = []
+        not_exist_group = []
+
+        for timewindow, filepath in registry.forecasts.items():
+            if registry.forecast_exists(timewindow):
+                exists_group.append(timewindow)
+            else:
+                not_exist_group.append(timewindow)
+
+        log.debug(f"    Existing forecasts: {len(exists_group)}")
+        log.debug(f"    Missing forecasts: {len(not_exist_group)}")
+        for timewindow in not_exist_group:
+            log.debug(f"      Time Window: {timewindow}")
+    log.debug("===================")
+
+
+def log_results_tree(log, experiment_registry):
+    """
+    Logs a summary of the results dictionary, sorted by test.
+    For each test and time window, it logs whether all models have results,
+    or if some results are missing, and specifies which models are missing.
+    """
+    log.debug("===================")
+
+    total_results = results_exist_count = results_not_exist_count = 0
+
+    # Get all unique test names and sort them
+    all_tests = sorted(
+        {test_name for tests in experiment_registry.results.values() for test_name in tests}
+    )
+
+    for test_name in all_tests:
+        log.debug(f"Test: {test_name}")
+        for timewindow, tests in experiment_registry.results.items():
+            if test_name in tests:
+                models = tests[test_name]
+                missing_models = []
+
+                for model_name, result_path in models.items():
+                    total_results += 1
+                    result_full_path = experiment_registry.get_result_key(timewindow, test_name, model_name)
+                    if os.path.exists(result_full_path):
+                        results_exist_count += 1
+                    else:
+                        results_not_exist_count += 1
+                        missing_models.append(model_name)
+
+                if not missing_models:
+                    log.debug(f"  Time Window: {timewindow} - All models evaluated.")
+                else:
+                    log.debug(
+                        f"  Time Window: {timewindow} - Missing results for models: "
+                        f"{', '.join(missing_models)}"
+                    )
+
+    log.debug(f"Total Results: {total_results}")
+    log.debug(f"Results that Exist: {results_exist_count}")
+    log.debug(f"Results that Do Not Exist: {results_not_exist_count}")
+    log.debug("===================")