forked from InsightSoftwareConsortium/ITKPythonPackage
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathBuildManager.py
More file actions
134 lines (118 loc) · 4.53 KB
/
BuildManager.py
File metadata and controls
134 lines (118 loc) · 4.53 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import json
import time
from datetime import datetime
from pathlib import Path
class BuildManager:
"""Manage a JSON build report for multistep runs.
Persists status and timing for each named step to a JSON file.
Steps already marked ``"done"`` are skipped on subsequent runs,
enabling resumable builds.
Parameters
----------
report_path : Path
Path to the JSON report file (created if absent).
step_names : list[str]
Ordered names of build steps to track.
Attributes
----------
report : dict
In-memory report structure with ``created_at``, ``updated_at``,
and ``steps`` keys.
"""
def __init__(self, report_path: Path, step_names: list[str]):
self.report_path = Path(report_path)
self._init_structure(step_names)
self._load_if_exists()
# ---- Public API ----
def run_step(self, step_name: str, func, force_rerun=False) -> None:
"""Execute a build step, recording timing and status.
Parameters
----------
step_name : str
Key identifying the step in the report.
func : callable
Zero-argument callable that performs the step's work.
force_rerun : bool, optional
If True, re-execute even when the step is already ``"done"``.
Raises
------
Exception
Re-raises any exception thrown by *func* after recording
the failure in the report.
"""
entry = self.report["steps"].setdefault(step_name, self._new_step_entry())
if entry.get("status") == "done" and not force_rerun:
# Already completed in a previous run; skip
return
# Mark start
entry["status"] = "running"
entry["started_at"] = self._now()
self.report["updated_at"] = entry["started_at"]
self.save()
start = time.perf_counter()
try:
func()
except Exception as e:
# Record failure and re-raise
entry["status"] = "failed"
entry["finished_at"] = self._now()
entry["duration_sec"] = round(time.perf_counter() - start, 3)
entry["error"] = f"{type(e).__name__}: {e}"
self.report["updated_at"] = entry["finished_at"]
self.save()
raise
else:
# Record success
entry["status"] = "done"
entry["finished_at"] = self._now()
entry["duration_sec"] = round(time.perf_counter() - start, 3)
self.report["updated_at"] = entry["finished_at"]
self.save()
def save(self) -> None:
"""Write the current report to disk atomically."""
self.report_path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.report_path.with_suffix(self.report_path.suffix + ".tmp")
with open(tmp, "w", encoding="utf-8") as f:
json.dump(self.report, f, indent=2, sort_keys=True)
tmp.replace(self.report_path)
# ---- Internal helpers ----
def _init_structure(self, step_names: list[str]) -> None:
steps = {name: self._new_step_entry() for name in step_names}
now = self._now()
self.report = {
"created_at": now,
"updated_at": now,
"steps": steps,
}
def _load_if_exists(self) -> None:
if not self.report_path.exists():
return
try:
with open(self.report_path, encoding="utf-8") as f:
existing = json.load(f)
# Merge existing with current set of steps, preserving statuses
existing_steps = existing.get("steps", {})
for name in self.report["steps"].keys():
if name in existing_steps:
self.report["steps"][name] = existing_steps[name]
# Bring over timestamps
self.report["created_at"] = existing.get(
"created_at", self.report["created_at"]
)
self.report["updated_at"] = existing.get(
"updated_at", self.report["updated_at"]
)
except Exception as e:
# Corrupt or unreadable file; keep freshly initialized structure
raise RuntimeError(f"Failed to load build report: {e}") from e
@staticmethod
def _now() -> str:
return datetime.now().isoformat(timespec="seconds")
@staticmethod
def _new_step_entry() -> dict:
return {
"status": "pending",
"started_at": None,
"finished_at": None,
"duration_sec": None,
}