Make the Airflow example portable

README.md

@@ -205,6 +205,12 @@ repository still contains legacy Airflow DAG helpers under `dpti/dags/` for
 users who need Airflow-based orchestration, but new users should start with the
 CLI.
 
+The legacy `workflow/DpFreeEnergy.py` file defines the `TI_taskflow` DAG. Place
+that file in Airflow's DAG directory before running `examples/airflow.sh`, or
+set `DPTI_AIRFLOW_DAG_ID` if the DAG is registered under another name. The
+example script resolves its input and work directory from the checkout instead
+of relying on a developer-specific absolute path.
+
 ## License
 
 `dpti` is distributed under the GNU Lesser General Public License v3.0. See

examples/FreeEnergy.json

@@ -1,7 +1,7 @@
 {
   "target_temp": 200,
   "target_pres": 50000,
-  "work_base_dir": "/home/fengbo/4_Sn/14_free_energy_airflow_test",
+  "work_base_dir": ".",
   "ti_path": "t",
   "conf_lmp":"beta.lmp",
   "ens": "npt-xy",

examples/airflow.sh

@@ -1 +1,21 @@
-airflow dags trigger  TI_taskflow  --conf $(printf "%s" $(cat FreeEnergy.json))
+#!/usr/bin/env bash
+set -euo pipefail
+
+# TI_taskflow is defined by workflow/DpFreeEnergy.py and must be loaded by Airflow.
+example_dir="$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
+config_file="${1:-${example_dir}/FreeEnergy.json}"
+dag_id="${DPTI_AIRFLOW_DAG_ID:-TI_taskflow}"
+work_base_dir="${DPTI_AIRFLOW_WORK_BASE_DIR:-${example_dir}}"
+
+config="$(DPTI_AIRFLOW_WORK_BASE_DIR="${work_base_dir}" python -c '
+import json
+import os
+import sys
+
+with open(sys.argv[1]) as fp:
+    data = json.load(fp)
+data["work_base_dir"] = os.environ["DPTI_AIRFLOW_WORK_BASE_DIR"]
+print(json.dumps(data))
+' "${config_file}")"
+
+airflow dags trigger "${dag_id}" --conf "${config}"

tests/test_airflow_example.py

@@ -0,0 +1,43 @@
+import json
+import os
+import stat
+import subprocess
+import tempfile
+import unittest
+
+
+class TestAirflowExample(unittest.TestCase):
+    def test_script_resolves_checkout_paths_and_dag_name(self):
+        """The example sends portable configuration to the documented DAG."""
+        repository_root = os.path.abspath("..")
+        script = os.path.join(repository_root, "examples", "airflow.sh")
+        with tempfile.TemporaryDirectory() as tempdir:
+            capture_path = os.path.join(tempdir, "args.txt")
+            airflow = os.path.join(tempdir, "airflow")
+            with open(airflow, "w") as fp:
+                fp.write('#!/bin/sh\nprintf "%s\\n" "$@" > "$AIRFLOW_CAPTURE"\n')
+            os.chmod(airflow, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+            env = os.environ.copy()
+            env["PATH"] = tempdir + os.pathsep + env["PATH"]
+            env["AIRFLOW_CAPTURE"] = capture_path
+
+            subprocess.run([script], cwd=repository_root, env=env, check=True)
+
+            with open(capture_path) as fp:
+                arguments = fp.read().splitlines()
+        self.assertEqual(arguments[:3], ["dags", "trigger", "TI_taskflow"])
+        config = json.loads(arguments[arguments.index("--conf") + 1])
+        self.assertEqual(
+            config["work_base_dir"], os.path.join(repository_root, "examples")
+        )
+
+    def test_json_contains_no_private_home_path(self):
+        config_path = os.path.join("..", "examples", "FreeEnergy.json")
+        with open(config_path) as fp:
+            config = json.load(fp)
+
+        self.assertEqual(config["work_base_dir"], ".")
+
+
+if __name__ == "__main__":
+    unittest.main()