Update Healer Man server

This commit is contained in:
phenom committed 2026-09-19 21:01:51 -04:00
1 parent f8314e7de1
commit 1fedb9368d
76 files changed
+20556 -2523

No files matched your search

+533
View File
@@ -0,0 +1,533 @@
#!/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")
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("--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())