import math import numpy as np import matplotlib.pyplot as plt from collections import defaultdict from matplotlib.transforms import blended_transform_factory as _blended_xform from reporting.team_colours import TEAM0_MPL, TEAM1_MPL import modelling.tails as tails from modelling.math_utils import natural_log from modelling.modeled_rows import compute_v1_team_surprise from reporting.report_model import collect_modeled_report_rows from modelling.row_schema import CumulativeSeriesPayload, UsableSurpriseRow from modelling.roll_events import collect_roll_events_compat from core.turn_allocation_adjustments import build_game_turn_map def extract_gamer_names(json_data): replay = json_data.get("Replay", {}) if isinstance(json_data, dict) else {} # Primary source in many BB3 replays. game_infos = replay.get("NotificationGameJoined", {}).get("GameInfos", {}) if isinstance(replay, dict) else {} gamers_infos = game_infos.get("GamersInfos", {}) if isinstance(game_infos, dict) else {} gamer_infos = gamers_infos.get("GamerInfos", []) if isinstance(gamers_infos, dict) else [] if isinstance(gamer_infos, dict): gamer_infos = [gamer_infos] names_by_slot = {} for gamer in gamer_infos: if not isinstance(gamer, dict): continue slot = gamer.get("Slot") name = gamer.get("Name") if slot is None or name in (None, ""): continue try: slot_int = int(str(slot)) except (TypeError, ValueError): continue names_by_slot[slot_int] = str(name) if 0 in names_by_slot and 1 in names_by_slot: return names_by_slot[0], names_by_slot[1] # Fallback source. rosters = replay.get("Rosters", {}) if isinstance(replay, dict) else {} team_rosters = rosters.get("TeamRoster", []) if isinstance(rosters, dict) else [] if isinstance(team_rosters, dict): team_rosters = [team_rosters] names = [] for idx, team_roster in enumerate(team_rosters): default_name = f"Team{idx}" if not isinstance(team_roster, dict): names.append(default_name) continue team = team_roster.get("Team", {}) if not isinstance(team, dict): names.append(default_name) continue gamer_name = team.get("GamerName") or team.get("CoachName") or team.get("Name") names.append(str(gamer_name) if gamer_name else default_name) while len(names) < 2: names.append(f"Team{len(names)}") return names[0], names[1] def format_one_in_games(p_tail): if not isinstance(p_tail, (int, float)): return "N/A" if p_tail <= 0: return "infinite" value = 1.0 / p_tail if value >= 1000: return f"{value:,.0f}" if value >= 100: return f"{value:.1f}" return f"{value:.2f}" def _display_team_name(name, team_index, team_meta=None): base = str(name).strip() if name is not None else "" if team_meta is not None: try: team_label = team_meta.team_label(str(team_index)) if team_label not in (None, ""): base = str(team_label) except Exception: pass if base == "": base = f"Team{team_index}" suffix = f"(team {team_index})" if suffix.lower() in base.lower(): return base return f"{base} {suffix}" def _extend_prefix_vector(prefix: np.ndarray, chunk: np.ndarray) -> np.ndarray: if chunk.size == 0: return prefix if prefix.size == 0: return chunk return np.concatenate((prefix, chunk)) def _to_turn_int(value): try: return int(str(value)) except (TypeError, ValueError): return None def _collect_touchdown_markers(game_state): """Return touchdown markers with turn and scoring team attribution. Turn numbers are derived from the step-turn counter (via the step_number lookup) so they stay consistent with the dice-roll rows used by the cumulative surprise series. """ # Build step-turn map once for the whole game. tt_map = build_game_turn_map(getattr(game_state, "raw_replay_steps", [])) def _step_turn(step_number, fallback_game_turn): """Return step-counter game-turn int for a step, falling back to board-state turn.""" if step_number is not None and tt_map: tt_str = tt_map.get(step_number) if tt_str is not None: turn = _to_turn_int(tt_str) if turn is not None: return turn return _to_turn_int(fallback_game_turn) markers = [] touchdown_events = getattr(game_state, "touchdown_events", None) if isinstance(touchdown_events, list) and touchdown_events: for touchdown in touchdown_events: if not isinstance(touchdown, dict): continue step_number = touchdown.get("step_number") turn_int = _step_turn(step_number, touchdown.get("game_turn")) if turn_int is None: continue team_value = touchdown.get("team_id") try: team_id = int(str(team_value)) if team_value is not None else None except (TypeError, ValueError): team_id = None if team_id not in (0, 1): team_id = None markers.append( { "turn": turn_int, "team_id": team_id, "step_number": step_number, "player_name": touchdown.get("player_name"), } ) else: for step in getattr(game_state, "steps", []) or []: touchdown = getattr(step, "touchdown", None) if not isinstance(touchdown, dict): continue step_number = getattr(step, "step_number", None) turn_int = _step_turn(step_number, getattr(step, "game_turn", None)) if turn_int is None: continue team_value = touchdown.get("team_id") try: team_id = int(str(team_value)) if team_value is not None else None except (TypeError, ValueError): team_id = None if team_id not in (0, 1): team_id = None markers.append( { "turn": turn_int, "team_id": team_id, "step_number": step_number, "player_name": touchdown.get("player_name"), } ) markers.sort( key=lambda item: ( int(item.get("turn", 0)), int(item.get("step_number", 0)) if str(item.get("step_number", "")).isdigit() else 0, ) ) return markers def _collect_usable_surprise_rows(game_state, team_meta=None, calculator=None) -> tuple[list[UsableSurpriseRow], list[dict]]: """Collect normalized surprise rows used by megadicing and plotting.""" modeled_payload = collect_modeled_report_rows( game_state, team_meta=team_meta, calculator=calculator, target_turn=None, ) events = list(modeled_payload.get("rows") or []) kickoff_events = list(modeled_payload.get("kickoff_rows") or []) usable_rows = [] for event in events: if not isinstance(event, dict): continue team0_surprise, team1_surprise = compute_v1_team_surprise(event) if team0_surprise is None and team1_surprise is None: continue turn_int = _to_turn_int(event.get("game_turn")) if turn_int is None: continue team_id_raw = event.get("dice_roller") if team_id_raw is None: team_id_raw = event.get("team_id") try: team_id = int(str(team_id_raw)) except (TypeError, ValueError): continue if team_id not in (0, 1): continue p_success = event.get("probability_success") p_fail = event.get("probability_fail") if not isinstance(p_success, (int, float)) or not isinstance(p_fail, (int, float)): continue surprise = float(team0_surprise if team_id == 0 else team1_surprise) usable_rows.append( { "turn": turn_int, "team_id": team_id, "surprise": surprise, "p_success": float(p_success), "p_fail": float(p_fail), } ) return usable_rows, kickoff_events def _infer_cumulative_series_max_turn(usable_rows, touchdown_markers, fallback=16): """Infer the last turn worth rendering for cumulative surprise outputs.""" candidate_turns = [] for row in usable_rows: if not isinstance(row, dict): continue turn_int = _to_turn_int(row.get("turn")) if turn_int is not None and turn_int > 0: candidate_turns.append(turn_int) for marker in touchdown_markers: if not isinstance(marker, dict): continue turn_int = _to_turn_int(marker.get("turn")) if turn_int is not None and turn_int > 0: candidate_turns.append(turn_int) if candidate_turns: return max(candidate_turns) return int(fallback) def _lr_quantile_centred(p_target, pvec, qvec, right_hint=None, left_hint=None): """Return (right_centred, left_centred) LR quantile thresholds in centred space. right_centred > 0: P(centred_score >= right_centred) ≈ p_target left_centred < 0: P(centred_score <= left_centred) ≈ p_target Returns (0.0, 0.0) on any failure or empty input. """ if len(pvec) == 0: return 0.0, 0.0 try: k1 = float(tails.K1(0, pvec, qvec)) right_raw_hint = None if right_hint is None else float(k1 + right_hint) left_raw_hint = None if left_hint is None else float(k1 + left_hint) right, left = tails.LR_quantile_pair( p_target, pvec, qvec, _mu=k1, right_hint=right_raw_hint, left_hint=left_raw_hint, ) return float(right - k1), float(left - k1) except Exception: return 0.0, 0.0 def _compute_team_prefix_stats(rows): """Compute score, p/q vectors, K1 mean, and centred score for one team's rows.""" score = float(sum(row["surprise"] for row in rows)) pvec = np.array([row["p_success"] for row in rows], dtype=float) qvec = np.array([row["p_fail"] for row in rows], dtype=float) k1 = float(tails.K1(0, pvec, qvec)) if len(rows) > 0 else 0.0 std = float(np.sqrt(tails.K2(0, pvec, qvec))) if len(rows) > 0 else 0.0 centred = float(score - k1) return { "score": score, "pvec": pvec, "qvec": qvec, "k1": k1, "std": std, "centred": centred, "count": len(rows), } def build_cumulative_surprise_series_v1( game_state, team_meta=None, calculator=None, max_turn=None, include_extra_bands: bool = True, ) -> CumulativeSeriesPayload: """Build cumulative V1 surprise series through each turn prefix. Definitions per team i and prefix turn n: - score_i_n: sum(log(1/p_success)) on success and log(p_fail) on fail - pvec_i_n: all p_success for team i up to turn n - qvec_i_n: all p_fail for team i up to turn n - centred_i_n: score_i_n - K1(0, pvec_i_n, qvec_i_n) Delta series: - delta_centred_n = centred_0_n - centred_1_n """ usable_rows, kickoff_events = _collect_usable_surprise_rows( game_state, team_meta=team_meta, calculator=calculator, ) touchdown_markers = _collect_touchdown_markers(game_state) touchdown_turns = sorted(set(int(marker["turn"]) for marker in touchdown_markers)) if max_turn is None: max_turn = _infer_cumulative_series_max_turn(usable_rows, touchdown_markers) else: max_turn = int(max_turn) # Pre-compute per-event K1 and K2 contributions at t=0 for all events at once # using vectorized numpy. K1 and K2 at t=0 are additive (separable per event), # so prefix cumulants can be accumulated in O(1) per turn instead of O(N). if usable_rows: _all_p = np.array([r["p_success"] for r in usable_rows], dtype=float) _all_q = np.array([r["p_fail"] for r in usable_rows], dtype=float) _lp = np.zeros_like(_all_p) _lq = np.zeros_like(_all_q) _pmask = _all_p > 0 _qmask = _all_q > 0 _lp[_pmask] = np.log(_all_p[_pmask]) _lq[_qmask] = np.log(_all_q[_qmask]) _k1c = -_all_p * _lp + _all_q * _lq _k2c = (_all_p * _lp**2 + _all_q * _lq**2) - _k1c**2 for _i, _row in enumerate(usable_rows): _row["_k1c"] = float(_k1c[_i]) _row["_k2c"] = float(_k2c[_i]) # Pre-group rows by turn so each turn's iteration is O(1) lookup rather # than an O(N) scan over all usable_rows. _rows_grouped = defaultdict(list) for row in usable_rows: _rows_grouped[row["turn"]].append(row) # Running accumulators for prefix K1/K2 (avoids O(N) recompute each turn). # k1_delta = k1_0_acc - k1_1_acc (K1 under p↔q swap flips sign) # k2_delta = k2_0_acc + k2_1_acc (K2 is symmetric under p↔q swap) k1_0_acc = 0.0 k2_0_acc = 0.0 k1_1_acc = 0.0 k2_1_acc = 0.0 score0_acc = 0.0 score1_acc = 0.0 team0_total = sum(1 for row in usable_rows if row.get("team_id") == 0) team1_total = sum(1 for row in usable_rows if row.get("team_id") == 1) pvec0_buf = np.empty(team0_total, dtype=float) qvec0_buf = np.empty(team0_total, dtype=float) pvec1_buf = np.empty(team1_total, dtype=float) qvec1_buf = np.empty(team1_total, dtype=float) p_delta_buf = np.empty(team0_total + team1_total, dtype=float) q_delta_buf = np.empty(team0_total + team1_total, dtype=float) idx0 = 0 idx1 = 0 prev_diff_q10_r = prev_diff_q10_l = None prev_diff_q01_r = prev_diff_q01_l = None prev_diff_q001_r = prev_diff_q001_l = None prev_t0_q10_r_n = prev_t0_q10_l_n = None prev_t1_q10_r_n = prev_t1_q10_l_n = None rows_by_turn = {} for n in range(1, max_turn + 1): current_rows = _rows_grouped.get(n, []) current_score0 = 0.0 current_score1 = 0.0 current_k1_0 = 0.0 current_k1_1 = 0.0 current_k2_0 = 0.0 current_k2_1 = 0.0 pvec_c0_list = [] qvec_c0_list = [] pvec_c1_list = [] qvec_c1_list = [] # Extend prefix accumulators with this turn's events. for row in current_rows: if row["team_id"] == 0: current_score0 += row["surprise"] current_k1_0 += row["_k1c"] current_k2_0 += row["_k2c"] p = row["p_success"] q = row["p_fail"] pvec_c0_list.append(p) qvec_c0_list.append(q) else: current_score1 += row["surprise"] current_k1_1 += row["_k1c"] current_k2_1 += row["_k2c"] p = row["p_success"] q = row["p_fail"] pvec_c1_list.append(p) qvec_c1_list.append(q) score0_acc += current_score0 score1_acc += current_score1 k1_0_acc += current_k1_0 k1_1_acc += current_k1_1 k2_0_acc += current_k2_0 k2_1_acc += current_k2_1 score0 = score0_acc score1 = score1_acc centred0 = score0_acc - k1_0_acc centred1 = score1_acc - k1_1_acc std0 = math.sqrt(max(k2_0_acc, 0.0)) std1 = math.sqrt(max(k2_1_acc, 0.0)) current_std0 = math.sqrt(max(current_k2_0, 0.0)) current_std1 = math.sqrt(max(current_k2_1, 0.0)) current_centred0 = float(current_score0 - current_k1_0) current_centred1 = float(current_score1 - current_k1_1) pvec_c0 = np.array(pvec_c0_list, dtype=float) if pvec_c0_list else np.array([], dtype=float) qvec_c0 = np.array(qvec_c0_list, dtype=float) if qvec_c0_list else np.array([], dtype=float) pvec_c1 = np.array(pvec_c1_list, dtype=float) if pvec_c1_list else np.array([], dtype=float) qvec_c1 = np.array(qvec_c1_list, dtype=float) if qvec_c1_list else np.array([], dtype=float) c0_len = int(pvec_c0.size) c1_len = int(pvec_c1.size) if c0_len: pvec0_buf[idx0: idx0 + c0_len] = pvec_c0 qvec0_buf[idx0: idx0 + c0_len] = qvec_c0 idx0 += c0_len if c1_len: pvec1_buf[idx1: idx1 + c1_len] = pvec_c1 qvec1_buf[idx1: idx1 + c1_len] = qvec_c1 idx1 += c1_len pvec0_acc = pvec0_buf[:idx0] qvec0_acc = qvec0_buf[:idx0] pvec1_acc = pvec1_buf[:idx1] qvec1_acc = qvec1_buf[:idx1] diff_centred = float(centred0 - centred1) # Delta p/q vectors still needed for LR_two_sided and quantile band calls. # k1/k2 for delta follow from linearity (K1 negates under p↔q; K2 is unchanged). delta_len = idx0 + idx1 if delta_len > 0: p_delta_buf[:idx0] = pvec0_acc p_delta_buf[idx0:delta_len] = qvec1_acc q_delta_buf[:idx0] = qvec0_acc q_delta_buf[idx0:delta_len] = pvec1_acc p_delta = p_delta_buf[:delta_len] q_delta = q_delta_buf[:delta_len] else: p_delta = np.array([], dtype=float) q_delta = np.array([], dtype=float) k1_delta = k1_0_acc - k1_1_acc k2_delta = k2_0_acc + k2_1_acc std_delta = math.sqrt(max(k2_delta, 0.0)) diff_raw = float(score0 - score1) # Per-turn prefix p-tail. # Sign: + (right tail) = team0 luckier relative to mean; - (left tail) = team1 luckier. if p_delta.size > 0 and q_delta.size > 0: try: p_tail_n, which_tail_n, abs_err_n = tails.LR_two_sided(diff_raw, p_delta, q_delta) if which_tail_n == "right": signed_p_n = float(p_tail_n) elif which_tail_n == "left": signed_p_n = -float(p_tail_n) else: signed_p_n = None except Exception: p_tail_n, which_tail_n, abs_err_n, signed_p_n = None, None, None, None else: p_tail_n, which_tail_n, abs_err_n, signed_p_n = None, None, None, None # LR quantile bands for diff panel (centred space, asymmetric right/left). # The light path keeps only the 1-in-10 band, which is all the app needs by default. if p_delta.size > 0: diff_q10_r, diff_q10_l = _lr_quantile_centred( 0.10, p_delta, q_delta, right_hint=prev_diff_q10_r, left_hint=prev_diff_q10_l, ) if include_extra_bands: diff_q01_r, diff_q01_l = _lr_quantile_centred( 0.01, p_delta, q_delta, right_hint=prev_diff_q01_r, left_hint=prev_diff_q01_l, ) diff_q001_r, diff_q001_l = _lr_quantile_centred( 0.001, p_delta, q_delta, right_hint=prev_diff_q001_r, left_hint=prev_diff_q001_l, ) else: diff_q01_r = diff_q01_l = diff_q001_r = diff_q001_l = None else: diff_q10_r = diff_q10_l = 0.0 diff_q01_r = diff_q01_l = diff_q001_r = diff_q001_l = None prev_diff_q10_r, prev_diff_q10_l = diff_q10_r, diff_q10_l if include_extra_bands: prev_diff_q01_r, prev_diff_q01_l = diff_q01_r, diff_q01_l prev_diff_q001_r, prev_diff_q001_l = diff_q001_r, diff_q001_l # LR quantile bands for per-turn per-team worm plot. t0_q10_r_turn, t0_q10_l_turn = _lr_quantile_centred(0.10, pvec_c0, qvec_c0) t1_q10_r_turn, t1_q10_l_turn = _lr_quantile_centred(0.10, pvec_c1, qvec_c1) # LR quantile bands for cumulative per-team worm plot. t0_q10_r_n, t0_q10_l_n = _lr_quantile_centred( 0.10, pvec0_acc, qvec0_acc, right_hint=prev_t0_q10_r_n, left_hint=prev_t0_q10_l_n, ) t1_q10_r_n, t1_q10_l_n = _lr_quantile_centred( 0.10, pvec1_acc, qvec1_acc, right_hint=prev_t1_q10_r_n, left_hint=prev_t1_q10_l_n, ) prev_t0_q10_r_n, prev_t0_q10_l_n = t0_q10_r_n, t0_q10_l_n prev_t1_q10_r_n, prev_t1_q10_l_n = t1_q10_r_n, t1_q10_l_n rows_by_turn[n] = { "turn": n, "usable_event_count": int(idx0 + idx1), "usable_event_count_turn": int(pvec_c0.size + pvec_c1.size), "team0_event_count": int(idx0), "team1_event_count": int(idx1), "team0_event_count_turn": int(pvec_c0.size), "team1_event_count_turn": int(pvec_c1.size), "score_0_turn": float(current_score0), "score_1_turn": float(current_score1), "k1_0_turn": float(current_k1_0), "k1_1_turn": float(current_k1_1), "std_0_turn": float(current_std0), "std_1_turn": float(current_std1), "team0_centred_turn": float(current_centred0), "team1_centred_turn": float(current_centred1), "score_0_n": score0, "score_1_n": score1, "k1_0_n": k1_0_acc, "k1_1_n": k1_1_acc, "std_0_n": std0, "std_1_n": std1, "team0_centred": centred0, "team1_centred": centred1, "diff_raw": diff_raw, "k1_delta_n": k1_delta, "std_delta_n": std_delta, "diff_centred": diff_centred, "p_tail_n": p_tail_n, "which_tail_n": which_tail_n, "abs_err_n": abs_err_n, "signed_p_n": signed_p_n, "diff_q10_right_n": diff_q10_r, "diff_q10_left_n": diff_q10_l, "t0_q10_right_turn": t0_q10_r_turn, "t0_q10_left_turn": t0_q10_l_turn, "t1_q10_right_turn": t1_q10_r_turn, "t1_q10_left_turn": t1_q10_l_turn, "t0_q10_right_n": t0_q10_r_n, "t0_q10_left_n": t0_q10_l_n, "t1_q10_right_n": t1_q10_r_n, "t1_q10_left_n": t1_q10_l_n, } if include_extra_bands: rows_by_turn[n]["diff_q01_right_n"] = diff_q01_r rows_by_turn[n]["diff_q01_left_n"] = diff_q01_l rows_by_turn[n]["diff_q001_right_n"] = diff_q001_r rows_by_turn[n]["diff_q001_left_n"] = diff_q001_l final_turn = int(max(rows_by_turn.keys())) if rows_by_turn else 0 final_centred_score = rows_by_turn.get(final_turn, {}).get("diff_centred", 0.0) return { "kickoff_events_included": len(kickoff_events), "touchdown_markers": touchdown_markers, "touchdown_turns": touchdown_turns, "usable_rows_total": len(usable_rows), "final_turn": final_turn, "final_centred_score": final_centred_score, "by_turn": rows_by_turn, } def build_signed_p_series_v1(series): """Extract signed p-value at each turn prefix from a cumulative series result. Sign convention (matches diff_centred / "Delta centred by turn prefix" plot): + (right tail): team0 had more surprising rolls relative to mean - (left tail): team1 had more surprising rolls relative to mean Returns a list of dicts ordered by turn, each with: turn, signed_p, p_tail, abs_err, which_tail, diff_centred, usable_event_count """ by_turn = series.get("by_turn", {}) result = [] for n in sorted(by_turn.keys()): row = by_turn[n] result.append({ "turn": n, "signed_p": row.get("signed_p_n"), "p_tail": row.get("p_tail_n"), "abs_err": row.get("abs_err_n"), "which_tail": row.get("which_tail_n"), "diff_centred": row.get("diff_centred"), "usable_event_count": row.get("usable_event_count"), }) return result def build_cumulative_surprise_series_meter_only( game_state, team_meta=None, calculator=None, max_turn=None, ) -> dict: """Fast meter-only path: cumulative series without quantile bands or figures. Returns a minimal series dict with only: - by_turn[n] with: turn, usable_event_count, team0_centred, team1_centred, diff_centred, std_delta_n, p_tail_n, which_tail_n, abs_err_n, signed_p_n Skips all _lr_quantile_centred calls (the expensive part) and all per-team quantile bands. Estimated speedup: ~90% faster than full builder (1-2s vs 9-10s). """ usable_rows, _ = _collect_usable_surprise_rows( game_state, team_meta=team_meta, calculator=calculator, ) touchdown_markers = _collect_touchdown_markers(game_state) if max_turn is None: max_turn = _infer_cumulative_series_max_turn(usable_rows, touchdown_markers) else: max_turn = int(max_turn) # Pre-compute per-event K1 and K2 contributions at t=0. if usable_rows: _all_p = np.array([r["p_success"] for r in usable_rows], dtype=float) _all_q = np.array([r["p_fail"] for r in usable_rows], dtype=float) _lp = np.zeros_like(_all_p) _lq = np.zeros_like(_all_q) _pmask = _all_p > 0 _qmask = _all_q > 0 _lp[_pmask] = np.log(_all_p[_pmask]) _lq[_qmask] = np.log(_all_q[_qmask]) _k1c = -_all_p * _lp + _all_q * _lq _k2c = (_all_p * _lp**2 + _all_q * _lq**2) - _k1c**2 for _i, _row in enumerate(usable_rows): _row["_k1c"] = float(_k1c[_i]) _row["_k2c"] = float(_k2c[_i]) # Pre-group rows by turn. _rows_grouped = defaultdict(list) for row in usable_rows: _rows_grouped[row["turn"]].append(row) # Running accumulators for prefix K1/K2. k1_0_acc = 0.0 k2_0_acc = 0.0 k1_1_acc = 0.0 k2_1_acc = 0.0 score0_acc = 0.0 score1_acc = 0.0 pvec0_acc = np.array([], dtype=float) qvec0_acc = np.array([], dtype=float) pvec1_acc = np.array([], dtype=float) qvec1_acc = np.array([], dtype=float) rows_by_turn = {} for n in range(1, max_turn + 1): current_rows = _rows_grouped.get(n, []) current0 = [row for row in current_rows if row["team_id"] == 0] current1 = [row for row in current_rows if row["team_id"] == 1] pvec_c0_list = [] qvec_c0_list = [] pvec_c1_list = [] qvec_c1_list = [] # Extend prefix accumulators with this turn's events. for row in current0: score0_acc += row["surprise"] k1_0_acc += row["_k1c"] k2_0_acc += row["_k2c"] pvec_c0_list.append(row["p_success"]) qvec_c0_list.append(row["p_fail"]) for row in current1: score1_acc += row["surprise"] k1_1_acc += row["_k1c"] k2_1_acc += row["_k2c"] pvec_c1_list.append(row["p_success"]) qvec_c1_list.append(row["p_fail"]) pvec_c0 = np.array(pvec_c0_list, dtype=float) if pvec_c0_list else np.array([], dtype=float) qvec_c0 = np.array(qvec_c0_list, dtype=float) if qvec_c0_list else np.array([], dtype=float) pvec_c1 = np.array(pvec_c1_list, dtype=float) if pvec_c1_list else np.array([], dtype=float) qvec_c1 = np.array(qvec_c1_list, dtype=float) if qvec_c1_list else np.array([], dtype=float) pvec0_acc = _extend_prefix_vector(pvec0_acc, pvec_c0) qvec0_acc = _extend_prefix_vector(qvec0_acc, qvec_c0) pvec1_acc = _extend_prefix_vector(pvec1_acc, pvec_c1) qvec1_acc = _extend_prefix_vector(qvec1_acc, qvec_c1) centred0 = score0_acc - k1_0_acc centred1 = score1_acc - k1_1_acc k2_delta = k2_0_acc + k2_1_acc std_delta = math.sqrt(max(k2_delta, 0.0)) diff_centred = float(centred0 - centred1) diff_raw = float(score0_acc - score1_acc) # Per-turn signed p-value (LR_two_sided is still called, but only once per turn). p_delta = np.concatenate((pvec0_acc, qvec1_acc)) if (pvec0_acc.size + qvec1_acc.size) > 0 else np.array([], dtype=float) q_delta = np.concatenate((qvec0_acc, pvec1_acc)) if (qvec0_acc.size + pvec1_acc.size) > 0 else np.array([], dtype=float) if p_delta.size > 0 and q_delta.size > 0: try: p_tail_n, which_tail_n, abs_err_n = tails.LR_two_sided(diff_raw, p_delta, q_delta) if which_tail_n == "right": signed_p_n = float(p_tail_n) elif which_tail_n == "left": signed_p_n = -float(p_tail_n) else: signed_p_n = None except Exception: p_tail_n, which_tail_n, abs_err_n, signed_p_n = None, None, None, None else: p_tail_n, which_tail_n, abs_err_n, signed_p_n = None, None, None, None # Minimal row for meter: only what's needed for gauge rendering and metadata. rows_by_turn[n] = { "turn": n, "usable_event_count": int(pvec0_acc.size + pvec1_acc.size), "team0_centred": float(centred0), "team1_centred": float(centred1), "diff_centred": diff_centred, "std_delta_n": std_delta, "p_tail_n": p_tail_n, "which_tail_n": which_tail_n, "abs_err_n": abs_err_n, "signed_p_n": signed_p_n, } final_turn = int(max(rows_by_turn.keys())) if rows_by_turn else 0 final_centred_score = rows_by_turn.get(final_turn, {}).get("diff_centred", 0.0) return { "by_turn": rows_by_turn, "final_turn": final_turn, "final_centred_score": float(final_centred_score), } def _add_td_markers(ax, touchdown_markers, team0_color, team1_color, team0_name, team1_name, x_min=None, x_max=None): """Place 'TD: team_name' annotations above the top frame, stacking multiple TDs per turn. Returns the maximum number of TDs in any single turn (0 if no markers). """ if not touchdown_markers: return 0 trans = _blended_xform(ax.transData, ax.transAxes) by_turn: dict = {} for marker in touchdown_markers: t = marker.get("turn") by_turn.setdefault(t, []).append(marker) max_per_turn = max(len(v) for v in by_turn.values()) _circle_y_base = 1.0 _text_y_base = 1.06 _step_y = 0.10 for td_turn, markers in sorted(by_turn.items()): # Clamp text alignment so it stays within the left/right frame lines. if x_min is not None and x_max is not None and x_min < x_max: _span = x_max - x_min _rel = (float(td_turn) - x_min) / _span # 0.0 = left edge, 1.0 = right edge if _rel <= 0.15: _ha = "left" elif _rel >= 0.85: _ha = "right" else: _ha = "center" else: _ha = "center" for i, marker in enumerate(markers): td_team_id = marker.get("team_id") if td_team_id == 0: color = team0_color name = team0_name elif td_team_id == 1: color = team1_color name = team1_name else: color = "crimson" name = "?" circle_y = _circle_y_base + i * _step_y text_y = _text_y_base + i * _step_y ax.plot( [float(td_turn)], [circle_y], marker="s", markersize=9, color=color, transform=trans, clip_on=False, linestyle="none", zorder=5, ) ax.text( float(td_turn), text_y, f"TD: {name}", transform=trans, ha=_ha, va="bottom", fontsize=10, color=color, clip_on=False, ) return max_per_turn def plot_cumulative_surprise_series_v1( game_state, team_meta=None, calculator=None, gamer0="Team0", gamer1="Team1", max_turn=None, include_extra_bands: bool = False, include_diff_figure: bool = True, ): """Plot cumulative centred surprise curves and centred team-difference curve.""" team0_label = _display_team_name(gamer0, 0, team_meta=team_meta) team1_label = _display_team_name(gamer1, 1, team_meta=team_meta) team0_bare = team0_label.replace(" (team 0)", "").strip() or team0_label team1_bare = team1_label.replace(" (team 1)", "").strip() or team1_label series = build_cumulative_surprise_series_v1( game_state, team_meta=team_meta, calculator=calculator, max_turn=max_turn, include_extra_bands=include_extra_bands, ) by_turn = series.get("by_turn", {}) touchdown_turns = series.get("touchdown_turns", []) touchdown_markers = series.get("touchdown_markers", []) turns = sorted(by_turn.keys()) team0_centred_turn = [by_turn[t]["team0_centred_turn"] for t in turns] team1_centred_turn = [by_turn[t]["team1_centred_turn"] for t in turns] team0_std_turn = [by_turn[t]["std_0_turn"] for t in turns] team1_std_turn = [by_turn[t]["std_1_turn"] for t in turns] team0_centred = [by_turn[t]["team0_centred"] for t in turns] team1_centred = [by_turn[t]["team1_centred"] for t in turns] team0_std = [by_turn[t]["std_0_n"] for t in turns] team1_std = [by_turn[t]["std_1_n"] for t in turns] diff_centred = [by_turn[t]["diff_centred"] for t in turns] diff_std = [by_turn[t]["std_delta_n"] for t in turns] if include_diff_figure: fig_diff, ax_diff = plt.subplots(1, 1, figsize=(11, 4.5)) else: fig_diff, ax_diff = None, None fig_worm, (ax_turn, ax_cum) = plt.subplots(2, 1, figsize=(11, 8), sharex=True) halftime_turn_boundary = 8.5 show_halftime = bool(turns) and min(turns) <= 8 and max(turns) >= 9 team0_color = TEAM0_MPL team1_color = TEAM1_MPL if show_halftime: _axes = [ax_turn, ax_cum] if ax_diff is not None: _axes.insert(0, ax_diff) for ax in _axes: ax.axvline( halftime_turn_boundary, color="0.35", linewidth=1.2, linestyle="--", alpha=0.45, zorder=0, ) _label_axes = [ax_turn] if ax_diff is not None: _label_axes.insert(0, ax_diff) for _ht_ax in _label_axes: _ht_ax.text( halftime_turn_boundary + 0.08, 0.96, "Half time", transform=_ht_ax.get_xaxis_transform(), color="0.35", fontsize=9, alpha=0.8, ha="left", va="top", ) _td_x_min = min(turns) if turns else None _td_x_max = max(turns) if turns else None if ax_diff is not None: _max_td_diff = _add_td_markers( ax_diff, touchdown_markers, team0_color, team1_color, team0_bare, team1_bare, x_min=_td_x_min, x_max=_td_x_max, ) else: _max_td_diff = 0 _max_td_worm = _add_td_markers( ax_turn, touchdown_markers, team0_color, team1_color, team0_bare, team1_bare, x_min=_td_x_min, x_max=_td_x_max, ) # LR quantile bands — pre-computed in build_cumulative_surprise_series_v1. # These are asymmetric: right (positive) and left (negative) thresholds differ # because the delta distribution is generally skewed by the teams' roll profiles. t0_q10_turn_r = [by_turn[t].get("t0_q10_right_turn", 0.0) for t in turns] t0_q10_turn_l = [by_turn[t].get("t0_q10_left_turn", 0.0) for t in turns] t1_q10_turn_r = [by_turn[t].get("t1_q10_right_turn", 0.0) for t in turns] t1_q10_turn_l = [by_turn[t].get("t1_q10_left_turn", 0.0) for t in turns] t0_q10_r = [by_turn[t].get("t0_q10_right_n", 0.0) for t in turns] t0_q10_l = [by_turn[t].get("t0_q10_left_n", 0.0) for t in turns] t1_q10_r = [by_turn[t].get("t1_q10_right_n", 0.0) for t in turns] t1_q10_l = [by_turn[t].get("t1_q10_left_n", 0.0) for t in turns] if ax_diff is not None: d_q10_r = [by_turn[t].get("diff_q10_right_n", 0.0) for t in turns] d_q10_l = [by_turn[t].get("diff_q10_left_n", 0.0) for t in turns] if include_extra_bands: d_q01_r = [by_turn[t].get("diff_q01_right_n", 0.0) for t in turns] d_q01_l = [by_turn[t].get("diff_q01_left_n", 0.0) for t in turns] d_q001_r = [by_turn[t].get("diff_q001_right_n", 0.0) for t in turns] d_q001_l = [by_turn[t].get("diff_q001_left_n", 0.0) for t in turns] # Per-turn bands. ax_turn.fill_between(turns, t0_q10_turn_l, t0_q10_turn_r, color=team0_color, alpha=0.12, label="_nolegend_") ax_turn.fill_between(turns, t1_q10_turn_l, t1_q10_turn_r, color=team1_color, alpha=0.12, label="_nolegend_") # Cumulative per-team bands. ax_cum.fill_between(turns, t0_q10_l, t0_q10_r, color=team0_color, alpha=0.12, label="_nolegend_") ax_cum.fill_between(turns, t1_q10_l, t1_q10_r, color=team1_color, alpha=0.12, label="_nolegend_") if ax_diff is not None: # Diff panel: gauge-style shading — grey centre with the 1-in-10 band by default. # The tighter bands are optional because they are the expensive part. _y_outer = max(max((abs(v) for v in diff_centred), default=1.0), max((abs(v) for v in d_q10_r + d_q10_l), default=1.0)) * 1.5 _large = [_y_outer] * len(turns) _neg_large = [-_y_outer] * len(turns) # Grey centre (between left-1/10 and right-1/10) ax_diff.fill_between(turns, d_q10_l, d_q10_r, color="0.75", alpha=0.35, zorder=0, label="_nolegend_") if include_extra_bands: # Light colour: 1/10 → 1/100 ax_diff.fill_between(turns, d_q10_r, d_q01_r, color=team0_color, alpha=0.18, zorder=0, label="_nolegend_") ax_diff.fill_between(turns, d_q01_l, d_q10_l, color=team1_color, alpha=0.18, zorder=0, label="_nolegend_") # Medium colour: 1/100 → 1/1000 ax_diff.fill_between(turns, d_q01_r, d_q001_r, color=team0_color, alpha=0.28, zorder=0, label="_nolegend_") ax_diff.fill_between(turns, d_q001_l, d_q01_l, color=team1_color, alpha=0.28, zorder=0, label="_nolegend_") # Dark colour: beyond 1/1000 ax_diff.fill_between(turns, d_q001_r, _large, color=team0_color, alpha=0.45, zorder=0, label="_nolegend_") ax_diff.fill_between(turns, _neg_large, d_q001_l, color=team1_color, alpha=0.45, zorder=0, label="_nolegend_") # Threshold lines ax_diff.plot(turns, d_q10_r, color="0.5", linewidth=0.8, linestyle="-", alpha=0.6, zorder=1, label="_nolegend_") ax_diff.plot(turns, d_q10_l, color="0.5", linewidth=0.8, linestyle="-", alpha=0.6, zorder=1, label="_nolegend_") if include_extra_bands: ax_diff.plot(turns, d_q01_r, color=team0_color, linewidth=0.8, linestyle="-", alpha=0.6, zorder=1, label="_nolegend_") ax_diff.plot(turns, d_q01_l, color=team1_color, linewidth=0.8, linestyle="-", alpha=0.6, zorder=1, label="_nolegend_") ax_diff.plot(turns, d_q001_r, color=team0_color, linewidth=0.8, linestyle="--", alpha=0.7, zorder=1, label="_nolegend_") ax_diff.plot(turns, d_q001_l, color=team1_color, linewidth=0.8, linestyle="--", alpha=0.7, zorder=1, label="_nolegend_") # Diff panel line — black so it reads clearly over coloured background. ax_diff.plot( turns, diff_centred, marker="o", color="black", zorder=3, ) ax_diff.axhline(0.0, color="black", linewidth=1, alpha=0.6) _diff_abs_max = max((abs(v) for v in diff_centred), default=1.0) # Show at least the full region boundary that the final point sits in. _final_diff = diff_centred[-1] if diff_centred else 0.0 _final_q10_r = d_q10_r[-1] if d_q10_r else 0.0 _final_q10_l = d_q10_l[-1] if d_q10_l else 0.0 if include_extra_bands: _final_q01_r = d_q01_r[-1] if d_q01_r else 0.0 _final_q01_l = d_q01_l[-1] if d_q01_l else 0.0 _final_q001_r = d_q001_r[-1] if d_q001_r else 0.0 _final_q001_l = d_q001_l[-1] if d_q001_l else 0.0 if _final_diff >= 0: _bounds = [_final_q10_r] if include_extra_bands: _bounds.extend([_final_q01_r, _final_q001_r]) else: _bounds = [abs(_final_q10_l)] if include_extra_bands: _bounds.extend([abs(_final_q01_l), abs(_final_q001_l)]) _region_bound = next((b for b in _bounds if abs(_final_diff) <= b), abs(_final_diff)) _diff_ylim = max(_diff_abs_max, _region_bound) _ylim_top = 1.2 * _diff_ylim ax_diff.set_ylim(-_ylim_top, _ylim_top) # Boundary labels — label each threshold line at the right frame edge (or where it exits). if turns: _label_pairs = [(d_q10_r, "1/10"), (d_q10_l, "1/10")] if include_extra_bands: _label_pairs.extend( [(d_q01_r, "1/100"), (d_q001_r, "1/1000"), (d_q01_l, "1/100"), (d_q001_l, "1/1000")] ) for _bvals, _lbl in _label_pairs: _bvals_arr = np.array(_bvals, dtype=float) _y_last = float(_bvals_arr[-1]) _at_top = _y_last >= 0 _lim = _ylim_top if _at_top else -_ylim_top if abs(_y_last) <= _ylim_top: ax_diff.text(turns[-1], _y_last, f" {_lbl}", color="black", fontsize=9, va=("bottom" if _at_top else "top"), ha="right", zorder=4, clip_on=False) else: _in_mask = np.abs(_bvals_arr) <= _ylim_top if _in_mask.any(): _li = int(np.where(_in_mask)[0][-1]) if _li + 1 < len(turns): _t0v, _t1v = float(turns[_li]), float(turns[_li + 1]) _y0v, _y1v = float(_bvals_arr[_li]), float(_bvals_arr[_li + 1]) _xc = _t0v + (_t1v - _t0v) * (_lim - _y0v) / (_y1v - _y0v) else: _xc = float(turns[-1]) ax_diff.text(_xc, _lim, f" {_lbl}", color="black", fontsize=9, va=("top" if _at_top else "bottom"), ha="left", zorder=4, clip_on=False) ax_diff.set_ylabel("Luck Difference") ax_diff.set_title("") # title set below after tight_layout ax_diff.grid(True, alpha=0.3) ax_turn.plot(turns, team0_centred_turn, marker="o", color=team0_color, label=team0_bare) ax_turn.plot(turns, team1_centred_turn, marker="o", color=team1_color, label=team1_bare) ax_turn.axhline(0.0, color="black", linewidth=1, alpha=0.6) _turn_abs_max = max((abs(v) for v in team0_centred_turn + team1_centred_turn), default=1.0) ax_turn.set_ylim(-1.2 * _turn_abs_max, 1.2 * _turn_abs_max) ax_turn.set_ylabel("Team Luck") ax_turn.set_title("") ax_turn.grid(True, alpha=0.3) ax_turn.legend(loc="upper left") ax_cum.plot(turns, team0_centred, marker="o", color=team0_color, label=team0_bare) ax_cum.plot(turns, team1_centred, marker="o", color=team1_color, label=team1_bare) ax_cum.axhline(0.0, color="black", linewidth=1, alpha=0.6) _cum_abs_max = max((abs(v) for v in team0_centred + team1_centred), default=1.0) ax_cum.set_ylim(-1.2 * _cum_abs_max, 1.2 * _cum_abs_max) ax_cum.set_ylabel("Cumulative Team Luck") ax_cum.set_title("") ax_cum.grid(True, alpha=0.3) ax_cum.legend(loc="upper left") # x-axis: integer tick marks for every turn, starting at 1; labels on all panels. if turns: x_ticks = list(range(max(1, min(turns)), max(turns) + 1)) _axes = [ax_turn, ax_cum] if ax_diff is not None: _axes.insert(0, ax_diff) for ax in _axes: ax.set_xlim(min(x_ticks), max(x_ticks)) ax.set_xticks(x_ticks) ax.tick_params(labelbottom=True) if ax_diff is not None: ax_diff.set_xlabel("Turn") ax_turn.set_xlabel("Turn") ax_cum.set_xlabel("Turn") _btm_diff = (0.10 * _max_td_diff + 0.05) if _max_td_diff > 0 else 0.0 _top_worm = 1.0 - (0.06 * _max_td_worm + 0.04) if _max_td_worm > 0 else 1.0 if fig_diff is not None: fig_diff.tight_layout(rect=[0, _btm_diff, 1, 1]) fig_worm.tight_layout(rect=[0, 0, 1, _top_worm]) fig_worm.subplots_adjust(hspace=0.35) if fig_diff is not None: # Multi-colour title on diff figure, placed after tight_layout so axes size is stable. _t0_bare = team0_bare _t1_bare = team1_bare _title_segs = [ ("(", "black"), (_t0_bare, team0_color), (" luck) \u2212 (", "black"), (_t1_bare, team1_color), (" luck)", "black"), ] _title_fs = 11 fig_diff.canvas.draw() _rdr = fig_diff.canvas.get_renderer() _ax_bbox = ax_diff.get_window_extent(renderer=_rdr) # Measure each segment width using a temporary invisible text object. _seg_widths = [] for _seg, _col in _title_segs: _tmp = ax_diff.text(0, 0, _seg, color=_col, fontsize=_title_fs, transform=ax_diff.transAxes, va="bottom", ha="left", clip_on=False) _seg_widths.append(_tmp.get_window_extent(renderer=_rdr).width) _tmp.remove() _total_title_px = sum(_seg_widths) _start_px = _ax_bbox.x0 + (_ax_bbox.width - _total_title_px) / 2.0 _cur_px = _start_px for (_seg, _col), _sw in zip(_title_segs, _seg_widths): _x_axes = (_cur_px - _ax_bbox.x0) / _ax_bbox.width ax_diff.text(_x_axes, 1.02, _seg, color=_col, fontsize=_title_fs, transform=ax_diff.transAxes, va="bottom", ha="left", clip_on=False) _cur_px += _sw return fig_diff, fig_worm, series def build_mega_dicing_v1_data(game_state, team_meta=None, calculator=None): """Pure data-layer mega dicing V1 computation (no printing).""" usable_rows, kickoff_events = _collect_usable_surprise_rows( game_state, team_meta=team_meta, calculator=calculator, ) team0surprise = [] team1surprise = [] pvec0_list = [] qvec0_list = [] pvec1_list = [] qvec1_list = [] for row in usable_rows: if not isinstance(row, dict): continue team_id = row.get("team_id") if team_id not in (0, 1): continue surprise = float(row.get("surprise", 0.0)) p_success = float(row.get("p_success", 0.0)) p_fail = float(row.get("p_fail", 0.0)) if team_id == 0: team0surprise.append(surprise) pvec0_list.append(p_success) qvec0_list.append(p_fail) else: team1surprise.append(surprise) pvec1_list.append(p_success) qvec1_list.append(p_fail) usable_events = len(usable_rows) if usable_events == 0: return None team0surprise = np.array(team0surprise, dtype=float) team1surprise = np.array(team1surprise, dtype=float) pvec0 = np.array(pvec0_list, dtype=float) qvec0 = np.array(qvec0_list, dtype=float) pvec1 = np.array(pvec1_list, dtype=float) qvec1 = np.array(qvec1_list, dtype=float) score0 = float(np.sum(team0surprise)) score1 = float(np.sum(team1surprise)) k1_0 = float(tails.K1(0, pvec0, qvec0)) if len(pvec0) > 0 else 0.0 k1_1 = float(tails.K1(0, pvec1, qvec1)) if len(pvec1) > 0 else 0.0 # User-defined final centred score: # score_0_N - K1(pvec_0_N,qvec_0_N) - score_1_N + K1(pvec_1_N,qvec_1_N) raw_delta = float(score0 - score1) centred_delta = float((score0 - k1_0) - (score1 - k1_1)) # User-defined p/q for tails analysis: # p = concat(pvec_0_N, qvec_1_N), q = concat(qvec_0_N, pvec_1_N) p_vec = np.concatenate((pvec0, qvec1)) if (len(pvec0) + len(qvec1)) > 0 else np.array([], dtype=float) q_vec = np.concatenate((qvec0, pvec1)) if (len(qvec0) + len(pvec1)) > 0 else np.array([], dtype=float) p_tail, which_tail, abs_err = tails.LR_two_sided(raw_delta, p_vec, q_vec) dist_mean = float(tails.K1(0, p_vec, q_vec)) dist_std = float(np.sqrt(tails.K2(0, p_vec, q_vec))) return { "usable_events": usable_events, "kickoff_events_included": len(kickoff_events), "score_0_n": score0, "score_1_n": score1, "raw_delta": raw_delta, "k1_0_n": k1_0, "k1_1_n": k1_1, "delta_surprise": centred_delta, "dist_mean": dist_mean, "dist_std": dist_std, "p_tail": p_tail, "which_tail": which_tail, "abs_err": abs_err, "one_in_games": (1.0 / p_tail) if isinstance(p_tail, (int, float)) and p_tail > 0 else None, } def build_details_summary(game_state, series, team_meta, gamer0, gamer1): """Assemble all data needed for the Details tab summary table. Returns a dict with keys: coach0, coach1 : gamer display names team0, team1 : team (roster) names race0, race1 : race names td0, td1 : touchdown counts (int) luck0, luck1 : centred luck score at final turn (float) diff_centred : luck0 - luck1 (float) luckier : 0 or 1 (index of team with higher centred luck), or None if equal p_tail : two-sided tail p-value (float or None) abs_err : p-value error estimate (float or None) which_tail : "left" | "right" | None one_in_games_str : "1 in X" formatted string (str) diced_label : short severity label (str), e.g. "regular", "mild", "mega" diced_team : 0 or 1 (which team was diced), or None if no significant dicing """ # ── Per-team luck from final turn of series ──────────────────────────────── by_turn = series.get("by_turn", {}) if series else {} final_turn = int(max(by_turn.keys())) if by_turn else None final_row = by_turn.get(final_turn, {}) if final_turn is not None else {} luck0 = final_row.get("team0_centred") luck1 = final_row.get("team1_centred") diff_centred = final_row.get("diff_centred") p_tail = final_row.get("p_tail_n") abs_err = final_row.get("abs_err_n") which_tail = final_row.get("which_tail_n") # Meter-only series can omit explicit tail metadata; reconstruct from signed p. if p_tail is None: signed_p = final_row.get("signed_p_n") if isinstance(signed_p, (int, float)): p_tail = abs(float(signed_p)) if which_tail is None: which_tail = "right" if signed_p >= 0 else "left" # If team-centred values are unavailable, derive a consistent split from diff. if luck0 is None and luck1 is None and isinstance(diff_centred, (int, float)): if diff_centred >= 0: luck0 = float(diff_centred) luck1 = 0.0 else: luck0 = 0.0 luck1 = float(-diff_centred) # ── Luckier team ─────────────────────────────────────────────────────────── if luck0 is not None and luck1 is not None: if luck0 > luck1: luckier = 0 elif luck1 > luck0: luckier = 1 else: luckier = None else: luckier = None # ── p-value → "1 in X" string ───────────────────────────────────────────── one_in_games_str = f"1 in {format_one_in_games(p_tail)}" if isinstance(p_tail, (int, float)) and p_tail > 0 else "N/A" # ── Severity / diced label (mirrors gauge.py tier logic) ────────────────── # which_tail "right" → team0 luckier → team1 was diced # which_tail "left" → team1 luckier → team0 was diced diced_team = None diced_label = "no significant dicing" if isinstance(p_tail, (int, float)) and p_tail > 0: abs_p = p_tail # p_tail is already two-sided if abs_p > 1.0 / 10: diced_label = "no significant dicing" diced_team = None elif abs_p > 1.0 / 32: diced_label = "mildly diced" elif abs_p > 1.0 / 100: diced_label = "diced" elif abs_p > 1.0 / 320: diced_label = "mega diced" elif abs_p > 1.0 / 1000: diced_label = "comedy mega diced" else: diced_label = "extra comedy mega diced" if diced_label != "no significant dicing": diced_team = 1 if which_tail == "right" else (0 if which_tail == "left" else None) # ── Touchdowns ──────────────────────────────────────────────────────────── td_events = getattr(game_state, "touchdown_events", None) or [] td0 = sum(1 for e in td_events if isinstance(e, dict) and str(e.get("team_id", "")) == "0") td1 = sum(1 for e in td_events if isinstance(e, dict) and str(e.get("team_id", "")) == "1") # ── Team / race names ───────────────────────────────────────────────────── team0 = team_meta.team_label("0") if team_meta else "Team 0" team1 = team_meta.team_label("1") if team_meta else "Team 1" race0 = team_meta.team_race("0") if team_meta else "" race1 = team_meta.team_race("1") if team_meta else "" return { "coach0": gamer0, "coach1": gamer1, "team0": team0, "team1": team1, "race0": race0, "race1": race1, "td0": td0, "td1": td1, "luck0": luck0, "luck1": luck1, "diff_centred": diff_centred, "luckier": luckier, "p_tail": p_tail, "abs_err": abs_err, "which_tail": which_tail, "one_in_games_str": one_in_games_str, "diced_label": diced_label, "diced_team": diced_team, } def _determine_diced_team(result, team0_label, team1_label): """Pure helper: determine which team was diced based on result thresholds.""" if not isinstance(result, dict): return None, None raw_delta = result.get("raw_delta") dist_mean = result.get("dist_mean") which_tail = result.get("which_tail") if not isinstance(raw_delta, (int, float)) or not isinstance(dist_mean, (int, float)): return None, None if which_tail not in ("left", "right"): return None, None if raw_delta > dist_mean and which_tail == "right": return (team0_label, team1_label) elif raw_delta < dist_mean and which_tail == "left": return (team1_label, team0_label) else: return (None, None) def print_mega_dicing_v1_results(result, dicer, dicee, team0_label, team1_label): """Print formatted mega dicing V1 result report.""" if result is None: return print("MEGA DICING V1") print("=" * 72) print(f"Usable roll events: {result['usable_events']}") print(f"Kickoff events included: {result['kickoff_events_included']}") print(f"{team0_label} score_N: {result['score_0_n']:.6f}") print(f"{team1_label} score_N: {result['score_1_n']:.6f}") print(f"{team0_label} K1_N: {result['k1_0_n']:.6f}") print(f"{team1_label} K1_N: {result['k1_1_n']:.6f}") print(f"Raw score difference (T0 - T1): {result['raw_delta']:.6f}") print(f"Combined K1 mean: {result['dist_mean']:.6f}") print(f"Final centred score: {result['delta_surprise']:.6f}") print(f"Distribution mean: {result['dist_mean']:.6f}") print(f"Distribution std : {result['dist_std']:.6f}") print(f"Tail p-value : {result['p_tail']:.6g} +/- {result['abs_err']:.2g}") print(f"Tail side : {result['which_tail']}") print("-" * 72) if dicer is not None and dicee is not None: print(f"{dicer} had more surprising dice relative to mean {result['dist_mean']:.2f}.") print(f"{dicee} was diced with p-value {result['p_tail']:.3g} +/- {result['abs_err']:.1g}.") print(f"A result at least as bad as this is expected roughly one in {format_one_in_games(result['p_tail'])} games.") else: print("Both gamers had same level of surprising dices") def mega_dicing_v1_from_game_state(game_state, gamer0="Team0", gamer1="Team1", team_meta=None, calculator=None): """Presentation wrapper for mega dicing V1 over pure data builder.""" result = build_mega_dicing_v1_data(game_state, team_meta=team_meta, calculator=calculator) if result is None: print("No usable roll events for Mega dicing V1 calculation.") return None team0_label = _display_team_name(gamer0, 0, team_meta=team_meta) team1_label = _display_team_name(gamer1, 1, team_meta=team_meta) dicer, dicee = _determine_diced_team(result, team0_label, team1_label) print_mega_dicing_v1_results(result, dicer, dicee, team0_label, team1_label) return result # ── Roll summary (Success-Fail distribution + dice distribution per player) ── def build_roll_summary(game_state, team_meta=None): """Build a summary of dice roll outcomes grouped by roll category and coach. Returns a dict: { "team_names": {"0": str, "1": str}, "by_category": { team_id: { category: [ { "prob_label": str, # e.g. "67%" "p_expected": float, "success": int, "neutral": int, "fail": int, "total": int, }, ... # sorted by p_expected descending ] } }, "by_coach": { team_id: { category: {die_face (int): count (int)} } }, } Included roll categories: 'action' (includes KO recov), 'armour', 'block', 'injury', 'casualty'. Ball, touchdown, and referee-event categories are excluded. """ import collections # Categories merged into 'action' _MERGE_TO_ACTION = frozenset({"KO recov"}) # Categories excluded from all output _EXCLUDE = frozenset({ "ball", "touchdown", "Officious Ref sent off", "OfficiousRefSendOff", "Argue Call", }) events = [ ev for ev in collect_roll_events_compat(game_state) if not (isinstance(ev, dict) and ev.get("exclude_from_detailed_report")) ] team_names = {"0": "Team 0", "1": "Team 1"} if team_meta is not None: try: team_names["0"] = team_meta.team_label("0") team_names["1"] = team_meta.team_label("1") except Exception: pass # Accumulate by category+probability bucket → success/neutral/fail counts _cat_acc = { "0": collections.defaultdict(lambda: collections.defaultdict(lambda: {"success": 0, "neutral": 0, "fail": 0, "p_expected": 0.0, "difficulty": None})), "1": collections.defaultdict(lambda: collections.defaultdict(lambda: {"success": 0, "neutral": 0, "fail": 0, "p_expected": 0.0, "difficulty": None})), } # Accumulate per coach (team) → category → die face → count _coach_acc = { "0": collections.defaultdict(collections.Counter), "1": collections.defaultdict(collections.Counter), } for ev in events: if ev.get("is_marker_only_action"): continue # Use dice_roller (who physically rolled) for attribution. # For action/block rolls dice_roller == team_id; they differ for armour # and injury rolls, where the attacker rolls against the defender. # Fall back to team_id only when dice_roller is not a known team. dr = str(ev.get("dice_roller") or "") team_id = dr if dr in ("0", "1") else str(ev.get("team_id") or "") if team_id not in ("0", "1"): continue raw_cat = str(ev.get("roll_category") or "other") if raw_cat in _EXCLUDE: continue cat = "action" if raw_cat in _MERGE_TO_ACTION else raw_cat rc = str(ev.get("result_classification") or "neutral") p_suc = ev.get("probability_success") dice_values = ev.get("dice_values") or [] # Round probability to nearest integer percent for bucketing p_pct = int(round((p_suc or 0.0) * 100)) bucket = _cat_acc[team_id][cat][p_pct] if bucket["p_expected"] == 0.0 and p_suc: bucket["p_expected"] = float(p_suc) if bucket["difficulty"] is None: bucket["difficulty"] = ev.get("difficulty") if rc == "success": bucket["success"] += 1 elif rc == "fail": bucket["fail"] += 1 else: bucket["neutral"] += 1 # Accumulate die face counts per coach and category. # Block dice use replay encoding 0-4; all other categories use D6 faces 1-6. for face in dice_values: if not isinstance(face, int): continue if cat == "block": if 0 <= face <= 4: _coach_acc[team_id][cat][face] += 1 else: if 1 <= face <= 6: _coach_acc[team_id][cat][face] += 1 def _action_prob_label(difficulty, p_pct): """Format 'N+ (X%)' or 'N++ (X%)' for a single-die action roll bucket.""" try: n = int(difficulty) except (TypeError, ValueError): return f"{p_pct}%" if not (2 <= n <= 6): return f"{p_pct}%" single_pct = round((7 - n) / 6 * 100) suffix = "+" if p_pct == single_pct else "++" return f"{n}{suffix} ({p_pct}%)" def _armour_prob_label(difficulty, p_pct): """Format 'N+ (X%)' for a 2D6 armour roll bucket.""" try: n = int(difficulty) return f"{n}+ ({p_pct}%)" except (TypeError, ValueError): return f"{p_pct}%" # Reference probabilities for block dice pools (MrMesmer labels) _BLOCK_LABEL_REF = [ (5/6, "1d [5 success] (83%)"), (4/6, "1d [4 success] (67%)"), (3/6, "1d [3 success] (50%)"), (2/6, "1d [2 success] (33%)"), (1/6, "1d [1 success] (16%)"), (35/36, "2d [5 success] (97%)"), (32/36, "2d [4 success] (89%)"), (27/36, "2d [3 success] (75%)"), (20/36, "2d [2 success] (55%)"), (11/36, "2d [1 success] (31%)"), (215/216, "3d [5 success] (100%)"), (208/216, "3d [4 success] (96%)"), (189/216, "3d [3 success] (87%)"), (152/216, "3d [2 success] (70%)"), (91/216, "3d [1 success] (42%)"), (9/36, ":red[red2d] [3 success] (25%)"), (4/36, ":red[red2d] [2 success] (11%)"), (1/36, ":red[red2d] [1 success] (3%)"), ] def _block_prob_label(p_expected): """Match a block roll's probability to the closest MrMesmer label.""" for ref_p, label in _BLOCK_LABEL_REF: if abs(p_expected - ref_p) < 0.015: return label return f"{int(p_expected * 100)}%" # 2D6 cumulative probabilities for computing injury thresholds _2D6_CUMUL = [ (2, 36/36), (3, 35/36), (4, 33/36), (5, 30/36), (6, 26/36), (7, 21/36), (8, 15/36), (9, 10/36), (10, 6/36), (11, 3/36), (12, 1/36), ] def _injury_prob_label(p_expected, p_pct): """Compute 2D6 target from probability and format 'N+ (X%)'.""" for n, ref_p in _2D6_CUMUL: if abs(p_expected - ref_p) < 0.015: return f"{n}+ ({p_pct}%)" return f"{p_pct}%" # Categories excluded from by_category (no meaningful pass/fail probability) _EXCLUDE_FROM_BY_CATEGORY = frozenset({"casualty"}) # Preferred display order for categories _CAT_ORDER = ["action", "block", "armour", "injury", "casualty"] # Build final by_category structure sorted by p_expected descending by_category = {} for tid in ("0", "1"): raw_cats = {} for cat, prob_dict in _cat_acc[tid].items(): if cat in _EXCLUDE_FROM_BY_CATEGORY: continue rows = [] for p_pct, counts in sorted(prob_dict.items(), reverse=True): # Action rows: drop 100% bucket; format remaining as N+/N++ if cat == "action": if p_pct == 100: continue prob_label = _action_prob_label(counts["difficulty"], p_pct) elif cat == "armour": prob_label = _armour_prob_label(counts["difficulty"], p_pct) elif cat == "block": prob_label = _block_prob_label(counts["p_expected"]) elif cat == "injury": prob_label = _injury_prob_label(counts["p_expected"], p_pct) else: prob_label = f"{p_pct}%" total = counts["success"] + counts["neutral"] + counts["fail"] rows.append({ "prob_label": prob_label, "p_expected": counts["p_expected"], "success": counts["success"], "neutral": counts["neutral"], "fail": counts["fail"], "total": total, }) raw_cats[cat] = rows # Apply preferred category ordering ordered = {} for cat in _CAT_ORDER: if cat in raw_cats: ordered[cat] = raw_cats[cat] for cat in raw_cats: if cat not in ordered: ordered[cat] = raw_cats[cat] by_category[tid] = ordered # Build final by_coach structure by_coach = { tid: {cat: dict(counter) for cat, counter in cat_dict.items()} for tid, cat_dict in _coach_acc.items() } return { "team_names": team_names, "by_category": by_category, "by_coach": by_coach, }