1484 lines
74 KiB
Python
1484 lines
74 KiB
Python
#!/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
|
||
|
||
|
||
def runtime_session_command(source: str, level: int) -> list[str]:
|
||
return [
|
||
str(vite_node_command()),
|
||
"--mode",
|
||
"balance-lab",
|
||
str(RUNTIME),
|
||
"session",
|
||
"--source",
|
||
source,
|
||
"--level",
|
||
str(level),
|
||
]
|
||
|
||
|
||
class RuntimeSession:
|
||
def __init__(self, source: str, level: int, workspace: Path) -> None:
|
||
command = runtime_session_command(source, 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"<th onclick=\"sortTable('{identifier}',{index})\">{html.escape(label)}</th>" for index, (_, label) in enumerate(columns))
|
||
body = []
|
||
for row in rows:
|
||
cells = "".join(f"<td>{html.escape(str(row.get(field, '')))}</td>" for field, _ in columns)
|
||
body.append(f"<tr>{cells}</tr>")
|
||
return (
|
||
f"<section><h2>{html.escape(title)}</h2>"
|
||
f"<input aria-label=\"Filter {html.escape(title)}\" placeholder=\"Filter rows…\" oninput=\"filterTable('{identifier}',this.value)\">"
|
||
f"<div class=\"table-wrap\"><table id=\"{identifier}\"><thead><tr>{headers}</tr></thead>"
|
||
f"<tbody>{''.join(body)}</tbody></table></div></section>"
|
||
)
|
||
|
||
|
||
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"""<!doctype html>
|
||
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
||
<title>Healer Man Balance Lab</title><style>
|
||
:root {{ color-scheme: dark; font-family: Inter, system-ui, sans-serif; background:#0d1117; color:#e6edf3; }}
|
||
body {{ margin:0 auto; max-width:1500px; padding:28px; }} h1,h2 {{ color:#f0f6fc; }}
|
||
.summary {{ color:#9da7b3; margin-bottom:28px; }} section {{ margin:32px 0; }}
|
||
input {{ width:min(420px,90%); padding:9px 12px; margin:0 0 12px; color:inherit; background:#161b22; border:1px solid #30363d; border-radius:6px; }}
|
||
.table-wrap {{ overflow:auto; border:1px solid #30363d; border-radius:8px; }} table {{ width:100%; border-collapse:collapse; font-size:13px; }}
|
||
th,td {{ padding:8px 10px; border-bottom:1px solid #21262d; text-align:right; white-space:nowrap; }} th {{ position:sticky; top:0; cursor:pointer; background:#161b22; }}
|
||
th:first-child,td:first-child,th:nth-child(2),td:nth-child(2) {{ text-align:left; }} tr:hover {{ background:#161b22; }}
|
||
</style></head><body><h1>Healer Man Balance Lab</h1>
|
||
<p class="summary">{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.</p>
|
||
{member_table}{boss_table}
|
||
<script>
|
||
function filterTable(id,q){{q=q.toLowerCase();document.querySelectorAll('#'+id+' tbody tr').forEach(r=>r.hidden=!r.textContent.toLowerCase().includes(q));}}
|
||
function sortTable(id,n){{const t=document.getElementById(id),b=t.tBodies[0],rows=[...b.rows],asc=t.dataset.sort!==(n+':asc');rows.sort((a,c)=>{{let x=a.cells[n].textContent.replace(/[$,%×]/g,'').replaceAll(',',''),y=c.cells[n].textContent.replace(/[$,%×]/g,'').replaceAll(',','');let xn=Number(x),yn=Number(y);return (Number.isFinite(xn)&&Number.isFinite(yn)?xn-yn:x.localeCompare(y))* (asc?1:-1);}});rows.forEach(r=>b.appendChild(r));t.dataset.sort=n+':' +(asc?'asc':'desc');}}
|
||
</script></body></html>"""
|
||
|
||
|
||
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/<party-id>/`. 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"""<!doctype html>
|
||
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
||
<title>Healer Man Automatic Balance Suite</title><style>
|
||
:root {{ color-scheme: dark; font-family: Inter, system-ui, sans-serif; background:#0d1117; color:#e6edf3; }}
|
||
body {{ margin:0 auto; max-width:1600px; padding:28px; }} h1,h2 {{ color:#f0f6fc; }}
|
||
.summary {{ color:#9da7b3; margin-bottom:28px; }} section {{ margin:32px 0; }}
|
||
input {{ width:min(420px,90%); padding:9px 12px; margin:0 0 12px; color:inherit; background:#161b22; border:1px solid #30363d; border-radius:6px; }}
|
||
.table-wrap {{ overflow:auto; border:1px solid #30363d; border-radius:8px; }} table {{ width:100%; border-collapse:collapse; font-size:13px; }}
|
||
th,td {{ padding:8px 10px; border-bottom:1px solid #21262d; text-align:right; white-space:nowrap; }} th {{ position:sticky; top:0; cursor:pointer; background:#161b22; }}
|
||
th:first-child,td:first-child,th:nth-child(2),td:nth-child(2),th:nth-child(3),td:nth-child(3) {{ text-align:left; }} tr:hover {{ background:#161b22; }}
|
||
</style></head><body><h1>Healer Man Automatic Balance Suite</h1>
|
||
<p class="summary">{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.</p>
|
||
{class_table}{ability_table}{party_table}{boss_table}
|
||
<script>
|
||
function filterTable(id,q){{q=q.toLowerCase();document.querySelectorAll('#'+id+' tbody tr').forEach(r=>r.hidden=!r.textContent.toLowerCase().includes(q));}}
|
||
function sortTable(id,n){{const t=document.getElementById(id),b=t.tBodies[0],rows=[...b.rows],asc=t.dataset.sort!==(n+':asc');rows.sort((a,c)=>{{let x=a.cells[n].textContent.replace(/[$,%×]/g,'').replaceAll(',',''),y=c.cells[n].textContent.replace(/[$,%×]/g,'').replaceAll(',','');let xn=Number(x),yn=Number(y);return (Number.isFinite(xn)&&Number.isFinite(yn)?xn-yn:x.localeCompare(y))* (asc?1:-1);}});rows.forEach(r=>b.appendChild(r));t.dataset.sort=n+':' +(asc?'asc':'desc');}}
|
||
</script></body></html>"""
|
||
|
||
|
||
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())
|