Coverage for src/footman/_progress.py: 95%
207 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Duration history and the run estimate behind the progress bar.
3Runs of the same invocation shape — the same chain with the same values and
4passthrough, serial or parallel, in the same directory — tend to take
5similar time. The store keeps the recent wall totals per shape in
6`<footman cache>/<cwd-key>.times.json`; the estimator turns them into a
7determinate expectation only when the history genuinely supports one:
8enough samples, and a controlled right tail. Anything less honest renders
9as the indeterminate pulse instead.
11Execution-path only, stdlib only. A missing, corrupt, or read-only store
12never fails (or even warns about) a run — timing is a nicety, never load
13bearing.
14"""
16from __future__ import annotations
18import hashlib
19import json
20import os
21import shutil
22import statistics
23import threading
24import time
25from dataclasses import dataclass
26from pathlib import Path
27from typing import TextIO
29from footman import _paths
30from footman.split import Segment
32SCHEMA = 1
33WINDOW = 50 # samples kept per chain — recency policy, sliding
34MAX_KEYS = 200 # chains kept per directory
35IDLE_DAYS = 60 # a chain not run for this long is forgotten
36MIN_SAMPLES = 5 # fewer → indeterminate
37TAIL_RATIO = 1.8 # p90 beyond this multiple of p50 → too erratic
40def default_jobs() -> int:
41 """The parallel width when nobody chose one: cores - 1, never below 2 —
42 the machine stays responsive, the fan-out stays real."""
43 return max(2, (os.cpu_count() or 3) - 1)
46def chain_key(segments: list[Segment], *, sequential: bool, jobs: int) -> str:
47 """A stable hash of the invocation shape.
49 Values and passthrough are part of the shape on purpose — `fm test --
50 -k one` is not `fm test` — and serial/parallel/width variants of the
51 same chain keep separate histories (different distributions entirely).
52 """
53 shape = [
54 {
55 "task": s.task,
56 "values": s.values,
57 "variadic": s.variadic,
58 "passthrough": s.passthrough,
59 }
60 for s in segments
61 ]
62 payload = json.dumps(
63 {"sequential": sequential, "jobs": jobs, "chain": shape},
64 sort_keys=True,
65 default=str,
66 )
67 return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
70@dataclass(frozen=True)
71class Estimate:
72 """A determinate expectation for one run.
74 The bar fills against *scale* (p90 — it usually completes just before
75 the end and clamps rather than overrunning); the label quotes
76 *typical* (p50).
77 """
79 typical: float
80 scale: float
83def estimate(runs: list[float]) -> Estimate | None:
84 """A determinate estimate from *runs*, or None when honesty forbids one."""
85 if len(runs) < MIN_SAMPLES:
86 return None
87 deciles = statistics.quantiles(runs, n=10, method="inclusive")
88 p50, p90 = statistics.median(runs), deciles[8]
89 if p50 <= 0 or p90 > TAIL_RATIO * p50:
90 return None # erratic history: an "estimate" would be a guess
91 return Estimate(typical=p50, scale=max(p90, 0.1))
94# --- the store ---------------------------------------------------------------
97def load_runs(cwd: Path, key: str) -> list[float]:
98 """The recent durations for *key*, oldest first; [] when unknown."""
99 entry = _load(cwd).get("chains", {}).get(key)
100 if not isinstance(entry, dict):
101 return []
102 runs = entry.get("runs")
103 if not isinstance(runs, list): 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 return []
105 return [float(r) for r in runs if isinstance(r, (int, float))]
108def load_cmd_width(cwd: Path, key: str) -> int:
109 """The widest command label of *key*'s previous run — step-line
110 alignment is right from the first line on a warm run."""
111 entry = _load(cwd).get("chains", {}).get(key)
112 width = entry.get("cmd") if isinstance(entry, dict) else 0
113 return width if isinstance(width, int) and width > 0 else 0
116def record(cwd: Path, key: str, seconds: float, cmd_width: int = 0) -> None:
117 """Append one green run's wall total; prune idle chains, cap sizes.
119 Best-effort by contract: an unwritable cache directory must never fail
120 the run that just succeeded.
121 """
122 now = time.time()
123 data = _load(cwd)
124 chains = data.get("chains")
125 if not isinstance(chains, dict):
126 chains = {}
127 entry = chains.get(key)
128 old = entry.get("runs") if isinstance(entry, dict) else None
129 runs = [r for r in old if isinstance(r, (int, float))] if old else []
130 runs.append(round(float(seconds), 3))
131 old_cmd = entry.get("cmd") if isinstance(entry, dict) else 0
132 chains[key] = {"last": now, "runs": runs[-WINDOW:]}
133 # The widest step label, for next run's alignment — a width-less record
134 # keeps what the chain already knew.
135 width = cmd_width if cmd_width > 0 else old_cmd
136 if isinstance(width, int) and width > 0:
137 chains[key]["cmd"] = width
139 horizon = now - IDLE_DAYS * 86400
140 chains = {
141 k: v
142 for k, v in chains.items()
143 if isinstance(v, dict) and v.get("last", 0) >= horizon
144 }
145 if len(chains) > MAX_KEYS:
146 for stale in sorted(chains, key=lambda k: chains[k].get("last", 0))[
147 : len(chains) - MAX_KEYS
148 ]:
149 del chains[stale]
151 path = _paths.times_path(cwd)
152 try:
153 # Atomic, like the manifest write: concurrent fm runs (a hook's
154 # `fm check` racing yours) may lose each other's *sample* — a
155 # nicety — but a torn file would lose the whole history.
156 path.parent.mkdir(parents=True, exist_ok=True)
157 tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
158 tmp.write_text(
159 json.dumps({"schema": SCHEMA, "chains": chains}), encoding="utf-8"
160 )
161 os.replace(tmp, path)
162 except OSError:
163 pass
166def _load(cwd: Path) -> dict:
167 try:
168 data = json.loads(_paths.times_path(cwd).read_text(encoding="utf-8"))
169 except (OSError, ValueError):
170 return {}
171 return data if isinstance(data, dict) else {}
174# --- the status line ---------------------------------------------------------
176_CLEAR = "\r\033[K"
177_BAR_CELLS = 15
178_PULSE_CELLS = 3
181def fmt_secs(t: float) -> str:
182 """`4.1s`, `42s`, `1m10s`, `4h35m` — as short as honesty allows."""
183 if t >= 3600: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 return f"{int(t) // 3600}h{(int(t) % 3600) // 60:02d}m"
185 if t >= 60:
186 return f"{int(t) // 60}m{int(t) % 60:02d}s"
187 return f"{t:.0f}s" if t >= 9.5 else f"{t:.1f}s"
190class StatusLine:
191 """The one live line for a run, drawn on the real stderr.
193 Fed by *both* parallel engines — scheduler nodes and `parallel()`
194 children are the same kind of unit — so a chain and a task-body
195 fan-out present identically. Determinate runs fill a bar against the
196 estimate's p90 (clamped at 98% until the run truly ends); anything
197 else pulses, with elapsed time either way.
199 Coexistence contract: the output routers report every real-terminal
200 write via `notify()`, which clears a painted line first and tracks
201 whether the terminal now sits at column 0. The ticker (and every
202 event repaint) paints **only at column 0**, so `run()`'s in-place
203 step rewrites and flushed blocks are never corrupted. The line writes
204 straight to the real stream, never through the routers.
205 """
207 def __init__(
208 self, err: TextIO, est: Estimate | None, *, color: bool = True
209 ) -> None:
210 self.err = err
211 self.est = est
212 self.color = color
213 self.lock = threading.RLock()
214 self.started = time.perf_counter()
215 self.total = 0
216 self.done = 0
217 self.failed = 0
218 self.running: list[str] = [] # a list: anonymous thunks may collide
219 # Counted progress: what running tasks have *reported* about their
220 # own work (name -> fraction). Counted beats estimated — a task
221 # that knows it is 23/150 through is better evidence than any
222 # history — so a reported fraction outranks the time-based fill.
223 self.counted: dict[str, tuple[int, int]] = {}
224 self.painted = False
225 self.at_col0 = True
226 self.ticks = 0
227 self._stop = threading.Event()
228 self._ticker: threading.Thread | None = None
230 # -- lifecycle
231 def open(self) -> None:
232 """Start the repaint ticker (elapsed must move without events)."""
233 self._ticker = threading.Thread(target=self._run_ticker, daemon=True)
234 self._ticker.start()
236 def close(self) -> None:
237 self._stop.set()
238 if self._ticker is not None:
239 self._ticker.join(timeout=1.0)
240 with self.lock:
241 self._clear_locked()
243 def _run_ticker(self) -> None:
244 while not self._stop.wait(0.2): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 self.paint()
247 # -- the engine feed (scheduler nodes and parallel() children alike)
248 def unit_added(self, count: int = 1) -> None:
249 with self.lock:
250 self.total += count
252 def unit_started(self, name: str) -> None:
253 with self.lock:
254 self.running.append(name)
255 self.paint()
257 def unit_counted(self, name: str, done: int, total: int) -> None:
258 """A running task reporting its own progress: `done` of `total`."""
259 with self.lock:
260 self.counted[name] = (done, total)
261 self.paint()
263 def unit_finished(self, name: str, ok: bool) -> None:
264 with self.lock:
265 if name in self.running:
266 self.running.remove(name)
267 self.counted.pop(name, None)
268 self.done += 1
269 self.failed += 0 if ok else 1
270 self.paint()
272 def unit_skipped(self, name: str) -> None:
273 """A dependent of a failure never ran: done for counting purposes —
274 the failure that caused it already counted itself."""
275 with self.lock:
276 self.done += 1
277 self.paint()
279 # -- the router feed
280 def notify(self, s: str) -> None:
281 """A real-terminal write is about to land: get out of its way."""
282 if not s:
283 return
284 with self.lock:
285 self._clear_locked()
286 self.at_col0 = s.endswith("\n")
288 # -- painting
289 def paint(self) -> None:
290 with self.lock:
291 if not self.at_col0:
292 return # someone's mid-line (a live `→ step`): stay away
293 self.ticks += 1
294 self.err.write(f"{_CLEAR}{self._render()}")
295 self.err.flush()
296 self.painted = True
298 def _clear_locked(self) -> None:
299 if self.painted:
300 self.err.write(_CLEAR)
301 self.err.flush()
302 self.painted = False
304 def _counted_fraction(self) -> tuple[float, str] | None:
305 """The run's progress from *reported* work, if any task reported.
307 A reporting task contributes a fractional unit — three tasks done
308 and a fourth 23/150 through is 3.15/4, not 3/4 — so a run of
309 reporters fills smoothly and a mixed run is smooth where it can
310 be. Returns `(fraction, label)`, or `None` when nobody reported.
311 """
312 if not self.counted or self.total <= 0:
313 return None
314 partial = 0.0
315 for done, total in self.counted.values():
316 if total > 0: 316 ↛ 315line 316 didn't jump to line 315 because the condition on line 316 was always true
317 partial += min(done / total, 1.0)
318 fraction = min((self.done + partial) / self.total, 1.0)
319 if len(self.counted) == 1: # one reporter: show its own counts 319 ↛ 322line 319 didn't jump to line 322 because the condition on line 319 was always true
320 done, total = next(iter(self.counted.values()))
321 return fraction, f" {done}/{total}"
322 return fraction, f" {fraction * 100:.0f}%"
324 def _render(self) -> str:
325 elapsed = time.perf_counter() - self.started
326 if (counted := self._counted_fraction()) is not None:
327 fraction, counts = counted
328 filled = int(fraction * _BAR_CELLS)
329 bar = "█" * filled + "░" * (_BAR_CELLS - filled)
330 label = f" {fmt_secs(elapsed)}{counts}"
331 elif self.est is not None:
332 frac = min(elapsed / self.est.scale, 0.98)
333 filled = int(frac * _BAR_CELLS)
334 bar = "█" * filled + "░" * (_BAR_CELLS - filled)
335 label = f" {fmt_secs(elapsed)} ~{fmt_secs(self.est.typical)}"
336 else: # indeterminate: a bouncing pulse
337 span = _BAR_CELLS - _PULSE_CELLS
338 bounce = self.ticks % (2 * span)
339 pos = bounce if bounce <= span else 2 * span - bounce
340 bar = "░" * pos + "█" * _PULSE_CELLS + "░" * (span - pos)
341 label = f" {fmt_secs(elapsed)}"
342 line = f"[{bar}]{label}"
343 if self.total > 1:
344 line += f" {self.done}/{self.total}"
345 if self.failed:
346 text = f"{self.failed} failed"
347 if self.color: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 text = f"\033[31m{text}\033[0m"
349 line += f" ({text})"
350 if self.running:
351 names = ", ".join(list(self.running)[:4])
352 if len(self.running) > 4: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 names += " ..."
354 line += f" running: {names}"
355 width = shutil.get_terminal_size((80, 24)).columns - 1
356 return line if len(line) <= width else line[:width]