Revision control
Copy as Markdown
# Copyright (C) 2026 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
"""Progress output for the framework tools, so people can see a long run working.
Messages go to stderr; stdout stays clean for the JSON that tools read.
Set DOC_AGENT_QUIET=1 (or call quiet()) to turn them off.
import progress
progress.step(1, 5, "Baseline build")
with progress.heartbeat("QDoc build", log_path):
subprocess.run(...)
progress.done("Verified: the warning is gone")
Stdlib only.
"""
import os
import re
import sys
import threading
import time
_QUIET = os.environ.get("DOC_AGENT_QUIET") == "1"
_START = time.monotonic()
WARN_RE = re.compile(r"\((?:qdoc)\) warning:|: warning:")
def quiet(on=True):
global _QUIET
_QUIET = on
os.environ["DOC_AGENT_QUIET"] = "1" if on else "0" # child tools inherit it
def _elapsed():
# Wall-clock time, so lines from nested tools (verify_fix -> build.py) line up.
return time.strftime("%H:%M:%S")
def say(msg):
if not _QUIET:
print(f"[{_elapsed()}] {msg}", file=sys.stderr, flush=True)
def step(n, total, msg):
say(f"Step {n}/{total}: {msg}")
def done(msg):
say(f"Done: {msg}")
def _log_stats(path):
try:
with open(path, errors="replace") as f:
lines = f.readlines()
except OSError:
return None
return len(lines), sum(1 for l in lines if WARN_RE.search(l))
class heartbeat:
"""Print a status line every `every` seconds until the block ends."""
def __init__(self, label, log=None, every=None):
self.label, self.log = label, log
self.every = every or float(os.environ.get("DOC_AGENT_HEARTBEAT", "15"))
self._stop = threading.Event()
def _run(self):
start = time.monotonic()
while not self._stop.wait(self.every):
s = int(time.monotonic() - start)
extra = ""
if self.log:
stats = _log_stats(self.log)
if stats:
extra = f", log {stats[0]:,} lines, {stats[1]:,} warnings so far"
say(f"{self.label} still running ({s // 60}m {s % 60:02d}s{extra})")
def __enter__(self):
self._t0 = time.monotonic()
say(f"{self.label} started")
if not _QUIET:
self._thread = threading.Thread(target=self._run, daemon=True)
self._thread.start()
return self
def __exit__(self, *exc):
self._stop.set()
s = int(time.monotonic() - self._t0)
say(f"{self.label} finished in {s // 60}m {s % 60:02d}s")
return False