"""Computation helpers for DPS/DPM summaries derived from TL combat logs."""
from __future__ import annotations
from collections import defaultdict
from statistics import mean, median
from typing import Dict, Iterable, List
from dps_logs.parser import LogEvent
def _safe_duration(events: List[LogEvent]) -> float:
if not events:
return 0.0
ordered = sorted(events, key=lambda event: event.timestamp)
span = (ordered[-1].timestamp - ordered[0].timestamp).total_seconds()
return span if span > 0 else 1.0
def _pct(part: int, whole: int) -> float:
return round((part / whole) * 100.0, 3) if whole else 0.0
def _damage_events(events: List[LogEvent]) -> List[LogEvent]:
return [event for event in events if event.event_type == "DamageDone"]
def _countable_damage_events(events: List[LogEvent]) -> List[LogEvent]:
return [event for event in events if event.damage > 0]
def summarize_run(run_id: str, events: List[LogEvent]) -> Dict:
damage_events = _damage_events(events)
countable_events = _countable_damage_events(damage_events)
duration_source = countable_events if countable_events else damage_events
duration_seconds = _safe_duration(duration_source)
total_events = len(damage_events)
total_damage = sum(event.damage for event in countable_events)
total_hits = len(countable_events)
crit_hits = sum(1 for event in countable_events if event.crit)
heavy_hits = sum(1 for event in countable_events if event.heavy)
dps = round(total_damage / duration_seconds, 3) if duration_seconds else 0.0
dpm = round(dps * 60.0, 3)
skill_totals: Dict[str, Dict[str, float]] = defaultdict(lambda: {
"total_events": 0,
"total_damage": 0,
"total_hits": 0,
"crit_hits": 0,
"heavy_hits": 0,
})
for event in damage_events:
bucket = skill_totals[event.skill]
bucket["total_events"] += 1
if event.damage <= 0:
continue
bucket["total_damage"] += event.damage
bucket["total_hits"] += 1
bucket["crit_hits"] += 1 if event.crit else 0
bucket["heavy_hits"] += 1 if event.heavy else 0
skills = []
for skill_name, stats in sorted(
skill_totals.items(), key=lambda item: item[1]["total_damage"], reverse=True
):
skill_dps = round(stats["total_damage"] / duration_seconds, 3) if duration_seconds else 0.0
skills.append({
"skill": skill_name,
"total_events": int(stats["total_events"]),
"total_damage": int(stats["total_damage"]),
"total_hits": int(stats["total_hits"]),
"crit_hits": int(stats["crit_hits"]),
"heavy_hits": int(stats["heavy_hits"]),
"dps": skill_dps,
"dpm": round(skill_dps * 60.0, 3),
"crit_rate_pct": _pct(int(stats["crit_hits"]), int(stats["total_hits"])),
"heavy_rate_pct": _pct(int(stats["heavy_hits"]), int(stats["total_hits"])),
})
return {
"run_id": run_id,
"total_damage": total_damage,
"total_events": total_events,
"duration_seconds": duration_seconds,
"dps": dps,
"dpm": dpm,
"total_hits": total_hits,
"crit_hits": crit_hits,
"heavy_hits": heavy_hits,
"crit_rate_pct": _pct(crit_hits, total_hits),
"heavy_rate_pct": _pct(heavy_hits, total_hits),
"skills": skills,
}
def build_summary(runs: Iterable[Dict]) -> Dict:
runs = list(runs)
if not runs:
return {
"total_runs": 0,
"total_damage": 0,
"combined_duration_seconds": 0.0,
"total_events": 0,
"overall_dps": 0.0,
"overall_dpm": 0.0,
"overall_crit_rate_pct": 0.0,
"overall_heavy_rate_pct": 0.0,
"top_skills": [],
"top_skills_by_damage": [],
"mean_dps": 0.0,
"median_dps": 0.0,
"mean_dpm": 0.0,
"median_dpm": 0.0,
}
total_damage = sum(run["total_damage"] for run in runs)
combined_duration = sum(run["duration_seconds"] for run in runs)
total_hits = sum(run["total_hits"] for run in runs)
crit_hits = sum(run["crit_hits"] for run in runs)
heavy_hits = sum(run["heavy_hits"] for run in runs)
duration_for_rate = combined_duration if combined_duration > 0 else 1.0
dps = round(total_damage / duration_for_rate, 3)
dpm = round(dps * 60.0, 3)
run_dps_values = [float(run.get("dps", 0.0) or 0.0) for run in runs]
mean_dps = round(mean(run_dps_values), 3) if run_dps_values else 0.0
median_dps = round(median(run_dps_values), 3) if run_dps_values else 0.0
mean_dpm = round(mean_dps * 60.0, 3)
median_dpm = round(median_dps * 60.0, 3)
aggregated_skills: Dict[str, Dict[str, float]] = defaultdict(lambda: {
"total_damage": 0,
"total_hits": 0,
"crit_hits": 0,
"heavy_hits": 0,
"total_events": 0,
})
for run in runs:
for skill in run["skills"]:
bucket = aggregated_skills[skill["skill"]]
bucket["total_damage"] += skill["total_damage"]
bucket["total_hits"] += skill["total_hits"]
bucket["crit_hits"] += skill["crit_hits"]
bucket["heavy_hits"] += skill["heavy_hits"]
bucket["total_events"] += skill.get("total_events", 0)
top_skills = []
for skill_name, stats in sorted(
aggregated_skills.items(), key=lambda item: item[1]["total_damage"], reverse=True
)[:5]:
skill_dps = round(stats["total_damage"] / duration_for_rate, 3)
top_skills.append({
"skill": skill_name,
"total_damage": int(stats["total_damage"]),
"total_hits": int(stats["total_hits"]),
"total_events": int(stats["total_events"]),
"dps": skill_dps,
"dpm": round(skill_dps * 60.0, 3),
"crit_rate_pct": _pct(int(stats["crit_hits"]), int(stats["total_hits"])),
"heavy_rate_pct": _pct(int(stats["heavy_hits"]), int(stats["total_hits"])),
})
return {
"total_runs": len(runs),
"total_damage": total_damage,
"combined_duration_seconds": combined_duration,
"total_events": sum(run.get("total_events", 0) for run in runs),
"overall_dps": dps,
"overall_dpm": dpm,
"overall_crit_rate_pct": _pct(crit_hits, total_hits),
"overall_heavy_rate_pct": _pct(heavy_hits, total_hits),
"top_skills": top_skills,
"top_skills_by_damage": top_skills,
"mean_dps": mean_dps,
"median_dps": median_dps,
"mean_dpm": mean_dpm,
"median_dpm": median_dpm,
}
__all__ = ["summarize_run", "build_summary"]
