Coverage for src/footman/_progress.py: 96%
217 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +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 threading
22import time
23from dataclasses import dataclass
24from pathlib import Path
25from typing import Any, TextIO
27from footman import _paths
28from footman._split import Segment
30SCHEMA = 1
31WINDOW = 50 # samples kept per chain — recency policy, sliding
32MAX_KEYS = 200 # chains kept per directory
33IDLE_DAYS = 60 # a chain not run for this long is forgotten
34MIN_SAMPLES = 5 # fewer → indeterminate
35TAIL_RATIO = 1.8 # p90 beyond this multiple of p50 → too erratic
38def default_jobs() -> int:
39 """The parallel width when nobody chose one: cores - 1, never below 2 —
40 the machine stays responsive, the fan-out stays real."""
41 return max(2, (os.cpu_count() or 3) - 1)
44def chain_key(segments: list[Segment], *, sequential: bool, jobs: int) -> str:
45 """A stable hash of the invocation shape.
47 Values and passthrough are part of the shape on purpose — `fm test --
48 -k one` is not `fm test` — and serial/parallel/width variants of the
49 same chain keep separate histories (different distributions entirely).
50 """
51 shape = [
52 {
53 "task": s.task,
54 "values": s.values,
55 "variadic": s.variadic,
56 "passthrough": s.passthrough,
57 }
58 for s in segments
59 ]
60 payload = json.dumps(
61 {"sequential": sequential, "jobs": jobs, "chain": shape},
62 sort_keys=True,
63 default=str,
64 )
65 return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
68@dataclass(frozen=True)
69class Estimate:
70 """A determinate expectation for one run.
72 The bar fills against *scale* (p90 — it usually completes just before
73 the end and clamps rather than overrunning); the label quotes
74 *typical* (p50).
75 """
77 typical: float
78 scale: float
81def estimate(runs: list[float]) -> Estimate | None:
82 """A determinate estimate from *runs*, or None when honesty forbids one."""
83 if len(runs) < MIN_SAMPLES:
84 return None
85 # Imported past the guard, not at module scope: `statistics` drags
86 # `fractions` and `random` for ~1.2 ms, and a run with no history to speak
87 # of — a fresh project, CI's first green — never reaches this line.
88 import statistics
90 deciles = statistics.quantiles(runs, n=10, method="inclusive")
91 p50, p90 = statistics.median(runs), deciles[8]
92 if p50 <= 0 or p90 > TAIL_RATIO * p50:
93 return None # erratic history: an "estimate" would be a guess
94 return Estimate(typical=p50, scale=max(p90, 0.1))
97# --- the store ---------------------------------------------------------------
100def load_runs(cwd: Path, key: str) -> list[float]:
101 """The recent durations for *key*, oldest first; [] when unknown."""
102 entry = _load(cwd).get("chains", {}).get(key)
103 if not isinstance(entry, dict):
104 return []
105 runs = entry.get("runs")
106 if not isinstance(runs, list): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 return []
108 return [float(r) for r in runs if isinstance(r, (int, float))]
111def load_cmd_width(cwd: Path, key: str) -> int:
112 """The widest command label of *key*'s previous run — step-line
113 alignment is right from the first line on a warm run."""
114 entry = _load(cwd).get("chains", {}).get(key)
115 width = entry.get("cmd") if isinstance(entry, dict) else 0
116 return width if isinstance(width, int) and width > 0 else 0
119def record(cwd: Path, key: str, seconds: float, cmd_width: int = 0) -> None:
120 """Append one green run's wall total; prune idle chains, cap sizes.
122 Best-effort by contract: an unwritable cache directory must never fail
123 the run that just succeeded.
124 """
125 now = time.time()
126 data = _load(cwd)
127 chains = data.get("chains")
128 if not isinstance(chains, dict):
129 chains = {}
130 entry = chains.get(key)
131 old = entry.get("runs") if isinstance(entry, dict) else None
132 runs = [r for r in old if isinstance(r, (int, float))] if old else []
133 runs.append(round(seconds, 3))
134 old_cmd = entry.get("cmd") if isinstance(entry, dict) else 0
135 chains[key] = {"last": now, "runs": runs[-WINDOW:]}
136 # The widest step label, for next run's alignment — a width-less record
137 # keeps what the chain already knew.
138 width = cmd_width if cmd_width > 0 else old_cmd
139 if isinstance(width, int) and width > 0:
140 chains[key]["cmd"] = width
142 horizon = now - IDLE_DAYS * 86400
143 chains = {
144 k: v
145 for k, v in chains.items()
146 if isinstance(v, dict) and v.get("last", 0) >= horizon
147 }
148 if len(chains) > MAX_KEYS:
149 for stale in sorted(chains, key=lambda k: chains[k].get("last", 0))[
150 : len(chains) - MAX_KEYS
151 ]:
152 del chains[stale]
154 path = _paths.times_path(cwd)
155 try:
156 # Atomic, like the manifest write: concurrent fm runs (a hook's
157 # `fm check` racing yours) may lose each other's *sample* — a
158 # nicety — but a torn file would lose the whole history.
159 path.parent.mkdir(parents=True, exist_ok=True)
160 tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
161 tmp.write_text(
162 json.dumps({"schema": SCHEMA, "chains": chains}), encoding="utf-8"
163 )
164 os.replace(tmp, path)
165 except OSError:
166 pass
169def _load(cwd: Path) -> dict[str, Any]:
170 try:
171 data = json.loads(_paths.times_path(cwd).read_text(encoding="utf-8"))
172 except (OSError, ValueError):
173 return {}
174 return data if isinstance(data, dict) else {}
177# --- the status line ---------------------------------------------------------
179_CLEAR = "\r\033[K"
180_BAR_CELLS = 15
181_PULSE_CELLS = 3
184def fmt_secs(t: float) -> str:
185 """`4.1s`, `42s`, `1m10s`, `4h35m` — as short as honesty allows."""
186 if t >= 3600: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 return f"{int(t) // 3600}h{(int(t) % 3600) // 60:02d}m"
188 if t >= 60:
189 return f"{int(t) // 60}m{int(t) % 60:02d}s"
190 return f"{t:.0f}s" if t >= 9.5 else f"{t:.1f}s"
193class StatusLine:
194 """The one live line for a run, drawn on the real stderr.
196 Fed by every engine that runs a request — scheduler nodes, `parallel()`
197 children, and body calls through the futures layer are the same kind of
198 unit — so a chain, a task-body fan-out and an inline `build()` present
199 identically. Determinate runs fill a bar against the estimate's p90
200 (clamped at 98% until the run truly ends); anything else pulses, with
201 elapsed time either way.
203 Coexistence contract: the output routers report every real-terminal
204 write via `notify()`, which clears a painted line first and tracks
205 whether the terminal now sits at column 0. The ticker (and every
206 event repaint) paints **only at column 0**, so `run()`'s in-place
207 step rewrites and flushed blocks are never corrupted. The line writes
208 straight to the real stream, never through the routers.
209 """
211 def __init__(
212 self, err: TextIO, est: Estimate | None, *, color: bool = True
213 ) -> None:
214 self.err = err
215 self.est = est
216 self.color = color
217 self.lock = threading.RLock()
218 self.started = time.perf_counter()
219 self.total = 0
220 self.done = 0
221 self.failed = 0
222 self.running: list[str] = [] # a list: anonymous thunks may collide
223 # Counted progress: what running tasks have *reported* about their
224 # own work (name -> fraction). Counted beats estimated — a task
225 # that knows it is 23/150 through is better evidence than any
226 # history — so a reported fraction outranks the time-based fill.
227 self.counted: dict[str, tuple[int, int]] = {}
228 self.painted = False
229 self.at_col0 = True
230 self.suspended = False
231 self.ticks = 0
232 self._stop = threading.Event()
233 self._ticker: threading.Thread | None = None
235 # -- lifecycle
236 def open(self) -> None:
237 """Start the repaint ticker (elapsed must move without events)."""
238 self._ticker = threading.Thread(target=self._run_ticker, daemon=True)
239 self._ticker.start()
241 def close(self) -> None:
242 self._stop.set()
243 if self._ticker is not None:
244 self._ticker.join(timeout=1.0)
245 with self.lock:
246 self._clear_locked()
248 def _run_ticker(self) -> None:
249 while not self._stop.wait(0.2):
250 self.paint()
252 # -- the engine feed (scheduler nodes and parallel() children alike)
253 def unit_added(self, count: int = 1) -> None:
254 with self.lock:
255 self.total += count
257 def unit_started(self, name: str) -> None:
258 with self.lock:
259 self.running.append(name)
260 self.paint()
262 def unit_counted(self, name: str, done: int, total: int) -> None:
263 """A running task reporting its own progress: `done` of `total`."""
264 with self.lock:
265 self.counted[name] = (done, total)
266 self.paint()
268 def unit_finished(self, name: str, ok: bool) -> None:
269 with self.lock:
270 if name in self.running:
271 self.running.remove(name)
272 self.counted.pop(name, None)
273 self.done += 1
274 self.failed += 0 if ok else 1
275 self.paint()
277 def unit_skipped(self, name: str) -> None:
278 """A dependent of a failure never ran: done for counting purposes —
279 the failure that caused it already counted itself."""
280 with self.lock:
281 self.done += 1
282 self.paint()
284 # -- the router feed
285 def notify(self, s: str) -> None:
286 """A real-terminal write is about to land: get out of its way."""
287 if not s:
288 return
289 with self.lock:
290 self._clear_locked()
291 self.at_col0 = s.endswith("\n")
293 # -- console coexistence
294 def suspend(self) -> None:
295 """A console owner (an interactive task) holds the real terminal:
296 clear the line and stop painting until `resume` — a repaint would
297 fight the prompt for the cursor. Unit events keep accumulating, so
298 the resumed line is immediately truthful."""
299 with self.lock:
300 self.suspended = True
301 self._clear_locked()
303 def resume(self) -> None:
304 with self.lock:
305 self.suspended = False
306 self.paint()
308 # -- painting
309 def paint(self) -> None:
310 with self.lock:
311 if self.suspended or not self.at_col0:
312 return # a wizard owns the terminal, or someone's mid-line
313 self.ticks += 1
314 self.err.write(f"{_CLEAR}{self._render()}")
315 self.err.flush()
316 self.painted = True
318 def _clear_locked(self) -> None:
319 if self.painted:
320 self.err.write(_CLEAR)
321 self.err.flush()
322 self.painted = False
324 def _counted_fraction(self) -> tuple[float, str] | None:
325 """The run's progress from *reported* work, if any task reported.
327 A reporting task contributes a fractional unit — three tasks done
328 and a fourth 23/150 through is 3.15/4, not 3/4 — so a run of
329 reporters fills smoothly and a mixed run is smooth where it can
330 be. Returns `(fraction, label)`, or `None` when nobody reported.
331 """
332 if not self.counted or self.total <= 0:
333 return None
334 partial = 0.0
335 for done, total in self.counted.values():
336 if total > 0: 336 ↛ 335line 336 didn't jump to line 335 because the condition on line 336 was always true
337 partial += min(done / total, 1.0)
338 fraction = min((self.done + partial) / self.total, 1.0)
339 if len(self.counted) == 1: # one reporter: show its own counts 339 ↛ 342line 339 didn't jump to line 342 because the condition on line 339 was always true
340 done, total = next(iter(self.counted.values()))
341 return fraction, f" {done}/{total}"
342 return fraction, f" {fraction * 100:.0f}%"
344 def _render(self) -> str:
345 elapsed = time.perf_counter() - self.started
346 if (counted := self._counted_fraction()) is not None:
347 fraction, counts = counted
348 filled = int(fraction * _BAR_CELLS)
349 bar = "█" * filled + "░" * (_BAR_CELLS - filled)
350 label = f" {fmt_secs(elapsed)}{counts}"
351 elif self.est is not None:
352 frac = min(elapsed / self.est.scale, 0.98)
353 filled = int(frac * _BAR_CELLS)
354 bar = "█" * filled + "░" * (_BAR_CELLS - filled)
355 label = f" {fmt_secs(elapsed)} ~{fmt_secs(self.est.typical)}"
356 else: # indeterminate: a bouncing pulse
357 span = _BAR_CELLS - _PULSE_CELLS
358 bounce = self.ticks % (2 * span)
359 pos = bounce if bounce <= span else 2 * span - bounce
360 bar = "░" * pos + "█" * _PULSE_CELLS + "░" * (span - pos)
361 label = f" {fmt_secs(elapsed)}"
362 line = f"[{bar}]{label}"
363 if self.total > 1:
364 line += f" {self.done}/{self.total}"
365 if self.failed:
366 text = f"{self.failed} failed"
367 if self.color: 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true
368 text = f"\033[31m{text}\033[0m"
369 line += f" ({text})"
370 if self.running:
371 names = ", ".join(list(self.running)[:4])
372 if len(self.running) > 4: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 names += " ..."
374 line += f" running: {names}"
375 import shutil
377 from footman._describe import fit
379 # The line carries ANSI when a failure is counted, so `len()` reads
380 # nine cells too many and a raw slice can cut an escape in half —
381 # leaving the terminal painted red by a line that never finished
382 # saying so.
383 width = shutil.get_terminal_size((80, 24)).columns - 1
384 return fit(line, width)