from collections import defaultdict from modelling.kickoff_helpers import ( KICKOFF_EVENT_NAME_MAP, calculate_kickoff_total_probabilities, classify_kickoff_event, get_team_kickoff_modifiers, is_blitz_kickoff_event, ) from modelling.modeled_rows import ( compute_v1_team_surprise, materialize_modeled_rows, ) from modelling.roll_events import collect_roll_events_compat from modelling.row_schema import ModeledRowsPayload from modelling.team_roll_helpers import ( is_known_team_id as _is_known_team_id, normalize_team_id, resolve_dice_roller, ) from modelling.turn_model import split_by_kickoff_step_infos from core.turn_allocation_adjustments import ( build_game_turn_map, build_step_turn_context, ) _KICKOFF_CATEGORY_ORDER = { "officious ref sent off": 0, "ko recov": 1, } def _compute_v1_team_surprise(event): return compute_v1_team_surprise(event) def _report_replay_json(team_meta): resolver = getattr(team_meta, "_resolver", None) json_data = getattr(resolver, "json_data", None) return json_data if isinstance(json_data, dict) else None def _extract_coin_toss_choice_info(team_meta): json_data = _report_replay_json(team_meta) replay = json_data.get("Replay", {}) if isinstance(json_data, dict) else {} replay_steps = replay.get("ReplayStep", []) if isinstance(replay, dict) else [] if isinstance(replay_steps, dict): replay_steps = [replay_steps] question_payload = None choice_payload = None choice_step = None for step in replay_steps: if not isinstance(step, dict): continue if question_payload is None and "EventQuestionKickingChoice" in step: question_payload = step.get("EventQuestionKickingChoice") if choice_payload is None and "EventKickingChoice" in step: choice_payload = step.get("EventKickingChoice") choice_step = step if question_payload is not None and choice_payload is not None: break chooser_team_id = None source = None if isinstance(choice_payload, dict): chooser_team_id = normalize_team_id(choice_payload.get("GamerId")) if _is_known_team_id(chooser_team_id): source = "EventKickingChoice" if chooser_team_id is None and isinstance(question_payload, dict): chooser_team_id = normalize_team_id(question_payload.get("GamerId")) if _is_known_team_id(chooser_team_id): source = "EventQuestionKickingChoice" if chooser_team_id is None and isinstance(choice_step, dict): active_gamer_changed = choice_step.get("EventActiveGamerChanged") if isinstance(active_gamer_changed, dict): chooser_team_id = normalize_team_id(active_gamer_changed.get("NewActiveGamer")) if _is_known_team_id(chooser_team_id): source = "EventActiveGamerChanged" else: chooser_team_id = None if not _is_known_team_id(chooser_team_id): chooser_team_id = None chose_receive = None if isinstance(choice_payload, dict): receive_raw = choice_payload.get("Receive") if receive_raw is not None: chose_receive = str(receive_raw) == "1" return { "chooser_team_id": chooser_team_id, "chose_receive": chose_receive, "source": source, } def build_coin_toss_report_row(turn_key, team_meta): info = _extract_coin_toss_choice_info(team_meta) chooser_team_id = info.get("chooser_team_id") if _is_known_team_id(chooser_team_id): team_label = team_meta.team_label(chooser_team_id) if team_meta is not None else f"Team {chooser_team_id}" note_parts = [f"choice by {team_label}"] chose_receive = info.get("chose_receive") if isinstance(chose_receive, bool): note_parts.append("chose receive" if chose_receive else "chose kick") source = info.get("source") if source: note_parts.append(source) return { "step_number": "0", "game_turn": turn_key, "roll_category": "kickoff", "display_action_type": "Coin toss", "report_result_classification": "success", "player_name": team_label, "team_id": chooser_team_id, "dice_roller": chooser_team_id, "target_player_name": "kick/recv", "dice_values": [], "dice_total": "-", "difficulty": "-", "probability_success": 0.5, "probability_neutral": 0.0, "probability_fail": 0.5, "result_value": 1, "payload_type": "SyntheticCoinToss", "result_name": "coin toss", "notes": "; ".join(note_parts), } return { "step_number": "0", "game_turn": turn_key, "roll_category": "kickoff", "display_action_type": "Coin toss", "report_result_classification": "unknown", "player_name": "-", "team_id": None, "dice_roller": None, "target_player_name": "-", "dice_values": [], "dice_total": "-", "difficulty": "-", "probability_success": None, "probability_neutral": None, "probability_fail": None, "result_value": None, "payload_type": "SyntheticCoinToss", "result_name": "coin toss", "notes": "coin toss info not available", "exclude_from_surprise": True, } def _build_step_context(game_state): context = {} for step in getattr(game_state, "steps", []) or []: step_num = getattr(step, "step_number", None) if step_num is None: continue context[step_num] = { "game_turn": getattr(step, "game_turn", None), "active_team_id": normalize_team_id(getattr(step, "active_team_id", None)), "turn_owner_team_id": normalize_team_id(getattr(step, "turn_owner_team_id", None)), "explicit_turn_end": bool(getattr(step, "explicit_turn_end", False)), "end_turn_reason": getattr(step, "end_turn_reason", None), "end_turn_type": getattr(step, "end_turn_type", None), "next_active_team_id": normalize_team_id(getattr(step, "next_active_team_id", None)), } return context def _build_kickoff_team_by_turn(game_state): result = {} for kickoff_event in getattr(game_state, "kick_off_events", []) or []: turn_key = str(getattr(kickoff_event, "game_turn", None)) team_id = normalize_team_id(getattr(kickoff_event, "team_id", None)) if turn_key != "None" and _is_known_team_id(team_id): result[turn_key] = team_id return result def _build_kickoff_steps_by_turn(game_state): all_steps = getattr(game_state, "steps", []) or [] result = defaultdict(list) for step in all_steps: if step.kick_off: turn_key = str(step.game_turn) if step.game_turn is not None else "None" step_num = step.step_number team_id = normalize_team_id(getattr(step.kick_off, "team_id", None)) result[turn_key].append((step_num, team_id)) # For kickoffs where kicker team_id is None, infer it from the first step # after the kickoff that carries a known active_team_id in a new game_turn. # The receiving team goes first after the kick; kicker = opponent(receiver). if any(team_id is None for items in result.values() for _, team_id in items): # Build step_number → active_team_id for quick lookup step_active = { s.step_number: normalize_team_id(s.active_team_id) for s in all_steps if s.active_team_id is not None } step_list = sorted(all_steps, key=lambda s: s.step_number) for turn_key, items in result.items(): for idx, (step_num, team_id) in enumerate(items): if team_id is not None: continue # Find the first subsequent step with a new game_turn and known active team kickoff_gt = None for s in step_list: if s.step_number == step_num: kickoff_gt = str(s.game_turn) if s.game_turn is not None else None break inferred_receiver = None for s in step_list: if s.step_number <= step_num: continue sgt = str(s.game_turn) if s.game_turn is not None else None if sgt != kickoff_gt and s.active_team_id is not None and not s.explicit_turn_end: inferred_receiver = normalize_team_id(s.active_team_id) break if inferred_receiver in ("0", "1"): inferred_kicker = "1" if inferred_receiver == "0" else "0" items[idx] = (step_num, inferred_kicker) return { key: sorted(value, key=lambda x: int(x[0]) if str(x[0]).isdigit() else 999999) for key, value in result.items() } def _turn_sort_key(turn_value): turn_string = str(turn_value) return int(turn_string) if turn_string.isdigit() else 999999 def _step_sort_key(event): step_number = event.get("step_number") try: return int(step_number) except (TypeError, ValueError): return 999999 def _build_step_possession_map(turn_events, fallback_active_team_id=None): by_step = defaultdict(list) for event in turn_events: by_step[event.get("step_number")].append(event) step_numbers = sorted( by_step.keys(), key=lambda value: int(value) if isinstance(value, (int, str)) and str(value).isdigit() else 999999, ) possession_by_step = {} current_possession = fallback_active_team_id if _is_known_team_id(fallback_active_team_id) else None for step_num in step_numbers: step_events = by_step[step_num] # Read the canonical owner set by the processor; carry forward on unset steps. for event in step_events: maybe_owner = normalize_team_id(event.get("turn_owner_team_id")) if _is_known_team_id(maybe_owner): current_possession = maybe_owner break possession_by_step[step_num] = current_possession return possession_by_step def _is_prob_one(value): return isinstance(value, (int, float)) and abs(float(value) - 1.0) < 1e-9 def _is_kickoff_related_event(event): payload = str(event.get("payload_type") or "").lower() result = str(event.get("result_name") or "").lower() text = f"{payload}|{result}" keywords = ( "kickoff", "kick_off", "korecovery", "ko_recovery", "ko-recov", "ko recov", "officiousref", "officious ref", "arguethecall", "argue", "bribe", ) return any(keyword in text for keyword in keywords) def _is_bounce_like_event(event): payload = str(event.get("payload_type") or "") result = str(event.get("result_name") or "") roll_category = str(event.get("roll_category") or "") player_name = str(event.get("player_name") or "") team_id = normalize_team_id(event.get("team_id")) target_name = event.get("target_player_name") is_neutral_bounce = _is_prob_one(event.get("probability_neutral")) is_success_bounce = _is_prob_one(event.get("probability_success")) return ( payload == "ResultRoll" and result == "ResultRoll" and roll_category in ("action", "ball") and (is_neutral_bounce or is_success_bounce) and not _is_known_team_id(team_id) and player_name in ("", "Unknown", "Ball Bounce") and target_name in (None, "", "Unknown") ) def _step_stays_in_kickoff_phase(step_events): if not step_events: return False if any(str(event.get("roll_category") or "") == "armour" for event in step_events): return False if any(_is_kickoff_related_event(event) for event in step_events): return True # A catch attempt immediately following a kickoff is the ball reception, # which is part of the kickoff procedure (not the start of regular play). # This check is only reached when kickoff_open=True in the caller. if any(str(event.get("roll_type") or "") == "7" for event in step_events): return True return all(_is_bounce_like_event(event) for event in step_events) def _absorb_and_split_blitz_kickoff_sections(turn_sections, step_context, tt_step_context=None): """For Blitz! kickoff results, perform two actions in one pass: 1. Absorb: the ball catch/scatter and the BLITZ activation sections are merged into the kickoff section. They are direct mechanical consequences of the kickoff table result, not independent team turns. The endpoint of absorption is the team_turn section whose last step has end_turn_type=6 and explicit_turn_end=True (the moment the game clock advances after the BLITZ activation). 2. Split: after absorption, any team_turn sections in the same entry that belong to the KICKING team are promoted to turn N+1. The kicking team already used their regular turn as the BLITZ activation, so their next section belongs to the following turn. _cascade_deduplicate_turn_sections then merges the promoted sections with the existing t+1 entry. """ if not step_context: return turn_sections result = [] for turn_entry in turn_sections: turn_key = turn_entry["turn_key"] sections = turn_entry.get("sections", []) try: turn_num = int(turn_key) except (TypeError, ValueError): result.append(turn_entry) continue kickoff_idx = next( (i for i, sec in enumerate(sections) if sec.get("phase") == "kickoff"), None, ) if kickoff_idx is None: result.append(turn_entry) continue kickoff_sec = sections[kickoff_idx] kicker_team = kickoff_sec.get("team_id") # If this kickoff was bumped from the previous turn (raw game_turn < # current turn_num) AND there is a second kickoff section later in this # entry, the first kickoff + any team_turn sections before the second # kickoff belong to turn N-1. Move them back. kickoff_step_num = kickoff_sec.get("kickoff_step_number") if ( kickoff_idx == 0 and kickoff_step_num is not None and step_context and result ): # A kickoff is "bumped" when TinTuna's lookahead-advance moved it # from gt_tt = turn_num-1 into the current turn. Compare the # pre-lookahead TinTuna turn (gt_tt) against turn_num. tt_gt = None if tt_step_context: tt_gt = (tt_step_context.get(kickoff_step_num) or {}).get("gt_tt") was_bumped = ( tt_gt is not None and isinstance(tt_gt, int) and tt_gt < turn_num ) if was_bumped and str(result[-1].get("turn_key", "")) == str(turn_num - 1): second_kickoff_idx = next( ( i for i in range(kickoff_idx + 1, len(sections)) if sections[i].get("phase") == "kickoff" ), None, ) if second_kickoff_idx is not None: sections_to_prev = sections[:second_kickoff_idx] sections_remaining = sections[second_kickoff_idx:] # If the kickoff section is immediately followed by a section # whose last step has end_turn_type=6, that section is the # tail of the kickoff phase (BLITZ blocks / armour resolved at # the phase-end clock advance). Merge it into the kickoff. if len(sections_to_prev) >= 2: ko_sec = sections_to_prev[0] next_sec = sections_to_prev[1] nxt_steps = [ e.get("step_number") for e in next_sec.get("events", []) if isinstance(e.get("step_number"), int) ] if nxt_steps: last_nxt = max(nxt_steps) if any( str(e.get("end_turn_type", "")) == "6" for e in next_sec.get("events", []) if e.get("step_number") == last_nxt ): merged_ko = dict(ko_sec) merged_ko["events"] = ( list(ko_sec.get("events", [])) + list(next_sec.get("events", [])) ) sections_to_prev = [merged_ko] + list(sections_to_prev[2:]) prev_entry = result[-1] result[-1] = { **prev_entry, "sections": prev_entry.get("sections", []) + sections_to_prev, } result.append({**turn_entry, "sections": sections_remaining}) continue # Find the BLITZ endpoint: first team_turn after kickoff whose last step # has end_turn_type=6 + explicit_turn_end for that team (the game-clock # advance signal at the end of the BLITZ activation). blitz_end_idx = None for i in range(kickoff_idx + 1, len(sections)): sec = sections[i] if sec.get("phase") != "team_turn": continue steps = [ e.get("step_number") for e in sec.get("events", []) if isinstance(e.get("step_number"), int) ] if not steps: continue last_sn = max(steps) sctx = step_context.get(last_sn, {}) if ( str(sctx.get("end_turn_type", "")) == "6" and sctx.get("explicit_turn_end", False) and sctx.get("turn_owner_team_id") == sec.get("team_id") ): blitz_end_idx = i break if blitz_end_idx is None: result.append(turn_entry) continue # Step 1: absorb sections from kickoff+1 through blitz_end_idx into # the kickoff section. merged_kickoff = dict(kickoff_sec) absorbed_events = list(kickoff_sec.get("events", [])) for i in range(kickoff_idx + 1, blitz_end_idx + 1): absorbed_events.extend(sections[i].get("events", [])) merged_kickoff["events"] = absorbed_events # All remaining sections after the absorbed block stay in the current # turn. The BLITZ activation is a kickoff-phase bonus; both teams # still play their regular turn in turn N. pre = sections[:kickoff_idx] remaining = sections[blitz_end_idx + 1 :] current_secs = pre + [merged_kickoff] + remaining result.append({**turn_entry, "turn_key": str(turn_num), "sections": current_secs}) return result def _cascade_deduplicate_turn_sections(turn_sections): """After a carryover-advance split, some turn_key values may appear in multiple entries where one team's section was promoted from the previous turn. This function merges all entries that share a turn_key, ensures each team appears at most once per turn (first-wins), and cascades any duplicate sections as pending overflow into the next turn. Only non-kickoff entries are candidates for the cascade merge; kickoff- containing entries are left untouched. """ # Quick exit: if every turn_key is unique there is nothing to do. seen: set = set() has_dup = False for entry in turn_sections: k = entry["turn_key"] if k in seen: has_dup = True break seen.add(k) if not has_dup: return turn_sections result: list = [] # pending: (turn_num_int, [sections], template_entry) pending_num: int | None = None pending_secs: list = [] pending_tpl: dict = {} i = 0 while i < len(turn_sections): entry = turn_sections[i] turn_key = entry["turn_key"] try: turn_num = int(turn_key) except (TypeError, ValueError): if pending_num is not None: result.append({**pending_tpl, "turn_key": str(pending_num), "sections": pending_secs}) pending_num = None result.append(entry) i += 1 continue sections = entry.get("sections", []) has_kickoff = any(s.get("phase") == "kickoff" for s in sections) if has_kickoff: # Do not cascade into or through kickoff entries. if pending_num is not None: result.append({**pending_tpl, "turn_key": str(pending_num), "sections": pending_secs}) pending_num = None result.append(entry) i += 1 continue # Collect all consecutive non-kickoff entries that share this turn_key. same_entries = [entry] j = i + 1 while j < len(turn_sections): nxt = turn_sections[j] if nxt["turn_key"] != turn_key: break if any(s.get("phase") == "kickoff" for s in nxt.get("sections", [])): break same_entries.append(nxt) j += 1 # Build the combined section list: pending overflow (if it belongs to # this turn) followed by all sections from same_entries. if pending_num is not None and pending_num == turn_num: all_sections = pending_secs + [s for e in same_entries for s in e.get("sections", [])] pending_num = None pending_secs = [] pending_tpl = {} else: if pending_num is not None: # Pending is for an earlier turn that had no matching entry — flush it. result.append({**pending_tpl, "turn_key": str(pending_num), "sections": pending_secs}) pending_num = None all_sections = [s for e in same_entries for s in e.get("sections", [])] # Assign first occurrence of each team to this turn; overflow to next. assigned_teams: set = set() current_secs: list = [] overflow_secs: list = [] for sec in all_sections: if sec.get("phase") != "team_turn": current_secs.append(sec) continue tid = sec.get("team_id") if tid not in assigned_teams: assigned_teams.add(tid) current_secs.append(sec) else: overflow_secs.append(sec) template = same_entries[0] result.append({**template, "turn_key": str(turn_num), "sections": current_secs}) if overflow_secs: pending_num = turn_num + 1 pending_secs = overflow_secs pending_tpl = template i = j # Flush any remaining pending at the end. if pending_num is not None: result.append({**pending_tpl, "turn_key": str(pending_num), "sections": pending_secs}) return result def _merge_kickoff_fragmented_sections(sections): """ Re-join team_turn fragments split by a kickoff boundary that belong to the same team's turn. Pattern 1 – pre-kickoff stray: [team_turn team=X] [kickoff] … [team_turn team=X] → merge pre-kickoff events into the later post-kickoff section. Pattern 1.5 – post-kickoff same-as-kicker (KO recovery / carryover): [team_turn team=X] [kickoff team=X, non-Blitz] [team_turn team=X] … → merge post-kickoff stray events back into the pre-kickoff section. Blitz kickoff events produce a correct second section for the kicker and are left untouched: Pattern 1.5 is skipped when the kickoff section contains a Blitz! event. Pattern 1 is safe for Blitz because the kicker's blitz-section appears *after* the kickoff, never before. """ if not any(s.get("phase") == "kickoff" for s in sections): return sections sections = list(sections) changed = True while changed: changed = False for i, sec in enumerate(sections): if sec.get("phase") != "kickoff": continue # Pattern 1: single-step stray team_turn immediately BEFORE this kickoff. # Only fires when the stray is within 2 steps of the kickoff — i.e. a # genuine one-step artifact of EventActiveGamerChanged just before the # kick (e.g. KO-recovery action fires on step N, kickoff fires on N+1). # Multi-step sections that are far from the kickoff are real turn play # (e.g. both teams completing their turns before a mid-game kickoff # recorded out of step order) and must NOT be merged, otherwise the # kickoff section ends up displayed before their earlier steps. if i > 0 and sections[i - 1].get("phase") == "team_turn": pre_team = sections[i - 1].get("team_id") if _is_known_team_id(pre_team): # Prefer the stored kickoff boundary step (recorded when the # section was created) over the min of event steps, because the # kickoff section may contain post-boundary events (e.g. the # ball reception catch) that are several steps past the actual # kick step but still part of the kickoff procedure. ko_step = sec.get("kickoff_step_number") if ko_step is None: ko_step = min( (e.get("step_number") for e in sec.get("events", []) if isinstance(e.get("step_number"), (int, float))), default=None, ) pre_steps = [ e.get("step_number") for e in sections[i - 1].get("events", []) if isinstance(e.get("step_number"), (int, float)) ] pre_min_step = min(pre_steps, default=None) # Use min step (not max) so the entire stray section must be # adjacent to the kickoff. Sections that start far from the # kickoff (e.g. TD/KO-recovery at step 223 before kickoff at # 267) are real inter-turn content, not strays. adjacent = ( ko_step is not None and pre_min_step is not None and ko_step - pre_min_step <= 2 ) if adjacent: for j in range(i + 1, len(sections)): if ( sections[j].get("phase") == "team_turn" and sections[j].get("team_id") == pre_team ): sections[j] = { **sections[j], "events": sections[i - 1]["events"] + sections[j]["events"], } sections.pop(i - 1) changed = True break if changed: break # Pattern 1.5: team_turn immediately AFTER a non-Blitz kickoff, # same team as kicker, with an earlier section for that team present ko_team = sec.get("team_id") if ( _is_known_team_id(ko_team) and i + 1 < len(sections) and sections[i + 1].get("phase") == "team_turn" and sections[i + 1].get("team_id") == ko_team ): ko_is_blitz = any(is_blitz_kickoff_event(e) for e in sec.get("events", [])) if not ko_is_blitz: for j in range(i - 1, -1, -1): if ( sections[j].get("phase") == "team_turn" and sections[j].get("team_id") == ko_team ): sections[j] = { **sections[j], "events": sections[j]["events"] + sections[i + 1]["events"], } sections.pop(i + 1) changed = True break if changed: break return sections def _split_turn_into_phases(turn_events, kickoff_team_id=None, fallback_active_team_id=None, kickoff_step_infos=None): ordered = sorted(turn_events, key=_step_sort_key) if not ordered: return [] possession_by_step = _build_step_possession_map(ordered, fallback_active_team_id=fallback_active_team_id) events_with_possession = [ {**event, "effective_active_team_id": possession_by_step.get(event.get("step_number"))} for event in ordered ] sections = split_by_kickoff_step_infos( events_with_possession, kickoff_step_infos or [], kickoff_team_id=kickoff_team_id, fallback_team_id=fallback_active_team_id, is_known_team_id=_is_known_team_id, is_kickoff_related_event=_is_kickoff_related_event, step_stays_in_kickoff_phase=_step_stays_in_kickoff_phase, ) return _merge_kickoff_fragmented_sections(sections) def _normalize_events_for_detailed_report(events, step_context): normalized_events = [] for event in events: if not isinstance(event, dict): continue event_copy = dict(event) event_copy["_source_event_id"] = id(event) step_number = event_copy.get("step_number") step_info = step_context.get(step_number, {}) if step_number is not None else {} if event_copy.get("game_turn") is None: event_copy["game_turn"] = step_info.get("game_turn") event_copy["active_team_id"] = normalize_team_id(step_info.get("active_team_id")) event_copy["turn_owner_team_id"] = normalize_team_id(step_info.get("turn_owner_team_id")) event_copy["explicit_turn_end"] = bool(step_info.get("explicit_turn_end", False)) event_copy["end_turn_reason"] = step_info.get("end_turn_reason") event_copy["end_turn_type"] = step_info.get("end_turn_type") event_copy["next_active_team_id"] = normalize_team_id(step_info.get("next_active_team_id")) event_copy["team_id"] = normalize_team_id(event_copy.get("team_id")) preserved_dice_roller = normalize_team_id(event_copy.get("dice_roller")) if preserved_dice_roller in ("0", "1"): event_copy["dice_roller"] = preserved_dice_roller else: event_copy["dice_roller"] = resolve_dice_roller( event_copy.get("team_id"), event_copy.get("roll_category"), ) normalized_events.append(event_copy) return normalized_events def _remap_kickoff_mappings_after_turn_reassignment( normalized_events, kickoff_steps_by_turn, tt_gt_map=None ): kickoff_team_by_step = {} for turn_key, step_infos in (kickoff_steps_by_turn or {}).items(): for step_num, team_id in step_infos: kickoff_team_by_step[step_num] = team_id event_turn_by_step = {} for event in normalized_events: step_num = event.get("step_number") if not isinstance(step_num, int) or step_num not in kickoff_team_by_step: continue event_turn_by_step.setdefault(step_num, str(event.get("game_turn"))) remapped_kickoff_steps_by_turn = defaultdict(list) remapped_kickoff_team_by_turn = {} for step_num, team_id in kickoff_team_by_step.items(): remapped_turn = event_turn_by_step.get(step_num) if remapped_turn is None: # No events at this step (e.g. kickoff table roll not captured as a # roll event). Fall back to the TinTuna game-turn map. if tt_gt_map: remapped_turn = tt_gt_map.get(step_num) if remapped_turn is None: continue remapped_kickoff_steps_by_turn[remapped_turn].append((step_num, team_id)) remapped_kickoff_team_by_turn.setdefault(remapped_turn, team_id) remapped_kickoff_steps_by_turn = { turn_key: sorted(infos, key=lambda x: int(x[0]) if str(x[0]).isdigit() else 999999) for turn_key, infos in remapped_kickoff_steps_by_turn.items() } remapped_kickoff_team_by_turn = dict(remapped_kickoff_team_by_turn) return remapped_kickoff_steps_by_turn, remapped_kickoff_team_by_turn def build_report_turn_sections(game_state, events=None, target_turn=None): if events is None: events = collect_roll_events_compat(game_state, with_step_defaults=True) raw_replay_steps = getattr(game_state, "raw_replay_steps", []) or [] # tt_step_ctx: per-step pre-lookahead step-turn context (used by was_bumped # check in _absorb_and_split_blitz_kickoff_sections). # tt_gt_map: post-lookahead step→turn map (used for the kickoff-mapping # fallback in _remap_kickoff_mappings_after_turn_reassignment). # Note: event game_turn is already step-turn-corrected by the processor # (run_step_turn_stage), so no post-hoc _apply_event_game_turns is needed. tt_step_ctx = build_step_turn_context(raw_replay_steps) # Reuse the already-built context so build_game_turn_map doesn't re-scan steps. tt_gt_map = build_game_turn_map(raw_replay_steps, _context=tt_step_ctx) # step_context (board-state) is still needed by # _absorb_and_split_blitz_kickoff_sections for BLITZ activation boundary # detection (end_turn_type, explicit_turn_end). step_context = _build_step_context(game_state) kickoff_team_by_turn = _build_kickoff_team_by_turn(game_state) kickoff_steps_by_turn = _build_kickoff_steps_by_turn(game_state) normalized_events = _normalize_events_for_detailed_report(events, step_context) kickoff_steps_by_turn, kickoff_team_by_turn = _remap_kickoff_mappings_after_turn_reassignment( normalized_events, kickoff_steps_by_turn, tt_gt_map=tt_gt_map, ) if target_turn is not None: normalized_events = [event for event in normalized_events if str(event.get("game_turn")) == str(target_turn)] by_turn = defaultdict(list) for event in normalized_events: by_turn[str(event.get("game_turn"))].append(event) turn_sections = [] for turn_key in sorted(by_turn.keys(), key=_turn_sort_key): turn_events = by_turn[turn_key] fallback_active_team_id = None for event in sorted(turn_events, key=_step_sort_key): if _is_known_team_id(event.get("active_team_id")): fallback_active_team_id = event.get("active_team_id") break sections = _split_turn_into_phases( turn_events, kickoff_team_id=kickoff_team_by_turn.get(turn_key), fallback_active_team_id=fallback_active_team_id, kickoff_step_infos=kickoff_steps_by_turn.get(turn_key, []), ) turn_sections.append( { "turn_key": turn_key, "kickoff_team_id": kickoff_team_by_turn.get(turn_key), "kickoff_step_infos": list(kickoff_steps_by_turn.get(turn_key, [])), "sections": sections, } ) # For Blitz! kickoff results: absorb the catch/scatter and BLITZ activation # into the kickoff section. turn_sections = _absorb_and_split_blitz_kickoff_sections( turn_sections, step_context, tt_step_context=tt_step_ctx ) # Merge duplicate turn_key entries (may arise from _absorb_and_split or # other edge cases). turn_sections = _cascade_deduplicate_turn_sections(turn_sections) return turn_sections def _build_v1_surprise_by_event_id(events): result = {} for event in events: t0, t1 = _compute_v1_team_surprise(event) if t0 is None and t1 is None: continue result[id(event)] = { "team0_surprise": t0, "team1_surprise": t1, } return result def _build_detailed_section_header(phase, turn_key, section_team_id, section_events, kickoff_report_rows, team_meta): def _team_label(team_id): if team_meta is None: return f"Team {team_id}" if _is_known_team_id(team_id) else "Unknown" return team_meta.team_label(team_id) if phase == "kickoff": kickoff_row_count = len(section_events) + len(kickoff_report_rows or []) return f"Kickoff phase turn {turn_key} (kicking {_team_label(section_team_id)}) - {kickoff_row_count} rows" if _is_known_team_id(section_team_id): return f"{_team_label(section_team_id)} turn {turn_key} - {len(section_events)} roll events" return f"Team Unknown turn {turn_key} - {len(section_events)} roll events" def _consume_pending_blitz_team_turn_label(header, phase, pending_blitz_team_turn_label): if phase == "team_turn" and pending_blitz_team_turn_label: return f"BLITZ ({header})", False return header, pending_blitz_team_turn_label def _kickoff_event_name(table_event_num): return KICKOFF_EVENT_NAME_MAP.get(str(table_event_num), "Unknown") def build_detailed_report_kickoff_row( kickoff_event, step_number, team_meta=None, calculator=None, kickoff_modifiers=None, kickoff_totals_cache=None, kicker_team_fallback=None, ): if kickoff_event is None: return None dice_values = kickoff_event.get_dice_values() dice_total = "-" try: dice_total_int = sum(int(value) for value in dice_values) dice_total = dice_total_int table_event_num = str(dice_total_int) except (TypeError, ValueError): table_event_num = "?" classification = classify_kickoff_event(table_event_num, kickoff_event) classification_value = classification.get("value") if classification_value == 1: result_value = 1 elif classification_value == -1: result_value = -1 else: result_value = 0 totals = calculate_kickoff_total_probabilities( kickoff_event, team_meta=team_meta, calculator=calculator, kickoff_modifiers=kickoff_modifiers, cache=kickoff_totals_cache, ) probability_success = ( totals.get("success", totals.get("pass")) if isinstance(totals, dict) else None ) probability_neutral = totals.get("neutral") if isinstance(totals, dict) else None probability_fail = totals.get("fail") if isinstance(totals, dict) else None event_name = _kickoff_event_name(table_event_num) next_event_name = kickoff_event.next_event or event_name target_name = event_name if next_event_name and next_event_name != event_name: target_name = f"{event_name} -> {next_event_name}" # Use the event's own team_id when available; fall back to the inferred # kicker team passed in from kickoff_steps_by_turn (e.g. for anonymous # opening kicks where the kicker player/team is not recorded). raw_team_id = kickoff_event.team_id effective_team_id = normalize_team_id(raw_team_id) if raw_team_id is not None else kicker_team_fallback return { "step_number": step_number, "roll_category": "kickoff", "report_result_classification": classification.get("label", "unknown"), "player_name": kickoff_event.player_name or "Unknown", "team_id": effective_team_id, "dice_roller": resolve_dice_roller(effective_team_id, "kickoff"), "target_player_name": target_name, "dice_values": dice_values, "dice_total": dice_total, "difficulty": table_event_num, "probability_success": probability_success, "probability_neutral": probability_neutral, "probability_fail": probability_fail, "result_value": result_value, "payload_type": "EventKickOffTable", "result_name": next_event_name, "display_action_type": "Kickoff", } def build_detailed_modeled_sections(game_state, team_meta, calculator=None, target_turn=None): events = collect_roll_events_compat(game_state) if not events: return [] kickoff_event_by_step = { step.step_number: step.kick_off for step in getattr(game_state, "steps", []) or [] if getattr(step, "kick_off", None) } kickoff_modifiers = get_team_kickoff_modifiers(team_meta) if team_meta is not None else {} kickoff_totals_cache = {} ottd_attempt_turns = getattr(game_state, "ottd_attempt_turns", set()) or set() ottd_evaluations = getattr(game_state, "ottd_evaluations", {}) or {} turn_sections = build_report_turn_sections(game_state, events=events, target_turn=target_turn) output = [] coin_toss_added = False for turn_info in turn_sections: turn_key = turn_info["turn_key"] sections = turn_info["sections"] turn_kickoff_steps = list(turn_info["kickoff_step_infos"]) section_outputs = [] # An OTTD attempt turn is a half-ending turn (8 or 16) that contains a # kickoff section – meaning a touchdown just triggered the kick, leaving # the receiving team exactly one turn in the half. # Only the team_turn section that appears AFTER the kickoff section is the # OTTD attempt; earlier team_turn sections (the scoring team's turn) are not. try: _turn_num = int(str(turn_key)) except (TypeError, ValueError): _turn_num = -1 _is_ottd_candidate_turn = _turn_num in ottd_attempt_turns # Build a per-section flag tracking kickoff visibility so the OTTD marker # is only placed on team_turn sections that follow the kickoff. _kickoff_seen_in_turn = False for section in sections: phase = section.get("phase") if phase == "kickoff": _kickoff_seen_in_turn = True section_team_id = section.get("team_id") section_events = [ event for event in (section.get("events") or []) if not (isinstance(event, dict) and event.get("exclude_from_detailed_report")) ] if phase == "kickoff": section_events = sorted( section_events, key=lambda e: ( _step_sort_key(e), _KICKOFF_CATEGORY_ORDER.get(str(e.get("roll_category", "")).lower(), 2), ), ) else: # Preserve original within-step order for all rows, but place # KO recovery rows at the end of the same step. section_events = sorted( enumerate(section_events), key=lambda pair: ( _step_sort_key(pair[1]), 1 if str(pair[1].get("roll_category", "")).lower() == "ko recov" else 0, pair[0], ), ) section_events = [ event for _, event in section_events ] surprise_by_event_id = _build_v1_surprise_by_event_id(section_events) kickoff_report_rows = [] if phase == "kickoff" and turn_kickoff_steps: if not coin_toss_added: kickoff_report_rows.append(build_coin_toss_report_row(turn_key, team_meta)) coin_toss_added = True kickoff_step_num, inferred_kicker_team = turn_kickoff_steps.pop(0) kickoff_event = kickoff_event_by_step.get(kickoff_step_num) kickoff_report_row = build_detailed_report_kickoff_row( kickoff_event, kickoff_step_num, team_meta=team_meta, calculator=calculator, kickoff_modifiers=kickoff_modifiers, kickoff_totals_cache=kickoff_totals_cache, kicker_team_fallback=inferred_kicker_team, ) if isinstance(kickoff_report_row, dict): kickoff_report_rows.append(kickoff_report_row) header = _build_detailed_section_header( phase, turn_key, section_team_id, section_events, kickoff_report_rows, team_meta, ) section_outputs.append( { "header": header, "phase": phase, "team_id": section_team_id, "events": section_events, "kickoff_report_rows": kickoff_report_rows, "surprise_by_event_id": surprise_by_event_id, "ottd_attempt": _is_ottd_candidate_turn and phase == "team_turn" and _kickoff_seen_in_turn, "ottd_mode": ( ottd_evaluations.get(_turn_num, {}).get("ottd_mode", {}) if _is_ottd_candidate_turn and phase == "team_turn" and _kickoff_seen_in_turn else {} ), } ) output.append({"turn_key": turn_key, "sections": section_outputs}) return output def collect_modeled_report_rows(game_state, team_meta, calculator=None, target_turn=None) -> ModeledRowsPayload: modeled_sections = build_detailed_modeled_sections( game_state, team_meta, calculator=calculator, target_turn=target_turn, ) return materialize_modeled_rows(modeled_sections) def build_step_section_lookup(game_state): """Return step -> {'turn','phase','team_id'} ownership from report sections.""" events = collect_roll_events_compat(game_state) turn_sections = build_report_turn_sections(game_state, events=events) step_lookup = {} for turn_info in turn_sections: turn_key = str(turn_info.get("turn_key")) for section in turn_info.get("sections", []): phase = section.get("phase") team_id = normalize_team_id(section.get("team_id")) for event in section.get("events") or []: step_num = event.get("step_number") if isinstance(step_num, int) and step_num not in step_lookup: step_lookup[step_num] = { "turn": turn_key, "phase": phase, "team_id": team_id, } return step_lookup def build_event_section_rows(game_state): """Return event-level ownership rows from report sections.""" events = collect_roll_events_compat(game_state) turn_sections = build_report_turn_sections(game_state, events=events) rows = [] for turn_info in turn_sections: turn_key = str(turn_info.get("turn_key")) for section in turn_info.get("sections", []): phase = section.get("phase") team_id = normalize_team_id(section.get("team_id")) for event in section.get("events") or []: step_num = event.get("step_number") if not isinstance(step_num, int): continue rows.append( { "step": step_num, "turn": turn_key, "phase": phase, "team_id": team_id, "cat": str(event.get("roll_category") or "-"), "payload": str(event.get("payload_type") or "-"), "result": str(event.get("result_name") or "-"), "player": str(event.get("player_name") or "-"), "roll_type": str(event.get("roll_type") or "-"), } ) return rows def event_matches_filter(event_row, event_filter): if not isinstance(event_row, dict): return False if not isinstance(event_filter, dict): return True for key in ("cat", "payload", "result", "roll_type"): expected = event_filter.get(key) if expected is None: continue if str(event_row.get(key) or "") != str(expected): return False player_contains = event_filter.get("player_contains") if player_contains is not None: if str(player_contains) not in str(event_row.get("player") or ""): return False return True def build_events_by_step_for_anchor_checks(game_state): """Return normalized detailed-report events grouped by step for anchor checks.""" events = collect_roll_events_compat(game_state) step_context = _build_step_context(game_state) normalized = _normalize_events_for_detailed_report(events, step_context) by_step = defaultdict(list) for event in normalized: step_num = event.get("step_number") if isinstance(step_num, int): by_step[step_num].append(event) return by_step