From b849eb657b48581226ef79b3ab6bccbdacbd925f Mon Sep 17 00:00:00 2001 From: mes123456 Date: Thu, 3 Sep 2026 19:21:09 +0800 Subject: [PATCH] feat(my-tools): show migration progress and duration estimate --- my-tools/show_blender_web_scheduler.py | 220 ++++++++++++++++++++----- 1 file changed, 182 insertions(+), 38 deletions(-) diff --git a/my-tools/show_blender_web_scheduler.py b/my-tools/show_blender_web_scheduler.py index 25d1ab1a7..ef9ef51cb 100755 --- a/my-tools/show_blender_web_scheduler.py +++ b/my-tools/show_blender_web_scheduler.py @@ -9,9 +9,12 @@ from __future__ import annotations import argparse import json +import statistics +import re +import time from pathlib import Path -from wbv2_runtime import CONTROL_ROOT, live_status, read_json +from wbv2_runtime import CONTROL_ROOT, ROOT, live_status, read_json def main() -> int: @@ -20,45 +23,186 @@ def main() -> int: parser.add_argument("--watch", action="store_true") parser.add_argument("--interval-ms", type=int, default=2000) args = parser.parse_args() - result = live_status(json_only=True, watch=args.watch, interval_ms=args.interval_ms) - if result.returncode != 0: - if result.stderr: - print(result.stderr, end="") - return result.returncode - if args.watch: - # Preserve the Node tool's ANSI refresh and output contract for long - # running terminal sessions. - print(result.stdout, end="") - return 0 + while True: + result = live_status(json_only=True, watch=False, interval_ms=args.interval_ms) + if result.returncode != 0: + if result.stderr: + print(result.stderr, end="") + return result.returncode + try: + status = json.loads(result.stdout) + except json.JSONDecodeError: + print(result.stdout, end="") + return 0 + status["myTools"] = { + "watchdogStatus": read_json(CONTROL_ROOT / "my-tools-watchdog.json"), + "schedulerWrapperStatus": read_json(CONTROL_ROOT / "my-tools-scheduler.json"), + } + status["progress"] = project_progress() + if args.json: + print(json.dumps(status, ensure_ascii=False, separators=(",", ":"))) + if not args.watch: + return 0 + try: + time.sleep(args.interval_ms / 1000) + except KeyboardInterrupt: + return 0 + continue + if args.watch: + print("\x1b[2J\x1b[H", end="") + print("Blender Web/WBV2 scheduler") + print(f" state: {status.get('state', 'UNKNOWN')}") + print(f" phase: {status.get('phase', 'unknown')}") + queue = status.get("queue", {}) + print(f" current task: {queue.get('task') or 'none'}") + print(f" parent manifest: {queue.get('parentManifest') or 'none'}") + scheduler = status.get("scheduler", {}) + print(f" scheduler: pid={scheduler.get('pid') or 'none'} alive={scheduler.get('alive', False)} state={scheduler.get('state') or 'unknown'}") + guardian = status.get("guardian", {}) + supervisor = status.get("supervisor", {}) + print(f" guardian: pid={guardian.get('pid') or 'none'} alive={guardian.get('alive', False)} state={guardian.get('state') or 'unknown'}") + print(f" supervisor: pid={supervisor.get('pid') or 'none'} alive={supervisor.get('alive', False)} state={supervisor.get('state') or 'unknown'}") + print(f" latest handoff: {status.get('latestHandoff') or 'none'}") + print(f" local HEAD: {status.get('localHead') or 'unknown'}") + for line in format_progress(status["progress"]): + print(line) + print(f" message: {status.get('message', '')}") + if not args.watch: + return 0 + try: + time.sleep(args.interval_ms / 1000) + except KeyboardInterrupt: + return 0 + + +def _read_json(path: Path) -> dict: try: - status = json.loads(result.stdout) - except json.JSONDecodeError: - print(result.stdout, end="") - return 0 - status["myTools"] = { - "watchdogStatus": read_json(CONTROL_ROOT / "my-tools-watchdog.json"), - "schedulerWrapperStatus": read_json(CONTROL_ROOT / "my-tools-scheduler.json"), + value = json.loads(path.read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError): + return {} + return value if isinstance(value, dict) else {} + + +def _queue_task() -> str | None: + try: + match = re.search(r"\| 当前任务 \| `([^`]+)` \|", (ROOT / "docs/EXECUTION_QUEUE.md").read_text(encoding="utf-8")) + except OSError: + return None + return match.group(1) if match else None + + +def project_progress() -> dict: + """Summarize progress from existing receipts without running any checks.""" + inventory = _read_json(CONTROL_ROOT / "dynamic-inventory.json") + total = int(inventory.get("summary", {}).get("unitCount", 0) or 0) + verified = int(inventory.get("coverage", {}).get("verifiedUnits", 0) or 0) + unverified = int(inventory.get("coverage", {}).get("unverifiedUnits", max(0, total - verified)) or 0) + index = _read_json(CONTROL_ROOT / "task-index.json") + source = index.get("source", {}).get("path") + plan = _read_json(ROOT / source) if source else {} + tasks = plan.get("tasks", []) if isinstance(plan.get("tasks", []), list) else [] + done_tasks = sum(1 for task in tasks if task.get("status") == "done") + active_tasks = [task.get("id") for task in tasks if task.get("ready") is True] + blocked_tasks = sum(1 for task in tasks if task.get("status") == "blocked") + + receipts = [] + historical_candidates = [] + for task_dir in ROOT.glob("tests/golden/WBV2-P*-001"): + receipt = _read_json(task_dir / "module-receipt.json") + historical = receipt.get("inventory", {}).get("historical", {}) + if historical: + historical_candidates.append((task_dir.name, historical)) + if receipt.get("status") != "PASS": + continue + dynamic = receipt.get("inventory", {}).get("dynamic", {}) + duration = receipt.get("efficiency", {}).get("totalDurationMs") + if not isinstance(duration, (int, float)) or duration < 0: + duration = sum(item.get("durationMs", 0) for item in receipt.get("commands", []) if isinstance(item, dict)) + current_verified = dynamic.get("verifiedUnits") + if isinstance(current_verified, int): + receipts.append({ + "task": receipt.get("task", task_dir.name), + "verifiedUnits": current_verified, + "durationMs": duration, + "reused": bool(receipt.get("efficiency", {}).get("existingEvidenceReused")), + "commands": receipt.get("commands", []), + }) + receipts.sort(key=lambda item: item["task"]) + productive = [item for item in receipts if not item["reused"] and item["durationMs"] > 0] + deltas = [] + previous = None + for item in receipts: + if previous is not None and item["verifiedUnits"] > previous and not item["reused"] and item["durationMs"] > 0: + deltas.append({"units": item["verifiedUnits"] - previous, "durationMs": item["durationMs"], "task": item["task"]}) + previous = max(previous or 0, item["verifiedUnits"]) + if deltas: + rates = [item["units"] / (item["durationMs"] / 3_600_000) for item in deltas if item["durationMs"] > 0] + unit_rate = statistics.median(rates) if rates else 0.0 + batch_units = statistics.median(item["units"] for item in deltas) + else: + unit_rate = 0.0 + batch_units = 0.0 + remaining_hours = unverified / unit_rate if unit_rate > 0 else None + stage_durations = {} + for item in productive: + for index, command in enumerate(item.get("commands", [])): + duration = command.get("durationMs") if isinstance(command, dict) else None + if isinstance(duration, (int, float)) and duration >= 0: + stage_durations.setdefault(str(index + 1), []).append(duration) + stage_medians = {stage: round(statistics.median(values)) for stage, values in stage_durations.items() if values} + historical = max(historical_candidates, default=("", {}), key=lambda item: item[0])[1] + return { + "scope": "dynamic-source-inventory", + "currentTask": _queue_task(), + "totalUnits": total, + "verifiedUnits": verified, + "unverifiedUnits": unverified, + "verifiedPercent": round((verified / total) * 100, 2) if total else 0.0, + "taskPlan": {"tasks": len(tasks), "done": done_tasks, "active": active_tasks, "blocked": blocked_tasks}, + "observedBatches": len(receipts), + "productiveBatches": len(deltas), + "reusedBatches": sum(1 for item in receipts if item["reused"]), + "observedExecutionMs": round(sum(item["durationMs"] for item in productive)), + "stageMedianDurationMs": stage_medians, + "medianIncrementalUnitsPerBatch": round(batch_units, 2), + "medianIncrementalUnitsPerHour": round(unit_rate, 2), + "estimatedRemainingHours": round(remaining_hours, 2) if remaining_hours is not None else None, + "estimatedTotalHours": round((verified / unit_rate) + remaining_hours, 2) if unit_rate > 0 and remaining_hours is not None else None, + "estimateBasis": "median successful non-reused module receipts; excludes blocked/unrunnable units and final full verification", + "lastObservedReceipt": receipts[-1]["task"] if receipts else None, + "lastIncrementalBatch": deltas[-1]["task"] if deltas else None, + "historicalInventory": { + "total": historical.get("total", 6900), + "effectiveCompleted": historical.get("effectiveCompleted", 290), + "effectivePending": historical.get("effectivePending", 6609), + "reproductionRequired": True, + }, } - if args.json: - print(json.dumps(status, ensure_ascii=False, separators=(",", ":"))) - return 0 - print("Blender Web/WBV2 scheduler") - print(f" state: {status.get('state', 'UNKNOWN')}") - print(f" phase: {status.get('phase', 'unknown')}") - queue = status.get("queue", {}) - print(f" current task: {queue.get('task') or 'none'}") - print(f" parent manifest: {queue.get('parentManifest') or 'none'}") - scheduler = status.get("scheduler", {}) - print(f" scheduler: pid={scheduler.get('pid') or 'none'} alive={scheduler.get('alive', False)} state={scheduler.get('state') or 'unknown'}") - guardian = status.get("guardian", {}) - supervisor = status.get("supervisor", {}) - print(f" guardian: pid={guardian.get('pid') or 'none'} alive={guardian.get('alive', False)} state={guardian.get('state') or 'unknown'}") - print(f" supervisor: pid={supervisor.get('pid') or 'none'} alive={supervisor.get('alive', False)} state={supervisor.get('state') or 'unknown'}") - print(f" latest handoff: {status.get('latestHandoff') or 'none'}") - print(f" local HEAD: {status.get('localHead') or 'unknown'}") - print(f" message: {status.get('message', '')}") - return 0 + + +def format_progress(progress: dict) -> list[str]: + estimate = progress.get("estimatedRemainingHours") + estimate_text = f"{estimate} h" if estimate is not None else "n/a" + total_estimate = progress.get("estimatedTotalHours") + total_text = f"{total_estimate} h" if total_estimate is not None else "n/a" + plan = progress.get("taskPlan", {}) + return [ + "", + "Project migration progress", + f" dynamic units: {progress.get('verifiedUnits', 0)}/{progress.get('totalUnits', 0)} verified ({progress.get('verifiedPercent', 0)}%), remaining={progress.get('unverifiedUnits', 0)}", + f" task plan: {plan.get('done', 0)}/{plan.get('tasks', 0)} done, active={','.join(plan.get('active', [])) or 'none'}, blocked={plan.get('blocked', 0)}", + f" historical ledger: {progress.get('historicalInventory', {}).get('effectiveCompleted', 0)}/{progress.get('historicalInventory', {}).get('total', 6900)} effective, pending={progress.get('historicalInventory', {}).get('effectivePending', 0)} (full reproduction required)", + f" observed batches: {progress.get('observedBatches', 0)} total, productive={progress.get('productiveBatches', 0)}, reused={progress.get('reusedBatches', 0)}", + f" efficiency: median +{progress.get('medianIncrementalUnitsPerBatch', 0)} units/batch, {progress.get('medianIncrementalUnitsPerHour', 0)} units/hour", + f" stage medians: {progress.get('stageMedianDurationMs') or 'n/a'} ms (1=build, 2=typecheck, 3=web build, 4=Chromium)", + f" estimate: remaining={estimate_text}, total={total_text}", + f" estimate basis: {progress.get('estimateBasis', 'n/a')}", + ] if __name__ == "__main__": - raise SystemExit(main()) + try: + raise SystemExit(main()) + except BrokenPipeError: + # Quietly support consumers such as ``head`` closing a JSON stream. + pass