First pass implementation

Author: rbeucher

pyproject.toml

@@ -27,17 +27,27 @@ dependencies = [
     "dask",
     "pyyaml",
     "tqdm",
-    "requests"
+    "requests",
+    "parsl"
 ]
 dynamic = ["version"]
 
 [project.optional-dependencies]
+# pip install access_mopper[dashboard]
+dashboard = [
+  "streamlit>=1.35.0"
+]
 test = [
     "pytest",
     "pytest-cov",
     "ruff"
 ]
 
+[project.scripts]
+mopper-cmorise = "access_mopper.batch_cmoriser:main"
+mopper-dashboard = "access_mopper.dashboard.cmor_dashboard:main"
+mopper-example-config = "access_mopper.examples.show_config:main"
+
 [build-system]
 build-backend = "setuptools.build_meta"
 requires = [
@@ -53,7 +63,9 @@ include-package-data = true
 
 [tool.setuptools.package-data]
 access_mopper = ["*.yml", "mappings/*.json",
-                 "vocabularies/**/*"]
+                 "vocabularies/**/*",
+                 "dashboard/*.py", "examples/*.yml"
+]
 
 [tool.versioneer]
 VCS = "git"

src/access_mopper/batch_cmoriser

@@ -0,0 +1,112 @@
+import sys
+import yaml
+from pathlib import Path
+import subprocess
+import os
+
+from access_mopper import ACCESS_ESM_CMORiser
+from access_mopper.tracking import TaskTracker
+from parsl import python_app, Config, HighThroughputExecutor
+from parsl.providers import PBSProProvider
+from parsl.addresses import address_by_hostname
+from importlib.resources import files
+import parsl
+
+
+def start_dashboard(dashboard_path: str, db_path: str):
+    env = os.environ.copy()
+    env["CMOR_TRACKER_DB"] = db_path
+    subprocess.Popen(
+        ["streamlit", "run", dashboard_path],
+        env=env,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+
+@python_app
+def run_cmor(variable, config, db_path):
+    from access_mopper import ACCESS_ESM_CMORiser
+    from access_mopper.tracking import TaskTracker
+    from pathlib import Path
+
+    exp = config["experiment_id"]
+    tracker = TaskTracker(Path(db_path))
+    tracker.add_task(variable, exp)
+
+    if tracker.is_done(variable, exp):
+        return f"Skipped: {variable} (already done)"
+
+    try:
+        tracker.mark_running(variable, exp)
+        cmoriser = ACCESS_ESM_CMORiser(
+            input_paths=Path(config["input_folder"]),
+            compound_name=variable,
+            experiment_id=config["experiment_id"],
+            source_id=config["source_id"],
+            variant_label=config["variant_label"],
+            grid_label=config["grid_label"],
+            activity_id=config.get("activity_id"),
+            output_path=config["output_folder"],
+            drs_root=config.get("drs_root"),
+        )
+        cmoriser.run()
+        tracker.mark_done(variable, exp)
+        return f"Completed: {variable}"
+    except Exception as e:
+        tracker.mark_failed(variable, exp, str(e))
+        raise
+
+
+def main():
+    if len(sys.argv) != 2:
+        print("Usage: mopper-cmorise path/to/batch_config.yml")
+        sys.exit(1)
+
+    config_path = Path(sys.argv[1])
+    if not config_path.exists():
+        print(f"Error: config file not found: {config_path}")
+        sys.exit(1)
+
+    with config_path.open() as f:
+        config_data = yaml.safe_load(f)
+
+    tracker = TaskTracker()
+    DB_PATH = tracker.db_path
+
+    # Start Streamlit dashboard
+    DASHBOARD_SCRIPT = files("access_mopper.dashboard").joinpath("cmor_dashboard.py")
+    start_dashboard(str(DASHBOARD_SCRIPT), str(DB_PATH))
+
+    # Configure Parsl
+    parsl_config = Config(
+        executors=[
+            HighThroughputExecutor(
+                label="htex_pbs",
+                address=address_by_hostname(),
+                max_workers=1,
+                provider=PBSProProvider(
+                    queue="normal",
+                    launcher=None,
+                    walltime="01:00:00",
+                    select_options="1:ncpus=4:mem=16GB",
+                    scheduler_options="#PBS -P your_project",
+                    worker_init="module load netcdf-python",
+                    nodes_per_block=1,
+                    init_blocks=1,
+                    max_blocks=10,
+                ),
+            )
+        ],
+        strategy="simple",
+    )
+
+    parsl.load(parsl_config)
+
+    futures = [run_cmor(var, config_data, str(DB_PATH)) for var in config_data["variables"]]
+    results = [f.result() for f in futures]
+    print("\n".join(results))
+
+
+if __name__ == "__main__":
+    main()

src/access_mopper/dashboard/__init__.py

@@ -0,0 +1,69 @@
+import os
+import sqlite3
+from pathlib import Path
+
+import pandas as pd
+import streamlit as st
+
+DB_PATH = Path(
+    os.getenv("CMOR_TRACKER_DB", Path.home() / ".mopper" / "db" / "cmor_tasks.db")
+)
+
+st.set_page_config(page_title="CMORisation Tracker", layout="wide")
+st.title("🧼 ACCESS CMORisation Dashboard")
+
+
+@st.cache_data(ttl=10)
+def load_data():
+    conn = sqlite3.connect(DB_PATH)
+    df = pd.read_sql_query("SELECT * FROM cmor_tasks", conn)
+    conn.close()
+    return df
+
+
+df = load_data()
+
+# Sidebar filters
+with st.sidebar:
+    st.header("Filters")
+    statuses = df["status"].unique().tolist()
+    selected_statuses = st.multiselect("Status", options=statuses, default=statuses)
+    experiments = df["experiment"].unique().tolist()
+    selected_experiments = st.multiselect(
+        "Experiment", options=experiments, default=experiments
+    )
+
+# Apply filters
+filtered_df = df[
+    df["status"].isin(selected_statuses) & df["experiment"].isin(selected_experiments)
+]
+
+st.markdown(f"### Showing {len(filtered_df)} task(s)")
+st.dataframe(filtered_df, use_container_width=True)
+
+# Summary stats
+st.markdown("### 📊 Summary")
+summary = df["status"].value_counts().rename_axis("status").reset_index(name="count")
+st.table(summary)
+
+# Errors
+if "failed" in df["status"].values:
+    st.markdown("### ❌ Failed Tasks")
+    st.dataframe(
+        df[df["status"] == "failed"][["variable", "experiment", "error_message"]],
+        use_container_width=True,
+    )
+
+
+def main():
+    import os
+    import subprocess
+    from pathlib import Path
+
+    db_path = Path(
+        os.getenv("CMOR_TRACKER_DB", Path.home() / ".mopper" / "db" / "cmor_tasks.db")
+    )
+    subprocess.run(
+        ["streamlit", "run", __file__],
+        env={**os.environ, "CMOR_TRACKER_DB": str(db_path)},
+    )

src/access_mopper/dashboard/cmor_dashboard.py

@@ -0,0 +1,19 @@
+# === General input/output paths
+input_folder: /g/data/xyz/experiment/piControl
+output_folder: /g/data/xyz/cmip6_output/piControl
+drs_root: /g/data/xyz/cmip6_output
+
+# === CMIP6 metadata
+experiment_id: piControl
+source_id: ACCESS-ESM1-6
+variant_label: r1i1p1f1
+grid_label: gn
+activity_id: CMIP
+
+# === List of variables to CMORise (table.variable)
+variables:
+  - Amon.tas
+  - Amon.ps
+  - Lmon.smc
+  - Omon.thetao
+  - Omon.so

src/access_mopper/examples/batch_config.yml

@@ -0,0 +1,16 @@
+import shutil
+import sys
+from importlib.resources import files
+from pathlib import Path
+
+
+def main():
+    example_file = files("access_mopper.examples").joinpath("batch_config.yml")
+
+    if len(sys.argv) == 2:
+        target_path = Path(sys.argv[1])
+        shutil.copy(example_file, target_path)
+        print(f"Example config copied to {target_path}")
+    else:
+        with example_file.open("r") as f:
+            print(f.read())

src/access_mopper/examples/show_config.py

@@ -0,0 +1,86 @@
+import sqlite3
+from pathlib import Path
+from typing import Optional
+
+
+class TaskTracker:
+    def __init__(self, db_path: Optional[Path] = None):
+        if db_path is None:
+            db_path = Path.home() / ".mopper" / "db" / "cmor_tasks.db"
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self.conn = sqlite3.connect(self.db_path)
+        self._init_db()
+
+    def _init_db(self):
+        with self.conn:
+            self.conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS cmor_tasks (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    variable TEXT NOT NULL,
+                    experiment TEXT NOT NULL,
+                    status TEXT CHECK(status IN ('pending', 'running', 'done', 'failed')) NOT NULL DEFAULT 'pending',
+                    start_time TEXT,
+                    end_time TEXT,
+                    error_message TEXT
+                )
+                """
+            )
+            self.conn.execute(
+                "CREATE UNIQUE INDEX IF NOT EXISTS idx_var_exp ON cmor_tasks(variable, experiment)"
+            )
+
+    def add_task(self, variable: str, experiment: str):
+        with self.conn:
+            self.conn.execute(
+                """
+                INSERT OR IGNORE INTO cmor_tasks (variable, experiment)
+                VALUES (?, ?)
+                """,
+                (variable, experiment),
+            )
+
+    def mark_running(self, variable: str, experiment: str):
+        with self.conn:
+            self.conn.execute(
+                """
+                UPDATE cmor_tasks
+                SET status='running', start_time=datetime('now')
+                WHERE variable=? AND experiment=?
+                """,
+                (variable, experiment),
+            )
+
+    def mark_done(self, variable: str, experiment: str):
+        with self.conn:
+            self.conn.execute(
+                """
+                UPDATE cmor_tasks
+                SET status='done', end_time=datetime('now'), error_message=NULL
+                WHERE variable=? AND experiment=?
+                """,
+                (variable, experiment),
+            )
+
+    def mark_failed(self, variable: str, experiment: str, error_message: str):
+        with self.conn:
+            self.conn.execute(
+                """
+                UPDATE cmor_tasks
+                SET status='failed', end_time=datetime('now'), error_message=?
+                WHERE variable=? AND experiment=?
+                """,
+                (error_message, variable, experiment),
+            )
+
+    def is_done(self, variable: str, experiment: str) -> bool:
+        cur = self.conn.cursor()
+        cur.execute(
+            """
+            SELECT status FROM cmor_tasks WHERE variable=? AND experiment=?
+            """,
+            (variable, experiment),
+        )
+        row = cur.fetchone()
+        return row is not None and row[0] == "done"