(run_dir: Path, plan: list[dict[str, Any]])
| 372 | |
| 373 | |
| 374 | def execute_plan(run_dir: Path, plan: list[dict[str, Any]]) -> dict[str, Any]: |
| 375 | for dirname in ["alignment", "qc", "peaks", "tracks", "logs", "motifs"]: |
| 376 | (run_dir / dirname).mkdir(parents=True, exist_ok=True) |
| 377 | result: dict[str, Any] = {"ok": True, "steps": []} |
| 378 | for index, item in enumerate(plan, start=1): |
| 379 | step = run_cmd(["bash", "-c", item["command"]], run_dir, timeout=7200) |
| 380 | safe = item["name"].replace(":", "").replace(" ", "_").replace("/", "_") |
| 381 | write_json(run_dir / "logs" / f"{index:02d}_{safe}.json", step) |
| 382 | result["steps"].append({"name": item["name"], "ok": step.get("ok")}) |
| 383 | result["ok"] = bool(result["ok"] and step.get("ok")) |
| 384 | if not step.get("ok"): |
| 385 | break |
| 386 | return result |
| 387 | |
| 388 | |
| 389 | def write_visuals( |
no test coverage detected