#!/usr/bin/env python3 """Run and compare repeatable local Balance Lab level brackets.""" from __future__ import annotations import argparse import csv import json import math import os import subprocess import sys from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path from typing import Any, Iterable, Sequence ROOT = Path(__file__).resolve().parents[1] BALANCE_LAB = ROOT / "scripts" / "balance_lab.py" def write_text(path: Path, value: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(value.rstrip() + "\n", encoding="utf-8", newline="\n") def write_json(path: Path, value: Any) -> None: write_text(path, json.dumps(value, indent=2, ensure_ascii=False)) def write_csv(path: Path, rows: Sequence[dict[str, Any]], fields: Sequence[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=fields, extrasaction="ignore", lineterminator="\n") writer.writeheader() writer.writerows(rows) def bracket_levels(step: int) -> list[int]: if step not in (1, 5, 10): raise ValueError("Bracket step must be 1, 5, or 10.") if step == 1: return list(range(1, 81)) return [1, *range(step, 81, step)] def parse_levels(value: str) -> list[int]: try: levels = sorted({int(part.strip()) for part in value.split(",") if part.strip()}) except ValueError as error: raise argparse.ArgumentTypeError("Levels must be comma-separated integers.") from error if not levels or levels[0] < 1 or levels[-1] > 80: raise argparse.ArgumentTypeError("Levels must be between 1 and 80.") return levels def build_balance_lab_command(args: argparse.Namespace, level: int, output_dir: Path) -> list[str]: command = [ sys.executable, str(BALANCE_LAB), "--source", args.source, "--level", str(level), "--party-mode", "automatic", "--spec-selection", args.spec_selection, "--party-seed", str(args.party_seed), "--party-configurations", str(args.party_configurations), "--simulations-per-party", str(args.simulations_per_party), "--duration", str(args.duration), "--boss-category", args.boss_category, "--output-dir", str(output_dir), ] if args.seeds: command.extend(("--seeds", args.seeds)) if args.active_slot is not None: command.extend(("--active-slot", str(args.active_slot))) if getattr(args, "seat_sweep", False): command.append("--seat-sweep") if getattr(args, "cooldown_policy", None): command.extend(("--cooldown-policy", str(args.cooldown_policy.resolve()))) if getattr(args, "companion_readiness_policy", None): command.extend(("--companion-readiness-policy", str(args.companion_readiness_policy.resolve()))) if getattr(args, "include_beta_specs", False): command.append("--include-beta-specs") if getattr(args, "ready_only", False): command.append("--ready-only") for selector in args.spec: command.extend(("--spec", selector)) for selector in args.boss: command.extend(("--boss", selector)) return command def run_level(args: argparse.Namespace, level: int, output_dir: Path) -> tuple[int, subprocess.CompletedProcess[str]]: result = subprocess.run( build_balance_lab_command(args, level, output_dir), cwd=ROOT, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) return level, result def weighted_mean(rows: Iterable[dict[str, Any]], field: str, weight: str = "runs") -> float: materialized = list(rows) total_weight = sum(float(row.get(weight, 0)) for row in materialized) if total_weight <= 0: return 0.0 return sum(float(row.get(field, 0) or 0) * float(row.get(weight, 0)) for row in materialized) / total_weight def raw_run_totals(report_path: Path) -> dict[str, Any] | None: raw_paths = sorted(report_path.parent.glob("parties/*/raw-runs.json")) if not raw_paths: return None totals: dict[str, Any] = { "runs": 0, "seconds": 0.0, "partyDamage": 0.0, "bossDamage": 0.0, "incomingDamage": 0.0, "rawIncomingDamage": 0.0, "defeats": 0, "wipes": 0, "incomingRuns": 0, "healingRequiredRuns": 0, "minimumHealthPercent": 0.0, "pressureWindows": 0, "healerThroughputWindows": 0, "healerResponseWindows": 0, "windowMetricsAvailable": False, } minimums = [] for raw_path in raw_paths: raw = json.loads(raw_path.read_text(encoding="utf-8")) for run in raw["runs"]: totals["runs"] += 1 totals["seconds"] += float(run["elapsedSeconds"]) totals["partyDamage"] += float(run["partyDamage"]) totals["bossDamage"] += float(run["bossDamage"]) totals["incomingDamage"] += float(run["incomingDamage"]) totals["rawIncomingDamage"] += float(run["rawIncomingDamage"]) totals["defeats"] += run["outcome"] == "defeated" totals["wipes"] += run["outcome"] == "wipe" totals["incomingRuns"] += float(run["incomingDamage"]) > 0 healer_throughput = sum( float(member.get("healing", 0)) + float(member.get("absorbs", 0)) for member in run["members"] if member["role"] == "healer" ) totals["healingRequiredRuns"] += healer_throughput > 0 if "pressureWindows" in run: totals["windowMetricsAvailable"] = True totals["pressureWindows"] += int(run["pressureWindows"]) totals["healerThroughputWindows"] += int(run["healerThroughputWindows"]) totals["healerResponseWindows"] += int(run["healerResponseWindows"]) minimums.append(min(float(member["minimumHealthPercent"]) for member in run["members"])) totals["minimumHealthPercent"] = sum(minimums) / len(minimums) if minimums else 1.0 return totals def csv_run_totals(report_path: Path) -> dict[str, Any] | None: candidates = [ report_path.with_name("suite_runs.csv"), report_path.parent / f"level-{json.loads(report_path.read_text(encoding='utf-8'))['level']}-suite_runs.csv", ] run_path = next((path for path in candidates if path.exists()), None) if run_path is None: return None with run_path.open("r", encoding="utf-8", newline="") as stream: rows = list(csv.DictReader(stream)) return { "runs": len(rows), "seconds": sum(float(row["elapsedSeconds"]) for row in rows), "partyDamage": sum(float(row["partyDamage"]) for row in rows), "bossDamage": sum(float(row["bossDamage"]) for row in rows), "incomingDamage": sum(float(row["incomingDamage"]) for row in rows), "rawIncomingDamage": sum(float(row["rawIncomingDamage"]) for row in rows), "defeats": sum(row["outcome"] == "defeated" for row in rows), "wipes": sum(row["outcome"] == "wipe" for row in rows), "incomingRuns": None, "healingRequiredRuns": None, "minimumHealthPercent": None, "pressureWindows": None, "healerThroughputWindows": None, "healerResponseWindows": None, "windowMetricsAvailable": False, } def summarize_report(report_path: Path) -> dict[str, Any]: report = json.loads(report_path.read_text(encoding="utf-8")) bosses = report["bosses"] totals = raw_run_totals(report_path) or csv_run_totals(report_path) if totals is None: raise ValueError(f"No raw-runs.json or suite_runs.csv found for {report_path}") seconds = max(0.1, float(totals["seconds"])) runs = max(1, int(totals["runs"])) healers = [row for row in report["classSpecs"] if row["role"] == "healer"] healer_healing = sum(float(row["totalHealing"]) for row in healers) healer_absorbs = sum(float(row["totalAbsorbs"]) for row in healers) party_dps = float(totals["partyDamage"]) / seconds boss_dps = float(totals["bossDamage"]) / seconds incoming_dtps = float(totals["incomingDamage"]) / seconds raw_incoming_dtps = float(totals["rawIncomingDamage"]) / seconds boss_health = weighted_mean(bosses, "bossMaxHealth") defeated_runs = sum(float(row["runs"]) * float(row["defeatRate"]) for row in bosses) defeated_seconds = sum( float(row["runs"]) * float(row["defeatRate"]) * float(row["averageTimeToDefeat"] or 0) for row in bosses ) return { "level": int(report["level"]), "reportPath": str(report_path), "partyConfigurations": int(report["summary"]["partyConfigurations"]), "bosses": int(report["summary"]["bosses"]), "runs": int(totals["runs"]), "bossHealth": boss_health, "partyDps": party_dps, "bossDps": boss_dps, "healthToDpsSeconds": boss_health / max(0.1, boss_dps), "incomingDtps": incoming_dtps, "rawIncomingDtps": raw_incoming_dtps, "mitigationRate": 1 - incoming_dtps / raw_incoming_dtps if raw_incoming_dtps > 0 else 0, "healerHps": healer_healing / seconds, "healerAbsorbsPerSecond": healer_absorbs / seconds, "healingCoverageRate": (healer_healing + healer_absorbs) / max(0.1, float(totals["incomingDamage"])), "incomingRunRate": ( float(totals["incomingRuns"]) / runs if totals["incomingRuns"] is not None else None ), "healingRequiredRunRate": ( float(totals["healingRequiredRuns"]) / runs if totals["healingRequiredRuns"] is not None else None ), "healerResponseRate": ( float(totals["healerResponseWindows"]) / max(1, int(totals["pressureWindows"])) if totals["windowMetricsAvailable"] else None ), "healerWindowsPerPressureWindow": ( float(totals["healerThroughputWindows"]) / max(1, int(totals["pressureWindows"])) if totals["windowMetricsAvailable"] else None ), "averageMinimumHealthPercent": totals["minimumHealthPercent"], "defeatRate": float(totals["defeats"]) / runs, "wipeRate": float(totals["wipes"]) / runs, "averageTimeToDefeat": defeated_seconds / defeated_runs if defeated_runs > 0 else None, } def add_growth_metrics(rows: list[dict[str, Any]]) -> None: rows.sort(key=lambda row: row["level"]) previous = None for row in rows: if previous is None: row["previousHealthToDpsSeconds"] = None row["bossHealthGrowth"] = None row["partyDpsGrowth"] = None row["durabilityGrowthVsDps"] = None else: row["previousHealthToDpsSeconds"] = previous["healthToDpsSeconds"] health_growth = row["bossHealth"] / max(0.1, previous["bossHealth"]) dps_growth = row["bossDps"] / max(0.1, previous["bossDps"]) row["bossHealthGrowth"] = health_growth row["partyDpsGrowth"] = dps_growth row["durabilityGrowthVsDps"] = health_growth / max(0.01, dps_growth) previous = row def evaluate_rows(rows: Sequence[dict[str, Any]], args: argparse.Namespace) -> bool: passed = True for row in rows: flags = [] if row["healthToDpsSeconds"] < args.min_health_to_dps_seconds: flags.append("boss durability below limit") if row["healthToDpsSeconds"] > args.max_health_to_dps_seconds: flags.append("boss durability above limit") growth = row["durabilityGrowthVsDps"] if ( growth is not None and growth > args.max_durability_growth and row["previousHealthToDpsSeconds"] >= args.min_health_to_dps_seconds ): flags.append("boss health outgrew party DPS") if row["incomingDtps"] <= 0: flags.append("boss dealt no effective damage") if row["healerHps"] <= 0 and row["incomingDtps"] > 0: flags.append("healer produced no effective healing") if row["incomingRunRate"] is not None and row["incomingRunRate"] < args.min_incoming_run_rate: flags.append("too many runs had no incoming damage") if ( row["healingRequiredRunRate"] is not None and row["healingRequiredRunRate"] < args.min_healing_required_run_rate ): flags.append("too many runs required no healer throughput") if ( row.get("healerResponseRate") is not None and row["healerResponseRate"] < args.min_healer_response_rate ): flags.append("healer did not answer enough pressure windows") row["flags"] = flags row["passed"] = not flags passed = passed and not flags return passed def discover_reports(directory: Path) -> list[Path]: candidates = [*directory.glob("level-*/suite-report.json"), *directory.glob("level-*.json")] reports = [] for path in sorted(set(candidates)): try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): continue if "level" in value and "bosses" in value and "classSpecs" in value: reports.append(path) return reports def weapon_dps(weapon: dict[str, Any] | None) -> float: if not weapon: return 0.0 return (float(weapon["damageMin"]) + float(weapon["damageMax"])) / 2 / max( 0.1, float(weapon["speedMs"]) / 1_000 ) def collect_heirlooms(report_paths: Sequence[Path]) -> list[dict[str, Any]]: rows = [] for report_path in report_paths: catalog_path = report_path.parent / "catalog.json" if not catalog_path.exists(): continue catalog = json.loads(catalog_path.read_text(encoding="utf-8")) for build in catalog["builds"]: gear = build.get("gear") if not gear: continue stats = gear["stats"] melee_weapon_dps = weapon_dps(stats.get("meleeWeapon")) ranged_weapon_dps = weapon_dps(stats.get("rangedWeapon")) rows.append({ "level": catalog["level"], "source": build["source"], "classId": build["classId"], "className": build["className"], "specialization": build["specialization"], "role": build["role"], "presetId": gear["presetId"], "model": gear["model"], "itemCount": gear["itemCount"], "averageItemLevel": gear["averageItemLevel"], "strength": stats["strength"], "agility": stats["agility"], "stamina": stats["stamina"], "intellect": stats["intellect"], "spirit": stats["spirit"], "attackPower": stats["attackPower"], "rangedAttackPower": stats["rangedAttackPower"], "spellPower": stats["spellPower"], "criticalStrikeRating": stats["criticalStrikeRating"], "hitRating": stats["hitRating"], "hasteRating": stats["hasteRating"], "expertiseRating": stats["expertiseRating"], "armorPenetrationRating": stats["armorPenetrationRating"], "defenseRating": stats["defenseRating"], "dodgeRating": stats["dodgeRating"], "parryRating": stats["parryRating"], "blockRating": stats["blockRating"], "blockValue": stats["blockValue"], "mp5": stats["mp5"], "armor": stats["armor"], "meleeWeaponDps": melee_weapon_dps, "rangedWeaponDps": ranged_weapon_dps, "weaponDps": max(melee_weapon_dps, ranged_weapon_dps), }) return rows def percent(value: Any) -> str: return "—" if value is None else f"{float(value) * 100:.1f}%" def ratio(value: Any) -> str: return "—" if value is None else f"{float(value):.2f}×" def markdown_report(rows: Sequence[dict[str, Any]]) -> str: lines = [ "# Balance Lab local bracket comparison", "", "| Level | Boss HP | Boss DPS taken | HP / DPS | Incoming DTPS | Healer HPS | Heal coverage | Pressure windows answered | Defeat | Wipe | HP growth / DPS growth | Result |", "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |", ] for row in rows: result = "pass" if row["passed"] else "; ".join(row["flags"]) lines.append( f"| {row['level']} | {row['bossHealth']:.0f} | {row['bossDps']:.1f} | " f"{row['healthToDpsSeconds']:.1f}s | {row['incomingDtps']:.1f} | {row['healerHps']:.1f} | " f"{percent(row['healingCoverageRate'])} | {percent(row['healerResponseRate'])} | {percent(row['defeatRate'])} | " f"{percent(row['wipeRate'])} | {ratio(row['durabilityGrowthVsDps'])} | {result} |" ) lines.extend([ "", "`HP / DPS` uses damage applied directly to the boss. `Heal coverage` is effective healer healing plus absorbs divided by effective incoming party damage. Pressure-window response counts a five-second incoming-damage window when healer throughput occurs in that window or an adjacent one. Growth compares each row with the preceding tested level.", ]) return "\n".join(lines) SUMMARY_FIELDS = ( "level", "partyConfigurations", "bosses", "runs", "bossHealth", "partyDps", "bossDps", "healthToDpsSeconds", "incomingDtps", "rawIncomingDtps", "mitigationRate", "healerHps", "healerAbsorbsPerSecond", "healingCoverageRate", "incomingRunRate", "healingRequiredRunRate", "healerResponseRate", "healerWindowsPerPressureWindow", "averageMinimumHealthPercent", "defeatRate", "wipeRate", "averageTimeToDefeat", "previousHealthToDpsSeconds", "bossHealthGrowth", "partyDpsGrowth", "durabilityGrowthVsDps", "passed", "flags", "reportPath", ) HEIRLOOM_FIELDS = ( "level", "source", "classId", "className", "specialization", "role", "presetId", "model", "itemCount", "averageItemLevel", "strength", "agility", "stamina", "intellect", "spirit", "attackPower", "rangedAttackPower", "spellPower", "criticalStrikeRating", "hitRating", "hasteRating", "expertiseRating", "armorPenetrationRating", "defenseRating", "dodgeRating", "parryRating", "blockRating", "blockValue", "mp5", "armor", "meleeWeaponDps", "rangedWeaponDps", "weaponDps", ) def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description="Run and compare local Balance Lab level brackets.") result.add_argument("--step", type=int, choices=(1, 5, 10), default=10, help="level bracket step; includes level 1 (default: 10)") result.add_argument("--levels", type=parse_levels, help="explicit comma-separated levels; overrides --step") result.add_argument("--source", choices=("wow", "coa", "rom", "all"), default="wow") result.add_argument("--spec-selection", choices=("random", "all", "specific"), default="all") result.add_argument("--spec", action="append", default=[]) result.add_argument("--party-seed", type=int, default=1) result.add_argument("--party-configurations", type=int, default=10) result.add_argument("--simulations-per-party", type=int, default=3) result.add_argument("--seeds", help="explicit comma-separated seeds") result.add_argument("--duration", type=int, default=80) result.add_argument("--active-slot", type=int) result.add_argument("--seat-sweep", action="store_true", help="run every party with each of its five active-seat assignments") result.add_argument("--cooldown-policy", type=Path) result.add_argument("--companion-readiness-policy", type=Path) result.add_argument("--include-beta-specs", action="store_true") result.add_argument("--ready-only", action="store_true", help="reject Needs Testing targets and retain only Ready/Beta automatic pools") result.add_argument("--boss-category", choices=("all", "wow", "rom"), default="all") result.add_argument("--boss", action="append", default=[]) result.add_argument("--jobs", type=int, default=min(4, os.cpu_count() or 1), help="level processes to run concurrently (default: up to 4)") result.add_argument("--output-dir", type=Path) result.add_argument("--analyze-existing", type=Path, help="skip simulations and compare reports already under this directory") result.add_argument("--check", action="store_true", help="exit nonzero when a regression limit is exceeded") result.add_argument("--min-health-to-dps-seconds", type=float, default=20) result.add_argument("--max-health-to-dps-seconds", type=float, default=70) result.add_argument("--max-durability-growth", type=float, default=1.5) result.add_argument("--min-incoming-run-rate", type=float, default=0.95) result.add_argument("--min-healing-required-run-rate", type=float, default=0.8) result.add_argument("--min-healer-response-rate", type=float, default=0.6) return result def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) if args.spec_selection == "specific" and not args.spec: raise SystemExit("--spec-selection specific requires at least one --spec value") if args.jobs < 1: raise SystemExit("--jobs must be at least 1") timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") output_dir = (args.analyze_existing or args.output_dir or ROOT / "artifacts" / "balance-lab-brackets" / timestamp).resolve() output_dir.mkdir(parents=True, exist_ok=True) if args.analyze_existing is None: levels = args.levels or bracket_levels(args.step) print(f"Running {len(levels)} local brackets with up to {args.jobs} concurrent level processes...") failures = [] with ThreadPoolExecutor(max_workers=min(args.jobs, len(levels))) as executor: futures = { executor.submit(run_level, args, level, output_dir / f"level-{level}"): level for level in levels } for future in as_completed(futures): level, result = future.result() if result.returncode == 0: print(f" level {level}: complete") else: failures.append(level) print(f" level {level}: failed", file=sys.stderr) if result.stdout.strip(): print(result.stdout.rstrip(), file=sys.stderr) if result.stderr.strip(): print(result.stderr.rstrip(), file=sys.stderr) if failures: raise SystemExit(f"Balance Lab failed at levels: {', '.join(map(str, sorted(failures)))}") report_paths = discover_reports(output_dir) if not report_paths: raise SystemExit(f"No suite reports found under {output_dir}") rows = [summarize_report(path) for path in report_paths] add_growth_metrics(rows) passed = evaluate_rows(rows, args) serializable_rows = [{**row, "flags": list(row["flags"])} for row in rows] write_json(output_dir / "bracket-summary.json", { "schemaVersion": 1, "generatedAt": datetime.now().astimezone().isoformat(), "passed": passed, "levels": [row["level"] for row in rows], "thresholds": { "minHealthToDpsSeconds": args.min_health_to_dps_seconds, "maxHealthToDpsSeconds": args.max_health_to_dps_seconds, "maxDurabilityGrowth": args.max_durability_growth, "minIncomingRunRate": args.min_incoming_run_rate, "minHealingRequiredRunRate": args.min_healing_required_run_rate, "minHealerResponseRate": args.min_healer_response_rate, }, "brackets": serializable_rows, }) write_csv(output_dir / "bracket-summary.csv", [ {**row, "flags": "; ".join(row["flags"])} for row in rows ], SUMMARY_FIELDS) write_text(output_dir / "bracket-summary.md", markdown_report(rows)) heirlooms = collect_heirlooms(report_paths) if heirlooms: write_json(output_dir / "heirloom-summary.json", heirlooms) write_csv(output_dir / "heirloom-summary.csv", heirlooms, HEIRLOOM_FIELDS) print(f"Comparison: {output_dir / 'bracket-summary.md'}") if heirlooms: print(f"Resolved heirlooms: {output_dir / 'heirloom-summary.csv'}") return 0 if passed or not args.check else 1 if __name__ == "__main__": raise SystemExit(main())