#!/usr/bin/env python3 """Interactive controller and report generator for live Healer Man boss simulations.""" from __future__ import annotations import argparse import csv import html import json import math import os import random import re import statistics import subprocess import sys from collections import defaultdict from datetime import datetime from pathlib import Path from typing import Any, Iterable, Sequence ROOT = Path(__file__).resolve().parents[1] RUNTIME = ROOT / "scripts" / "balance-lab" / "runtime.ts" 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 vite_node_command() -> Path: suffix = ".cmd" if os.name == "nt" else "" executable = ROOT / "node_modules" / ".bin" / f"vite-node{suffix}" if not executable.exists(): raise SystemExit("vite-node is unavailable. Run 'npm install' in the HealerMan project first.") return executable class RuntimeSession: def __init__(self, source: str, level: int, workspace: Path) -> None: command = [ str(vite_node_command()), str(RUNTIME), "session", "--source", source, "--level", str(level), ] self.process = subprocess.Popen( command, cwd=ROOT, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, encoding="utf-8", ) assert self.process.stdout is not None catalog_line = self.process.stdout.readline() if not catalog_line: code = self.process.wait() raise SystemExit(code or "The live balance runtime ended before returning its catalog.") self.catalog = json.loads(catalog_line) write_json(workspace / "catalog.json", self.catalog) def simulate(self, config: Path, output: Path) -> None: if self.process.stdin is None or self.process.stdout is None: raise SystemExit("The live balance runtime has no input/output channel.") request = {"config": str(config), "output": str(output)} self.process.stdin.write(json.dumps(request, separators=(",", ":")) + "\n") self.process.stdin.flush() response_line = self.process.stdout.readline() code = self.process.poll() if code: raise SystemExit(code) if not response_line or not json.loads(response_line).get("ok"): raise SystemExit("The live balance runtime did not confirm completion.") def close(self) -> None: if self.process.poll() is not None: return if self.process.stdin is not None and self.process.stdout is not None: try: self.process.stdin.write('{"close":true}\n') self.process.stdin.flush() self.process.stdout.readline() except (BrokenPipeError, OSError): pass self.process.stdin.close() try: self.process.wait(timeout=5) except subprocess.TimeoutExpired: self.process.terminate() self.process.wait(timeout=5) def normalized(value: str) -> str: return "".join(character for character in value.casefold() if character.isalnum()) def console_text(value: str) -> str: return value.replace("—", "-").replace("…", "...").replace("×", "x") def build_aliases(build: dict[str, Any]) -> set[str]: class_id = build["classId"] class_name = build["className"] role = build["role"] specialization = build["specialization"] return { normalized(build["id"]), normalized(f"{class_id}:{specialization}"), normalized(f"{class_name}:{specialization}"), normalized(f"{class_id}:{role}:{specialization}"), normalized(f"{class_name}:{role}:{specialization}"), normalized(build["label"]), } def resolve_build(selector: str, builds: Sequence[dict[str, Any]]) -> dict[str, Any]: needle = normalized(selector) matches = [build for build in builds if needle in build_aliases(build)] if len(matches) == 1: return matches[0] if not matches: raise SystemExit(f"No party member matches '{selector}'. Use --list-members to see valid choices.") labels = ", ".join(match["label"] for match in matches) raise SystemExit(f"Party member selector '{selector}' is ambiguous: {labels}") def resolve_boss(selector: str, bosses: Sequence[dict[str, Any]]) -> dict[str, Any]: needle = normalized(selector) matches = [boss for boss in bosses if needle in { normalized(boss["key"]), normalized(boss["name"]), normalized(f"{boss['dungeonTitle']}:{boss['name']}"), }] if len(matches) == 1: return matches[0] if not matches: raise SystemExit(f"No boss matches '{selector}'. Use --list-bosses to see valid keys.") labels = ", ".join(f"{match['key']} ({match['name']})" for match in matches) raise SystemExit(f"Boss selector '{selector}' is ambiguous: {labels}") def random_party(builds: Sequence[dict[str, Any]], seed: int) -> list[dict[str, Any]]: generator = random.Random(seed) by_role: dict[str, list[dict[str, Any]]] = defaultdict(list) for build in builds: by_role[build["role"]].append(build) if not by_role["tank"] or not by_role["healer"] or len(by_role["damage"]) < 3: raise SystemExit("This source does not expose enough tank, healer, and damage builds for a 1/1/3 party.") damage = generator.sample(by_role["damage"], 3) return [generator.choice(by_role["tank"]), generator.choice(by_role["healer"]), *damage] def automatic_parties( builds: Sequence[dict[str, Any]], sources: Sequence[str], configurations_per_source: int, seed: int, ) -> list[dict[str, Any]]: offsets = {"wow": 100_000, "coa": 200_000, "rom": 300_000} configurations: list[dict[str, Any]] = [] for source in sources: source_builds = [build for build in builds if build["source"] == source] seen: set[tuple[str, str, tuple[str, ...]]] = set() attempts = 0 while len(seen) < configurations_per_source: attempts += 1 if attempts > max(1_000, configurations_per_source * 100): raise SystemExit( f"Could not generate {configurations_per_source} unique {source} parties. " "Request fewer party configurations." ) party = random_party(source_builds, seed + offsets[source] + attempts - 1) signature = (party[0]["id"], party[1]["id"], tuple(sorted(member["id"] for member in party[2:]))) if signature in seen: continue seen.add(signature) configurations.append({ "id": f"{source}-party-{len(seen):03d}", "source": source, "party": party, "activeSlot": next( (index for index, member in enumerate(party) if member["role"] == "healer"), 0, ), }) return configurations def coverage_parties( builds: Sequence[dict[str, Any]], sources: Sequence[str], selected_build_ids: Sequence[str] | None, seed: int, ) -> list[dict[str, Any]]: """Build the minimum deterministic party set that covers every requested build.""" requested = set(selected_build_ids) if selected_build_ids is not None else None offsets = {"wow": 100_000, "coa": 200_000, "rom": 300_000} configurations: list[dict[str, Any]] = [] for source in sources: source_builds = [build for build in builds if build["source"] == source] targets = source_builds if requested is None else [ build for build in source_builds if build["id"] in requested ] if not targets: continue available_by_role: dict[str, list[dict[str, Any]]] = defaultdict(list) targets_by_role: dict[str, list[dict[str, Any]]] = defaultdict(list) for build in source_builds: available_by_role[build["role"]].append(build) for build in targets: targets_by_role[build["role"]].append(build) if not available_by_role["tank"] or not available_by_role["healer"] or len(available_by_role["damage"]) < 3: raise SystemExit( f"The {source} catalog does not expose enough tank, healer, and damage builds for a 1/1/3 party." ) party_count = max( 1, len(targets_by_role["tank"]), len(targets_by_role["healer"]), math.ceil(len(targets_by_role["damage"]) / 3), ) generator = random.Random(seed + offsets[source]) for rows in (*available_by_role.values(), *targets_by_role.values()): rows.sort(key=lambda build: build["id"]) generator.shuffle(rows) def filler(role: str, party_index: int, slot_index: int, used: set[str]) -> dict[str, Any]: pool = available_by_role[role] start = (party_index * (3 if role == "damage" else 1) + slot_index) % len(pool) for avoid_requested in (True, False): for offset in range(len(pool)): candidate = pool[(start + offset) % len(pool)] if candidate["id"] in used: continue if avoid_requested and requested is not None and candidate["id"] in requested: continue return candidate raise SystemExit(f"Could not fill a unique {role} slot for a {source} coverage party.") for party_index in range(party_count): used: set[str] = set() tank = targets_by_role["tank"][party_index] if party_index < len(targets_by_role["tank"]) else filler( "tank", party_index, 0, used ) used.add(tank["id"]) healer = targets_by_role["healer"][party_index] if party_index < len(targets_by_role["healer"]) else filler( "healer", party_index, 0, used ) used.add(healer["id"]) damage = list(targets_by_role["damage"][party_index * 3:(party_index + 1) * 3]) used.update(build["id"] for build in damage) while len(damage) < 3: build = filler("damage", party_index, len(damage), used) damage.append(build) used.add(build["id"]) party = [tank, healer, *damage] configurations.append({ "id": f"{source}-party-{party_index + 1:03d}", "source": source, "party": party, "activeSlot": 1, }) if not configurations: raise SystemExit("No class/spec builds matched the requested coverage selection.") return configurations def search_builds(query: str, builds: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: needle = normalized(query) return [build for build in builds if needle in normalized( f"{build['id']} {build['label']} {build['source']} {build['role']}" )] def search_bosses(query: str, bosses: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: needle = normalized(query) return [boss for boss in bosses if needle in normalized( f"{boss['key']} {boss['name']} {boss['dungeonTitle']} {boss['category']}" )] def parse_seeds(value: str) -> list[int]: try: seeds = [int(entry.strip()) for entry in value.split(",") if entry.strip()] except ValueError as error: raise argparse.ArgumentTypeError("seeds must be comma-separated integers") from error if not seeds: raise argparse.ArgumentTypeError("at least one seed is required") return seeds def prompt_choice(prompt: str, options: Sequence[str], default: str) -> str: rendered = "/".join(options) while True: value = input(f"{prompt} [{rendered}] ({default}): ").strip().lower() or default if value in options: return value print(f"Choose one of: {', '.join(options)}") def prompt_integer(prompt: str, default: int, minimum: int, maximum: int) -> int: while True: raw = input(f"{prompt} ({default}): ").strip() if not raw: return default try: value = int(raw) except ValueError: print("Enter a whole number.") continue if minimum <= value <= maximum: return value print(f"Enter a value from {minimum} through {maximum}.") def prompt_specific_bosses(bosses: Sequence[dict[str, Any]]) -> list[str]: selected: dict[str, dict[str, Any]] = {} print("\nBoss picker: search by boss, dungeon, key, or source. Add selections across multiple searches.") while True: suffix = f" ({len(selected)} selected; type 'done' to finish)" if selected else "" query = input(f"Boss search{suffix}: ").strip() if query.casefold() == "done": if selected: return list(selected) print("Select at least one boss first.") continue matches = search_bosses(query, bosses) if not matches: print("No bosses matched that search.") continue if len(matches) > 50: print(f"{len(matches)} bosses matched. Use a more specific search (50 maximum per result page).") continue for index, boss in enumerate(matches, 1): marker = " [selected]" if boss["key"] in selected else "" print(f" {index:>2}. {boss['dungeonTitle']} - {boss['name']} ({boss['category']}){marker}") raw = input("Add numbers separated by commas, or 'all': ").strip().casefold() if raw == "all": indexes = list(range(len(matches))) else: try: indexes = [int(entry.strip()) - 1 for entry in raw.split(",") if entry.strip()] except ValueError: indexes = [] if not indexes or any(index < 0 or index >= len(matches) for index in indexes): print("Choose valid result numbers.") continue for index in indexes: selected[matches[index]["key"]] = matches[index] print(f"Selected bosses: {len(selected)}") def prompt_specific_builds(builds: Sequence[dict[str, Any]]) -> list[str]: selected: dict[str, dict[str, Any]] = {} print("\nClass/spec picker: search by class, specialization, role, source, or build ID.") while True: suffix = f" ({len(selected)} selected; type 'done' to finish)" if selected else "" query = input(f"Class/spec search{suffix}: ").strip() if query.casefold() == "done": if selected: return list(selected) print("Select at least one class/spec first.") continue matches = search_builds(query, builds) if not matches: print("No class/spec builds matched that search.") continue if len(matches) > 50: print(f"{len(matches)} builds matched. Use a more specific search (50 maximum per result page).") continue for index, build in enumerate(matches, 1): marker = " [selected]" if build["id"] in selected else "" print(f" {index:>2}. {console_text(build['label'])} [{build['source']}]{marker}") raw = input("Add numbers separated by commas, or 'all': ").strip().casefold() if raw == "all": indexes = list(range(len(matches))) else: try: indexes = [int(entry.strip()) - 1 for entry in raw.split(",") if entry.strip()] except ValueError: indexes = [] if not indexes or any(index < 0 or index >= len(matches) for index in indexes): print("Choose valid result numbers.") continue for index in indexes: selected[matches[index]["id"]] = matches[index] print(f"Selected class/spec builds: {len(selected)}") def interactive_arguments( args: argparse.Namespace, workspace: Path, ) -> tuple[argparse.Namespace, dict[str, Any], RuntimeSession]: print("Healer Man Balance Lab") print("The Python controller uses the game's live TypeScript combat runtime; it does not duplicate combat formulas.") args.party_mode = prompt_choice("Party selection", ("random", "manual", "automatic"), "automatic") source_options = ("wow", "coa", "rom", "all") if args.party_mode == "automatic" else ("wow", "coa", "rom") args.source = prompt_choice("Party source", source_options, "all" if args.party_mode == "automatic" else "wow") args.level = prompt_integer("Party level", 60, 1, 80) session = RuntimeSession(args.source, args.level, workspace) catalog = session.catalog if args.party_mode == "random": args.party_seed = prompt_integer("Random party seed", 1, -2_147_483_648, 2_147_483_647) elif args.party_mode == "automatic": args.spec_selection = prompt_choice("Class/spec selection", ("random", "all", "specific"), "all") if args.spec_selection == "random": args.party_configurations = prompt_integer("Unique party configurations per source", 10, 1, 1_000) elif args.spec_selection == "specific": args.spec = prompt_specific_builds(catalog["builds"]) args.party_seed = prompt_integer("Automatic party seed", 1, -2_147_483_648, 2_147_483_647) else: print("\nAvailable members:") for index, build in enumerate(catalog["builds"], 1): print(f" {index:>3}. {console_text(build['label'])}") while True: raw = input("Pick exactly five numbers, comma-separated: ").strip() try: indexes = [int(entry.strip()) - 1 for entry in raw.split(",") if entry.strip()] except ValueError: indexes = [] if len(indexes) == 5 and all(0 <= index < len(catalog["builds"]) for index in indexes): args.member = [catalog["builds"][index]["id"] for index in indexes] break print("Choose exactly five valid list numbers.") boss_scope = prompt_choice("Boss selection", ("all", "wow", "rom", "specific"), "all") if boss_scope == "specific": args.boss = prompt_specific_bosses(catalog["bosses"]) args.boss_category = "all" else: args.boss_category = boss_scope repetitions = prompt_integer("Simulations per party configuration (for each selected boss)", 3, 1, 100) args.simulations_per_party = repetitions args.seeds = list(range(1, repetitions + 1)) args.duration = prompt_integer("Maximum seconds per fight", 60, 5, 600) return args, catalog, session def weighted_rate(rows: Iterable[dict[str, Any]], amount: str) -> float: materialized = list(rows) seconds = sum(float(row["elapsedSeconds"]) for row in materialized) return sum(float(row.get(amount, 0)) for row in materialized) / max(0.1, seconds) def mean(values: Iterable[float]) -> float: materialized = list(values) return sum(materialized) / len(materialized) if materialized else 0.0 def tuning_label(ratio: float | None) -> str: if ratio is None or not math.isfinite(ratio): return "no baseline" if ratio >= 1.5: return "very high" if ratio >= 1.25: return "high" if ratio <= 0.5: return "very low" if ratio <= 0.75: return "low" return "baseline" def aggregate_boss_runs(runs: Sequence[dict[str, Any]]) -> tuple[list[dict[str, Any]], float]: by_boss: dict[str, list[dict[str, Any]]] = defaultdict(list) for run in runs: by_boss[run["bossKey"]].append(run) bosses = [] for boss_runs in by_boss.values(): first = boss_runs[0] defeated = [run for run in boss_runs if run["outcome"] == "defeated"] bosses.append({ "bossKey": first["bossKey"], "bossName": first["bossName"], "category": first["bossCategory"], "dungeonId": first["dungeonId"], "dungeonTitle": first["dungeonTitle"], "runs": len(boss_runs), "incomingDtps": weighted_rate(boss_runs, "incomingDamage"), "averageIncomingDamage": mean(run["incomingDamage"] for run in boss_runs), "partyDps": weighted_rate(boss_runs, "partyDamage"), "bossDpsTaken": weighted_rate(boss_runs, "bossDamage"), "defeatRate": len(defeated) / len(boss_runs), "wipeRate": sum(run["outcome"] == "wipe" for run in boss_runs) / len(boss_runs), "averageDeaths": mean(run["deaths"] for run in boss_runs), "averageTimeToDefeat": mean(run["timeToDefeatSeconds"] for run in defeated) if defeated else None, "bossMaxHealth": mean(run["bossMaxHealth"] for run in boss_runs), }) nonzero_pressure = [boss["incomingDtps"] for boss in bosses if boss["incomingDtps"] > 0] pressure_median = statistics.median(nonzero_pressure) if nonzero_pressure else 0.0 for boss in bosses: ratio = boss["incomingDtps"] / pressure_median if pressure_median > 0 else None boss["pressureVsMedian"] = ratio boss["pressureFlag"] = tuning_label(ratio) bosses.sort(key=lambda row: (-row["incomingDtps"], row["bossName"], row["dungeonTitle"])) return bosses, pressure_median def aggregate(raw: dict[str, Any]) -> dict[str, Any]: runs: list[dict[str, Any]] = raw["runs"] by_slot: dict[int, list[dict[str, Any]]] = defaultdict(list) by_member_boss: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list) for run in runs: for member in run["members"]: enriched = {**member, "elapsedSeconds": run["elapsedSeconds"], "bossKey": run["bossKey"]} by_slot[int(member["slot"])].append(enriched) by_member_boss[(int(member["slot"]), run["bossKey"])].append(enriched) bosses, pressure_median = aggregate_boss_runs(runs) members = [] for slot, rows in by_slot.items(): first = rows[0] members.append({ "slot": slot + 1, "activePlayer": first["activePlayer"], "classId": first["classId"], "className": first["className"], "specialization": first["specialization"], "role": first["role"], "dps": weighted_rate(rows, "damage"), "bossDps": weighted_rate(rows, "bossDamage"), "hps": weighted_rate(rows, "healing"), "hpsReceived": weighted_rate(rows, "healingReceived"), "dtps": weighted_rate(rows, "damageTaken"), "totalDamage": sum(row["damage"] for row in rows), "totalHealing": sum(row["healing"] for row in rows), "totalHealingReceived": sum(row.get("healingReceived", 0) for row in rows), "totalAbsorbs": sum(row["absorbs"] for row in rows), "deathEvents": sum(row["deathEvents"] for row in rows), "minimumHealthPercent": min(row["minimumHealthPercent"] for row in rows), }) role_medians = { role: statistics.median(member["dps"] for member in members if member["role"] == role) for role in {member["role"] for member in members} } for member in members: baseline = role_medians[member["role"]] ratio = member["dps"] / baseline if baseline > 0 else None member["damageVsRoleMedian"] = ratio member["damageFlag"] = tuning_label(ratio) members.sort(key=lambda row: (-row["dps"], row["slot"])) member_boss = [] for (slot, boss_key), rows in by_member_boss.items(): first = rows[0] boss = next(run for run in runs if run["bossKey"] == boss_key) member_boss.append({ "bossKey": boss_key, "bossName": boss["bossName"], "dungeonTitle": boss["dungeonTitle"], "slot": slot + 1, "className": first["className"], "specialization": first["specialization"], "role": first["role"], "dps": weighted_rate(rows, "damage"), "bossDps": weighted_rate(rows, "bossDamage"), "hps": weighted_rate(rows, "healing"), "hpsReceived": weighted_rate(rows, "healingReceived"), "dtps": weighted_rate(rows, "damageTaken"), "deathEvents": sum(row["deathEvents"] for row in rows), }) member_boss.sort(key=lambda row: (row["bossName"], row["dungeonTitle"], row["slot"])) return { "schemaVersion": 1, "generatedAt": raw["generatedAt"], "engine": raw["engine"], "policyId": raw["policyId"], "rotationVersion": raw["rotationVersion"], "config": raw["config"], "party": raw["party"], "summary": { "bosses": len(bosses), "runs": len(runs), "bossPressureMedianDtps": pressure_median, }, "bosses": bosses, "members": members, "memberByBoss": member_boss, } def format_number(value: Any, digits: int = 1) -> str: if value is None: return "—" return f"{float(value):,.{digits}f}" def format_ratio(value: Any) -> str: return "—" if value is None else f"{format_number(value, 2)}×" 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 flattened_abilities(raw: dict[str, Any], key: str) -> list[dict[str, Any]]: rows = [] for run in raw["runs"]: if key == "members": for member in run["members"]: for ability in member["abilities"]: rows.append({ "bossKey": run["bossKey"], "bossName": run["bossName"], "seed": run["seed"], "slot": member["slot"] + 1, "className": member["className"], "specialization": member["specialization"], **ability, }) else: for ability in run["bossAbilities"]: rows.append({ "bossKey": run["bossKey"], "bossName": run["bossName"], "dungeonTitle": run["dungeonTitle"], "seed": run["seed"], **ability, }) return rows def markdown_report(report: dict[str, Any]) -> str: config = report["config"] lines = [ "# Healer Man balance-lab report", "", f"Generated: {report['generatedAt']}", "", f"Policy: `{report['policyId']}` · rotation: `{report['rotationVersion']}`", "", f"Party source: **{config['source']}** · level **{config['level']}** · " f"{report['summary']['bosses']} bosses · {report['summary']['runs']} simulations", "", "Pressure flags compare each boss's effective party DTPS with the non-zero median for this exact party. " "Damage flags compare a member only with other selected members in the same role.", "", "## Every party member", "", "| Slot | Member | Role | DPS | Boss DPS | HPS done | HPS received | DTPS | Role median | Flag | Death events |", "| ---: | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- | ---: |", ] for member in report["members"]: marker = " (active)" if member["activePlayer"] else "" lines.append( f"| {member['slot']} | {member['className']} — {member['specialization']}{marker} | " f"{member['role']} | {format_number(member['dps'])} | {format_number(member['bossDps'])} | " f"{format_number(member['hps'])} | {format_number(member['hpsReceived'])} | " f"{format_number(member['dtps'])} | " f"{format_ratio(member['damageVsRoleMedian'])} | {member['damageFlag']} | {member['deathEvents']} |" ) lines.extend([ "", "## Every boss, highest party pressure first", "", "| Boss | Dungeon | Source | DTPS | Median | Flag | Avg total damage | Party DPS | Defeat | Wipe | Avg deaths | Avg TTK |", "| --- | --- | --- | ---: | ---: | --- | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for boss in report["bosses"]: lines.append( f"| {boss['bossName']} | {boss['dungeonTitle']} | {boss['category']} | " f"{format_number(boss['incomingDtps'])} | {format_ratio(boss['pressureVsMedian'])} | " f"{boss['pressureFlag']} | {format_number(boss['averageIncomingDamage'])} | " f"{format_number(boss['partyDps'])} | {format_number(boss['defeatRate'] * 100, 0)}% | " f"{format_number(boss['wipeRate'] * 100, 0)}% | {format_number(boss['averageDeaths'], 2)} | " f"{format_number(boss['averageTimeToDefeat'])} |" ) lines.extend([ "", "## Reading the results", "", "- `incomingDtps` measures effective damage received by all five members per active fight second, including adds.", "- `averageIncomingDamage` includes fight length, so it exposes healer workload that DTPS alone can hide.", "- `partyDps` includes boss and add damage; `bossDps` excludes add damage.", "- A flag is a screening signal, not an automatic balance verdict. Confirm outliers with the per-boss, ability, wipe, and TTK CSVs.", "- Headless runs execute combat and authored attack effects, but do not render visual/audio effects or mount scene-only encounter objects.", ]) return "\n".join(lines) def html_table(identifier: str, title: str, rows: Sequence[dict[str, Any]], columns: Sequence[tuple[str, str]]) -> str: headers = "".join(f"{html.escape(label)}" for index, (_, label) in enumerate(columns)) body = [] for row in rows: cells = "".join(f"{html.escape(str(row.get(field, '')))}" for field, _ in columns) body.append(f"{cells}") return ( f"

{html.escape(title)}

" f"" f"
{headers}" f"{''.join(body)}
" ) def html_report(report: dict[str, Any]) -> str: bosses = [{ **row, "incomingDtps": format_number(row["incomingDtps"]), "pressureVsMedian": format_number(row["pressureVsMedian"], 2), "averageIncomingDamage": format_number(row["averageIncomingDamage"]), "partyDps": format_number(row["partyDps"]), "defeatRate": format_number(row["defeatRate"] * 100, 0) + "%", "wipeRate": format_number(row["wipeRate"] * 100, 0) + "%", "averageDeaths": format_number(row["averageDeaths"], 2), "averageTimeToDefeat": format_number(row["averageTimeToDefeat"]), } for row in report["bosses"]] members = [{ **row, "member": f"{row['className']} — {row['specialization']}", "dps": format_number(row["dps"]), "bossDps": format_number(row["bossDps"]), "hps": format_number(row["hps"]), "hpsReceived": format_number(row["hpsReceived"]), "dtps": format_number(row["dtps"]), "damageVsRoleMedian": format_number(row["damageVsRoleMedian"], 2), } for row in report["members"]] member_table = html_table("members", "Every party member", members, ( ("slot", "Slot"), ("member", "Member"), ("role", "Role"), ("dps", "DPS"), ("bossDps", "Boss DPS"), ("hps", "HPS done"), ("hpsReceived", "HPS received"), ("dtps", "DTPS"), ("damageVsRoleMedian", "Role median ×"), ("damageFlag", "Flag"), ("deathEvents", "Death events"), )) boss_table = html_table("bosses", "Every boss — highest pressure first", bosses, ( ("bossName", "Boss"), ("dungeonTitle", "Dungeon"), ("category", "Source"), ("incomingDtps", "Party DTPS"), ("pressureVsMedian", "Median ×"), ("pressureFlag", "Flag"), ("averageIncomingDamage", "Avg total damage"), ("partyDps", "Party DPS"), ("defeatRate", "Defeat"), ("wipeRate", "Wipe"), ("averageDeaths", "Avg deaths"), ("averageTimeToDefeat", "Avg TTK"), )) return f""" Healer Man Balance Lab

Healer Man Balance Lab

{html.escape(report['generatedAt'])} · policy {html.escape(report['policyId'])} · rotation {html.escape(report['rotationVersion'])} · level {report['config']['level']} {html.escape(report['config']['source'])} party · {report['summary']['bosses']} bosses · {report['summary']['runs']} simulations. Click a heading to sort; filter boxes keep every row accessible.

{member_table}{boss_table} """ def create_reports(raw_path: Path, output_dir: Path) -> dict[str, Any]: raw = json.loads(raw_path.read_text(encoding="utf-8")) report = aggregate(raw) write_json(output_dir / "report.json", report) write_text(output_dir / "report.md", markdown_report(report)) write_text(output_dir / "report.html", html_report(report)) write_csv(output_dir / "bosses.csv", report["bosses"], ( "bossKey", "bossName", "category", "dungeonId", "dungeonTitle", "runs", "incomingDtps", "pressureVsMedian", "pressureFlag", "averageIncomingDamage", "partyDps", "bossDpsTaken", "defeatRate", "wipeRate", "averageDeaths", "averageTimeToDefeat", "bossMaxHealth", )) write_csv(output_dir / "party_members.csv", report["members"], ( "slot", "activePlayer", "classId", "className", "specialization", "role", "dps", "bossDps", "hps", "hpsReceived", "dtps", "damageVsRoleMedian", "damageFlag", "totalDamage", "totalHealing", "totalHealingReceived", "totalAbsorbs", "deathEvents", "minimumHealthPercent", )) write_csv(output_dir / "member_by_boss.csv", report["memberByBoss"], ( "bossKey", "bossName", "dungeonTitle", "slot", "className", "specialization", "role", "dps", "bossDps", "hps", "hpsReceived", "dtps", "deathEvents", )) run_rows = [{ "policyId": raw["policyId"], "rotationVersion": raw["rotationVersion"], **{key: value for key, value in run.items() if key not in {"members", "bossAbilities"}}, } for run in raw["runs"]] write_csv(output_dir / "runs.csv", run_rows, ( "policyId", "rotationVersion", "bossKey", "bossName", "bossCategory", "dungeonId", "dungeonTitle", "seed", "outcome", "elapsedSeconds", "timeToDefeatSeconds", "bossLevel", "bossMaxHealth", "bossRemainingHealth", "partyDamage", "bossDamage", "partyDps", "bossDpsTaken", "incomingDamage", "rawIncomingDamage", "incomingDtps", "deaths", "deathEvents", "bossAttacks", )) member_abilities = flattened_abilities(raw, "members") write_csv(output_dir / "member_abilities.csv", member_abilities, ( "bossKey", "bossName", "seed", "slot", "className", "specialization", "abilityId", "casts", "rawDamage", "effectiveDamage", "rawHealing", "effectiveHealing", "overheal", "absorbs", "hits", )) boss_abilities = flattened_abilities(raw, "bosses") write_csv(output_dir / "boss_abilities.csv", boss_abilities, ( "bossKey", "bossName", "dungeonTitle", "seed", "abilityId", "casts", "rawDamage", "effectiveDamage", "rawHealing", "effectiveHealing", "overheal", "absorbs", "hits", )) return report def aggregate_class_specs( groups: dict[tuple[str, ...], list[dict[str, Any]]], ) -> list[dict[str, Any]]: rows = [] for key, observations in groups.items(): source, build_id, class_id, class_name, specialization, role = key rows.append({ "source": source, "buildId": build_id, "classId": class_id, "className": class_name, "specialization": specialization, "role": role, "partyConfigurations": len({row["partyConfigurationId"] for row in observations}), "runs": len(observations), "dps": weighted_rate(observations, "damage"), "bossDps": weighted_rate(observations, "bossDamage"), "hps": weighted_rate(observations, "healing"), "hpsReceived": weighted_rate(observations, "healingReceived"), "dtps": weighted_rate(observations, "damageTaken"), "totalDamage": sum(row["damage"] for row in observations), "totalDamageTaken": sum(row["damageTaken"] for row in observations), "totalHealing": sum(row["healing"] for row in observations), "totalHealingReceived": sum(row.get("healingReceived", 0) for row in observations), "totalAbsorbs": sum(row["absorbs"] for row in observations), "deathEvents": sum(row["deathEvents"] for row in observations), }) baselines = { (source, role): statistics.median( row["dps"] for row in rows if row["source"] == source and row["role"] == role ) for source, role in {(row["source"], row["role"]) for row in rows} } for row in rows: baseline = baselines[(row["source"], row["role"])] ratio = row["dps"] / baseline if baseline > 0 else None row["damageVsSourceRoleMedian"] = ratio row["damageFlag"] = tuning_label(ratio) rows.sort(key=lambda row: (row["source"], row["role"], -row["dps"], row["className"], row["specialization"])) return rows def humanize_ability_id(ability_id: str) -> str: if ability_id == "party-auto-attack": return "Auto Attack" if ability_id == "unknown": return "Unknown effect" words = re.sub(r"^(?:wow335|coa|rom)-", "", ability_id) words = re.sub(r"^[a-z]+-", "", words) words = re.sub(r"(?:-\d+)+$", "", words) return " ".join(word.capitalize() for word in words.split("-") if word) or ability_id def aggregate_class_abilities( groups: dict[tuple[str, ...], list[dict[str, Any]]], class_groups: dict[tuple[str, ...], list[dict[str, Any]]], ) -> list[dict[str, Any]]: rows = [] for key, observations in groups.items(): source, build_id, class_id, class_name, specialization, role, ability_id = key class_observations = class_groups[(source, build_id, class_id, class_name, specialization, role)] elapsed_seconds = sum(float(row["elapsedSeconds"]) for row in class_observations) effective_damage = sum(float(row["effectiveDamage"]) for row in observations) hits = sum(int(row["hits"]) for row in observations) casts = sum(int(row["casts"]) for row in observations) active_runs = sum(bool(row["used"]) for row in observations) available_runs = sum(bool(row["available"]) for row in observations) class_damage = sum(float(row["damage"]) for row in class_observations) metadata = next((row for row in observations if row["available"]), observations[0]) effective_healing = sum(float(row["effectiveHealing"]) for row in observations) absorbs = sum(float(row["absorbs"]) for row in observations) if active_runs == 0 and available_runs > 0: status = "unused" elif effective_damage > 0: status = "damage" elif effective_healing > 0 or absorbs > 0: status = "support" else: status = "used" rows.append({ "source": source, "buildId": build_id, "classId": class_id, "className": class_name, "specialization": specialization, "role": role, "abilityId": ability_id, "abilityName": metadata.get("abilityName") or humanize_ability_id(ability_id), "icon": metadata.get("icon") or "", "categories": ", ".join(metadata.get("categories") or []), "cooldownSeconds": metadata.get("cooldownSeconds"), "partyConfigurations": len({row["partyConfigurationId"] for row in class_observations}), "bosses": len({row["bossKey"] for row in class_observations}), "runs": len(class_observations), "availableRuns": available_runs, "activeRuns": active_runs, "usageRate": active_runs / len(class_observations) if class_observations else 0, "casts": casts, "castsPerMinute": casts / max(0.1, elapsed_seconds) * 60, "hits": hits, "hitsPerRun": hits / len(class_observations) if class_observations else 0, "rawDamage": sum(float(row["rawDamage"]) for row in observations), "effectiveDamage": effective_damage, "dps": effective_damage / max(0.1, elapsed_seconds), "damageShare": effective_damage / class_damage if class_damage > 0 else 0, "averageDamagePerHit": effective_damage / hits if hits > 0 else 0, "rawHealing": sum(float(row["rawHealing"]) for row in observations), "effectiveHealing": effective_healing, "overheal": sum(float(row["overheal"]) for row in observations), "absorbs": absorbs, "status": status, "unusedAvailable": status == "unused" and available_runs > 0, }) rows.sort(key=lambda row: ( row["source"], row["className"], row["specialization"], row["unusedAvailable"], -row["effectiveDamage"], row["abilityName"], )) return rows def suite_markdown(report: dict[str, Any]) -> str: summary = report["summary"] lines = [ "# Healer Man automatic balance suite", "", f"Generated: {report['generatedAt']}", "", f"Policy: `{report['policyId']}` · rotation: `{report['rotationVersion']}`", "", f"**{summary['partyConfigurations']} party configurations**, **{summary['bosses']} bosses**, " f"and **{summary['runs']} simulations** at level **{report['level']}**.", "", "Each configuration has a complete report under `parties//`. The suite tables retain every " "tested configuration, class/spec, and boss.", "", "## Every tested class/spec", "", "| Source | Class/spec | Role | Parties | Runs | DPS | Boss DPS | HPS done | HPS received | DTPS | Total taken | Role median | Flag | Deaths |", "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- | ---: |", ] for row in report["classSpecs"]: lines.append( f"| {row['source']} | {row['className']} — {row['specialization']} | {row['role']} | " f"{row['partyConfigurations']} | {row['runs']} | {format_number(row['dps'])} | " f"{format_number(row['bossDps'])} | {format_number(row['hps'])} | " f"{format_number(row['hpsReceived'])} | {format_number(row['dtps'])} | " f"{format_number(row['totalDamageTaken'])} | " f"{format_ratio(row['damageVsSourceRoleMedian'])} | {row['damageFlag']} | {row['deathEvents']} |" ) lines.extend([ "", "## Every class/spec ability", "", "Available abilities with no observed cast, hit, damage, healing, or absorb event are retained as `unused`.", "", "| Source | Class/spec | Ability | AI category | Status | Active runs | Casts | Hits | Effective damage | DPS | Damage share | Healing | Overheal | Absorbs |", "| --- | --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for row in report["classAbilities"]: lines.append( f"| {row['source']} | {row['className']} — {row['specialization']} | " f"{row['abilityName']} (`{row['abilityId']}`) | {row['categories'] or '—'} | {row['status']} | " f"{row['activeRuns']}/{row['runs']} | {row['casts']} | {row['hits']} | " f"{format_number(row['effectiveDamage'])} | {format_number(row['dps'])} | " f"{format_number(row['damageShare'] * 100, 1)}% | {format_number(row['effectiveHealing'])} | " f"{format_number(row['overheal'])} | {format_number(row['absorbs'])} |" ) lines.extend([ "", "## Every party configuration", "", "| Party | Source | Members | Runs | Party DPS | DTPS | Defeat | Wipe | Avg deaths | Report |", "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |", ]) for row in report["partyConfigurations"]: lines.append( f"| {row['partyConfigurationId']} | {row['source']} | {row['members']} | {row['runs']} | " f"{format_number(row['partyDps'])} | {format_number(row['incomingDtps'])} | " f"{format_number(row['defeatRate'] * 100, 0)}% | {format_number(row['wipeRate'] * 100, 0)}% | " f"{format_number(row['averageDeaths'], 2)} | {row['reportPath']} |" ) lines.extend([ "", "## Every boss across all parties", "", "| Boss | Dungeon | Source | Runs | DTPS | Median | Flag | Avg total damage | Party DPS | Defeat | Wipe | Avg deaths |", "| --- | --- | --- | ---: | ---: | ---: | --- | ---: | ---: | ---: | ---: | ---: |", ]) for row in report["bosses"]: lines.append( f"| {row['bossName']} | {row['dungeonTitle']} | {row['category']} | {row['runs']} | " f"{format_number(row['incomingDtps'])} | {format_ratio(row['pressureVsMedian'])} | " f"{row['pressureFlag']} | {format_number(row['averageIncomingDamage'])} | " f"{format_number(row['partyDps'])} | {format_number(row['defeatRate'] * 100, 0)}% | " f"{format_number(row['wipeRate'] * 100, 0)}% | {format_number(row['averageDeaths'], 2)} |" ) return "\n".join(lines) def suite_html(report: dict[str, Any]) -> str: class_specs = [{ **row, "classSpec": f"{row['className']} — {row['specialization']}", "dps": format_number(row["dps"]), "bossDps": format_number(row["bossDps"]), "hps": format_number(row["hps"]), "hpsReceived": format_number(row["hpsReceived"]), "dtps": format_number(row["dtps"]), "totalDamageTaken": format_number(row["totalDamageTaken"]), "damageVsSourceRoleMedian": format_ratio(row["damageVsSourceRoleMedian"]), } for row in report["classSpecs"]] class_abilities = [{ **row, "classSpec": f"{row['className']} — {row['specialization']}", "activeRuns": f"{row['activeRuns']}/{row['runs']}", "castsPerMinute": format_number(row["castsPerMinute"]), "effectiveDamage": format_number(row["effectiveDamage"]), "dps": format_number(row["dps"]), "damageShare": format_number(row["damageShare"] * 100, 1) + "%", "averageDamagePerHit": format_number(row["averageDamagePerHit"]), "effectiveHealing": format_number(row["effectiveHealing"]), "overheal": format_number(row["overheal"]), "absorbs": format_number(row["absorbs"]), } for row in report["classAbilities"]] parties = [{ **row, "partyDps": format_number(row["partyDps"]), "incomingDtps": format_number(row["incomingDtps"]), "defeatRate": format_number(row["defeatRate"] * 100, 0) + "%", "wipeRate": format_number(row["wipeRate"] * 100, 0) + "%", "averageDeaths": format_number(row["averageDeaths"], 2), } for row in report["partyConfigurations"]] bosses = [{ **row, "incomingDtps": format_number(row["incomingDtps"]), "pressureVsMedian": format_ratio(row["pressureVsMedian"]), "averageIncomingDamage": format_number(row["averageIncomingDamage"]), "partyDps": format_number(row["partyDps"]), "defeatRate": format_number(row["defeatRate"] * 100, 0) + "%", "wipeRate": format_number(row["wipeRate"] * 100, 0) + "%", "averageDeaths": format_number(row["averageDeaths"], 2), } for row in report["bosses"]] class_table = html_table("class-specs", "Every tested class/spec", class_specs, ( ("source", "Source"), ("classSpec", "Class/spec"), ("role", "Role"), ("partyConfigurations", "Parties"), ("runs", "Runs"), ("dps", "DPS"), ("bossDps", "Boss DPS"), ("hps", "HPS done"), ("hpsReceived", "HPS received"), ("dtps", "DTPS"), ("totalDamageTaken", "Total taken"), ("damageVsSourceRoleMedian", "Role median"), ("damageFlag", "Flag"), ("deathEvents", "Deaths"), )) ability_table = html_table("class-abilities", "Every class/spec ability", class_abilities, ( ("source", "Source"), ("classSpec", "Class/spec"), ("abilityName", "Ability"), ("abilityId", "Exact ID"), ("categories", "AI category"), ("status", "Status"), ("activeRuns", "Active runs"), ("casts", "Casts"), ("castsPerMinute", "Casts/min"), ("hits", "Hits"), ("effectiveDamage", "Damage"), ("dps", "DPS"), ("damageShare", "Damage share"), ("averageDamagePerHit", "Damage/hit"), ("effectiveHealing", "Healing"), ("overheal", "Overheal"), ("absorbs", "Absorbs"), )) party_table = html_table("parties", "Every party configuration", parties, ( ("partyConfigurationId", "Party"), ("source", "Source"), ("members", "Members"), ("runs", "Runs"), ("partyDps", "Party DPS"), ("incomingDtps", "DTPS"), ("defeatRate", "Defeat"), ("wipeRate", "Wipe"), ("averageDeaths", "Avg deaths"), ("reportPath", "Detailed report"), )) boss_table = html_table("suite-bosses", "Every boss across all parties", bosses, ( ("bossName", "Boss"), ("dungeonTitle", "Dungeon"), ("category", "Source"), ("runs", "Runs"), ("incomingDtps", "DTPS"), ("pressureVsMedian", "Median"), ("pressureFlag", "Flag"), ("averageIncomingDamage", "Avg total damage"), ("partyDps", "Party DPS"), ("defeatRate", "Defeat"), ("wipeRate", "Wipe"), ("averageDeaths", "Avg deaths"), )) summary = report["summary"] return f""" Healer Man Automatic Balance Suite

Healer Man Automatic Balance Suite

{html.escape(report['generatedAt'])} · policy {html.escape(report['policyId'])} · rotation {html.escape(report['rotationVersion'])} · level {report['level']} · {summary['partyConfigurations']} parties · {summary['bosses']} bosses · {summary['runs']} simulations. Every row is retained; click headings to sort or filter each table.

{class_table}{ability_table}{party_table}{boss_table} """ def create_suite_reports(entries: Sequence[dict[str, Any]], output_dir: Path, level: int) -> dict[str, Any]: compact_runs: list[dict[str, Any]] = [] party_rows: list[dict[str, Any]] = [] boss_by_party: list[dict[str, Any]] = [] class_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) class_boss_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) class_ability_groups: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list) policy_ids: set[str] = set() rotation_versions: set[str] = set() for entry in entries: raw = json.loads(Path(entry["rawPath"]).read_text(encoding="utf-8")) policy_ids.add(raw["policyId"]) rotation_versions.add(raw["rotationVersion"]) report = entry["report"] runs = raw["runs"] compact_runs.extend({ "partyConfigurationId": entry["id"], "partySource": entry["source"], "policyId": raw["policyId"], "rotationVersion": raw["rotationVersion"], **{key: value for key, value in run.items() if key not in {"members", "bossAbilities"}}, } for run in runs) defeated = [run for run in runs if run["outcome"] == "defeated"] members = "; ".join( f"{member['className']} {member['specialization']} ({member['role']})" for member in raw["party"] ) party_rows.append({ "partyConfigurationId": entry["id"], "source": entry["source"], "members": members, "runs": len(runs), "partyDps": weighted_rate(runs, "partyDamage"), "incomingDtps": weighted_rate(runs, "incomingDamage"), "defeatRate": len(defeated) / len(runs), "wipeRate": sum(run["outcome"] == "wipe" for run in runs) / len(runs), "averageDeaths": mean(run["deaths"] for run in runs), "reportPath": (Path("parties") / entry["id"] / "report.html").as_posix(), }) for boss in report["bosses"]: boss_by_party.append({"partyConfigurationId": entry["id"], "partySource": entry["source"], **boss}) for run in runs: for member in run["members"]: observation = { **member, "source": entry["source"], "partyConfigurationId": entry["id"], "elapsedSeconds": run["elapsedSeconds"], "bossKey": run["bossKey"], "bossName": run["bossName"], "dungeonTitle": run["dungeonTitle"], } key = ( entry["source"], member["buildId"], member["classId"], member["className"], member["specialization"], member["role"], ) class_groups[key].append(observation) class_boss_groups[(*key, run["bossKey"], run["bossName"], run["dungeonTitle"])].append(observation) available = { ability["abilityId"]: ability for ability in member.get("availableAbilities", []) } observed = {ability["abilityId"]: ability for ability in member.get("abilities", [])} for ability_id in sorted((set(available) | set(observed)) - {"balance-lab-pull"}): expected = available.get(ability_id, {}) metrics = observed.get(ability_id, {}) used = any(float(metrics.get(field, 0)) > 0 for field in ( "casts", "hits", "rawDamage", "effectiveDamage", "rawHealing", "effectiveHealing", "overheal", "absorbs", )) class_ability_groups[(*key, ability_id)].append({ "partyConfigurationId": entry["id"], "bossKey": run["bossKey"], "elapsedSeconds": run["elapsedSeconds"], "abilityName": expected.get("abilityName"), "icon": expected.get("icon"), "categories": expected.get("categories", []), "cooldownSeconds": expected.get("cooldownSeconds"), "available": ability_id in available, "used": used, "casts": metrics.get("casts", 0), "hits": metrics.get("hits", 0), "rawDamage": metrics.get("rawDamage", 0), "effectiveDamage": metrics.get("effectiveDamage", 0), "rawHealing": metrics.get("rawHealing", 0), "effectiveHealing": metrics.get("effectiveHealing", 0), "overheal": metrics.get("overheal", 0), "absorbs": metrics.get("absorbs", 0), }) bosses, pressure_median = aggregate_boss_runs(compact_runs) class_specs = aggregate_class_specs(class_groups) class_abilities = aggregate_class_abilities(class_ability_groups, class_groups) class_spec_by_boss = [] for key, observations in class_boss_groups.items(): source, build_id, class_id, class_name, specialization, role, boss_key, boss_name, dungeon_title = key class_spec_by_boss.append({ "source": source, "buildId": build_id, "classId": class_id, "className": class_name, "specialization": specialization, "role": role, "bossKey": boss_key, "bossName": boss_name, "dungeonTitle": dungeon_title, "partyConfigurations": len({row["partyConfigurationId"] for row in observations}), "runs": len(observations), "dps": weighted_rate(observations, "damage"), "bossDps": weighted_rate(observations, "bossDamage"), "hps": weighted_rate(observations, "healing"), "hpsReceived": weighted_rate(observations, "healingReceived"), "dtps": weighted_rate(observations, "damageTaken"), "totalDamageTaken": sum(row["damageTaken"] for row in observations), "deathEvents": sum(row["deathEvents"] for row in observations), }) class_spec_by_boss.sort(key=lambda row: ( row["source"], row["className"], row["specialization"], row["bossName"], row["dungeonTitle"] )) party_rows.sort(key=lambda row: (row["source"], row["partyConfigurationId"])) boss_by_party.sort(key=lambda row: (row["partyConfigurationId"], -row["incomingDtps"], row["bossName"])) if len(policy_ids) != 1 or len(rotation_versions) != 1: raise RuntimeError("Balance Lab suite mixed combat policy or rotation versions") report = { "schemaVersion": 1, "generatedAt": datetime.now().astimezone().isoformat(), "engine": "live-typescript-runtime", "policyId": next(iter(policy_ids)), "rotationVersion": next(iter(rotation_versions)), "level": level, "summary": { "partyConfigurations": len(entries), "bosses": len(bosses), "runs": len(compact_runs), "classSpecs": len(class_specs), "classAbilities": len(class_abilities), "bossPressureMedianDtps": pressure_median, }, "classSpecs": class_specs, "classAbilities": class_abilities, "partyConfigurations": party_rows, "bosses": bosses, "bossByParty": boss_by_party, "classSpecByBoss": class_spec_by_boss, } write_json(output_dir / "suite-report.json", report) write_text(output_dir / "suite-report.md", suite_markdown(report)) write_text(output_dir / "suite-report.html", suite_html(report)) write_csv(output_dir / "party_configurations.csv", party_rows, ( "partyConfigurationId", "source", "members", "runs", "partyDps", "incomingDtps", "defeatRate", "wipeRate", "averageDeaths", "reportPath", )) write_csv(output_dir / "class_specs_across_parties.csv", class_specs, ( "source", "buildId", "classId", "className", "specialization", "role", "partyConfigurations", "runs", "dps", "bossDps", "hps", "hpsReceived", "dtps", "damageVsSourceRoleMedian", "damageFlag", "totalDamage", "totalDamageTaken", "totalHealing", "totalHealingReceived", "totalAbsorbs", "deathEvents", )) write_csv(output_dir / "class_abilities_across_parties.csv", class_abilities, ( "source", "buildId", "classId", "className", "specialization", "role", "abilityId", "abilityName", "categories", "cooldownSeconds", "partyConfigurations", "bosses", "runs", "availableRuns", "activeRuns", "usageRate", "casts", "castsPerMinute", "hits", "hitsPerRun", "rawDamage", "effectiveDamage", "dps", "damageShare", "averageDamagePerHit", "rawHealing", "effectiveHealing", "overheal", "absorbs", "status", "unusedAvailable", )) write_csv(output_dir / "bosses_across_parties.csv", bosses, ( "bossKey", "bossName", "category", "dungeonId", "dungeonTitle", "runs", "incomingDtps", "pressureVsMedian", "pressureFlag", "averageIncomingDamage", "partyDps", "bossDpsTaken", "defeatRate", "wipeRate", "averageDeaths", "averageTimeToDefeat", "bossMaxHealth", )) write_csv(output_dir / "boss_by_party.csv", boss_by_party, ( "partyConfigurationId", "partySource", "bossKey", "bossName", "category", "dungeonId", "dungeonTitle", "runs", "incomingDtps", "pressureVsMedian", "pressureFlag", "averageIncomingDamage", "partyDps", "bossDpsTaken", "defeatRate", "wipeRate", "averageDeaths", "averageTimeToDefeat", "bossMaxHealth", )) write_csv(output_dir / "class_spec_by_boss.csv", class_spec_by_boss, ( "source", "buildId", "classId", "className", "specialization", "role", "bossKey", "bossName", "dungeonTitle", "partyConfigurations", "runs", "dps", "bossDps", "hps", "hpsReceived", "dtps", "totalDamageTaken", "deathEvents", )) write_csv(output_dir / "suite_runs.csv", compact_runs, ( "partyConfigurationId", "partySource", "policyId", "rotationVersion", "bossKey", "bossName", "bossCategory", "dungeonId", "dungeonTitle", "seed", "outcome", "elapsedSeconds", "timeToDefeatSeconds", "bossLevel", "bossMaxHealth", "bossRemainingHealth", "partyDamage", "bossDamage", "partyDps", "bossDpsTaken", "incomingDamage", "rawIncomingDamage", "incomingDtps", "deaths", "deathEvents", "bossAttacks", )) return report def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description="Simulate a selected WoW, CoA, or RoM party against authored bosses.") result.add_argument("--source", choices=("wow", "coa", "rom", "all"), help="party catalog; all requires automatic mode") result.add_argument("--level", type=int, default=60, help="party level (default: 60)") result.add_argument("--party-mode", choices=("random", "manual", "automatic"), default="random") result.add_argument("--spec-selection", choices=("random", "all", "specific"), default="random", help="automatic party strategy (default: random)") result.add_argument("--spec", action="append", default=[], help="required build ID or class:spec selector for --spec-selection specific") result.add_argument("--party-seed", type=int, default=1, help="random party seed") result.add_argument("--party-configurations", type=int, default=10, help="automatic parties per selected source") result.add_argument("--member", action="append", default=[], help="manual build ID or class:spec selector; repeat five times") result.add_argument("--active-slot", type=int, help="1-based member controlled by the benchmark policy; defaults to healer") result.add_argument("--boss-category", choices=("all", "wow", "rom"), default="all") result.add_argument("--boss", action="append", default=[], help="boss key/name selector; repeat as needed; default is every boss") result.add_argument("--simulations-per-party", type=int, default=3, help="deterministic runs for each party/boss pairing") result.add_argument("--seeds", type=parse_seeds, help="explicit comma-separated seeds; overrides --simulations-per-party") result.add_argument("--duration", type=int, default=60, help="maximum fight duration in seconds") result.add_argument("--output-dir", type=Path, help="report directory") result.add_argument("--list-members", action="store_true") result.add_argument("--list-bosses", action="store_true") return result def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") output_dir = (args.output_dir or ROOT / "artifacts" / "balance-lab" / timestamp).resolve() output_dir.mkdir(parents=True, exist_ok=True) interactive = argv is None and len(sys.argv) == 1 catalog: dict[str, Any] session: RuntimeSession if interactive: args, catalog, session = interactive_arguments(args, output_dir) else: if not args.source: raise SystemExit("--source is required for non-interactive runs") session = RuntimeSession(args.source, args.level, output_dir) catalog = session.catalog if args.list_members: for build in catalog["builds"]: print(f"{build['id']}\t{console_text(build['label'])}") session.close() return 0 if args.list_bosses: for boss in catalog["bosses"]: print(f"{boss['key']}\t{boss['category']}\t{boss['dungeonTitle']} - {boss['name']}") session.close() return 0 builds = catalog["builds"] if args.source == "all" and args.party_mode != "automatic": session.close() raise SystemExit("--source all is supported only with --party-mode automatic") if args.party_configurations < 1: session.close() raise SystemExit("--party-configurations must be at least 1") if args.simulations_per_party < 1: session.close() raise SystemExit("--simulations-per-party must be at least 1") seeds = args.seeds or list(range(1, args.simulations_per_party + 1)) if args.party_mode == "automatic": sources = ("wow", "coa", "rom") if args.source == "all" else (args.source,) if args.spec_selection != "specific" and args.spec: session.close() raise SystemExit("--spec values require --spec-selection specific") if args.spec_selection == "random": configurations = automatic_parties(builds, sources, args.party_configurations, args.party_seed) selected_build_ids: list[str] = [] else: if args.spec_selection == "specific" and not args.spec: session.close() raise SystemExit("--spec-selection specific requires at least one --spec value") selected_build_ids = list(dict.fromkeys( resolve_build(selector, builds)["id"] for selector in args.spec )) configurations = coverage_parties( builds, sources, None if args.spec_selection == "all" else selected_build_ids, args.party_seed, ) if args.active_slot is not None: if not 1 <= args.active_slot <= 5: session.close() raise SystemExit("--active-slot must be from 1 through 5") configurations = [{**configuration, "activeSlot": args.active_slot - 1} for configuration in configurations] elif args.party_mode == "manual": if len(args.member) != 5: session.close() raise SystemExit("Manual mode requires exactly five --member values.") party = [resolve_build(selector, builds) for selector in args.member] active_slot = args.active_slot or next( (index + 1 for index, build in enumerate(party) if build["role"] == "healer"), 1 ) configurations = [{"id": f"{args.source}-party-001", "source": args.source, "party": party, "activeSlot": active_slot - 1}] else: party = random_party(builds, args.party_seed) active_slot = args.active_slot or next( (index + 1 for index, build in enumerate(party) if build["role"] == "healer"), 1 ) configurations = [{"id": f"{args.source}-party-001", "source": args.source, "party": party, "activeSlot": active_slot - 1}] if any(not 0 <= configuration["activeSlot"] < 5 for configuration in configurations): session.close() raise SystemExit("--active-slot must be from 1 through 5") bosses = [resolve_boss(selector, catalog["bosses"]) for selector in args.boss] boss_count = len(bosses) if bosses else sum( args.boss_category == "all" or boss["category"] == args.boss_category for boss in catalog["bosses"] ) total_runs = len(configurations) * boss_count * len(seeds) print( f"Running {total_runs} simulations ({len(configurations)} parties x {boss_count} bosses " f"x {len(seeds)} runs per party/boss)..." ) entries = [] automatic = args.party_mode == "automatic" if automatic: write_json(output_dir / "suite-config.json", { "schemaVersion": 1, "policyId": catalog["policyId"], "rotationVersion": catalog["rotationVersion"], "mode": "automatic", "specSelection": args.spec_selection, "selectedBuildIds": selected_build_ids, "source": args.source, "level": args.level, "partySeed": args.party_seed, "partyConfigurationsPerSource": args.party_configurations if args.spec_selection == "random" else None, "partyConfigurations": len(configurations), "simulationsPerParty": len(seeds), "seeds": seeds, "durationSeconds": args.duration, "bossCategory": args.boss_category, **({"bossKeys": [boss["key"] for boss in bosses]} if bosses else {}), "parties": [{ "id": configuration["id"], "source": configuration["source"], "activeSlot": configuration["activeSlot"], "partyBuildIds": [build["id"] for build in configuration["party"]], } for configuration in configurations], }) try: for configuration_index, configuration in enumerate(configurations, 1): party = configuration["party"] party_dir = output_dir / "parties" / configuration["id"] if automatic else output_dir config = { "schemaVersion": 1, "source": configuration["source"], "level": args.level, "durationSeconds": args.duration, "seeds": seeds, "partyBuildIds": [build["id"] for build in party], "activeSlot": configuration["activeSlot"], "bossCategory": args.boss_category, **({"bossKeys": [boss["key"] for boss in bosses]} if bosses else {}), } config_path = party_dir / "config.json" raw_path = party_dir / "raw-runs.json" write_json(config_path, config) print(f"\nParty {configuration_index}/{len(configurations)}: {configuration['id']}") for index, build in enumerate(party): marker = " [active]" if index == configuration["activeSlot"] else "" print(f" {index + 1}. {console_text(build['label'])}{marker}") session.simulate(config_path, raw_path) report = create_reports(raw_path, party_dir) entries.append({ "id": configuration["id"], "source": configuration["source"], "rawPath": str(raw_path), "report": report, }) finally: session.close() if automatic: suite_report = create_suite_reports(entries, output_dir, args.level) print( f"\nComplete: {suite_report['summary']['partyConfigurations']} parties, " f"{suite_report['summary']['bosses']} bosses, and {suite_report['summary']['runs']} runs." ) print(f"Open: {output_dir / 'suite-report.html'}") else: report = entries[0]["report"] print(f"\nComplete: {report['summary']['bosses']} bosses and {report['summary']['runs']} runs.") print(f"Open: {output_dir / 'report.html'}") print(f"Raw and CSV data: {output_dir}") return 0 if __name__ == "__main__": raise SystemExit(main())