ft: now docker environment manager handles custom volumes, with the purpose of giving different input folders for different containers, so there is no file conflict for parallelization. allowed the possibility of storing input data in the results folder for transparency, or simply in /tmp/

refac: td models now does not pass "args" as argument to run command. that means that model src code should automatically read the input/{args} file, and specify {args} in its configuration if necessary. modified pymock from case_g for the new requirements, as well as fixed its Dockerfile to be compatible with new floatcsep versions.

Author: pabloitu

floatcsep/commands/main.py

@@ -158,7 +158,7 @@ def reproduce(config: str, **kwargs) -> None:
     reproduced_exp.run()
 
     original_config = reproduced_exp.original_config
-    original_exp = Experiment.from_yml(original_config, rundir=reproduced_exp.original_run_dir)
+    original_exp = Experiment.from_yml(original_config, run_dir=reproduced_exp.original_run_dir)
     original_exp.stage_models()
     original_exp.set_tasks()
 

floatcsep/experiment.py

@@ -101,8 +101,9 @@ def __init__(
         tests: str = None,
         postprocess: str = None,
         default_test_kwargs: dict = None,
-        rundir: str = "results",
+        run_dir: str = "results",
         run_mode: str = "serial",
+        stage_dir: ... = "results",
         report_hook: dict = None,
         **kwargs,
     ) -> None:
@@ -116,20 +117,21 @@ def __init__(
 
         workdir = Path(kwargs.get("path", os.getcwd())).resolve()
         if kwargs.get("timestamp", False):
-            rundir = Path(rundir, f"run_{datetime.datetime.utcnow().date().isoformat()}")
-        os.makedirs(Path(workdir, rundir), exist_ok=True)
+            run_dir = Path(run_dir, f"run_{datetime.datetime.utcnow().date().isoformat()}")
+        os.makedirs(Path(workdir, run_dir), exist_ok=True)
 
         self.name = name if name else "floatingExp"
-        self.registry = ExperimentRegistry.factory(workdir=workdir, run_dir=rundir)
+        self.registry = ExperimentRegistry.factory(workdir=workdir, run_dir=run_dir)
         self.results_repo = ResultsRepository(self.registry)
         self.catalog_repo = CatalogRepository(self.registry)
         self.run_id = "run"
 
         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.original_run_dir = kwargs.get("original_run_dir", None)
+        self.run_dir = run_dir
         self.run_mode = run_mode
+        self.stage_dir = stage_dir
         self.seed = kwargs.get("seed", None)
         self.time_config = read_time_cfg(time_config, **kwargs)
         self.region_config = read_region_cfg(region_config, **kwargs)
@@ -139,7 +141,7 @@ def __init__(
         logger = kwargs.get("logging", False)
         if logger:
             filename = "experiment.log" if logger is True else logger
-            self.registry.logger = os.path.join(workdir, rundir, filename)
+            self.registry.logger = os.path.join(workdir, run_dir, filename)
             log.info(f"Logging at {self.registry.logger}")
             add_fhandler(self.registry.logger)
 
@@ -304,7 +306,7 @@ def stage_models(self) -> None:
             i.stage(
                 self.time_windows,
                 run_mode=self.run_mode,
-                stage_dir=self.registry.run_dir,
+                stage_dir=self.stage_dir,
                 run_id=self.run_id,
             )
             self.registry.add_model_registry(i)
@@ -587,8 +589,6 @@ def make_repr(self) -> None:
         if not exists(target_cat):
             shutil.copy2(self.registry.abs(self.catalog_repo.cat_path), target_cat)
 
-        # relative_path = self.registry.rel(self.registry.run_dir)
-        # print(self.registry.workdir.__class__, self.registry.run_dir.__class__)
         relative_path = Path(
             os.path.relpath(self.registry.workdir.as_posix(), self.registry.run_dir.as_posix())
         )
@@ -687,14 +687,13 @@ def from_yml(cls, config_yml: str, repr_dir=None, **kwargs):
             # Only ABSOLUTE PATH
             _dict["path"] = abspath(join(_dir_yml, _dict.get("path", "")))
 
-            # replaces rundir case reproduce option is used
+            # replaces run_dir case reproduce option is used
             if repr_dir:
-                _dict["original_rundir"] = _dict.get("rundir", "results")
-                _dict["rundir"] = relpath(join(_dir_yml, repr_dir), _dict["path"])
+                _dict["original_run_dir"] = _dict.get("run_dir", "results")
+                _dict["run_dir"] = relpath(join(_dir_yml, repr_dir), _dict["path"])
                 _dict["original_config"] = abspath(join(_dict["path"], _dict["config_file"]))
             else:
-
-                _dict["rundir"] = _dict.get("rundir", kwargs.pop("rundir", "results"))
+                _dict["run_dir"] = _dict.get("run_dir", kwargs.pop("run_dir", "results"))
             _dict["config_file"] = relpath(config_yml, _dir_yml)
             if "logging" in _dict:
                 kwargs.pop("logging")

floatcsep/infrastructure/environments.py

@@ -7,6 +7,7 @@
 import sys
 import venv
 from abc import ABC, abstractmethod
+from pathlib import Path
 from typing import Union
 
 import docker
@@ -58,7 +59,7 @@ def env_exists(self):
         pass
 
     @abstractmethod
-    def run_command(self, command):
+    def run_command(self, command, **kwargs):
         """
         Executes a command within the context of the environment.
 
@@ -267,7 +268,7 @@ def install_dependencies(self) -> None:
         ]
         subprocess.run(cmd, check=True)
 
-    def run_command(self, command) -> None:
+    def run_command(self, command, **kwargs) -> None:
         """
         Runs a specified command within the conda environment.
 
@@ -350,7 +351,7 @@ def install_dependencies(self) -> None:
         cmd = f"{pip_executable} install -e {os.path.abspath(self.model_directory)}"
         self.run_command(cmd)
 
-    def run_command(self, command) -> None:
+    def run_command(self, command, **kwargs) -> None:
         """
         Executes a specified command in the virtual environment and logs the output.
 
@@ -459,15 +460,17 @@ def env_exists(self) -> bool:
         except ImageNotFound:
             return False
 
-    def run_command(self, command=None) -> None:
+    def run_command(self, command=None, input_dir=None, forecast_dir=None) -> None:
         """
         Runs the model’s Docker container with input/ and forecasts/ mounted.
         Streams logs and checks for non-zero exit codes.
         """
-        model_root = os.path.abspath(self.model_directory)
+        model_root = Path(self.model_directory).resolve()
+        host_volume_input = input_dir or model_root / "input"
+        host_volume_forecasts = forecast_dir or model_root / "forecasts"
         mounts = {
-            os.path.join(model_root, "input"): {"bind": "/app/input", "mode": "rw"},
-            os.path.join(model_root, "forecasts"): {"bind": "/app/forecasts", "mode": "rw"},
+            host_volume_input: {"bind": "/app/input", "mode": "rw"},
+            host_volume_forecasts: {"bind": "/app/forecasts", "mode": "rw"},
         }
 
         uid, gid = os.getuid(), os.getgid()

floatcsep/model.py

@@ -1,7 +1,7 @@
-import shutil
 import json
 import logging
 import os
+import shutil
 from abc import ABC, abstractmethod
 from datetime import datetime
 from pathlib import Path
@@ -10,10 +10,10 @@
 import yaml
 from csep.core.forecasts import GriddedForecast, CatalogForecast
 
-from floatcsep.utils.accessors import from_zenodo, from_git
 from floatcsep.infrastructure.environments import EnvironmentFactory
 from floatcsep.infrastructure.registries import ModelRegistry
 from floatcsep.infrastructure.repositories import ForecastRepository
+from floatcsep.utils.accessors import from_zenodo, from_git
 from floatcsep.utils.helpers import timewindow2str, str2timewindow, parse_nested_dicts
 
 log = logging.getLogger("floatLogger")
@@ -393,7 +393,8 @@ def create_forecast(self, tstring: str, **kwargs) -> None:
             f"Running {self.name} using {self.environment.__class__.__name__}:"
             f" {timewindow2str([start_date, end_date])}"
         )
-        self.environment.run_command(f"{self.func} {self.registry.get_args_key(tstring)}")
+        input_dir = self.registry.get_input_dir(tstring)
+        self.environment.run_command(command=f"{self.func}", input_dir=input_dir)
 
     def prepare_args(self, start: datetime, end: datetime, **kwargs) -> None:
         """

tutorials/case_g/custom_plot_script.py

@@ -13,7 +13,7 @@ def main(experiment):
     """
 
     # Get all the timewindows
-    timewindows = experiment.timewindows
+    timewindows = experiment.time_windows
 
     # Get the pymock model
     model = experiment.get_model("pymock")

tutorials/case_g/pymock/Dockerfile

@@ -5,29 +5,37 @@ FROM python:3.8.13
 ARG USERNAME=modeler
 ARG USER_UID=1100
 ARG USER_GID=$USER_UID
-RUN groupadd --non-unique -g $USER_GID $USERNAME \
-    && useradd -u $USER_UID -g $USER_GID -s /bin/sh -m $USERNAME
 
-# Set up work directory in the Docker container.
-## *Change {pymock} to {model_name} when used as template*
-WORKDIR /usr/src/pymock/
+# User setup
+RUN groupadd --gid $USER_GID $USERNAME \
+ && useradd --uid $USER_UID --gid $USER_GID --shell /bin/bash --create-home $USERNAME
+
+# Create floatcsep IO interface
+RUN mkdir -p /app/input /app/forecasts && chown -R $USERNAME:$USERNAME /app
+
+# Create user and venv
+USER $USERNAME
+WORKDIR /app
+
+# Set up Python venv
+ENV VIRTUAL_ENV=/home/$USERNAME/venv
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+ENV PYTHONUNBUFFERED=1
+
+# Create virtual environment and install python basic packages
+RUN python3 -m venv $VIRTUAL_ENV && pip install --upgrade pip setuptools wheel
 
 # Copy the repository from the local machine to the Docker container.
 ## *Only the needed folders/files for the model build*
-COPY --chown=$USER_UID:$USER_GID pymock ./pymock/
-COPY --chown=$USER_UID:$USER_GID tests ./tests/
+COPY --chown=$USER_UID:$USER_GID pymock/ ./pymock/
 COPY --chown=$USER_UID:$USER_GID setup.cfg run.py setup.py ./
 
-# Set up and create python virtual environment
-ENV VIRTUAL_ENV=/opt/venv
-RUN python3 -m venv $VIRTUAL_ENV
-ENV PATH="$VIRTUAL_ENV/bin:$PATH"
-
 # Install the pymock package.
 ## *Uses pip to install setup.cfg and requirements/instructions therein*
-RUN pip install --no-cache-dir --upgrade pip
-RUN pip install .
+RUN pip3 install --no-cache-dir -e .
 
 # Docker can now be initialized as user
 USER $USERNAME
 
+ENTRYPOINT ["pymock"]
+

tutorials/case_g/pymock/pymock/libs.py

@@ -11,9 +11,7 @@ def syncat_path(start, end, folder):
         start = start.date()
         end = end.date()
 
-    return os.path.join(folder,
-                        f"pymock_{start.isoformat()}_{end.isoformat()}.csv"
-                        )
+    return os.path.join(folder, f"pymock_{start.isoformat()}_{end.isoformat()}.csv")
 
 
 def load_cat(path):
@@ -30,10 +28,16 @@ def load_cat(path):
     catalog = []
     with open(path) as f_:
         for line in f_.readlines()[1:]:
-            line = line.split(',')
-            event = [float(line[0]), float(line[1]), float(line[2]),
-                     datetime.fromisoformat(line[3]),
-                     float(line[4]), int(line[5]), int(line[6])]
+            line = line.split(",")
+            event = [
+                float(line[0]),
+                float(line[1]),
+                float(line[2]),
+                datetime.fromisoformat(line[3]),
+                float(line[4]),
+                int(line[5]),
+                int(line[6]),
+            ]
             catalog.append(event)
 
     return catalog
@@ -46,13 +50,15 @@ def write_forecast(start, end, forecast, folder=None):
     """
 
     if folder is None:
-        folder = 'forecasts'
+        folder = "forecasts"
     os.makedirs(folder, exist_ok=True)
-    with open(syncat_path(start, end, folder), 'w') as file_:
-        file_.write('lon, lat, M, time_string, depth, catalog_id, event_id\n')
+    with open(syncat_path(start, end, folder), "w") as file_:
+        file_.write("lon, lat, M, time_string, depth, catalog_id, event_id\n")
         for event in forecast:
-            line = f'{event[0]},{event[1]},{event[2]:.2f},' \
-                   f'{event[3].isoformat()},{event[4]},{event[5]},{event[6]}\n'
+            line = (
+                f"{event[0]},{event[1]},{event[2]:.2f},"
+                f"{event[3].isoformat()},{event[4]},{event[5]},{event[6]}\n"
+            )
             file_.write(line)
 
 
@@ -62,23 +68,23 @@ def read_args(path):
     {argument} = {argument_value}
     """
     import os, sys
-    params = {'start_date': None, 'end_date': None}
 
-    with open(path, 'r') as f_:
+    params = {"start_date": None, "end_date": None}
+
+    with open(path, "r") as f_:
         for line in f_.readlines():
-            line_ = [i.strip() for i in line.split('=')]
-            if line_[0] == 'start_date':
-                params['start_date'] = datetime.fromisoformat(line_[1])
-            elif line_[0] == 'end_date':
-                params['end_date'] = datetime.fromisoformat(line_[1])
-            elif line_[0] == 'catalog':
-                params['catalog'] = os.path.join(os.path.dirname(path),
-                                                 line_[1])
-            elif line_[0] == 'mag_min':
-                params['mag_min'] = float(line_[1])
-            elif line_[0] == 'n_sims':
-                params['n_sims'] = int(line_[1])
-            elif line_[0] == 'seed':
-                params['seed'] = int(line_[1])
+            line_ = [i.strip() for i in line.split("=")]
+            if line_[0] == "start_date":
+                params["start_date"] = datetime.fromisoformat(line_[1])
+            elif line_[0] == "end_date":
+                params["end_date"] = datetime.fromisoformat(line_[1])
+            elif line_[0] == "catalog":
+                params["catalog"] = os.path.join(os.path.dirname(path), line_[1])
+            elif line_[0] == "mag_min":
+                params["mag_min"] = float(line_[1])
+            elif line_[0] == "n_sims":
+                params["n_sims"] = int(line_[1])
+            elif line_[0] == "seed":
+                params["seed"] = int(line_[1])
 
     return params