(run_dir: Path, plan: list[dict[str, Any]])
| 822 | |
| 823 | |
| 824 | def execute_plan(run_dir: Path, plan: list[dict[str, Any]]) -> dict[str, Any]: |
| 825 | for dirname in ["qiime2", "tables", "logs", "workflow"]: |
| 826 | (run_dir / dirname).mkdir(parents=True, exist_ok=True) |
| 827 | result: dict[str, Any] = {"ok": True, "steps": []} |
| 828 | for index, item in enumerate(plan, start=1): |
| 829 | step = run_cmd(["bash", "-c", item["command"]], run_dir, timeout=7200) |
| 830 | safe = item["name"].replace(":", "").replace(" ", "_").replace("/", "_") |
| 831 | write_json(run_dir / "logs" / f"{index:02d}_{safe}.json", step) |
| 832 | result["steps"].append({"name": item["name"], "ok": step.get("ok")}) |
| 833 | result["ok"] = bool(result["ok"] and step.get("ok")) |
| 834 | if not step.get("ok"): |
| 835 | break |
| 836 | return result |
| 837 | |
| 838 | |
| 839 | def write_visuals( |
no test coverage detected