# -*- coding: utf-8 -*- import xmltodict import base64 import binascii from collections import defaultdict from dataclasses import dataclass from xml.parsers.expat import ExpatError from core.game_state import GameState from core.kick_off_event import KickOffEvent from core.team_state import TeamState from core.calculator import Calculator from modelling.action_probability import ActionProbabilityEnricher from modelling.skill_transition import SkillTransitionBackend from modelling.row_schema import build_roll_event_row from modelling.synthetic_roll_sources import ( build_ejection_roll_events, build_ko_recovery_roll_events, build_officious_ref_roll_events, build_touchdown_roll_events, extract_step_player_situations, is_ko_situation, ) from core.roll_category_constants import ( CONSEQUENCE_ROLL_CATEGORIES_LOWER, INITIATIVE_ROLL_CATEGORIES_LOWER, ) from core.lookup_registry import ( get_normalized_block_face, get_skill_rule, get_skill_lookup, get_special_action_rule, get_special_skill_usage_aliases, ) from core.processor_parse_stage import ( run_parse_stage, run_post_parse_synthetic_stage, ) from core.processor_stages import ( run_annotation_stage, run_probability_enrichment_stage, run_step_turn_stage, ) from core.ruleset import RuleSet SKILL_ID_TO_NAME = get_skill_lookup() PRO_RULE = get_skill_rule("pro") BRAWLER_RULE = get_skill_rule("brawler") SPECIAL_SKILL_USAGE_ALIASES = get_special_skill_usage_aliases() BREATHE_FIRE_RULE = get_special_action_rule("Breathe Fire") BOMB_HIT_RULE = get_special_action_rule("Bomb Hit") THROW_BOMB_RULE = get_special_action_rule("Throw Bomb") INTERCEPTION_RULE = get_special_action_rule("Interception") FIREBALL_RULE = get_special_action_rule("Fireball") TTM_RULE = get_special_action_rule("Throw Team Mate") ALWAYS_HUNGRY_RULE = get_special_action_rule("Always Hungry") TTM_SCATTER_RULE = get_special_action_rule("TTM Scatter") TTM_LANDING_RULE = get_special_action_rule("TTM Landing") LONER_RULE = get_special_action_rule("Loner") PICK_ME_UP_RULE = get_special_action_rule("Pick Me Up") ARGUE_THE_CALL_RULE = get_special_action_rule("Argue the Call") @dataclass class ReplayStepRollExtractor: """Dedicated extractor object for per-step roll event extraction.""" processor: "GameStateProcessor" def extract_roll_events_from_step(self, step_data, game_turn, forced_blitz_player_id=None, step_number=None): return self.processor._extract_roll_events_from_step_impl( step_data, game_turn, forced_blitz_player_id=forced_blitz_player_id, step_number=step_number, ) class GameStateProcessor: """Processes replay data to extract game state, steps, and events.""" def __init__(self, game_state=None, ruleset=None): """Initialize replay processor. Processes replay data to extract game state, steps, and events. """ self.game_state = game_state or GameState() self.step_counter = 0 self.calculator = Calculator() self._warning_keys = set() # Create team state trackers self.team_states = { 0: TeamState(team_id=0), # Home team 1: TeamState(team_id=1) # Away team } self.ruleset = ruleset if ruleset is not None else RuleSet() self.step_roll_extractor = ReplayStepRollExtractor(processor=self) self.action_probability_enricher = ActionProbabilityEnricher(self, ruleset=self.ruleset) self.skill_transition_backend = SkillTransitionBackend(self) def _event_player_has_skill_id(self, event, skill_id): """Return True when the acting player's skill_ids set contains skill_id. Uses the central player_lookup (populated from the roster at parse time), which stores raw ID strings as-is from the replay data. This is the canonical ID-based skill check; prefer it over name-based comparisons. """ if not skill_id or not isinstance(event, dict): return False player_lookup = self._safe_player_lookup(event.get("player_id")) skill_ids = player_lookup.get("skill_ids") if not isinstance(skill_ids, list): return False return str(skill_id) in skill_ids def _event_player_has_skill(self, event, skill_name): """Return True when the acting player's skill list contains skill_name (name-based). Checks event-level player_skills snapshot first; falls back to player_lookup. Prefer _event_player_has_skill_id for skills that have a known ID. """ if not skill_name or not isinstance(event, dict): return False raw_skills = event.get("player_skills") if not isinstance(raw_skills, list): player_lookup = self._safe_player_lookup(event.get("player_id")) raw_skills = player_lookup.get("skills") if not isinstance(raw_skills, list): return False normalized_target = str(skill_name).strip().lower() for skill in raw_skills: if str(skill).strip().lower() == normalized_target: return True return False def _enrich_action_probabilities(self): """Apply canonical action reroll assumptions to processed roll events. This is run in processor-space so reports remain read-only presentation. """ self.action_probability_enricher.enrich_action_probabilities() def _maybe_decode_base64_string(self, value): """Decode base64 string content when applicable; otherwise return as-is.""" if not isinstance(value, str): return value try: decoded = base64.b64decode(value, validate=True).decode("utf-8") # Avoid converting ordinary short numeric values. if decoded and any(ch.isalpha() for ch in decoded): return decoded except (binascii.Error, UnicodeDecodeError, ValueError): pass return value def _skill_id_to_name(self, skill_id): """Map BB skill ids to display names used across reports and modelling.""" return SKILL_ID_TO_NAME.get(str(skill_id)) def _classify_special_skill_usage(self, skill_value): """Return normalized special weapon skill label for ResultSkillUsage payloads.""" normalized = str(skill_value or "").strip().lower() mapped = SPECIAL_SKILL_USAGE_ALIASES.get(normalized) if mapped: return mapped if "chain" in normalized and "saw" in normalized: return SPECIAL_SKILL_USAGE_ALIASES.get("chainsaw", "Chainsaw") return None def _collect_breathe_fire_usage_hints(self, replay_steps): """Collect step-local Breathe Fire usage anchors from ResultSkillUsage payloads.""" hints = [] if not isinstance(replay_steps, list): return hints for step_index, step_data in enumerate(replay_steps, start=1): if not isinstance(step_data, dict): continue exec_sequences = self._ensure_list(step_data.get("EventExecuteSequence")) for seq_item in exec_sequences: if not isinstance(seq_item, dict): continue sequence = seq_item.get("Sequence") if not isinstance(sequence, dict): continue step_results = self._ensure_list(sequence.get("StepResult")) for step_result in step_results: if not isinstance(step_result, dict): continue step_node = step_result.get("Step") step_player_id = None step_target_id = None if isinstance(step_node, dict): step_message = self._parse_message_data_xml(step_node.get("MessageData")) if isinstance(step_message, dict) and len(step_message) > 0: root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if isinstance(root, dict): step_player_id = root.get("PlayerId") step_target_id = root.get("TargetId") string_messages = step_result.get("Results", {}).get("StringMessage") parsed_messages = self._parse_string_messages(self._ensure_list(string_messages)) for parsed in parsed_messages: if not isinstance(parsed, dict): continue if str(parsed.get("payload_type") or "") != "ResultSkillUsage": continue root = parsed.get("payload_root") if not isinstance(root, dict): continue skill_label = self._classify_special_skill_usage(root.get("Skill")) if skill_label != "Breathe Fire": continue used_value = str(root.get("Used") or "").strip().lower() is_used = used_value in ("", "1", "true", "yes") if not is_used: continue hints.append( { "step_number": step_index, "player_id": ( root.get("PlayerId") or root.get("Id") or root.get("SourceId") or root.get("AttackerId") or step_player_id ), "target_id": ( root.get("TargetId") or root.get("Target") or root.get("VictimId") or step_target_id ), } ) return hints def _annotate_breathe_fire_actions(self, replay_steps): """Label action rows that correspond to anchored Breathe Fire usages. Strong matching strategy: 1) Anchor from replay-level ResultSkillUsage where Skill=105 and Used=1. 2) Map to nearby action rows with the known Breathe Fire roll signature. """ usage_hints = self._collect_breathe_fire_usage_hints(replay_steps) if len(usage_hints) == 0: return # Strong row signature observed in replay payloads for Breathe Fire resolution. breathe_fire_signature = BREATHE_FIRE_RULE.get("signature", {}) breathe_fire_skill_id = BREATHE_FIRE_RULE.get("skill_id") breathe_fire_label = BREATHE_FIRE_RULE.get("synthetic_action_label", "Breathe Fire") action_events = getattr(self, "_events_by_cat", {}).get( breathe_fire_signature.get("roll_category", "action"), () ) signature_rows = [] for idx, event in enumerate(action_events): if str(event.get("step_name") or "") != breathe_fire_signature.get("step_name", ""): continue if str(event.get("step_type") or "") != breathe_fire_signature.get("step_type", ""): continue if str(event.get("roll_type") or "") != breathe_fire_signature.get("roll_type", ""): continue player_id = event.get("player_id") if player_id is None: continue skill_ids = self._safe_player_lookup(player_id).get("skill_ids", []) if not isinstance(skill_ids, list) or breathe_fire_skill_id not in skill_ids: continue signature_rows.append(idx) if len(signature_rows) == 0: return claimed_rows = set() sorted_hints = sorted(usage_hints, key=lambda h: self._to_int(h.get("step_number"), default=0) or 0) for hint_pos, hint in enumerate(sorted_hints): hint_step = self._to_int(hint.get("step_number"), default=None) if hint_step is None: continue hint_player = hint.get("player_id") hint_target = hint.get("target_id") next_hint_step = None if hint_pos + 1 < len(sorted_hints): next_hint_step = self._to_int(sorted_hints[hint_pos + 1].get("step_number"), default=None) max_step = hint_step + 3 if next_hint_step is not None: max_step = min(max_step, next_hint_step - 1) matched_any = False for row_idx in signature_rows: if row_idx in claimed_rows: continue event = action_events[row_idx] event_step = self._to_int(event.get("step_number"), default=None) if event_step is None or event_step < hint_step or event_step > max_step: continue if hint_player is not None and str(event.get("player_id")) != str(hint_player): continue if hint_target is not None and event.get("target_player_id") is not None: if str(event.get("target_player_id")) != str(hint_target): continue event["is_breathe_fire_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = breathe_fire_label claimed_rows.add(row_idx) matched_any = True if not matched_any: self._emit_processing_warning( ( f"Breathe Fire usage on step {hint_step} was detected but no matching " "action row was found nearby." ), key=("breathe-fire-unmatched", hint_step, str(hint_player), str(hint_target)), ) def _annotate_bomb_hit_actions(self): """Label bomb hit-resolution action rows using the observed roll type only. Current detection rule: - action rows with roll_type=88 """ bomb_hit_signature = BOMB_HIT_RULE.get("signature", {}) events = getattr(self, "_events_by_cat", {}).get( bomb_hit_signature.get("roll_category", "action"), () ) for event in events: if str(event.get("roll_type") or "") != bomb_hit_signature.get("roll_type", ""): continue dice_roller_raw = event.get("dice_roller") if dice_roller_raw is None: dice_roller_raw = event.get("team_id") flipped_dice_roller = self._to_int(dice_roller_raw, default=None) if flipped_dice_roller in (0, 1): event["dice_roller"] = 1 - flipped_dice_roller event["is_bomb_hit_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = BOMB_HIT_RULE.get("synthetic_action_label", "Bomb Hit") def _annotate_throw_bomb_actions(self): """Label throw bomb action rows for Bombardier players. Detection rule: - action rows with step_type=32 performed by a player with Bombardier skill (skill_id=80) """ throw_bomb_signature = THROW_BOMB_RULE.get("signature", {}) skill_id = THROW_BOMB_RULE.get("skill_id") label = THROW_BOMB_RULE.get("synthetic_action_label", "Throw Bomb") events = getattr(self, "_events_by_cat", {}).get( throw_bomb_signature.get("roll_category", "action"), () ) for event in events: if str(event.get("step_type") or "") != throw_bomb_signature.get("step_type", ""): continue player_id = event.get("player_id") if player_id is None: continue skill_ids = self._safe_player_lookup(player_id).get("skill_ids", []) if not isinstance(skill_ids, list) or skill_id not in skill_ids: continue event["is_throw_bomb_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = label def _annotate_interception_actions(self): """Label interception attempt action rows. Detection rule: - action rows with step_type=29 and roll_type=6 """ interception_signature = INTERCEPTION_RULE.get("signature", {}) label = INTERCEPTION_RULE.get("synthetic_action_label", "Interception") events = getattr(self, "_events_by_cat", {}).get( interception_signature.get("roll_category", "action"), () ) for event in events: if str(event.get("step_type") or "") != interception_signature.get("step_type", ""): continue if str(event.get("roll_type") or "") != interception_signature.get("roll_type", ""): continue event["is_interception_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = label def _annotate_pass_initiation_actions(self): """Label the actual throw test row as Pass for report display. Replay payloads can tag the follow-up scatter row (`roll_type=8`) as Pass while leaving the initiating throw test (`roll_type=5`, `step_type=11`) as a generic Action row. Fill only the display label on non-synthetic rows so Throw Bomb / Throw Team Mate sequences keep their existing synthetic labels. """ events = getattr(self, "_events_by_cat", {}).get("action", ()) for event in events: if str(event.get("roll_type") or "") != "5": continue if str(event.get("step_type") or "") != "11": continue if bool(event.get("is_synthetic_skill_action")): continue if str(event.get("synthetic_action_label") or "").strip(): continue if str(event.get("display_action_type") or "").strip(): continue event["display_action_type"] = "Pass" def _annotate_pass_scatter_actions(self): """Label follow-up pass direction rows as Scatter and ignore scoring math. For ordinary passes, replay payloads emit the actual throw test as `roll_type=5` and then a follow-up direction row as `roll_type=8` with three directional dice. That second row is not a success/fail action attempt and should not contribute to probability summaries. Bomb and Throw Team Mate scatters already carry synthetic labels, so this method only targets non-synthetic rows whose same-step earlier row was identified as a Pass initiation. """ events = getattr(self, "_events_by_cat", {}).get("action", ()) events_by_step = {} for event in events: step_number = event.get("step_number") if isinstance(step_number, int): events_by_step.setdefault(step_number, []).append(event) for step_events in events_by_step.values(): ordered = sorted( step_events, key=lambda event: int(event.get("roll_ordinal_in_step") or 0), ) seen_pass_initiation = False for event in ordered: if str(event.get("display_action_type") or "").strip() == "Pass": seen_pass_initiation = True continue if not seen_pass_initiation: continue if str(event.get("roll_type") or "") != "8": continue if bool(event.get("is_synthetic_skill_action")): continue if str(event.get("synthetic_action_label") or "").strip(): continue event["display_action_type"] = "Scatter" event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = "Scatter" event["is_marker_only_action"] = True event["exclude_from_surprise"] = True event["result_classification"] = "unknown" event["report_result_classification"] = "unknown" event["probability_success"] = None event["probability_neutral"] = None event["probability_fail"] = None event.pop("action_attempt_label", None) event.pop("action_attempt_ordinal_in_activation", None) def _enrich_pick_me_up_probabilities(self): """Override probabilities for Pick Me Up! rolls. Pick Me Up! succeeds on 5+ (p_success=1/3, p_neutral=2/3, p_fail=0). A missed roll is neutral (boar stays prone but no harm done), not a failure. """ pmu_probs = PICK_ME_UP_RULE.get("probabilities", {}) p_success = float(pmu_probs.get("success", 1.0 / 3.0)) p_neutral = float(pmu_probs.get("neutral", 2.0 / 3.0)) p_fail = float(pmu_probs.get("fail", 0.0)) for event in getattr(self, "_events_by_cat", {}).get("action", ()): if not event.get("is_pick_me_up_action"): continue event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail # A missed roll is neutral (p_fail=0); reclassify game-reported 'fail' to 'neutral'. current = str(event.get("result_classification") or "") if current == "fail": event["result_classification"] = "neutral" event["report_result_classification"] = "neutral" event["result_value"] = 0 def _enrich_argue_the_call_probabilities(self): """Override probabilities for Argue the Call rolls. Argue the Call succeeds on 6+ (p_success=1/6, p_fail=5/6, no neutral). """ argue_probs = ARGUE_THE_CALL_RULE.get("probabilities", {}) p_success = float(argue_probs.get("success", 1.0 / 6.0)) p_neutral = float(argue_probs.get("neutral", 0.0)) p_fail = float(argue_probs.get("fail", 5.0 / 6.0)) for event in getattr(self, "_events_by_cat", {}).get("action", ()): if not event.get("is_argue_the_call_action"): continue event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail # Keep Argue-the-Call binary for reporting and surprise stats. current = str(event.get("result_classification") or "") if current == "neutral": difficulty = event.get("difficulty") dice_values = event.get("dice_values") or [] try: target = int(difficulty) best_die = max(int(v) for v in dice_values) if dice_values else None except (TypeError, ValueError): target = None best_die = None if best_die is not None and target is not None: if best_die >= target: event["result_classification"] = "success" event["report_result_classification"] = "success" event["result_value"] = 1 else: event["result_classification"] = "fail" event["report_result_classification"] = "fail" event["result_value"] = -1 def _enrich_loner_probabilities(self): """Override probabilities for Loner reroll-check rolls. Loner is binary (success/fail, no neutral outcome). The success probability varies by skill variant: - Loner (4+) id=44: 3/6 = 0.5 - Loner (3+) id=1008: 4/6 = 0.6667 - Loner (5+) id=1022: 2/6 = 0.3333 When the player's skill_ids are unavailable, defaults to Loner (4+) as the most common variant. Probabilities are defined in action_probability.yaml under special_action_rules.Loner.skill_probabilities. """ skill_probs = LONER_RULE.get("skill_probabilities") or {} # Ordered fallback chain: try Loner (4+) first as default. default_probs = ( skill_probs.get("44") or next(iter(skill_probs.values()), None) or {"success": 0.5, "fail": 0.5} ) loner_skill_ids = set(LONER_RULE.get("skill_ids") or skill_probs.keys()) for event in getattr(self, "_events_by_cat", {}).get("action", ()): if not event.get("is_loner_action"): continue # Determine the Loner variant from the player's skill_ids. player_id = event.get("player_id") probs = default_probs if player_id is not None: player_skill_ids = self._safe_player_lookup(player_id).get("skill_ids") or [] for sid in player_skill_ids: if str(sid) in loner_skill_ids: probs = skill_probs[str(sid)] break event["probability_success"] = float(probs["success"]) event["probability_fail"] = float(probs["fail"]) event["probability_neutral"] = 0.0 def _collect_foul_hints_from_board_state(self, replay_steps): """Return dict mapping step_number -> team_id_str for steps where a team's FoulAvailable decreased in the board state (board-state cross-check signal). TeamState is ordered [team-0, team-1] with no explicit Id field; array index is used directly as team_id. """ result = {} prev_fa = {} # team_idx -> was_available (bool) for raw_idx, step in enumerate(replay_steps): step_number = raw_idx + 1 if not isinstance(step, dict): prev_fa = {} continue board_state = step.get("BoardState") if not isinstance(board_state, dict): prev_fa = {} continue list_teams = board_state.get("ListTeams") or {} team_states = list_teams.get("TeamState") or [] if isinstance(team_states, dict): team_states = [team_states] # A Blitz kickoff step (EventBlitz present) clears FoulAvailable as a # phase-transition artifact — fouling is not permitted on the Blitz # extra turn. Do not record a foul hint here; update prev_fa so # subsequent steps can still detect real fouls normally. is_blitz_kickoff_step = "EventBlitz" in step for team_idx, team_state in enumerate(team_states): if not isinstance(team_state, dict): continue fa_raw = team_state.get("FoulAvailable") is_available = fa_raw is not None and str(fa_raw).strip() not in ("", "0") was_available = prev_fa.get(team_idx) if ( was_available is not None and was_available and not is_available and not is_blitz_kickoff_step ): result[step_number] = str(team_idx) prev_fa[team_idx] = is_available return result def _annotate_foul_armour_injury_rolls(self, replay_steps): """Re-attribute armour and injury rolls that follow a foul to the fouling player. During parse, these rolls pick up the step actor (the fouled player). This method corrects player_id, player_name, team_id, and dice_roller on those rolls to reflect the fouler, and cross-checks detection against the board-state FoulAvailable counter. """ ev_by_cat = getattr(self, "_events_by_cat", {}) board_state_fouls = self._collect_foul_hints_from_board_state(replay_steps) foul_events = [e for e in ev_by_cat.get("action", ()) if e.get("is_foul_action")] detected_step_numbers = {e.get("step_number") for e in foul_events} for bs_step, bs_team in board_state_fouls.items(): if bs_step not in detected_step_numbers: self._emit_processing_warning( f"Board state FoulAvailable decrease for team {bs_team} at step {bs_step} " "but no synthetic foul event was emitted for this step.", key=("foul-bs-unmatched", bs_step), ) if not foul_events: return by_step = {} for event in (*ev_by_cat.get("armour", ()), *ev_by_cat.get("injury", ())): sn = event.get("step_number") by_step.setdefault(sn, []).append(event) def _has_double(ev): vals = ev.get("dice_values") or [] if ev else [] return len(vals) >= 2 and len(set(vals)) < len(vals) for foul_event in foul_events: step_number = foul_event.get("step_number") fouler_player_id = foul_event.get("player_id") fouler_player_name = foul_event.get("player_name") fouler_team_id = foul_event.get("team_id") foul_ordinal = foul_event.get("roll_ordinal_in_step") or 0 fouler_skills = foul_event.get("player_skills") or [] has_sneaky_git = any( str(s).lower().startswith("sneaky git") for s in fouler_skills ) dirty_player_skills = [] for raw_skill in fouler_skills: skill_text = str(raw_skill).strip() normalized = skill_text.lower().replace(" ", "") if normalized in ("dirtyplayer(+1)", "dirtyplayer(+2)"): dirty_player_skills.append(skill_text) bs_team = board_state_fouls.get(step_number) if bs_team is not None and str(bs_team) != str(fouler_team_id): self._emit_processing_warning( f"Foul at step {step_number}: board state indicates team {bs_team} " f"but foul event identifies team {fouler_team_id}.", key=("foul-bs-mismatch", step_number), ) foul_armour_event = None foul_injury_event = None for event in by_step.get(step_number, []): if event is foul_event: continue roll_cat = str(event.get("roll_category") or "") if roll_cat not in ("armour", "injury"): continue ordinal = event.get("roll_ordinal_in_step") or 0 if ordinal <= foul_ordinal: continue event["player_id"] = fouler_player_id event["player_name"] = fouler_player_name event["team_id"] = fouler_team_id event["dice_roller"] = str(fouler_team_id) if fouler_team_id is not None else event.get("dice_roller") if roll_cat == "armour": event["is_foul_armour_roll"] = True event["action_type"] = "foul" event["propagated_foul_relevant_skills"] = list(dirty_player_skills) foul_armour_event = event else: event["is_foul_injury_roll"] = True event["action_type"] = "foul" event["propagated_foul_relevant_skills"] = list(dirty_player_skills) foul_injury_event = event # Sent-off check: a double on armour (unless Sneaky Git) or injury triggers ejection. sent_off = _has_double(foul_injury_event) or ( not has_sneaky_git and _has_double(foul_armour_event) ) if sent_off: foul_event["result_value"] = -1 foul_event["result_classification"] = "fail" foul_event["report_result_classification"] = "fail" foul_event["notes"] = "sent off" # Set foul action probabilities based on injury-roll presence and Sneaky Git. # For synthetic foul actions, any non-fail outcome is treated as success. # p_success = 1 - p_fail, p_neutral = 0, p_fail = P(sent off). if foul_injury_event is not None: p_fail = 1.0 / 6.0 if has_sneaky_git else 1.0 / 3.0 else: p_fail = 0.0 if has_sneaky_git else 1.0 / 6.0 foul_event["probability_success"] = round(1.0 - p_fail, 10) foul_event["probability_fail"] = p_fail foul_event["probability_neutral"] = 0.0 foul_event.pop("exclude_from_surprise", None) # Default result is success (not sent off); overridden to fail above if sent off. if not sent_off: foul_event["result_value"] = 1 foul_event["result_classification"] = "success" foul_event["report_result_classification"] = "success" def _annotate_block_armour_rolls(self): """Re-attribute attack-caused armour rows to the causal attacker/target. Some armour rows are parsed with the damaged player as both actor and target. This pass rewrites those rows when the nearest preceding non-armour event is an offensive attack action (block, stab, chainsaw, breathe fire, vomit). """ events = getattr(self.game_state, "roll_events", []) or [] if not events: return index_by_id = {id(event): idx for idx, event in enumerate(events) if isinstance(event, dict)} for event in getattr(self, "_events_by_cat", {}).get("armour", ()): if not isinstance(event, dict): continue if event.get("is_foul_armour_roll"): continue event_idx = index_by_id.get(id(event)) if event_idx is None: continue current_victim_id = event.get("player_id") source_attack_event = self._find_nearest_armour_source_event( events, event_idx, current_victim_id, ) if source_attack_event is None: continue event["player_id"] = source_attack_event.get("player_id") event["player_name"] = source_attack_event.get("player_name") event["team_id"] = source_attack_event.get("team_id") event["dice_roller"] = source_attack_event.get("dice_roller") event["player_skills"] = source_attack_event.get("player_skills") or [] event["target_player_id"] = source_attack_event.get("target_player_id") event["target_player_name"] = source_attack_event.get("target_player_name") event["target_player_skills"] = source_attack_event.get("target_player_skills") or [] event["target_has_ball"] = bool(source_attack_event.get("target_has_ball")) event["armour_source_category"] = str(source_attack_event.get("roll_category") or "").lower() event["armour_source_action_label"] = str( source_attack_event.get("synthetic_action_label") or source_attack_event.get("display_action_type") or source_attack_event.get("action_type") or "" ).strip() def _collect_fireball_usage_hints(self, replay_steps): """Collect step numbers where EventUseSpecialCard matches the Fireball card_id.""" hints = [] if not isinstance(replay_steps, list): return hints card_id = FIREBALL_RULE.get("card_id") if card_id is None: return hints for step_index, step_data in enumerate(replay_steps, start=1): if not isinstance(step_data, dict): continue ev = step_data.get("EventUseSpecialCard") if not isinstance(ev, dict): continue if str(ev.get("CardId") or "") != str(card_id): continue hints.append({"step_number": step_index, "gamer_id": ev.get("GamerId")}) return hints def _annotate_fireball_actions(self, replay_steps): """Label fireball hit-check action rows anchored to EventUseSpecialCard.""" usage_hints = self._collect_fireball_usage_hints(replay_steps) if len(usage_hints) == 0: return hint_steps = {hint["step_number"] for hint in usage_hints} fireball_signature = FIREBALL_RULE.get("signature", {}) label = FIREBALL_RULE.get("synthetic_action_label", "Fireball") events = getattr(self, "_events_by_cat", {}).get( fireball_signature.get("roll_category", "action"), () ) for event in events: if str(event.get("step_type") or "") != fireball_signature.get("step_type", ""): continue if str(event.get("roll_type") or "") != fireball_signature.get("roll_type", ""): continue event_step = self._to_int(event.get("step_number"), default=None) if event_step not in hint_steps: continue event["is_fireball_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = label event["dice_roller"] = 1 def _collect_ttm_hints(self, replay_steps): """Collect ThrowerId/ThrownMateId from ThrowTeamMateStep XML, keyed by step number.""" hints = {} if not isinstance(replay_steps, list): return hints for step_index, step_data in enumerate(replay_steps, start=1): if not isinstance(step_data, dict): continue exec_sequences = self._ensure_list(step_data.get("EventExecuteSequence")) for seq_item in exec_sequences: if not isinstance(seq_item, dict): continue sequence = seq_item.get("Sequence") if not isinstance(sequence, dict): continue step_results = self._ensure_list(sequence.get("StepResult")) for sr in step_results: if not isinstance(sr, dict): continue step_node = sr.get("Step", {}) if not isinstance(step_node, dict): continue if step_node.get("Name") != "ThrowTeamMateStep": continue msg_data = step_node.get("MessageData", "") if not msg_data: continue parsed = self._parse_message_data_xml(msg_data) if not isinstance(parsed, dict): continue root = parsed.get("ThrowTeamMateStep", {}) if not isinstance(root, dict): continue thrower_id = root.get("ThrowerId") thrown_id = root.get("ThrownMateId") if thrower_id is not None: hints[step_index] = { "thrower_id": str(thrower_id), "thrown_id": str(thrown_id) if thrown_id is not None else None, } return hints def _annotate_ttm_actions(self, replay_steps): """Annotate all rolls within Throw Team Mate action sequences.""" events = getattr(self.game_state, "roll_events", None) if not isinstance(events, list): return hints = self._collect_ttm_hints(replay_steps) if not hints: return hint_step_numbers = set(hints.keys()) ttm_roll_type = TTM_RULE.get("signature", {}).get("roll_type", "41") ttm_label = TTM_RULE.get("synthetic_action_label", "Throw Team Mate") ah_roll_type = ALWAYS_HUNGRY_RULE.get("signature", {}).get("roll_type", "43") ah_label = ALWAYS_HUNGRY_RULE.get("synthetic_action_label", "Always Hungry") scatter_label = TTM_SCATTER_RULE.get("synthetic_action_label", "Scatter") landing_step_type = TTM_LANDING_RULE.get("signature", {}).get("step_type", "14") landing_roll_type = TTM_LANDING_RULE.get("signature", {}).get("roll_type", "42") landing_label = TTM_LANDING_RULE.get("synthetic_action_label", "Landing") loner_roll_type = LONER_RULE.get("signature", {}).get("roll_type", "71") for event in events: if not isinstance(event, dict): continue step_num = self._to_int(event.get("step_number"), default=None) if step_num not in hint_step_numbers: continue step_name = str(event.get("step_name") or "") roll_type = str(event.get("roll_type") or "") roll_cat = str(event.get("roll_category") or "") hint = hints[step_num] if step_name == "ThrowTeamMateStep": thrower_id = hint.get("thrower_id") if thrower_id and event.get("player_id") is None: thrower_lookup = self._safe_player_lookup(thrower_id) event["player_id"] = thrower_id event["player_name"] = thrower_lookup.get("name", f"Player {thrower_id}") event["team_id"] = thrower_lookup.get("team_id", event.get("team_id")) event["dice_roller"] = thrower_lookup.get("team_id", event.get("dice_roller")) event["player_skills"] = thrower_lookup.get("skills", []) if roll_cat != "action": continue if roll_type == ah_roll_type: event["is_always_hungry_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = ah_label elif roll_type == ttm_roll_type: event["is_ttm_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = ttm_label elif roll_type == loner_roll_type: pass else: event["is_ttm_scatter_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = scatter_label event["player_name"] = "-" elif step_name == "PlayerStep": if str(event.get("step_type") or "") != landing_step_type: continue if roll_type != landing_roll_type: continue if roll_cat != "action": continue if event.get("player_id") is None: thrown_id = hint.get("thrown_id") if thrown_id: thrown_lookup = self._safe_player_lookup(thrown_id) event["player_id"] = thrown_id event["player_name"] = thrown_lookup.get("name", f"Player {thrown_id}") event["team_id"] = thrown_lookup.get("team_id", event.get("team_id")) event["player_skills"] = thrown_lookup.get("skills", []) event["is_ttm_landing_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = landing_label def _annotate_loner_actions(self): """Annotate Loner reroll-check rows anywhere in the replay.""" loner_signature = LONER_RULE.get("signature", {}) label = LONER_RULE.get("synthetic_action_label", "Loner") events = getattr(self, "_events_by_cat", {}).get(loner_signature.get("roll_category", "action"), ()) for event in events: if str(event.get("roll_type") or "") != loner_signature.get("roll_type", "71"): continue event["is_loner_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = label def _annotate_pick_me_up_actions(self): """Annotate Pick Me Up! rolls using the dedicated roll signature.""" pick_me_up_signature = PICK_ME_UP_RULE.get("signature", {}) label = PICK_ME_UP_RULE.get("synthetic_action_label", "Pick Me Up!") events = getattr(self, "_events_by_cat", {}).get(pick_me_up_signature.get("roll_category", "action"), ()) expected_step_type = pick_me_up_signature.get("step_type") expected_roll_type = str(pick_me_up_signature.get("roll_type", "92")) for event in events: if expected_step_type is not None and str(event.get("step_type") or "") != str(expected_step_type): continue if str(event.get("roll_type") or "") != expected_roll_type: continue # Diagnostics only: PMU should generally appear as 5+ in replay payload. raw_payload = event.get("raw") if isinstance(event.get("raw"), dict) else {} requirement = str(raw_payload.get("Requirement") or "").strip() difficulty = str(event.get("difficulty") or "").strip() if requirement not in ("", "5") or difficulty not in ("", "5"): self._emit_processing_warning( ( f"Pick Me Up candidate at step {event.get('step_number')} has " f"unexpected requirement/difficulty ({requirement}/{difficulty}); " "classifying as PMU by roll_type=92." ), key=("pmu-unexpected-target", event.get("step_number"), event.get("roll_ordinal_in_step")), ) event["is_pick_me_up_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = label def _annotate_argue_the_call_actions(self): """Annotate Argue the Call rolls using its dedicated roll signature.""" argue_signature = ARGUE_THE_CALL_RULE.get("signature", {}) argue_label = ARGUE_THE_CALL_RULE.get("synthetic_action_label", "Argue the Call") events = getattr(self, "_events_by_cat", {}).get(argue_signature.get("roll_category", "action"), ()) expected_step_type = argue_signature.get("step_type") expected_roll_type = str(argue_signature.get("roll_type", "30")) for event in events: if expected_step_type is not None and str(event.get("step_type") or "") != str(expected_step_type): continue if str(event.get("roll_type") or "") != expected_roll_type: continue if event.get("is_pick_me_up_action"): continue event["is_argue_the_call_action"] = True event["is_synthetic_skill_action"] = True event["synthetic_action_label"] = argue_label def _is_offensive_armour_source_action(self, event): """Return True for action rows that can directly cause an armour roll.""" if not isinstance(event, dict): return False if bool(event.get("is_foul_action")): return False if bool(event.get("is_stab_action")): return True if bool(event.get("is_chainsaw_action")): return True if bool(event.get("is_breathe_fire_action")): return True labels = ( event.get("synthetic_action_label"), event.get("display_action_type"), event.get("action_type"), ) for label in labels: normalized = str(label or "").strip().lower() if normalized in ("stab", "chainsaw", "breathe fire"): return True if "vomit" in normalized: return True return False def _find_nearest_armour_source_event(self, events, event_idx, victim_player_id): """Return nearest preceding causal attack event for an armour row. The nearest preceding non-armour event must be a supported attack action, and must target the armour victim when both ids are known. """ for prior_idx in range(event_idx - 1, -1, -1): prior_event = events[prior_idx] if not isinstance(prior_event, dict): continue prior_category = str(prior_event.get("roll_category") or "").lower() if prior_category == "armour": continue is_supported_attack = ( prior_category == "block" or ( prior_category == "action" and self._is_offensive_armour_source_action(prior_event) ) ) if not is_supported_attack: break attack_target_id = prior_event.get("target_player_id") if ( victim_player_id is not None and attack_target_id is not None and str(attack_target_id) != str(victim_player_id) ): break return prior_event return None def _annotate_injury_mighty_blow_from_armour(self): """Propagate Mighty Blow from block-sourced armour rows to following injury rows.""" events = getattr(self.game_state, "roll_events", []) or [] if not events: return index_by_id = {id(event): idx for idx, event in enumerate(events) if isinstance(event, dict)} mighty_blow_allow = {"mighty blow (+1)", "mighty blow (+2)"} for injury_event in getattr(self, "_events_by_cat", {}).get("injury", ()): if not isinstance(injury_event, dict): continue if injury_event.get("is_foul_injury_roll"): continue injury_idx = index_by_id.get(id(injury_event)) if injury_idx is None: continue source_armour_event = None for prior_idx in range(injury_idx - 1, -1, -1): prior_event = events[prior_idx] if not isinstance(prior_event, dict): continue prior_category = str(prior_event.get("roll_category") or "").lower() if prior_category == "injury": continue if prior_category != "armour": break source_armour_event = prior_event break if source_armour_event is None: continue if str(source_armour_event.get("armour_source_category") or "").lower() != "block": continue mb_skills = [] for skill in source_armour_event.get("player_skills") or []: skill_text = str(skill).strip() if skill_text.lower() in mighty_blow_allow: mb_skills.append(skill_text) if not mb_skills: continue injury_event["propagated_armour_relevant_skills"] = mb_skills def _annotate_crowd_surf_injuries(self): """Detect crowd-surf injury rows and attribute them to the causal attacker. Crowd-surf injuries skip armour entirely. In the raw payload they carry ``Source='2'`` on the injury roll, so we use that marker to identify them. """ events = getattr(self.game_state, "roll_events", []) or [] if not events: return index_by_id = {id(event): idx for idx, event in enumerate(events) if isinstance(event, dict)} for injury_event in getattr(self, "_events_by_cat", {}).get("injury", ()): if not isinstance(injury_event, dict): continue if injury_event.get("is_foul_injury_roll"): continue raw = injury_event.get("raw") if not isinstance(raw, dict) or str(raw.get("Source") or "") != "2": continue injury_event["is_crowd_surf_injury"] = True injury_idx = index_by_id.get(id(injury_event)) if injury_idx is None: continue source_event = None for prior_idx in range(injury_idx - 1, -1, -1): prior_event = events[prior_idx] if not isinstance(prior_event, dict): continue prior_category = str(prior_event.get("roll_category") or "").lower() if prior_category in ("injury", "armour"): continue if prior_category in ("block", "action"): source_event = prior_event break if source_event is not None: injury_event["player_id"] = source_event.get("player_id") injury_event["player_name"] = source_event.get("player_name") injury_event["team_id"] = source_event.get("team_id") injury_event["dice_roller"] = source_event.get("dice_roller") injury_event["player_skills"] = list(source_event.get("player_skills") or []) injury_event["target_player_id"] = source_event.get("target_player_id") injury_event["target_player_name"] = source_event.get("target_player_name") injury_event["target_player_skills"] = list(source_event.get("target_player_skills") or []) injury_event["target_has_ball"] = bool(source_event.get("target_has_ball")) injury_event["injury_source_category"] = str(source_event.get("roll_category") or "").lower() notes = str(injury_event.get("notes") or "-").strip() if notes in ("", "-"): injury_event["notes"] = "crowd surf" elif "crowd surf" not in notes: injury_event["notes"] = notes + "; crowd surf" def _annotate_injury_attacker_from_armour(self): """Re-attribute injury rows to the causal attacker via the preceding armour row. For each non-foul injury row, find the nearest preceding armour row and copy the attacker identity (player_id, player_name, team_id, dice_roller, player_skills) and target identity from it. Also sets injury_source_category to mirror the armour row's armour_source_category so the renderer can apply the right skill policy. """ events = getattr(self.game_state, "roll_events", []) or [] if not events: return index_by_id = {id(event): idx for idx, event in enumerate(events) if isinstance(event, dict)} for injury_event in getattr(self, "_events_by_cat", {}).get("injury", ()): if not isinstance(injury_event, dict): continue if injury_event.get("is_foul_injury_roll"): continue if injury_event.get("is_crowd_surf_injury"): continue injury_idx = index_by_id.get(id(injury_event)) if injury_idx is None: continue source_armour_event = None for prior_idx in range(injury_idx - 1, -1, -1): prior_event = events[prior_idx] if not isinstance(prior_event, dict): continue prior_category = str(prior_event.get("roll_category") or "").lower() if prior_category == "injury": continue if prior_category != "armour": break source_armour_event = prior_event break if source_armour_event is None: continue injury_event["player_id"] = source_armour_event.get("player_id") injury_event["player_name"] = source_armour_event.get("player_name") injury_event["team_id"] = source_armour_event.get("team_id") injury_event["dice_roller"] = source_armour_event.get("dice_roller") injury_event["player_skills"] = list(source_armour_event.get("player_skills") or []) injury_event["target_player_id"] = source_armour_event.get("target_player_id") injury_event["target_player_name"] = source_armour_event.get("target_player_name") injury_event["target_player_skills"] = list(source_armour_event.get("target_player_skills") or []) injury_event["target_has_ball"] = bool(source_armour_event.get("target_has_ball")) injury_event["injury_source_category"] = str(source_armour_event.get("armour_source_category") or "").lower() def _enrich_breathe_fire_probabilities(self): """Override probabilities and result classification for Breathe Fire action rows. Fixed outcome model: die == 1 → fail (always possible with probability 1/6) die in (2, 3) → neutral (probability 2/6) die in (4..6) → success (probability 3/6) The prior probability distribution is fixed regardless of outcome: P(S) = 3/6 = 0.5 P(N) = 2/6 ≈ 0.333… P(F) = 1/6 ≈ 0.167… """ for event in getattr(self, "_events_by_cat", {}).get("action", ()): if not event.get("is_breathe_fire_action"): continue breathe_fire_results = BREATHE_FIRE_RULE.get("result_by_first_die", {}) breathe_fire_probabilities = BREATHE_FIRE_RULE.get("probabilities", {}) dice_values = event.get("dice_values") or [] if not isinstance(dice_values, list) or len(dice_values) == 0: self._emit_processing_warning( "Breathe Fire action row has no dice_values; skipping probability override.", key=("breathe-fire-no-dice", event.get("step_number"), event.get("player_id")), ) continue result_classification, probability_success, probability_neutral, probability_fail = ( self.calculator.calc_fixed_outcome_action( dice_values[0], breathe_fire_results, breathe_fire_probabilities, default_result="success", ) ) event["result_classification"] = result_classification event["report_result_classification"] = result_classification event["probability_success"] = probability_success event["probability_neutral"] = probability_neutral event["probability_fail"] = probability_fail def _emit_processing_warning(self, message, key=None): """Record and print a deduplicated processing warning.""" dedupe_key = key if key is not None else message if dedupe_key in self._warning_keys: return self._warning_keys.add(dedupe_key) self.game_state.processing_warnings.append(message) print(f"[GameStateProcessor][warning] {message}") def _normalize_player_name(self, player_name): """Normalize known replay name corruption artifacts to readable display names.""" if not isinstance(player_name, str): return player_name normalized = player_name.strip() known_name_fixes = { "#<�": "izzy", } return known_name_fixes.get(normalized, normalized) def _extract_player_skill_ids(self, player_data): """Extract and normalize raw player skill IDs from roster/BoardState payload.""" skill_ids = [] if not isinstance(player_data, dict): return skill_ids acquired_skills = player_data.get("AcquiredSkills") or {} innate_skills = player_data.get("InnateSkills") or {} skill_containers = [ acquired_skills.get("AcquiredSkillsItem"), innate_skills.get("InnateSkillsItem"), ] for skill_items in skill_containers: if skill_items is None: continue if not isinstance(skill_items, list): skill_items = [skill_items] for skill_id in skill_items: normalized = str(skill_id).strip() if normalized and normalized not in skill_ids: skill_ids.append(normalized) return skill_ids def _extract_player_skills(self, player_data): """Extract and normalize player skills from roster payload.""" skills = [] skill_ids = self._extract_player_skill_ids(player_data) for skill_id in skill_ids: skill_name = self._skill_id_to_name(skill_id) skills.append(skill_name if skill_name else "") return skills def _to_int(self, value, default=None): """Convert common numeric payload values to int.""" try: return int(value) except (TypeError, ValueError): return default def _normalize_team_id(self, value): if value is None: return None try: return str(int(value)) except (TypeError, ValueError): return str(value) def _opponent_team_id(self, team_id): normalized = self._normalize_team_id(team_id) if normalized == "0": return "1" if normalized == "1": return "0" return normalized def _resolve_dice_roller(self, team_id, roll_category): """Return the team that rolled the die for this event. For armour/injury, dice roller is the opponent of the event team. For action/block/kickoff/casualty and all other categories, dice roller equals team_id. """ normalized_team = self._normalize_team_id(team_id) category = str(roll_category or "").lower() if category in ("armour", "injury"): return self._opponent_team_id(normalized_team) return normalized_team def _normalize_regulation_game_turn(self, value): """Cap non-overtime game turns to the BB2020 regulation max of 16.""" try: normalized = int(value) except (TypeError, ValueError): return value if normalized < 0: return "0" if normalized > 16: return "16" return str(normalized) def _weather_name_from_roll_total(self, roll_total): """Map 2d6 weather roll totals to BB2020 weather names.""" try: value = int(roll_total) except (TypeError, ValueError): return "Unknown" if value == 2: return "Sweltering Heat" if value == 3: return "Very Sunny" if 4 <= value <= 10: return "Nice" if value == 11: return "Pouring Rain" if value == 12: return "Blizzard" return "Unknown" def extract_weather_event(self, step_data, step_number, game_turn): """Extract weather roll from a replay step when available. Primary source is EventWeatherRoll. We also allow a one-time bootstrap from BoardState.LastWeatherRoll if no weather has been seen yet. """ if not isinstance(step_data, dict): return None weather_payload = step_data.get("EventWeatherRoll") source = "EventWeatherRoll" if not isinstance(weather_payload, dict) and self.game_state.current_weather_roll_total is None: board_state = step_data.get("BoardState", {}) if isinstance(board_state, dict): weather_payload = board_state.get("LastWeatherRoll") source = "BoardState.LastWeatherRoll" if not isinstance(weather_payload, dict): return None dice_obj = weather_payload.get("Dice", {}) dice_entries = dice_obj.get("Die") if isinstance(dice_obj, dict) else None if dice_entries is None: dice_entries = weather_payload.get("Die") if dice_entries is None: return None if not isinstance(dice_entries, list): dice_entries = [dice_entries] dice_values = [] for die in dice_entries: if isinstance(die, dict): raw = die.get("Value") else: raw = die try: dice_values.append(int(raw)) except (TypeError, ValueError): continue if not dice_values: return None roll_total = sum(dice_values) return { "step_number": step_number, "game_turn": game_turn, "source": source, "dice_values": dice_values, "roll_total": roll_total, "weather_name": self._weather_name_from_roll_total(roll_total), } def _is_ko_situation(self, value): """Thin delegate — logic lives in synthetic_roll_sources.""" return is_ko_situation(value) def _extract_step_player_situations(self, step_data): """Thin delegate — logic lives in synthetic_roll_sources.""" return extract_step_player_situations(step_data) def _safe_player_lookup(self, player_id): """Return player lookup entry when present, else an empty dict.""" if player_id is None: return {} return self.game_state.player_lookup.get(str(player_id), {}) def _build_synthetic_roll_event( self, *, game_turn, step_number, player_id, team_id, roll_category, result_name, payload_type, result_value, probability_success, probability_neutral=0.0, probability_fail=None, dice_values=None, dice_total="-", difficulty="-", roll_type="-", outcome="-", modifier_values=None, target_player_id=None, target_player_name=None, target_player_skills=None, step_name="-", step_type="-", armour_holds="-", raw=None, player_name=None, player_skills=None, dice_roller_label=None, exclude_from_surprise=False, ): """Build a normalized synthetic roll event row.""" lookup = self._safe_player_lookup(player_id) team_id_str = str(team_id) if team_id is not None else "Unknown" if probability_fail is None: probability_fail = round(1.0 - float(probability_success) - float(probability_neutral), 10) if dice_values is None: dice_values = [] if modifier_values is None: modifier_values = [] if target_player_skills is None: target_player_skills = [] resolved_player_name = player_name if player_name is not None else lookup.get("name", "Unknown") if isinstance(player_skills, list): resolved_player_skills = player_skills else: lookup_skills = lookup.get("skills", []) resolved_player_skills = lookup_skills if isinstance(lookup_skills, list) else [] result_classification = "success" if result_value > 0 else "fail" if result_value < 0 else "neutral" event = { "game_turn": game_turn, "step_number": step_number, "player_id": player_id, "player_name": resolved_player_name, "team_id": team_id_str, "dice_roller": self._resolve_dice_roller(team_id_str, dice_roller_label or roll_category), "player_skills": resolved_player_skills, "target_player_id": target_player_id, "target_player_name": target_player_name, "target_player_skills": target_player_skills, "step_name": step_name, "step_type": step_type, "result_name": result_name, "payload_type": payload_type, "roll_category": roll_category, "dice_values": dice_values, "dice_total": dice_total, "difficulty": difficulty, "roll_type": roll_type, "outcome": outcome, "modifier_values": modifier_values, "result_value": result_value, "result_classification": result_classification, "report_result_classification": result_classification, "armour_holds": armour_holds, "probability_success": probability_success, "probability_neutral": probability_neutral, "probability_fail": probability_fail, "raw": raw if isinstance(raw, dict) else {}, } if exclude_from_surprise: event["exclude_from_surprise"] = True return build_roll_event_row(event) def _build_ko_recovery_roll_events(self, replay_steps): """Infer KO recovery outcomes from Situation transitions at touchdown kickoffs and halftime.""" return build_ko_recovery_roll_events(self, replay_steps) def _parse_player_characteristics(self, player_data): """Extract roster characteristic values keyed by replay characteristic id.""" if not isinstance(player_data, dict): return {} characteristics_node = player_data.get("Characteristics") if not isinstance(characteristics_node, dict): return {} raw_items = characteristics_node.get("PlayerCharacteristic", []) if isinstance(raw_items, dict): raw_items = [raw_items] parsed = {} for raw_item in raw_items: if not isinstance(raw_item, dict): continue characteristic_id = raw_item.get("Characteristic") if characteristic_id in (None, ""): continue key = str(characteristic_id) parsed[key] = { "value": str(raw_item.get("Value")) if raw_item.get("Value") is not None else None, "bonuses": str(raw_item.get("NbBonuses")) if raw_item.get("NbBonuses") is not None else None, "maluses": str(raw_item.get("NbMaluses")) if raw_item.get("NbMaluses") is not None else None, } return parsed def _build_player_lookup_los_index(self, replay_steps): """Return kickoff LOS participation keyed by player id from kickoff board snapshots.""" if not isinstance(replay_steps, list): return {} pitch_length = 26 pitch_width = 15 centre_zone_width = 7 los_left_x = (pitch_length // 2) - 1 los_right_x = pitch_length // 2 los_y_min = (pitch_width - centre_zone_width) // 2 los_y_max = los_y_min + centre_zone_width - 1 side_to_los_x = { "32": los_left_x, "64": los_right_x, } los_index = {} for step_number, step_data in enumerate(replay_steps, 1): if not isinstance(step_data, dict) or "EventKickOffTable" not in step_data: continue board_state = step_data.get("BoardState") if not isinstance(board_state, dict): continue team_states = board_state.get("ListTeams", {}).get("TeamState", []) if not isinstance(team_states, list): team_states = [team_states] for team_idx, team_state in enumerate(team_states[:2]): if not isinstance(team_state, dict): continue team_id = str(team_idx) side = str(team_state.get("Side")) if team_state.get("Side") is not None else None los_x = side_to_los_x.get(side) if los_x is None: los_x = los_left_x if team_id == "0" else los_right_x players = team_state.get("ListPitchPlayers", {}).get("PlayerState", []) if not isinstance(players, list): players = [players] for player in players: if not isinstance(player, dict): continue player_id = player.get("Id") if player_id in (None, ""): continue cell = player.get("Cell") if not isinstance(cell, dict): continue try: cell_x = int(cell.get("X")) cell_y = int(cell.get("Y")) except (TypeError, ValueError): continue if cell_x == -1 and cell_y == -1: continue if cell_x != los_x: continue if cell_y < los_y_min or cell_y > los_y_max: continue player_id_str = str(player_id) entries = los_index.setdefault(player_id_str, []) entries.append( { "step_number": step_number, "team_id": team_id, "x": cell_x, "y": cell_y, } ) return los_index def _iter_kickoff_snapshot_team_players(self, kickoff_step_number, team_id): """Yield on-pitch player snapshots for one team from a kickoff BoardState.""" raw_steps = getattr(self.game_state, "raw_replay_steps", None) if not isinstance(raw_steps, list): return [] step_index = self._to_int(kickoff_step_number, default=None) if step_index is None: return [] step_index -= 1 if step_index < 0 or step_index >= len(raw_steps): return [] step_data = raw_steps[step_index] if not isinstance(step_data, dict): return [] board_state = step_data.get("BoardState") if not isinstance(board_state, dict): return [] team_states = board_state.get("ListTeams", {}).get("TeamState", []) if not isinstance(team_states, list): team_states = [team_states] normalized_team_id = self._normalize_team_id(team_id) team_index = self._to_int(normalized_team_id, default=None) if team_index is None or team_index < 0 or team_index >= len(team_states): return [] team_state = team_states[team_index] if not isinstance(team_state, dict): return [] side = str(team_state.get("Side")) if team_state.get("Side") is not None else None los_x = 12 if side == "32" else 13 if side == "64" else 12 if normalized_team_id == "0" else 13 back_x = los_x - 1 if los_x == 12 else los_x + 1 players = team_state.get("ListPitchPlayers", {}).get("PlayerState", []) if not isinstance(players, list): players = [players] snapshots = [] for player in players: if not isinstance(player, dict): continue player_id = player.get("Id") if player_id in (None, ""): continue cell = player.get("Cell") if not isinstance(cell, dict): continue cell_x = self._to_int(cell.get("X"), default=None) cell_y = self._to_int(cell.get("Y"), default=None) if cell_x is None or cell_y is None: continue if cell_x < 0 or cell_y < 0: continue player_id_str = str(player_id) lookup = self._safe_player_lookup(player_id_str) skill_names = lookup.get("skills") if isinstance(lookup.get("skills"), list) else [] skill_names_lower = {str(skill).strip().lower() for skill in skill_names if skill not in (None, "")} ma_value = lookup.get("ma") effective_ma = self._to_int(ma_value, default=None) has_sprint = "sprint" in skill_names_lower if effective_ma is not None and has_sprint: effective_ma += 1 snapshots.append( { "player_id": player_id_str, "player_name": lookup.get("name", f"Player {player_id_str}"), "team_id": normalized_team_id, "x": cell_x, "y": cell_y, "side": side, "los_x": los_x, "back_x": back_x, "ma": self._to_int(ma_value, default=None), "effective_ma": effective_ma, "has_sprint": has_sprint, "has_stand_firm": "stand firm" in skill_names_lower, "has_sidestep": "sidestep" in skill_names_lower, "has_grab": "grab" in skill_names_lower, "has_juggernaut": "juggernaut" in skill_names_lower, "has_frenzy": "frenzy" in skill_names_lower, "is_fast": effective_ma is not None and effective_ma >= 7, "on_halfway_line": cell_x == los_x, "on_halfway_or_one_back": cell_x in (los_x, back_x), "skills": skill_names, } ) return snapshots def _build_ottd_turn_kickoff_index(self): """Return turn -> earliest kickoff context for OTTD candidate evaluation.""" index = {} for step in getattr(self.game_state, "steps", []) or []: if not getattr(step, "kick_off", None): continue turn_num = self._to_int(getattr(step, "game_turn", None), default=None) if turn_num not in (8, 16): continue existing = index.get(turn_num) if existing is not None and existing.get("kickoff_step_number") <= step.step_number: continue kicker_team_id = self._normalize_team_id(getattr(step.kick_off, "team_id", None)) index[turn_num] = { "kickoff_step_number": step.step_number, "kicker_team_id": kicker_team_id, "receiving_team_id": self._opponent_team_id(kicker_team_id) if kicker_team_id in ("0", "1") else None, } return index def _collect_turn_touchdowns_after_step(self, turn_num, step_number, team_id=None): """Return touchdown markers that happen after a kickoff within the same game turn.""" matches = [] for step in getattr(self.game_state, "steps", []) or []: if getattr(step, "touchdown", None) is None: continue if self._to_int(getattr(step, "game_turn", None), default=None) != turn_num: continue if self._to_int(getattr(step, "step_number", None), default=None) is None: continue if step.step_number <= step_number: continue touchdown = step.touchdown if isinstance(step.touchdown, dict) else {} touchdown_team_id = self._normalize_team_id(touchdown.get("team_id")) if team_id in ("0", "1") and touchdown_team_id != team_id: continue matches.append( { "step_number": step.step_number, "team_id": touchdown_team_id, "player_id": touchdown.get("player_id"), "player_name": touchdown.get("player_name", "Unknown"), "source": touchdown.get("source", "unknown"), } ) return sorted(matches, key=lambda row: row.get("step_number") or 0) def _collect_turn_ttm_actions_after_step(self, turn_num, step_number, team_id=None): """Return Throw Team Mate throw actions after a kickoff within the same game turn.""" matches = [] for event in getattr(self.game_state, "roll_events", []) or []: if not isinstance(event, dict) or not event.get("is_ttm_action"): continue if self._to_int(event.get("game_turn"), default=None) != turn_num: continue event_step = self._to_int(event.get("step_number"), default=None) if event_step is None or event_step <= step_number: continue event_team_id = self._normalize_team_id(event.get("team_id")) if team_id in ("0", "1") and event_team_id != team_id: continue matches.append( { "step_number": event_step, "team_id": event_team_id, "player_id": event.get("player_id"), "player_name": event.get("player_name", "Unknown"), "roll_type": event.get("roll_type"), } ) return sorted(matches, key=lambda row: row.get("step_number") or 0) def _minimum_ottd_players_required(self, pushes_needed, has_sidestep): """Return required players on pitch for a push OTTD setup.""" if pushes_needed == 1: return 4 if has_sidestep else 6 if pushes_needed == 2: return 6 if has_sidestep else 7 if pushes_needed == 3: return 6 if has_sidestep else 8 if pushes_needed == 4: return 8 if has_sidestep else 9 return None def _evaluate_ottd_defender_skill_blocker(self, receiving_players, kicking_players, pushes_needed): """Return whether LOS defenders fully block push OTTD by anti-push skills. Sidestep protection is ignored when the attacking team has Grab. Stand Firm protection is ignored when the attacking team has: - Juggernaut and exactly 1 push is needed, or - Juggernaut plus Frenzy and exactly 2 pushes are needed. """ los_defenders = [player for player in kicking_players if player.get("on_halfway_line")] attacking_has_grab = any(player.get("has_grab") for player in receiving_players) attacking_has_juggernaut = any(player.get("has_juggernaut") for player in receiving_players) attacking_has_juggernaut_frenzy = any( player.get("has_juggernaut") and player.get("has_frenzy") for player in receiving_players ) ignore_sidestep = attacking_has_grab ignore_stand_firm = ( pushes_needed == 1 and attacking_has_juggernaut ) or ( pushes_needed == 2 and attacking_has_juggernaut_frenzy ) evaluated_defenders = [] fully_blocked = bool(los_defenders) for defender in los_defenders: active_protections = [] ignored_protections = [] if defender.get("has_sidestep"): if ignore_sidestep: ignored_protections.append("Sidestep") else: active_protections.append("Sidestep") if defender.get("has_stand_firm"): if ignore_stand_firm: ignored_protections.append("Stand Firm") else: active_protections.append("Stand Firm") if not active_protections: fully_blocked = False evaluated_defenders.append( { "player_id": defender.get("player_id"), "player_name": defender.get("player_name"), "x": defender.get("x"), "y": defender.get("y"), "active_protections": active_protections, "ignored_protections": ignored_protections, "has_active_protection": bool(active_protections), } ) return { "los_defenders": evaluated_defenders, "los_defender_count": len(evaluated_defenders), "attacking_has_grab": attacking_has_grab, "attacking_has_juggernaut": attacking_has_juggernaut, "attacking_has_juggernaut_frenzy": attacking_has_juggernaut_frenzy, "ignore_sidestep": ignore_sidestep, "ignore_stand_firm": ignore_stand_firm, "fully_blocked": fully_blocked, } def _evaluate_ottd_candidate_turn(self, turn_num, kickoff_context): """Build a diagnostic OTTD decision for one structurally valid candidate turn.""" kickoff_step_number = kickoff_context.get("kickoff_step_number") receiving_team_id = kickoff_context.get("receiving_team_id") kicker_team_id = kickoff_context.get("kicker_team_id") receiving_players = self._iter_kickoff_snapshot_team_players(kickoff_step_number, receiving_team_id) receiving_players = sorted(receiving_players, key=lambda row: (row.get("x", 999), row.get("y", 999), row.get("player_name", ""))) kicking_players = self._iter_kickoff_snapshot_team_players(kickoff_step_number, kicker_team_id) kicking_players = sorted(kicking_players, key=lambda row: (row.get("x", 999), row.get("y", 999), row.get("player_name", ""))) touchdowns_after_kickoff = self._collect_turn_touchdowns_after_step( turn_num, kickoff_step_number, team_id=receiving_team_id, ) ttm_actions_after_kickoff = self._collect_turn_ttm_actions_after_step( turn_num, kickoff_step_number, team_id=receiving_team_id, ) touchdown_override = len(touchdowns_after_kickoff) > 0 and len(ttm_actions_after_kickoff) == 0 fastest_halfway_players = [ player for player in receiving_players if player.get("is_fast") and player.get("on_halfway_line") ] fastest_halfway_players.sort( key=lambda row: ( -(row.get("effective_ma") or -999), -1 if row.get("has_sidestep") else 0, row.get("y") or 999, row.get("player_name", ""), ) ) fastest_player = fastest_halfway_players[0] if fastest_halfway_players else None pushes_needed = None minimum_players_required = None minimum_players_met = False if fastest_player is not None and fastest_player.get("effective_ma") is not None: pushes_needed = 11 - fastest_player["effective_ma"] minimum_players_required = self._minimum_ottd_players_required( pushes_needed, fastest_player.get("has_sidestep", False), ) if minimum_players_required is not None: minimum_players_met = len(receiving_players) >= minimum_players_required defender_skill_block = self._evaluate_ottd_defender_skill_blocker( receiving_players, kicking_players, pushes_needed, ) all_players_shallow = bool(receiving_players) and all( player.get("on_halfway_or_one_back") for player in receiving_players ) has_fast_player_on_halfway_line = fastest_player is not None setup_based_decision = ( has_fast_player_on_halfway_line and minimum_players_required is not None and minimum_players_met and not all_players_shallow and not defender_skill_block.get("fully_blocked") ) final_decision = touchdown_override or setup_based_decision reason_lines = [ f"Structural candidate: yes (turn {turn_num} contains a kickoff at step {kickoff_step_number}).", f"Receiving team: {receiving_team_id if receiving_team_id is not None else '?'}; kicking team: {kicker_team_id if kicker_team_id is not None else '?' }.", ( "Touchdown override: yes - a touchdown was scored after the kickoff and no Throw Team Mate action " "was detected in the receiving team's turn." if touchdown_override else "Touchdown override: no." ), ( f"Receiving players on pitch at kickoff: {len(receiving_players)}." if receiving_players else "Receiving players on pitch at kickoff: 0." ), ] if fastest_player is None: reason_lines.append("Fast player on halfway line: no qualifying player found (effective MA >= 7 required).") else: sidestep_text = "yes" if fastest_player.get("has_sidestep") else "no" sprint_text = "yes" if fastest_player.get("has_sprint") else "no" reason_lines.append( "Fastest qualifying player on halfway line: " f"{fastest_player.get('player_name')} [id={fastest_player.get('player_id')}] " f"at ({fastest_player.get('x')},{fastest_player.get('y')}), " f"MA={fastest_player.get('ma')}, effective_MA={fastest_player.get('effective_ma')}, " f"Sprint={sprint_text}, Sidestep={sidestep_text}." ) reason_lines.append( f"Pushes needed: {pushes_needed}; minimum players required: " f"{minimum_players_required if minimum_players_required is not None else 'unavailable'}; " f"criterion met: {'yes' if minimum_players_met else 'no'}." ) reason_lines.append( "All receiving players are on the halfway line or one square back: " f"{'yes' if all_players_shallow else 'no'}." ) if defender_skill_block.get("los_defender_count"): blocker_parts = [] if defender_skill_block.get("ignore_sidestep"): blocker_parts.append("Sidestep ignored because attacking team has Grab") if defender_skill_block.get("ignore_stand_firm"): blocker_parts.append("Stand Firm ignored because attacking team has qualifying Juggernaut support") blocker_context = "; ".join(blocker_parts) if blocker_parts else "no anti-push skill overrides applied" reason_lines.append( f"Kicking LOS defenders with anti-push skills: {defender_skill_block.get('los_defender_count')} checked; " f"all protected after overrides: {'yes' if defender_skill_block.get('fully_blocked') else 'no'}; " f"{blocker_context}." ) else: reason_lines.append("Kicking LOS defenders with anti-push skills: none on the halfway line.") if ttm_actions_after_kickoff: ttm_summary = ", ".join( f"step {row.get('step_number')}:{row.get('player_name')}" for row in ttm_actions_after_kickoff ) reason_lines.append(f"Throw Team Mate actions after kickoff: {ttm_summary}.") else: reason_lines.append("Throw Team Mate actions after kickoff: none.") if touchdowns_after_kickoff: touchdown_summary = ", ".join( f"step {row.get('step_number')}:{row.get('player_name')}" for row in touchdowns_after_kickoff ) reason_lines.append(f"Touchdowns after kickoff: {touchdown_summary}.") else: reason_lines.append("Touchdowns after kickoff: none.") reason_lines.append( f"Final OTTD decision: {'yes' if final_decision else 'no'}." ) fast_halfway_players = [ { "player_id": player.get("player_id"), "player_name": player.get("player_name"), "effective_ma": player.get("effective_ma"), "x": player.get("x"), "y": player.get("y"), "los_x": player.get("los_x"), "side": player.get("side"), } for player in fastest_halfway_players ] return { "turn": turn_num, "is_candidate": True, "kickoff_step_number": kickoff_step_number, "kicker_team_id": kicker_team_id, "receiving_team_id": receiving_team_id, "touchdowns_after_kickoff": touchdowns_after_kickoff, "ttm_actions_after_kickoff": ttm_actions_after_kickoff, "touchdown_override": touchdown_override, "receiving_players": receiving_players, "kicking_players": kicking_players, "receiving_player_count": len(receiving_players), "fast_halfway_players": fast_halfway_players, "fastest_halfway_player": fastest_player, "pushes_needed": pushes_needed, "minimum_players_required": minimum_players_required, "minimum_players_met": minimum_players_met, "all_players_shallow": all_players_shallow, "defender_skill_block": defender_skill_block, "has_fast_player_on_halfway_line": has_fast_player_on_halfway_line, "setup_based_decision": setup_based_decision, "final_decision": final_decision, "criterion_1_pending": False, "reason_lines": reason_lines, } def _extract_step_team_positions(self, step_number, team_id): """Return on-pitch player positions for one team from a raw replay step.""" raw_steps = getattr(self.game_state, "raw_replay_steps", None) if not isinstance(raw_steps, list): return [] step_index = self._to_int(step_number, default=None) team_index = self._to_int(team_id, default=None) if step_index is None or team_index is None: return [] step_index -= 1 if step_index < 0 or step_index >= len(raw_steps): return [] step_data = raw_steps[step_index] if not isinstance(step_data, dict): return [] board_state = step_data.get("BoardState") if not isinstance(board_state, dict): return [] team_states = board_state.get("ListTeams", {}).get("TeamState", []) if not isinstance(team_states, list): team_states = [team_states] if team_index < 0 or team_index >= len(team_states): return [] team_state = team_states[team_index] if not isinstance(team_state, dict): return [] players = team_state.get("ListPitchPlayers", {}).get("PlayerState", []) if not isinstance(players, list): players = [players] positions = [] for player in players: if not isinstance(player, dict): continue player_id = player.get("Id") if player_id in (None, ""): continue cell = player.get("Cell") if isinstance(player.get("Cell"), dict) else None if not isinstance(cell, dict): continue x = self._to_int(cell.get("X"), default=None) y = self._to_int(cell.get("Y"), default=None) if x is None or y is None or x < 0 or y < 0: continue positions.append({"player_id": str(player_id), "x": x, "y": y}) return positions def _squares_into_opponent_half(self, player_position, los_x): """Return how many squares a player is into the opponent half from the halfway line.""" if not isinstance(player_position, dict): return None x = self._to_int(player_position.get("x"), default=None) los_x_int = self._to_int(los_x, default=None) if x is None or los_x_int is None: return None if los_x_int <= 12: return max(0, x - los_x_int) return max(0, los_x_int - x) def _evaluate_ottd_mode_lifecycle_for_turn(self, evaluation): """Evaluate OTTD mode on/off lifecycle for one accepted OTTD attempt turn. Implemented mode-off criteria: - C1: selected block die is non-push - C2: a fast, not-yet-activated starter reaches pushes_needed squares into opponent half - C3: all fast starters from the halfway line have activated """ if not isinstance(evaluation, dict) or not evaluation.get("final_decision"): return { "enabled": False, "reason": "turn is not classified as a final OTTD attempt", } turn_num = self._to_int(evaluation.get("turn"), default=None) kickoff_step = self._to_int(evaluation.get("kickoff_step_number"), default=None) receiving_team_id = self._normalize_team_id(evaluation.get("receiving_team_id")) pushes_needed = self._to_int(evaluation.get("pushes_needed"), default=None) fast_starters = evaluation.get("fast_halfway_players") or [] monitored = { str(player.get("player_id")): dict(player) for player in fast_starters if isinstance(player, dict) and player.get("player_id") is not None } monitored_ids = set(monitored.keys()) candidate_steps = [] for step in getattr(self.game_state, "steps", []) or []: if self._to_int(getattr(step, "game_turn", None), default=None) != turn_num: continue if kickoff_step is not None and step.step_number <= kickoff_step: continue if self._normalize_team_id(getattr(step, "turn_owner_team_id", None)) != receiving_team_id: continue candidate_steps.append(step) candidate_steps = sorted(candidate_steps, key=lambda s: s.step_number) mode_on_step = candidate_steps[0].step_number if candidate_steps else None touchdown_steps_after_kickoff = [ self._to_int(row.get("step_number"), default=None) for row in evaluation.get("touchdowns_after_kickoff") or [] if isinstance(row, dict) ] touchdown_steps_after_kickoff = [step for step in touchdown_steps_after_kickoff if step is not None] first_touchdown_step = min(touchdown_steps_after_kickoff) if touchdown_steps_after_kickoff else None activated = set() activation_step_by_player = {} turned_off = False turned_off_step = None turned_off_timing = None turned_off_criterion = None turned_off_reason = None # Replay block-face encoding: 2 is Push (0=ATT down, 1=Both down, 2=Push, 3=Stumbles, 4=Pow). push_face_value = 2 for step in candidate_steps: # C1: any selected non-push block face by the receiving team turns mode off. if not turned_off: step_number_int = self._to_int(step.step_number, default=None) for event in getattr(self.game_state, "roll_events", []) or []: if not isinstance(event, dict): continue if self._to_int(event.get("step_number"), default=None) != step_number_int: continue if str(event.get("roll_category") or "").lower() != "block": continue if self._normalize_team_id(event.get("team_id")) != receiving_team_id: continue selected_value = self._to_int(event.get("selected_block_die_value"), default=None) if selected_value is None: continue attacker_is_blitzing = bool(event.get("attacker_is_blitzing")) has_juggernaut = self._event_player_has_skill(event, "Juggernaut") # C1 exception: Juggernaut on a blitzing block with selected Both Down does not disable OTTD mode. both_down_face_value = 1 if attacker_is_blitzing and has_juggernaut and selected_value == both_down_face_value: continue if selected_value != push_face_value: rolled_values_raw = event.get("dice_values") if isinstance(event.get("dice_values"), list) else [] rolled_values = [ self._to_int(value, default=None) for value in rolled_values_raw ] rolled_values = [value for value in rolled_values if value is not None] push_was_available = push_face_value in rolled_values turned_off = True turned_off_step = step.step_number turned_off_timing = "before_step" if push_was_available else "after_step" turned_off_criterion = "C1" if push_was_available: turned_off_reason = ( f"step {step.step_number}: non-push selected block face ({selected_value}) by " f"{event.get('player_name', 'Unknown')} despite push being available in dice {rolled_values_raw}" ) else: turned_off_reason = ( f"step {step.step_number}: non-push selected block face ({selected_value}) by " f"{event.get('player_name', 'Unknown')} with no push available in dice {rolled_values_raw}" ) break if turned_off: break step_positions = self._extract_step_team_positions(step.step_number, receiving_team_id) positions_by_player_id = { str(pos.get("player_id")): pos for pos in step_positions if isinstance(pos, dict) and pos.get("player_id") is not None } # C2: player reaches required depth before they have activated. if not turned_off and pushes_needed is not None and pushes_needed >= 0: for player_id in sorted(monitored_ids - activated): pos = positions_by_player_id.get(player_id) if not isinstance(pos, dict): continue player_los_x = monitored.get(player_id, {}).get("los_x") into_half = self._squares_into_opponent_half(pos, player_los_x) if into_half is None: continue if into_half >= pushes_needed: turned_off = True turned_off_step = step.step_number turned_off_timing = "after_step" turned_off_criterion = "C2" turned_off_reason = ( f"step {step.step_number}: {monitored.get(player_id, {}).get('player_name', player_id)} " f"reached {into_half} squares into opponent half before activating " f"(pushes_needed={pushes_needed})" ) break if turned_off: break acting_player_id = getattr(step, "acting_player_id", None) acting_team_id = self._normalize_team_id(getattr(step, "acting_player_team_id", None)) if acting_player_id is not None and acting_team_id == receiving_team_id: acting_player_id_str = str(acting_player_id) if acting_player_id_str in monitored_ids and acting_player_id_str not in activated: activated.add(acting_player_id_str) activation_step_by_player[acting_player_id_str] = step.step_number # C3: all monitored fast starters have activated. if monitored_ids and activated == monitored_ids: turned_off = True turned_off_step = step.step_number turned_off_timing = "after_step" turned_off_criterion = "C3" turned_off_reason = ( f"step {step.step_number}: all fast starters from the halfway line have activated" ) break if turned_off and first_touchdown_step is not None and turned_off_step is not None and turned_off_step <= first_touchdown_step: turned_off_step = first_touchdown_step + 1 turned_off_timing = "after_step" touchdown_reason = f"touchdown scored at step {first_touchdown_step}" if turned_off_reason: turned_off_reason = f"{turned_off_reason}; delayed until after {touchdown_reason}" else: turned_off_reason = f"delayed until after {touchdown_reason}" return { "enabled": True, "mode_on_step": mode_on_step, "criterion_1_pending": False, "monitored_player_ids": sorted(monitored_ids), "monitored_players": [monitored[player_id] for player_id in sorted(monitored_ids)], "activated_player_ids": sorted(activated), "activation_step_by_player": activation_step_by_player, "turned_off": turned_off, "turned_off_step": turned_off_step, "turned_off_timing": turned_off_timing, "turned_off_criterion": turned_off_criterion, "turned_off_reason": turned_off_reason, } def _detect_ottd_attempt_turns(self): """Detect and classify OTTD candidate turns with diagnostic reasoning.""" kickoff_index = self._build_ottd_turn_kickoff_index() evaluations = {} final_attempt_turns = set() for turn_num in sorted(kickoff_index.keys()): evaluation = self._evaluate_ottd_candidate_turn(turn_num, kickoff_index[turn_num]) evaluation["ottd_mode"] = self._evaluate_ottd_mode_lifecycle_for_turn(evaluation) evaluations[turn_num] = evaluation if evaluation.get("final_decision"): final_attempt_turns.add(turn_num) self.game_state.ottd_candidate_turns = set(kickoff_index.keys()) self.game_state.ottd_attempt_turns = final_attempt_turns self.game_state.ottd_evaluations = evaluations def _annotate_player_lookup_kickoff_los(self, replay_steps): """Attach kickoff LOS participation summaries to player lookup entries.""" los_index = self._build_player_lookup_los_index(replay_steps) if not los_index: return for player_id, entries in los_index.items(): existing = self.game_state.player_lookup.get(player_id, {}) if not isinstance(existing, dict): existing = {} merged = dict(existing) merged["kickoff_los_entries"] = entries merged["kickoff_los_count"] = len(entries) merged["kickoff_los_step_numbers"] = [entry["step_number"] for entry in entries] self.game_state.player_lookup[player_id] = merged def _upsert_player_lookup( self, player_id, team_id=None, player_name=None, skills=None, skill_ids=None, number=None, characteristics=None, ma=None, ): """Create or update a player lookup entry with best-known data.""" if player_id is None: return player_id_str = str(player_id) existing = self.game_state.player_lookup.get(player_id_str, {}) merged = { 'name': existing.get('name', 'Unknown'), 'team_id': existing.get('team_id', 'Unknown'), 'skills': existing.get('skills', []), 'skill_ids': existing.get('skill_ids', []), 'number': existing.get('number'), 'characteristics': existing.get('characteristics', {}), 'ma': existing.get('ma'), 'kickoff_los_entries': existing.get('kickoff_los_entries', []), 'kickoff_los_count': existing.get('kickoff_los_count', 0), 'kickoff_los_step_numbers': existing.get('kickoff_los_step_numbers', []), } if team_id is not None: merged['team_id'] = str(team_id) if player_name not in (None, '', 'Unknown'): decoded_name = self._maybe_decode_base64_string(player_name) merged['name'] = self._normalize_player_name(decoded_name) if isinstance(skills, list) and len(skills) > 0: merged['skills'] = skills if isinstance(skill_ids, list) and len(skill_ids) > 0: merged['skill_ids'] = skill_ids if number not in (None, ''): merged['number'] = str(number) if isinstance(characteristics, dict) and len(characteristics) > 0: merged['characteristics'] = characteristics if ma not in (None, ''): try: merged['ma'] = int(str(ma)) except (TypeError, ValueError): merged['ma'] = str(ma) self.game_state.player_lookup[player_id_str] = merged def _extract_team_inducements(self, replay_steps): """Scan the pre-match inducement phase and populate team_states[i].inducements_purchased. Each purchase is appended as an InducementIdTypes string (e.g. "1"=keg, "7"=star player). Multiple purchases of the same type each add a separate entry so the list length directly counts how many of that type were bought. Known purchase events -------------------- EventBuyMercenary – player-type inducements (journeyman=6, star player=7). The bought category is identified by matching MercenaryType (an encoded player template ID) against the EventInducementsData category entries. Any other EventBuy* – forward-looking hook for non-player inducements (e.g. kegs). We expect a GamerId and Type/InducementType field. EventInducementsData – authoritative fallback for purchased counts. In this corpus, purchases are often reflected via category InPossession values rather than explicit EventBuy* payloads. We top up per-type counts from the maximum observed InPossession in the inducement phase, while preserving explicit EventBuy captures. """ # Cache the latest EventInducementsData; it is re-emitted each inducement turn. event_inducements_data = None current_gamer_id = "0" in_possession_by_team_and_type = { "0": defaultdict(int), "1": defaultdict(int), } def _record_in_possession_from_categories(team_key, categories): if team_key not in ("0", "1"): return if not isinstance(categories, list): categories = [categories] for cat in categories: if not isinstance(cat, dict): continue ind_type = str(cat.get("Type") or "").strip() if ind_type == "": continue in_possession = self._to_int(cat.get("InPossession"), default=0) or 0 if in_possession > in_possession_by_team_and_type[team_key][ind_type]: in_possession_by_team_and_type[team_key][ind_type] = in_possession def _signature_from_team_inducements_data(team_entry): if not isinstance(team_entry, dict): return set() cats = team_entry.get("InducementsCategories", {}).get("InducementsCategory", []) if not isinstance(cats, list): cats = [cats] signature = set() for cat in cats: if not isinstance(cat, dict): continue ind_type = str(cat.get("Type") or "").strip() cost = str(cat.get("Cost") or "").strip() max_count = str(cat.get("Max") or "").strip() if ind_type: signature.add((ind_type, cost, max_count)) return signature def _resolve_gamer_id(event_data, fallback_gamer_id): if not isinstance(event_data, dict): return fallback_gamer_id explicit = str(event_data.get("GamerId", "")).strip() if explicit in ("0", "1"): return explicit team_data = event_data.get("TeamInducementsData") if not isinstance(team_data, dict) or not isinstance(event_inducements_data, dict): return fallback_gamer_id event_signature = _signature_from_team_inducements_data(team_data) if not event_signature: return fallback_gamer_id team_list = ( event_inducements_data .get("TeamInducements", {}) .get("TeamInducements", []) ) if not isinstance(team_list, list): team_list = [team_list] best_team = None best_score = -1 for team_idx, team_entry in enumerate(team_list): if team_idx not in (0, 1): continue team_signature = _signature_from_team_inducements_data(team_entry) if not team_signature: continue score = len(event_signature & team_signature) if score > best_score: best_score = score best_team = str(team_idx) return best_team if best_team in ("0", "1") else fallback_gamer_id for step in replay_steps: if not isinstance(step, dict): continue # Track which team is buying this turn. new_ind_turn = step.get("EventNewInducementsTurn") if isinstance(new_ind_turn, dict): current_gamer_id = str(new_ind_turn.get("GamerId", current_gamer_id)) if step.get("EventInducementsData"): event_inducements_data = step["EventInducementsData"] team_list = ( event_inducements_data .get("TeamInducements", {}) .get("TeamInducements", []) if isinstance(event_inducements_data, dict) else [] ) if not isinstance(team_list, list): team_list = [team_list] for team_idx, team_entry in enumerate(team_list): if team_idx not in (0, 1) or not isinstance(team_entry, dict): continue cats = ( team_entry .get("InducementsCategories", {}) .get("InducementsCategory", []) ) _record_in_possession_from_categories(str(team_idx), cats) add_inducement = step.get("EventAddInducement") if isinstance(add_inducement, dict): inferred_gamer_id = _resolve_gamer_id(add_inducement, current_gamer_id) team_data = add_inducement.get("TeamInducementsData") if isinstance(team_data, dict): cats = ( team_data .get("InducementsCategories", {}) .get("InducementsCategory", []) ) _record_in_possession_from_categories(inferred_gamer_id, cats) # ── EventBuyMercenary (journeymen & star players) ────────────────── buy_merc = step.get("EventBuyMercenary") if isinstance(buy_merc, dict): gamer_id = str(buy_merc.get("GamerId", current_gamer_id)) mercenary_type = buy_merc.get("MercenaryType") inducement_type = None if event_inducements_data and mercenary_type is not None: team_list = ( event_inducements_data .get("TeamInducements", {}) .get("TeamInducements", []) ) if not isinstance(team_list, list): team_list = [team_list] team_idx = int(gamer_id) if gamer_id in ("0", "1") else 0 if team_idx < len(team_list): cats = ( team_list[team_idx] .get("InducementsCategories", {}) .get("InducementsCategory", []) ) if not isinstance(cats, list): cats = [cats] for cat in cats: if not isinstance(cat, dict): continue players = cat.get("Players") or {} if not isinstance(players, dict): continue pd = players.get("PlayerData") or {} if isinstance(pd, dict) and pd.get("IdPlayerTypes") == mercenary_type: inducement_type = cat.get("Type") break if inducement_type and gamer_id in ("0", "1"): self.team_states[int(gamer_id)].inducements_purchased.append(str(inducement_type)) # ── Any other EventBuy* (non-player inducements e.g. kegs) ───────── for key in step: if key.startswith("EventBuy") and key != "EventBuyMercenary": event_data = step[key] if isinstance(event_data, dict): gamer_id = _resolve_gamer_id(event_data, current_gamer_id) ind_type = ( event_data.get("Type") or event_data.get("InducementType") or event_data.get("Inducement") ) if ind_type and gamer_id in ("0", "1"): self.team_states[int(gamer_id)].inducements_purchased.append(str(ind_type)) # Stop once the inducement phase is over. if step.get("EventEndInducements") is not None: break for team_key, type_counts in in_possession_by_team_and_type.items(): if team_key not in ("0", "1"): continue state = self.team_states[int(team_key)] for ind_type, possessed_count in type_counts.items(): if possessed_count <= 0: continue already_recorded = state.inducements_purchased.count(str(ind_type)) missing = possessed_count - already_recorded if missing > 0: state.inducements_purchased.extend([str(ind_type)] * missing) def _extract_step_turn_boundary(self, step_data): """Extract explicit turn-end metadata from a replay step.""" boundary = { "explicit_turn_end": False, "end_turn_reason": None, "end_turn_type": None, "next_active_team_id": None, } if not isinstance(step_data, dict): return boundary end_turn = step_data.get("EventEndTurn") if isinstance(end_turn, dict): boundary["explicit_turn_end"] = True boundary["end_turn_reason"] = end_turn.get("Reason") boundary["end_turn_type"] = end_turn.get("FinishingTurnType") next_playing_gamer = self._normalize_team_id(end_turn.get("NextPlayingGamer")) if next_playing_gamer in ("0", "1"): boundary["next_active_team_id"] = next_playing_gamer active_gamer_changed = step_data.get("EventActiveGamerChanged") if isinstance(active_gamer_changed, dict): new_active_gamer = self._normalize_team_id(active_gamer_changed.get("NewActiveGamer")) if new_active_gamer in ("0", "1"): boundary["next_active_team_id"] = new_active_gamer return boundary def _infer_game_turn_owner(self): try: t0_gt = int(self.team_states[0].game_turn) t1_gt = int(self.team_states[1].game_turn) if t0_gt > t1_gt: return "0" if t1_gt > t0_gt: return "1" return None except (TypeError, ValueError): return None def _extract_observed_step_roll_context(self, roll_events): observed_categories = { str(event.get("roll_category") or "").strip().lower() for event in (roll_events or []) if isinstance(event, dict) } is_consequence_only_step = bool(observed_categories) and observed_categories.issubset(CONSEQUENCE_ROLL_CATEGORIES_LOWER) observed_step_names = { str(event.get("step_name") or "").strip() for event in (roll_events or []) if isinstance(event, dict) } is_damage_step_only = bool(observed_step_names) and observed_step_names.issubset({"DamageStep"}) is_consequence_like_step = is_consequence_only_step or is_damage_step_only return observed_categories, is_consequence_like_step def _infer_initiative_owner_from_roll_events(self, roll_events): initiative_teams = { self._normalize_team_id(event.get("team_id")) for event in (roll_events or []) if isinstance(event, dict) and str(event.get("roll_category") or "").strip().lower() in INITIATIVE_ROLL_CATEGORIES_LOWER and self._normalize_team_id(event.get("team_id")) in ("0", "1") } return next(iter(initiative_teams)) if len(initiative_teams) == 1 else None def _classify_step_turn_owner(self, step, step_data, kick_off, current_turn_owner, roll_events=None): """Return canonical turn ownership for a step and the next carried owner.""" kickoff_team_id = self._normalize_team_id(kick_off.team_id) if kick_off else None step_eag = step_data.get("EventActiveGamerChanged") eag_new = ( self._normalize_team_id(step_eag.get("NewActiveGamer")) if isinstance(step_eag, dict) else None ) step_et = step_data.get("EventEndTurn") et_next = ( self._normalize_team_id(step_et.get("NextPlayingGamer")) if isinstance(step_et, dict) else None ) has_end_turn = step.explicit_turn_end gt_owner = self._infer_game_turn_owner() step_owner = None next_owner_state = current_turn_owner active_fallback = self._normalize_team_id(step.active_team_id) # Consequence-only fallout rows (armour/injury/casualty) should stay on the # current activation even if EventActiveGamerChanged appears in this step. observed_categories, is_consequence_like_step = self._extract_observed_step_roll_context(roll_events) # Generic ball rolls are not reliable initiative evidence because bounce / # scatter fallout can be attributed to the non-active team mid-activation. initiative_owner = self._infer_initiative_owner_from_roll_events(roll_events) if kickoff_team_id in ("0", "1"): # Kickoff step: the kicker owns this step; set receiver as next owner. step_owner = kickoff_team_id receiver = eag_new or et_next if receiver not in ("0", "1"): receiver = self._opponent_team_id(kickoff_team_id) next_owner_state = receiver elif eag_new in ("0", "1") and not has_end_turn: # Mid-turn gamer switch: transfer ownership from the next step. # Keep the current step on the prior owner when it is consequence-only. if ( current_turn_owner in ("0", "1") and active_fallback == current_turn_owner and eag_new != current_turn_owner and ( not observed_categories or initiative_owner not in ("0", "1") or initiative_owner == current_turn_owner ) ): # Timer/UI-only or reaction-window active-gamer changes: the # opponent briefly takes UI control (e.g. to confirm a bomb # reaction or an opponent-scoped interaction) but the turn still # belongs to the current owner. Guard fires when there are no # rolls, or when roll evidence supports the current owner rather # than the incoming gamer. step_owner = current_turn_owner next_owner_state = current_turn_owner else: next_owner_state = eag_new if is_consequence_like_step and current_turn_owner in ("0", "1"): step_owner = current_turn_owner else: step_owner = eag_new elif has_end_turn: # End of someone's turn: this step belongs to whoever was active. step_owner = next_owner_state if next_owner_state in ("0", "1") else gt_owner # Determine next active team from explicit signals or GameTurn delta. next_owner = eag_new or et_next or gt_owner if next_owner in ("0", "1"): next_owner_state = next_owner if next_owner not in ("0", "1"): # When GT delta is ambiguous (equal counters), fall back to the # active_team_id, which tracks which team's counter changed. if active_fallback in ("0", "1"): next_owner = active_fallback if next_owner in ("0", "1"): next_owner_state = next_owner elif initiative_owner in ("0", "1") and ( active_fallback not in ("0", "1") or active_fallback == initiative_owner ): # No explicit handoff signal in this step; trust clear initiative rolls # only when they align with the active-team tracker. step_owner = initiative_owner next_owner_state = initiative_owner elif next_owner_state in ("0", "1"): # No change: carry forward. step_owner = next_owner_state elif gt_owner in ("0", "1"): # Bootstrap from GameTurn when no explicit signal has fired yet. step_owner = gt_owner next_owner_state = gt_owner return step_owner, next_owner_state def build_player_lookup(self, jsonData): """Build a lookup dictionary for quick player ID -> (name, team_id) access.""" rosters = jsonData["Replay"]["Rosters"] team_rosters = rosters.get("TeamRoster", []) if not isinstance(team_rosters, list): team_rosters = [team_rosters] for team_idx, team_roster in enumerate(team_rosters): team_id = team_roster.get("Team", {}).get("TeamId", team_idx) player_data_list = team_roster.get("Players", {}).get("PlayerData", []) if not isinstance(player_data_list, list): player_data_list = [player_data_list] for player_data in player_data_list: if isinstance(player_data, dict): player_id = player_data.get("Id", "Unknown") player_name = self._maybe_decode_base64_string(player_data.get("Name", "Unknown")) player_skill_ids = self._extract_player_skill_ids(player_data) player_skills = self._extract_player_skills(player_data) player_number = player_data.get("Number") player_characteristics = self._parse_player_characteristics(player_data) player_ma = None if isinstance(player_characteristics, dict): ma_entry = player_characteristics.get("0") if isinstance(ma_entry, dict): player_ma = ma_entry.get("value") self._upsert_player_lookup( player_id, team_id=team_id, player_name=player_name, skills=player_skills, skill_ids=player_skill_ids, number=player_number, characteristics=player_characteristics, ma=player_ma, ) # Add induced/star players that are not always present in Rosters. replay = jsonData.get("Replay", {}) replay_steps = replay.get("ReplayStep", []) if isinstance(replay_steps, dict): replay_steps = [replay_steps] for step_data in replay_steps: if not isinstance(step_data, dict): continue # EventBuyMercenary can include selected star player details. buy_mercenary = step_data.get("EventBuyMercenary") if isinstance(buy_mercenary, dict): mercenary_id = buy_mercenary.get("MercenaryId") gamer_id = buy_mercenary.get("GamerId") mercenary_name = None mercenary_number = None team_inducements_data = buy_mercenary.get("TeamInducementsData") if isinstance(team_inducements_data, dict): categories = team_inducements_data.get("InducementsCategories", {}).get("InducementsCategory", []) if not isinstance(categories, list): categories = [categories] for category in categories: if not isinstance(category, dict): continue players_node = category.get("Players") or {} if not isinstance(players_node, dict): continue player_data = players_node.get("PlayerData") if isinstance(player_data, dict): mercenary_name = player_data.get("Name", mercenary_name) mercenary_number = player_data.get("Number", mercenary_number) break self._upsert_player_lookup( mercenary_id, team_id=gamer_id, player_name=mercenary_name, number=mercenary_number, ) # BoardState often has all currently active players, including induced stars. board_state = step_data.get("BoardState") if not isinstance(board_state, dict): continue team_states = board_state.get("ListTeams", {}).get("TeamState", []) if not isinstance(team_states, list): team_states = [team_states] for team_idx, team_state in enumerate(team_states): if not isinstance(team_state, dict): continue players_state = team_state.get("ListPitchPlayers", {}).get("PlayerState", []) if not isinstance(players_state, list): players_state = [players_state] for player_state in players_state: if not isinstance(player_state, dict): continue player_id = player_state.get("Id") player_data = player_state.get("Data", {}) name = player_data.get("Name") if isinstance(player_data, dict) else None number = player_data.get("Number") if isinstance(player_data, dict) else None skill_ids = self._extract_player_skill_ids(player_data) if isinstance(player_data, dict) else [] skills = self._extract_player_skills(player_data) if isinstance(player_data, dict) else [] self._upsert_player_lookup( player_id, team_id=team_idx, player_name=name, skills=skills, skill_ids=skill_ids, number=number, ) # Populate per-team inducement lists from the pre-match phase. self._extract_team_inducements(replay_steps) def _ensure_list(self, value): """Normalize a value to list form for replay fields that are variably shaped.""" if value is None: return [] if isinstance(value, list): return value return [value] def normalize_replay_step_shapes(self, replay_steps): """Normalize replay step payload shapes in-place for consistent downstream parsing.""" for step_data in replay_steps: if not isinstance(step_data, dict): continue # Normalize TeamState to always be a list. board_state = step_data.get("BoardState") if isinstance(board_state, dict): list_teams = board_state.get("ListTeams") if isinstance(list_teams, dict) and "TeamState" in list_teams: list_teams["TeamState"] = self._ensure_list(list_teams.get("TeamState")) # Normalize EventExecuteSequence and nested StepResult/StringMessage fields. if "EventExecuteSequence" in step_data: step_data["EventExecuteSequence"] = self._ensure_list(step_data.get("EventExecuteSequence")) for sequence_item in step_data["EventExecuteSequence"]: if not isinstance(sequence_item, dict): continue sequence = sequence_item.get("Sequence") if not isinstance(sequence, dict) or "StepResult" not in sequence: continue sequence["StepResult"] = self._ensure_list(sequence.get("StepResult")) for step_result in sequence["StepResult"]: if not isinstance(step_result, dict): continue results = step_result.get("Results") if isinstance(results, dict) and "StringMessage" in results: results["StringMessage"] = self._ensure_list(results.get("StringMessage")) def extract_player_from_event_execute_sequence(self, exec_seq): """Extract player ID from an EventExecuteSequence.""" # Replay payloads can provide EventExecuteSequence as either a list or a dict. sequence_items = [] if isinstance(exec_seq, list): sequence_items = exec_seq elif isinstance(exec_seq, dict): sequence_items = [exec_seq] for seq_item in sequence_items: if not isinstance(seq_item, dict) or "Sequence" not in seq_item: continue sequence = seq_item["Sequence"] if not isinstance(sequence, dict) or "StepResult" not in sequence: continue step_results = sequence["StepResult"] if isinstance(step_results, dict): step_results = [step_results] if not isinstance(step_results, list): continue for step_result in step_results: if not isinstance(step_result, dict): continue step = step_result.get("Step") if isinstance(step, dict) and step.get("Name") == "PlayerStep" and "MessageData" in step: message_data = step["MessageData"] if isinstance(message_data, str): try: player_step_data = xmltodict.parse(message_data) return player_step_data["PlayerStep"]["PlayerId"] except (ExpatError, TypeError, KeyError): pass return None def extract_kick_off_event(self, data, game_turn): """Extract a kickoff event from data if EventKickOffTable is present.""" if "EventKickOffTable" not in data: return None # Get the EventKickOffTable data kickoff_table = data["EventKickOffTable"] # Extract player info from EventExecuteSequence player_id = None player_name = None team_id = None exec_seq = data.get("EventExecuteSequence") if exec_seq: player_id = self.extract_player_from_event_execute_sequence(exec_seq) # Look up player name and team ID if isinstance(player_id, str): lookup = self._safe_player_lookup(player_id) player_name = lookup.get('name', 'Not found') team_id = lookup.get('team_id', 'Not found') # Find following Event* event_keys = [k for k in data.keys() if k.startswith("Event")] next_event = None next_event_data = None if "EventKickOffTable" in event_keys: idx = event_keys.index("EventKickOffTable") if idx + 1 < len(event_keys): next_event = event_keys[idx + 1] next_event_data = data.get(next_event) # Build kickoff-specific context used by classification/reporting rules. kickoff_context = { "secret_weapon_eligible_counts": {"0": 0, "1": 0}, "secret_weapon_eligible_player_ids": {"0": [], "1": []}, "team_game_turns": {}, "weather_total_before": None, "weather_dice_before": [], } board_state = data.get("BoardState") if isinstance(board_state, dict): last_weather_roll = board_state.get("LastWeatherRoll") if isinstance(last_weather_roll, dict): weather_dice = last_weather_roll.get("Die") if weather_dice is None: weather_dice = (last_weather_roll.get("Dice") or {}).get("Die") if weather_dice is None: weather_dice = [] if not isinstance(weather_dice, list): weather_dice = [weather_dice] parsed_weather_dice = [] for die in weather_dice: if isinstance(die, dict): raw_value = die.get("Value") else: raw_value = die try: parsed_weather_dice.append(int(raw_value)) except (TypeError, ValueError): continue if parsed_weather_dice: kickoff_context["weather_dice_before"] = parsed_weather_dice kickoff_context["weather_total_before"] = sum(parsed_weather_dice) secret_weapon_ids = board_state.get("ListSecretWeapons", {}).get("ListSecretWeaponsItem", []) if secret_weapon_ids is None: secret_weapon_ids = [] if not isinstance(secret_weapon_ids, list): secret_weapon_ids = [secret_weapon_ids] secret_weapon_ids = set(str(pid) for pid in secret_weapon_ids if pid is not None) team_states = board_state.get("ListTeams", {}).get("TeamState", []) if not isinstance(team_states, list): team_states = [team_states] for team_idx, team_state in enumerate(team_states[:2]): if not isinstance(team_state, dict): continue tid = str(team_idx) kickoff_context["team_game_turns"][tid] = team_state.get("GameTurn") players = team_state.get("ListPitchPlayers", {}).get("PlayerState", []) if not isinstance(players, list): players = [players] for player in players: if not isinstance(player, dict): continue pid = str(player.get("Id")) if player.get("Id") is not None else None if not pid or pid not in secret_weapon_ids: continue cell = player.get("Cell") if isinstance(player.get("Cell"), dict) else {} cell_x = str(cell.get("X")) if cell.get("X") is not None else None cell_y = str(cell.get("Y")) if cell.get("Y") is not None else None is_on_pitch = not (cell_x == "-1" and cell_y == "-1") if is_on_pitch: kickoff_context["secret_weapon_eligible_counts"][tid] += 1 kickoff_context["secret_weapon_eligible_player_ids"][tid].append(pid) # Create and return the KickOffEvent kick_off = KickOffEvent( player_id=player_id, player_name=player_name, team_id=team_id, game_turn=game_turn, kickoff_table=kickoff_table, next_event=next_event, next_event_data=next_event_data, kickoff_context=kickoff_context, ) return kick_off def extract_touchdown_event(self, data): """Extract touchdown details from a replay step if present.""" touchdown_data = data.get("EventTouchdown") if not isinstance(touchdown_data, dict): return None player_id = touchdown_data.get("PlayerId") lookup = self._safe_player_lookup(player_id) return { 'player_id': player_id, 'player_name': lookup.get('name', 'Unknown'), 'team_id': lookup.get('team_id', 'Unknown'), 'source': 'EventTouchdown', } def _infer_touchdown_from_score_updates(self, score_updates, extracted_player_id=None): """Fallback touchdown inference when EventTouchdown is missing in a scoring step.""" if not isinstance(score_updates, list): return None scoring_updates = [] for update in score_updates: if not isinstance(update, dict): continue team_id = update.get("team_id") points = update.get("points") if team_id is None: continue try: points_int = int(str(points)) except (TypeError, ValueError): continue if points_int > 0: scoring_updates.append({"team_id": str(team_id), "points": points_int}) if not scoring_updates: return None inferred_team_id = scoring_updates[0]["team_id"] inferred_player_id = None inferred_player_name = "Unknown" if extracted_player_id is not None: lookup = self._safe_player_lookup(extracted_player_id) lookup_team_id = str(lookup.get("team_id")) if lookup.get("team_id") is not None else None if lookup_team_id == inferred_team_id: inferred_player_id = extracted_player_id inferred_player_name = lookup.get("name", "Unknown") self._emit_processing_warning( f"[touchdown] inferred touchdown from EventScore fallback for team {inferred_team_id} (no EventTouchdown payload in step)", key=("touchdown-score-fallback", inferred_team_id), ) return { "player_id": inferred_player_id, "player_name": inferred_player_name, "team_id": inferred_team_id, "source": "EventScoreFallback", } def _build_touchdown_roll_events(self): """Create synthetic touchdown marker events for detailed roll reporting.""" return build_touchdown_roll_events(self) def extract_score_from_event(self, event_data): """Extract score information from EventScore.""" # EventScore typically has Team and Points fields if isinstance(event_data, dict): team_id = event_data.get('Team', event_data.get('TeamId')) points = event_data.get('Points', 1) return team_id, points return None, 0 def _parse_message_data_xml(self, message_data): """Parse message data that may be plain XML or base64-encoded XML.""" if not isinstance(message_data, str): return None # Try direct XML parse first (common in current workflow). try: return xmltodict.parse(message_data) except (ExpatError, TypeError): pass # Fallback: payloads in some parser modes can be base64-encoded once or twice. for _ in range(2): try: message_data = base64.b64decode(message_data).decode("utf-8", errors="replace") except (binascii.Error, UnicodeDecodeError, ValueError): return None try: return xmltodict.parse(message_data) except (ExpatError, TypeError): continue return None def _extract_dice_values(self, dice_node): """Normalize Dice.Die payloads into a list of integer values.""" if not isinstance(dice_node, dict): return [] dies = dice_node.get("Die") if dies is None: return [] if not isinstance(dies, list): dies = [dies] values = [] for die in dies: if isinstance(die, dict) and "Value" in die: try: value = die.get("Value") if value is not None: values.append(int(value)) except (TypeError, ValueError): pass return values def _normalize_block_dice_values(self, dice_values): """Map BB3 block-face encoding (0..4) to Calculator.calc_blocks codes.""" if not isinstance(dice_values, list): return [] normalized = [] for value in dice_values: normalized_die = get_normalized_block_face(value) if normalized_die is None: continue normalized.append(normalized_die) return normalized def _extract_modifier_values(self, modifiers_node): """Extract integer modifier values from Modifiers payload.""" if not isinstance(modifiers_node, dict): return [] modifier_node = modifiers_node.get("Modifier") if modifier_node is None: return [] if not isinstance(modifier_node, list): modifier_node = [modifier_node] values = [] for modifier in modifier_node: if isinstance(modifier, dict) and "Value" in modifier: try: value = modifier.get("Value") if value is not None: values.append(int(value)) except (TypeError, ValueError): pass return values def _extract_player_id_from_step_message(self, step_message): """Extract PlayerId from parsed Step.MessageData payload.""" if not isinstance(step_message, dict) or len(step_message) == 0: return None root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if not isinstance(root, dict): return None return root.get("PlayerId") def _extract_ball_carrier_id_from_step(self, step_data): """Return ball carrier player id for the replay step, when available.""" if not isinstance(step_data, dict): return None board_state = step_data.get("BoardState") if not isinstance(board_state, dict): return None ball_state = board_state.get("Ball") if not isinstance(ball_state, dict): return None is_held = ball_state.get("IsHeld") if str(is_held).strip().lower() not in ("1", "true", "yes"): return None carrier_id = ball_state.get("Carrier") if carrier_id in (None, "", "-1"): return None return carrier_id def _extract_bool_like(self, value): """Convert common replay truthy/falsey encodings to bool or None.""" if isinstance(value, bool): return value normalized = str(value).strip().lower() if normalized in ("1", "true", "yes"): return True if normalized in ("0", "false", "no"): return False return None def _infer_block_is_blitzing(self, payload_root, step_message): """Infer whether current block belongs to a Blitz action. Prefer explicit replay markers only; avoid speculative heuristics that may mislabel ordinary blocks. """ explicit_keys = ( "IsBlitz", "IsBlitzing", "Blitz", "Blitzing", "IsBlitzAction", ) if isinstance(payload_root, dict): for key in explicit_keys: if key in payload_root: parsed = self._extract_bool_like(payload_root.get(key)) if parsed is not None: return parsed if isinstance(step_message, dict) and len(step_message) > 0: root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if isinstance(root, dict): for key in explicit_keys: if key in root: parsed = self._extract_bool_like(root.get(key)) if parsed is not None: return parsed action_like = root.get("ActionType") or root.get("Action") if isinstance(action_like, str) and "blitz" in action_like.lower(): return True return False def _label_ball_bounce_events(self, roll_events): """Classify BallStep events as 'ball' category with context from originating player.""" if not isinstance(roll_events, list): return roll_events # Find all bounce events and try to backfill from previous roll. for i, event in enumerate(roll_events): if (event.get('step_name') == 'BallStep' and event.get('player_name') == 'Unknown' and event.get('player_id') is None): # This is a ball bounce. Try to find the originating player. step_number = event.get('step_number') roll_ordinal = event.get('roll_ordinal_in_step') or 0 # Look for a previous roll in the same step. origin_player = None origin_team_id = None for j in range(i - 1, -1, -1): prev_event = roll_events[j] if str(prev_event.get('step_number')) != str(step_number): break if int(prev_event.get('roll_ordinal_in_step') or 0) < roll_ordinal: prev_player = prev_event.get('player_name', '') if prev_player and prev_player != 'Unknown': origin_player = prev_player origin_team_id = prev_event.get('team_id') break # Update ball bounce classification and semantics. if origin_player: event['player_name'] = f"Ball Bounce (from {origin_player})" else: event['player_name'] = "Ball Bounce" # Set team_id to originating player's team (so dice_roller resolves correctly). if origin_team_id is not None: event['team_id'] = origin_team_id # Set roll category to 'ball' instead of 'action'. event['roll_category'] = 'ball' # Set result to neutral. event['result_value'] = 0 event['result_classification'] = 'neutral' event['report_result_classification'] = 'neutral' # Set probabilities: 100% neutral, 0% success/fail. event['probability_success'] = 0.0 event['probability_neutral'] = 1.0 event['probability_fail'] = 0.0 return roll_events def _parse_string_messages(self, string_messages): """Parse step-result StringMessage payloads once and preserve replay order.""" parsed = [] if not isinstance(string_messages, list): return parsed for string_message_index, result in enumerate(string_messages, start=1): if not isinstance(result, dict): continue payload = self._parse_message_data_xml(result.get("MessageData")) if not isinstance(payload, dict) or len(payload) == 0: continue payload_type = next(iter(payload.keys())) payload_root = payload.get(payload_type) if not isinstance(payload_root, dict): continue parsed.append( { "string_message_index": string_message_index, "result": result, "result_name": result.get("Name"), "payload": payload, "payload_type": payload_type, "payload_root": payload_root, } ) return parsed def extract_roll_events_from_step(self, step_data, game_turn, forced_blitz_player_id=None, step_number=None): """Compatibility entrypoint that delegates to dedicated step extractor.""" return self.step_roll_extractor.extract_roll_events_from_step( step_data, game_turn, forced_blitz_player_id=forced_blitz_player_id, step_number=step_number, ) def _extract_roll_events_from_step_impl(self, step_data, game_turn, forced_blitz_player_id=None, step_number=None): """Extract all non-kickoff dice roll events from a replay step.""" roll_events = [] is_kickoff_step = isinstance(step_data, dict) and "EventKickOffTable" in step_data step_ball_carrier_id = self._extract_ball_carrier_id_from_step(step_data) roll_ordinal_in_step = 0 roll_type_ordinal_in_step = defaultdict(int) # Step-scoped context for better Blitz block detection. step_blitz_players = set() step_stood_up_players = set() step_special_skill_usages = [] step_consumed_special_usage_indexes = set() step_foul_player_id = None step_foul_target_id = None step_foul_emitted = False # Some replay steps carry roll payloads as top-level EventRoll/ResultRoll # (with optional EventSkillUsage), either with or without # EventExecuteSequence. top_level_parsed_messages = [] top_event_skill_usage_raw = step_data.get("EventSkillUsage") if isinstance(step_data, dict) else None top_event_roll_raw = step_data.get("EventRoll") if isinstance(step_data, dict) else None top_result_roll_raw = step_data.get("ResultRoll") if isinstance(step_data, dict) else None top_event_skill_usages = self._ensure_list(top_event_skill_usage_raw) for usage in top_event_skill_usages: if not isinstance(usage, dict): continue top_level_parsed_messages.append( { "string_message_index": len(top_level_parsed_messages) + 1, "result": {"Name": "ResultSkillUsage", "MessageData": ""}, "result_name": "ResultSkillUsage", "payload_type": "ResultSkillUsage", "payload_root": usage, } ) top_event_rolls = self._ensure_list(top_event_roll_raw) top_result_rolls = self._ensure_list(top_result_roll_raw) top_level_rolls = [] for roll in top_event_rolls: if isinstance(roll, dict): top_level_rolls.append(roll) for roll in top_result_rolls: if isinstance(roll, dict): top_level_rolls.append(roll) for roll in top_level_rolls: top_level_parsed_messages.append( { "string_message_index": len(top_level_parsed_messages) + 1, "result": {"Name": "ResultRoll", "MessageData": ""}, "result_name": "ResultRoll", "payload_type": "ResultRoll", "payload_root": roll, } ) def _event_signature(event): if not isinstance(event, dict): return None return ( str(event.get("roll_type") or "-"), str(event.get("difficulty") or "-"), tuple(event.get("dice_values") or []), str(event.get("player_id") or "-"), str(event.get("team_id") or "-"), ) exec_sequences = step_data.get("EventExecuteSequence") if isinstance(exec_sequences, list): for sequence_index, seq_item in enumerate(exec_sequences, start=1): if not isinstance(seq_item, dict): continue sequence = seq_item.get("Sequence") if not isinstance(sequence, dict): continue step_results = sequence.get("StepResult") if not isinstance(step_results, list): continue for step_result_index, step_result in enumerate(step_results, start=1): if not isinstance(step_result, dict): continue step_node = step_result.get("Step") step_name = step_node.get("Name") if isinstance(step_node, dict) else None step_message = None actor_player_id = None step_type = None if isinstance(step_node, dict) and isinstance(step_node.get("MessageData"), str): step_message = self._parse_message_data_xml(step_node.get("MessageData")) actor_player_id = self._extract_player_id_from_step_message(step_message) if isinstance(step_message, dict) and len(step_message) > 0: root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if isinstance(root, dict): step_type = root.get("StepType") # Observed in BB3: StepType 7 appears when a prone player stands up. if str(step_type) == "7" and actor_player_id is not None: step_stood_up_players.add(str(actor_player_id)) actor_lookup = self._safe_player_lookup(actor_player_id) actor_player_name = actor_lookup.get("name", "Unknown") actor_team_id = actor_lookup.get("team_id", "Unknown") actor_skills = actor_lookup.get("skills", []) defender_player_id = None defender_player_name = None defender_skills = [] if isinstance(step_message, dict) and len(step_message) > 0: root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if isinstance(root, dict): defender_player_id = root.get("TargetId") # TargetId of '-1' (or any non-positive value) is a sentinel meaning "no target". try: _did_int = int(str(defender_player_id)) if defender_player_id is not None else -1 except (TypeError, ValueError): _did_int = -1 if _did_int > 0: defender_lookup = self._safe_player_lookup(defender_player_id) defender_player_name = defender_lookup.get("name", "Unknown") defender_skills = defender_lookup.get("skills", []) else: defender_player_id = None results = step_result.get("Results") if not isinstance(results, dict): continue string_messages = results.get("StringMessage") parsed_messages = self._parse_string_messages(string_messages) if len(parsed_messages) == 0: continue step_foul_player_id, step_foul_target_id = self._collect_step_sidechannel_hints( parsed_messages, actor_player_id, defender_player_id, step_blitz_players, step_special_skill_usages, step_foul_player_id, step_foul_target_id, ) extracted_events, roll_ordinal_in_step, step_foul_emitted = self._extract_roll_events_from_string_messages( parsed_messages, game_turn=game_turn, step_number=step_number, sequence_index=sequence_index, step_result_index=step_result_index, step_name=step_name, step_type=step_type, step_message=step_message, actor_player_id=actor_player_id, actor_player_name=actor_player_name, actor_team_id=actor_team_id, actor_skills=actor_skills, defender_player_id=defender_player_id, defender_player_name=defender_player_name, defender_skills=defender_skills, forced_blitz_player_id=forced_blitz_player_id, step_ball_carrier_id=step_ball_carrier_id, is_kickoff_step=is_kickoff_step, step_blitz_players=step_blitz_players, step_stood_up_players=step_stood_up_players, step_special_skill_usages=step_special_skill_usages, step_consumed_special_usage_indexes=step_consumed_special_usage_indexes, step_foul_player_id=step_foul_player_id, step_foul_target_id=step_foul_target_id, step_foul_emitted=step_foul_emitted, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) roll_events.extend(extracted_events) if len(top_level_parsed_messages) > 0: actor_player_id = None if len(top_level_rolls) > 0 and isinstance(top_level_rolls[0], dict): actor_player_id = top_level_rolls[0].get("PlayerId") if actor_player_id is None and len(top_event_skill_usages) > 0 and isinstance(top_event_skill_usages[0], dict): actor_player_id = top_event_skill_usages[0].get("PlayerId") actor_lookup = self._safe_player_lookup(actor_player_id) actor_player_name = actor_lookup.get("name", "Unknown") actor_team_id = actor_lookup.get("team_id") if actor_team_id is None and len(top_level_rolls) > 0 and isinstance(top_level_rolls[0], dict): actor_team_id = top_level_rolls[0].get("TeamId") actor_team_id = actor_team_id if actor_team_id is not None else "Unknown" actor_skills = actor_lookup.get("skills", []) step_foul_player_id, step_foul_target_id = self._collect_step_sidechannel_hints( top_level_parsed_messages, actor_player_id, None, step_blitz_players, step_special_skill_usages, step_foul_player_id, step_foul_target_id, ) extracted_top_level_events, roll_ordinal_in_step, step_foul_emitted = self._extract_roll_events_from_string_messages( top_level_parsed_messages, game_turn=game_turn, step_number=step_number, sequence_index=0, step_result_index=0, step_name="TopLevelEventRoll", step_type=top_level_rolls[0].get("StepType") if len(top_level_rolls) > 0 and isinstance(top_level_rolls[0], dict) else None, step_message=None, actor_player_id=actor_player_id, actor_player_name=actor_player_name, actor_team_id=actor_team_id, actor_skills=actor_skills, defender_player_id=None, defender_player_name=None, defender_skills=[], forced_blitz_player_id=forced_blitz_player_id, step_ball_carrier_id=step_ball_carrier_id, is_kickoff_step=is_kickoff_step, step_blitz_players=step_blitz_players, step_stood_up_players=step_stood_up_players, step_special_skill_usages=step_special_skill_usages, step_consumed_special_usage_indexes=step_consumed_special_usage_indexes, step_foul_player_id=step_foul_player_id, step_foul_target_id=step_foul_target_id, step_foul_emitted=step_foul_emitted, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) existing_signatures = { sig for sig in (_event_signature(event) for event in roll_events) if sig is not None } for event in extracted_top_level_events: sig = _event_signature(event) if sig is None or sig in existing_signatures: continue roll_events.append(event) existing_signatures.add(sig) if len(step_special_skill_usages) > 0: for usage_index, usage in enumerate(step_special_skill_usages): if usage_index in step_consumed_special_usage_indexes: continue if str(usage.get("skill_label") or "") != "Chainsaw": continue self._emit_processing_warning( ( f"Chainsaw usage on step {step_number if step_number is not None else '?'} " "was detected but no linked armour roll was found in the same step." ), key=("chainsaw-unlinked", step_number, usage_index), ) # Post-process: identify and label ball bounce events. roll_events = self._label_ball_bounce_events(roll_events) return roll_events def _collect_step_sidechannel_hints( self, parsed_messages, actor_player_id, defender_player_id, step_blitz_players, step_special_skill_usages, step_foul_player_id, step_foul_target_id, ): """Collect non-roll hints (blitz/foul/special skill usage) from parsed messages.""" for parsed in parsed_messages: if not isinstance(parsed, dict): continue maybe_type = parsed.get("payload_type") maybe_root = parsed.get("payload_root") if not isinstance(maybe_root, dict): continue if maybe_type == "ResultUseAction": action_value = maybe_root.get("Action") action_text = str(action_value).strip().lower() is_blitz_action = action_text == "3" or "blitz" in action_text if is_blitz_action and actor_player_id is not None: step_blitz_players.add(str(actor_player_id)) if action_text == "6" and actor_player_id is not None: step_foul_player_id = actor_player_id step_foul_target_id = defender_player_id continue if maybe_type == "ResultSkillUsage": skill_raw = maybe_root.get("Skill") skill_label = self._classify_special_skill_usage(skill_raw) used_value = str(maybe_root.get("Used") or "").strip().lower() is_used = used_value in ("", "1", "true", "yes") if skill_label and is_used: step_special_skill_usages.append({ "skill_label": skill_label, "skill_raw": skill_raw, "player_id": ( maybe_root.get("PlayerId") or maybe_root.get("Id") or maybe_root.get("SourceId") or maybe_root.get("AttackerId") or actor_player_id ), "target_id": ( maybe_root.get("TargetId") or maybe_root.get("Target") or maybe_root.get("VictimId") or defender_player_id ), "raw": maybe_root, }) return step_foul_player_id, step_foul_target_id def _resolve_string_message_roll_root(self, payload_root): if isinstance(payload_root.get("Dice"), dict): return payload_root roll_infos = payload_root.get("RollInfos") if isinstance(roll_infos, dict) and isinstance(roll_infos.get("Dice"), dict): return roll_infos # QuestionApothecaryCasualtyChoice carries the accepted reroll under NewRoll. new_roll = payload_root.get("NewRoll") if isinstance(new_roll, dict) and isinstance(new_roll.get("Dice"), dict): return new_roll return None def _classify_string_message_roll(self, payload_type, result_name, step_name, roll_root): roll_type_str = str(roll_root.get("RollType")) if roll_root.get("RollType") is not None else "" payload_lower = str(payload_type or "").lower() result_lower = str(result_name or "").lower() step_name_lower = str(step_name or "").lower() if payload_type == "QuestionBlockDice": return "block" if payload_type == "QuestionChooseDice": # Pro-driven block resolution commonly arrives as QuestionChooseDice # with block faces and an attacker choice instead of QuestionBlockDice. if roll_root.get("AttackerChoice") is not None: return "block" if payload_type == "ResultInjuryRoll": return "injury" if ( "casualty" in payload_lower or "casualty" in result_lower or "casualty" in step_name_lower or ( roll_type_str == "12" and payload_type in ("QuestionApothecaryCasualtyUsage", "ResultCasualtyRoll") ) ): return "casualty" if ( "korecovery" in payload_lower or "korecovery" in result_lower or ( roll_type_str == "11" and payload_type == "QuestionKORecovery" ) ): return "ko_recovery_injury" if roll_root.get("Difficulty") is not None: return "armour" if roll_type_str == "10" else "action" return "other" def _resolve_block_string_message_roll( self, *, dice_values, payload_root, roll_root, step_message, actor_player_id, actor_skills, defender_player_id, defender_skills, forced_blitz_player_id, step_ball_carrier_id, step_blitz_players, step_stood_up_players, ): def _resolve_attacker_choice(root, actor_pid): if not isinstance(root, dict): return True, "default_attacker" attacker_choice_raw = root.get("AttackerChoice") if attacker_choice_raw is not None: return str(attacker_choice_raw) == "1", "attacker_choice" defender_choice_raw = root.get("DefenderChoice") if defender_choice_raw is not None: return str(defender_choice_raw) != "1", "defender_choice" gamer_raw = root.get("GamerId") if gamer_raw is not None and actor_pid is not None: chooser_team_id = self._normalize_team_id(gamer_raw) actor_team_id = self._normalize_team_id(self._safe_player_lookup(actor_pid).get("team_id")) if chooser_team_id in ("0", "1") and actor_team_id in ("0", "1"): return chooser_team_id == actor_team_id, "gamer_id" return True, "default_attacker" def _parse_dice_outcomes(root): if not isinstance(root, dict): return [] node = root.get("DiceOutcomes") if not isinstance(node, dict): return [] item = node.get("DiceOutcomesItem") if item is None: return [] values = item if isinstance(item, list) else [item] parsed = [] for value in values: parsed_val = self._to_int(value, default=None) if parsed_val is None: return [] parsed.append(parsed_val) return parsed def _select_block_die_metadata(raw_dice_values, root, attacker_choose): if not isinstance(raw_dice_values, list) or len(raw_dice_values) == 0: return { "selected_block_die_index": None, "selected_block_die_value": None, "selected_block_die_outcome": None, "selected_block_die_source": None, } if len(raw_dice_values) == 1: return { "selected_block_die_index": 0, "selected_block_die_value": raw_dice_values[0], "selected_block_die_outcome": None, "selected_block_die_source": "single_die", } outcome_values = _parse_dice_outcomes(root) if len(outcome_values) == len(raw_dice_values): selected_outcome = max(outcome_values) if attacker_choose else min(outcome_values) selected_index = outcome_values.index(selected_outcome) return { "selected_block_die_index": selected_index, "selected_block_die_value": raw_dice_values[selected_index], "selected_block_die_outcome": selected_outcome, "selected_block_die_source": "dice_outcomes_attacker_choice", } return { "selected_block_die_index": None, "selected_block_die_value": None, "selected_block_die_outcome": None, "selected_block_die_source": None, } block_dice_values = self._normalize_block_dice_values(dice_values) attacker_chooses, attacker_choice_source = _resolve_attacker_choice( roll_root if isinstance(roll_root, dict) else payload_root, actor_player_id, ) attacker_choice = 1 if attacker_chooses else 0 attacker_is_blitzing = self._infer_block_is_blitzing(payload_root, step_message) if (not attacker_is_blitzing) and actor_player_id is not None and str(actor_player_id) in step_blitz_players: attacker_is_blitzing = True if (not attacker_is_blitzing) and actor_player_id is not None and str(actor_player_id) in step_stood_up_players: attacker_is_blitzing = True if (not attacker_is_blitzing) and forced_blitz_player_id is not None: attacker_is_blitzing = str(actor_player_id) == str(forced_blitz_player_id) target_has_ball = False for possession_key in ("TargetHasBall", "DefenderHasBall", "HasBall", "IsBallCarrier"): if possession_key in payload_root: possession_raw = payload_root.get(possession_key) target_has_ball = str(possession_raw).strip().lower() in ("1", "true", "yes") break if not target_has_ball and defender_player_id is not None and step_ball_carrier_id is not None: target_has_ball = str(defender_player_id) == str(step_ball_carrier_id) block_result, p_success, p_neutral, p_fail = self.calculator.calc_blocks( block_dice_values, actor_skills, defender_skills, isAttackerChoose=attacker_choice, defenderHasBall=target_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) selected_die_meta = _select_block_die_metadata( dice_values, roll_root if isinstance(roll_root, dict) else payload_root, bool(attacker_choice), ) return { 'roll_category': 'block', 'result_value': block_result, 'probability_success': p_success, 'probability_neutral': p_neutral, 'probability_fail': p_fail, 'target_has_ball': target_has_ball, 'attacker_is_blitzing': attacker_is_blitzing, 'armour_holds': None, 'selected_block_die_index': selected_die_meta.get('selected_block_die_index'), 'selected_block_die_value': selected_die_meta.get('selected_block_die_value'), 'selected_block_die_outcome': selected_die_meta.get('selected_block_die_outcome'), 'selected_block_die_source': selected_die_meta.get('selected_block_die_source'), 'block_choice_source': attacker_choice_source, } def _resolve_injury_string_message_roll(self, roll_root, modifier_values, defender_skills): injury_outcome = self._to_int(roll_root.get("Outcome"), default=0) modifier_total = sum(modifier_values) if modifier_values else 0 defender_skill_names = [str(skill).lower() for skill in defender_skills] has_stunty = "stunty" in defender_skill_names has_thick_skull = "thick skull" in defender_skill_names injury_result, p_success, p_neutral, p_fail = self.calculator.calc_injury( injury_outcome, modifierVal=modifier_total, is_stunty=has_stunty, is_thick_skull=has_thick_skull, ) return { 'roll_category': 'injury', 'result_value': injury_result, 'probability_success': p_success, 'probability_neutral': p_neutral, 'probability_fail': p_fail, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': None, } def _build_special_usage_roll_events( self, *, game_turn, step_number, sequence_index, step_result_index, string_message_index, step_name, step_type, actor_player_id, actor_player_name, actor_team_id, defender_player_name, defender_skills, step_special_skill_usages, step_consumed_special_usage_indexes, preferred_usage_index=None, roll_ordinal_in_step, roll_type_ordinal_in_step, ): if len(step_special_skill_usages) == 0: return [], roll_ordinal_in_step, roll_type_ordinal_in_step matched_usage_index = None matched_usage = None actor_key = str(actor_player_id) if actor_player_id is not None else None if isinstance(preferred_usage_index, int): if 0 <= preferred_usage_index < len(step_special_skill_usages): if preferred_usage_index not in step_consumed_special_usage_indexes: preferred_usage = step_special_skill_usages[preferred_usage_index] if isinstance(preferred_usage, dict): matched_usage_index = preferred_usage_index matched_usage = preferred_usage if matched_usage is None: for usage_index, usage in enumerate(step_special_skill_usages): if usage_index in step_consumed_special_usage_indexes: continue usage_target = usage.get("target_id") if actor_key is not None and usage_target is not None and str(usage_target) == actor_key: matched_usage_index = usage_index matched_usage = usage break if matched_usage is None: for usage_index, usage in enumerate(step_special_skill_usages): if usage_index in step_consumed_special_usage_indexes: continue matched_usage_index = usage_index matched_usage = usage break if not isinstance(matched_usage, dict): return [], roll_ordinal_in_step, roll_type_ordinal_in_step step_consumed_special_usage_indexes.add(matched_usage_index) usage_label = str(matched_usage.get("skill_label") or "SkillAction") usage_player_id = matched_usage.get("player_id") usage_target_id = matched_usage.get("target_id") if usage_target_id is None: usage_target_id = actor_player_id if usage_label == "Chainsaw" and actor_key is not None: usage_target = matched_usage.get("target_id") if usage_target is None or str(usage_target) != actor_key: self._emit_processing_warning( ( f"Chainsaw usage on step {step_number if step_number is not None else '?'} " f"did not explicitly target armour victim {actor_key}; linked by fallback." ), key=("chainsaw-fallback-link", step_number, actor_key), ) if usage_label == "Chainsaw" and usage_player_id is None: self._emit_processing_warning( ( f"Chainsaw usage on step {step_number if step_number is not None else '?'} " "is missing source player id in payload." ), key=("chainsaw-missing-source", step_number), ) usage_player_lookup = self._safe_player_lookup(usage_player_id) usage_target_lookup = self._safe_player_lookup(usage_target_id) usage_team_id = usage_player_lookup.get("team_id", actor_team_id) usage_roll_type_key = str(usage_label).lower() usage_player_skills = usage_player_lookup.get("skills", []) if not isinstance(usage_player_skills, list): usage_player_skills = [] usage_target_skills = usage_target_lookup.get("skills", defender_skills) if not isinstance(usage_target_skills, list): usage_target_skills = defender_skills if isinstance(defender_skills, list) else [] roll_ordinal_in_step += 1 roll_type_ordinal_in_step[usage_roll_type_key] += 1 synthetic_event = { 'game_turn': game_turn, 'player_id': usage_player_id, 'player_name': usage_player_lookup.get("name", actor_player_name), 'team_id': usage_team_id, 'dice_roller': self._resolve_dice_roller(usage_team_id, "action"), 'player_skills': usage_player_skills, 'target_player_id': usage_target_id, 'target_player_name': usage_target_lookup.get("name", defender_player_name), 'target_player_skills': usage_target_skills, 'target_has_ball': False, 'attacker_is_blitzing': False, 'is_synthetic_skill_action': True, 'synthetic_action_label': usage_label, 'step_name': step_name, 'step_type': step_type, 'sequence_index': sequence_index, 'step_result_index': step_result_index, 'string_message_index': string_message_index, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step[usage_roll_type_key], 'result_name': "ResultSkillUsage", 'payload_type': "ResultSkillUsage", 'roll_category': "action", 'dice_values': [], 'dice_total': 0, 'difficulty': None, 'roll_type': usage_roll_type_key, 'outcome': None, 'modifier_values': [], 'result_value': 1, 'result_classification': "success", 'report_result_classification': "success", 'armour_holds': None, 'probability_success': 1.0, 'probability_neutral': 0.0, 'probability_fail': 0.0, 'raw': matched_usage.get("raw") if isinstance(matched_usage, dict) else {}, } if usage_label == "Stab": synthetic_event['is_stab_action'] = True synthetic_event['is_marker_only_action'] = True synthetic_event['result_value'] = None synthetic_event['result_classification'] = "unknown" synthetic_event['report_result_classification'] = "unknown" synthetic_event['probability_success'] = None synthetic_event['probability_neutral'] = None synthetic_event['probability_fail'] = None synthetic_event['exclude_from_surprise'] = True if usage_label == "Chainsaw": synthetic_event['is_chainsaw_action'] = True synthetic_event['is_marker_only_action'] = True synthetic_event['result_value'] = None synthetic_event['result_classification'] = "unknown" synthetic_event['report_result_classification'] = "unknown" synthetic_event['probability_success'] = None synthetic_event['probability_neutral'] = None synthetic_event['probability_fail'] = None synthetic_event['exclude_from_surprise'] = True return [synthetic_event], roll_ordinal_in_step, roll_type_ordinal_in_step def _build_foul_action_roll_event( self, *, game_turn, sequence_index, step_result_index, string_message_index, step_name, step_type, actor_player_name, actor_team_id, defender_player_name, defender_skills, step_foul_player_id, step_foul_target_id, roll_ordinal_in_step, roll_type_ordinal_in_step, ): foul_player_lookup = self._safe_player_lookup(step_foul_player_id) foul_target_lookup = self._safe_player_lookup(step_foul_target_id) foul_team_id = foul_player_lookup.get("team_id", actor_team_id) foul_player_skills = foul_player_lookup.get("skills", []) if not isinstance(foul_player_skills, list): foul_player_skills = [] foul_target_skills = foul_target_lookup.get("skills", defender_skills) if not isinstance(foul_target_skills, list): foul_target_skills = defender_skills if isinstance(defender_skills, list) else [] roll_ordinal_in_step += 1 roll_type_ordinal_in_step["foul"] += 1 return { 'game_turn': game_turn, 'player_id': step_foul_player_id, 'player_name': foul_player_lookup.get("name", actor_player_name), 'team_id': foul_team_id, 'dice_roller': self._resolve_dice_roller(foul_team_id, "action"), 'player_skills': foul_player_skills, 'target_player_id': step_foul_target_id, 'target_player_name': foul_target_lookup.get("name", defender_player_name), 'target_player_skills': foul_target_skills, 'target_has_ball': False, 'attacker_is_blitzing': False, 'is_synthetic_skill_action': True, 'synthetic_action_label': 'Foul', 'is_foul_action': True, 'step_name': step_name, 'step_type': step_type, 'sequence_index': sequence_index, 'step_result_index': step_result_index, 'string_message_index': string_message_index, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step["foul"], 'result_name': 'ResultUseAction', 'payload_type': 'ResultUseAction', 'roll_category': 'action', 'dice_values': [], 'dice_total': 0, 'difficulty': None, 'roll_type': 'foul', 'outcome': None, 'modifier_values': [], 'result_value': 1, 'result_classification': 'success', 'report_result_classification': 'success', 'armour_holds': None, 'probability_success': 1.0, 'probability_neutral': 0.0, 'probability_fail': 0.0, 'exclude_from_surprise': True, 'raw': {'synthetic': 'foul_action', 'action': '6'}, }, roll_ordinal_in_step, roll_type_ordinal_in_step def _resolve_difficulty_string_message_roll( self, *, dice_values, roll_root, game_turn, step_number, sequence_index, step_result_index, string_message_index, step_name, step_type, actor_player_id, actor_player_name, actor_team_id, defender_player_name, defender_skills, step_special_skill_usages, step_consumed_special_usage_indexes, step_foul_player_id, step_foul_target_id, step_foul_emitted, roll_ordinal_in_step, roll_type_ordinal_in_step, ): difficulty_int = self._to_int(roll_root.get("Difficulty"), default=0) roll_type_str = str(roll_root.get("RollType")) if roll_root.get("RollType") is not None else "" if roll_type_str == "10": synthetic_events, roll_ordinal_in_step, roll_type_ordinal_in_step = self._build_special_usage_roll_events( game_turn=game_turn, step_number=step_number, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, step_name=step_name, step_type=step_type, actor_player_id=actor_player_id, actor_player_name=actor_player_name, actor_team_id=actor_team_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_special_skill_usages=step_special_skill_usages, step_consumed_special_usage_indexes=step_consumed_special_usage_indexes, preferred_usage_index=None, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) if step_foul_player_id is not None and not step_foul_emitted: step_foul_emitted = True foul_event, roll_ordinal_in_step, roll_type_ordinal_in_step = self._build_foul_action_roll_event( game_turn=game_turn, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, step_name=step_name, step_type=step_type, actor_player_name=actor_player_name, actor_team_id=actor_team_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_foul_player_id=step_foul_player_id, step_foul_target_id=step_foul_target_id, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) synthetic_events.append(foul_event) is_success, p_success, p_fail = self.calculator.calc_armour(dice_values, difficulty_int) return { 'roll_category': 'armour', 'result_value': 1 if is_success else -1, 'probability_success': p_fail, 'probability_neutral': 0.0, 'probability_fail': p_success, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': bool(is_success), 'synthetic_events': synthetic_events, 'step_foul_emitted': step_foul_emitted, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step, } # Chainsaw kickback uses RollType=67 (2+ D6). Emit pending Chainsaw marker # before the kickback row so display ordering matches replay semantics. if roll_type_str == "67": synthetic_events = [] chainsaw_usage_idx = next( ( idx for idx, u in enumerate(step_special_skill_usages) if str(u.get("skill_label") or "") == "Chainsaw" and idx not in step_consumed_special_usage_indexes ), None, ) if chainsaw_usage_idx is not None: chainsaw_synthetic_events, roll_ordinal_in_step, roll_type_ordinal_in_step = self._build_special_usage_roll_events( game_turn=game_turn, step_number=step_number, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, step_name=step_name, step_type=step_type, actor_player_id=actor_player_id, actor_player_name=actor_player_name, actor_team_id=actor_team_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_special_skill_usages=step_special_skill_usages, step_consumed_special_usage_indexes=step_consumed_special_usage_indexes, preferred_usage_index=chainsaw_usage_idx, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) synthetic_events.extend(chainsaw_synthetic_events) if step_foul_player_id is not None and not step_foul_emitted: step_foul_emitted = True foul_event, roll_ordinal_in_step, roll_type_ordinal_in_step = self._build_foul_action_roll_event( game_turn=game_turn, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, step_name=step_name, step_type=step_type, actor_player_name=actor_player_name, actor_team_id=actor_team_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_foul_player_id=step_foul_player_id, step_foul_target_id=step_foul_target_id, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) is_success, p_success, p_fail = self.calculator.calc_action(dice_values, difficulty_int) return { 'roll_category': 'action', 'result_value': 1 if is_success else -1, 'probability_success': p_success, 'probability_neutral': 0.0, 'probability_fail': p_fail, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': None, 'synthetic_events': synthetic_events + [foul_event], 'step_foul_emitted': step_foul_emitted, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step, } is_success, p_success, p_fail = self.calculator.calc_action(dice_values, difficulty_int) return { 'roll_category': 'action', 'result_value': 1 if is_success else -1, 'probability_success': p_success, 'probability_neutral': 0.0, 'probability_fail': p_fail, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': None, 'synthetic_events': synthetic_events, 'step_foul_emitted': step_foul_emitted, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step, } is_success, p_success, p_fail = self.calculator.calc_action(dice_values, difficulty_int) return { 'roll_category': 'action', 'result_value': 1 if is_success else -1, 'probability_success': p_success, 'probability_neutral': 0.0, 'probability_fail': p_fail, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': None, 'synthetic_events': [], 'step_foul_emitted': step_foul_emitted, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step, } def _build_roll_event_from_string_message( self, *, game_turn, actor_player_id, actor_player_name, actor_team_id, actor_skills, defender_player_id, defender_player_name, defender_skills, step_name, step_type, sequence_index, step_result_index, string_message_index, roll_ordinal_in_step, roll_type_ordinal_in_step, result_name, payload_type, roll_category, dice_values, roll_root, modifier_values, result_value, result_classification, report_result_classification, armour_holds, probability_success, probability_neutral, probability_fail, target_has_ball, attacker_is_blitzing, action_type=None, ): normalized_actor_skills = actor_skills if isinstance(actor_skills, list) else [] normalized_defender_skills = defender_skills if isinstance(defender_skills, list) else [] return { 'game_turn': game_turn, 'player_id': actor_player_id, 'player_name': actor_player_name, 'team_id': actor_team_id, 'dice_roller': self._resolve_dice_roller(actor_team_id, roll_category), 'player_skills': normalized_actor_skills, 'target_player_id': defender_player_id, 'target_player_name': defender_player_name, 'target_player_skills': normalized_defender_skills, 'target_has_ball': target_has_ball, 'attacker_is_blitzing': attacker_is_blitzing, 'step_name': step_name, 'step_type': step_type, 'sequence_index': sequence_index, 'step_result_index': step_result_index, 'string_message_index': string_message_index, 'roll_ordinal_in_step': roll_ordinal_in_step, 'roll_type_ordinal_in_step': roll_type_ordinal_in_step, 'result_name': result_name, 'payload_type': payload_type, 'roll_category': roll_category, 'dice_values': dice_values, 'dice_total': sum(dice_values), 'difficulty': roll_root.get('Difficulty'), 'roll_type': roll_root.get('RollType'), 'outcome': roll_root.get('Outcome'), 'modifier_values': modifier_values, 'result_value': result_value, 'result_classification': result_classification, 'report_result_classification': report_result_classification, 'armour_holds': armour_holds, 'probability_success': probability_success, 'probability_neutral': probability_neutral, 'probability_fail': probability_fail, 'raw': roll_root, **({"action_type": action_type} if action_type is not None else {}), } def _extract_roll_events_from_string_messages( self, parsed_messages, *, game_turn, step_number, sequence_index, step_result_index, step_name, step_type, step_message, actor_player_id, actor_player_name, actor_team_id, actor_skills, defender_player_id, defender_player_name, defender_skills, forced_blitz_player_id, step_ball_carrier_id, is_kickoff_step, step_blitz_players, step_stood_up_players, step_special_skill_usages, step_consumed_special_usage_indexes, step_foul_player_id, step_foul_target_id, step_foul_emitted, roll_ordinal_in_step, roll_type_ordinal_in_step, ): """Extract roll events from parsed StepResult string messages, preserving replay order.""" roll_events = [] for parsed in parsed_messages: if not isinstance(parsed, dict): continue string_message_index = parsed.get("string_message_index") result = parsed.get("result") result_name = parsed.get("result_name") payload_type = parsed.get("payload_type") payload_root = parsed.get("payload_root") if not isinstance(payload_root, dict): continue roll_root = self._resolve_string_message_roll_root(payload_root) if roll_root is None: continue dice_values = self._extract_dice_values(roll_root.get("Dice")) if len(dice_values) == 0: continue if payload_type == "EventKickOffTable": continue modifier_values = self._extract_modifier_values(roll_root.get("Modifiers")) roll_family = self._classify_string_message_roll(payload_type, result_name, step_name, roll_root) roll_resolution = { 'roll_category': 'other', 'probability_success': None, 'probability_neutral': None, 'probability_fail': None, 'result_value': 0, 'target_has_ball': False, 'attacker_is_blitzing': False, 'armour_holds': None, } if roll_family == "block": roll_resolution = self._resolve_block_string_message_roll( dice_values=dice_values, payload_root=payload_root, roll_root=roll_root, step_message=step_message, actor_player_id=actor_player_id, actor_skills=actor_skills, defender_player_id=defender_player_id, defender_skills=defender_skills, forced_blitz_player_id=forced_blitz_player_id, step_ball_carrier_id=step_ball_carrier_id, step_blitz_players=step_blitz_players, step_stood_up_players=step_stood_up_players, ) elif roll_family in ("injury", "ko_recovery_injury"): roll_resolution = self._resolve_injury_string_message_roll( roll_root, modifier_values, defender_skills, ) elif roll_family == "casualty": die_val = dice_values[0] if dice_values else 0 cas_result, cas_p_success, cas_p_neutral, cas_p_fail = self.calculator.calc_casualty(die_val) roll_resolution['roll_category'] = 'casualty' roll_resolution['result_value'] = cas_result roll_resolution['probability_success'] = cas_p_success roll_resolution['probability_neutral'] = cas_p_neutral roll_resolution['probability_fail'] = cas_p_fail elif roll_family in ("armour", "action"): roll_resolution = self._resolve_difficulty_string_message_roll( dice_values=dice_values, roll_root=roll_root, game_turn=game_turn, step_number=step_number, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, step_name=step_name, step_type=step_type, actor_player_id=actor_player_id, actor_player_name=actor_player_name, actor_team_id=actor_team_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_special_skill_usages=step_special_skill_usages, step_consumed_special_usage_indexes=step_consumed_special_usage_indexes, step_foul_player_id=step_foul_player_id, step_foul_target_id=step_foul_target_id, step_foul_emitted=step_foul_emitted, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step, ) roll_events.extend(roll_resolution.get('synthetic_events', [])) step_foul_emitted = roll_resolution.get('step_foul_emitted', step_foul_emitted) roll_ordinal_in_step = roll_resolution.get('roll_ordinal_in_step', roll_ordinal_in_step) roll_type_ordinal_in_step = roll_resolution.get('roll_type_ordinal_in_step', roll_type_ordinal_in_step) roll_category = roll_resolution['roll_category'] probability_success = roll_resolution['probability_success'] probability_neutral = roll_resolution['probability_neutral'] probability_fail = roll_resolution['probability_fail'] result_value = roll_resolution['result_value'] target_has_ball = roll_resolution['target_has_ball'] attacker_is_blitzing = roll_resolution['attacker_is_blitzing'] armour_holds = roll_resolution['armour_holds'] result_classification = "neutral" report_result_classification = "neutral" if result_value == 1: result_classification = "success" elif result_value == -1: result_classification = "fail" roll_type_value = roll_root.get("RollType") roll_type_key = str(roll_type_value) if roll_type_value is not None else "-" roll_ordinal_in_step += 1 roll_type_ordinal_in_step[roll_type_key] += 1 report_result_classification = result_classification if roll_category == "armour": if result_classification == "success": report_result_classification = "fail" elif result_classification == "fail": report_result_classification = "success" event_actor_player_id = actor_player_id event_actor_player_name = actor_player_name event_actor_team_id = actor_team_id event_actor_skills = actor_skills if step_name == "TopLevelEventRoll" and payload_type == "ResultRoll": roll_player_id = roll_root.get("PlayerId") if roll_player_id is not None: lookup = self._safe_player_lookup(roll_player_id) event_actor_player_id = roll_player_id event_actor_player_name = lookup.get("name", actor_player_name) event_actor_team_id = lookup.get("team_id", actor_team_id) event_actor_skills = lookup.get("skills", actor_skills) elif roll_root.get("TeamId") is not None: event_actor_team_id = roll_root.get("TeamId") roll_event = self._build_roll_event_from_string_message( game_turn=game_turn, actor_player_id=event_actor_player_id, actor_player_name=event_actor_player_name, actor_team_id=event_actor_team_id, actor_skills=event_actor_skills, defender_player_id=defender_player_id, defender_player_name=defender_player_name, defender_skills=defender_skills, step_name=step_name, step_type=step_type, sequence_index=sequence_index, step_result_index=step_result_index, string_message_index=string_message_index, roll_ordinal_in_step=roll_ordinal_in_step, roll_type_ordinal_in_step=roll_type_ordinal_in_step[roll_type_key], result_name=result_name, payload_type=payload_type, roll_category=roll_category, dice_values=dice_values, roll_root=roll_root, modifier_values=modifier_values, result_value=result_value, result_classification=result_classification, report_result_classification=report_result_classification, armour_holds=armour_holds, probability_success=probability_success, probability_neutral=probability_neutral, probability_fail=probability_fail, target_has_ball=target_has_ball, attacker_is_blitzing=attacker_is_blitzing, action_type="APO" if payload_type == "QuestionApothecaryCasualtyChoice" else None, ) # Armour rows emitted right after an attack should inherit attacker and # target identity from that attack, rather than step actor fallback. if roll_category == "armour": current_victim_id = roll_event.get("player_id") source_attack_event = self._find_nearest_armour_source_event( roll_events, len(roll_events), current_victim_id, ) if source_attack_event is not None: roll_event["player_id"] = source_attack_event.get("player_id") roll_event["player_name"] = source_attack_event.get("player_name") roll_event["team_id"] = source_attack_event.get("team_id") roll_event["dice_roller"] = source_attack_event.get("dice_roller") roll_event["player_skills"] = source_attack_event.get("player_skills") or [] roll_event["target_player_id"] = source_attack_event.get("target_player_id") roll_event["target_player_name"] = source_attack_event.get("target_player_name") roll_event["target_player_skills"] = source_attack_event.get("target_player_skills") or [] roll_event["armour_source_category"] = str(source_attack_event.get("roll_category") or "").lower() roll_event["armour_source_action_label"] = str( source_attack_event.get("synthetic_action_label") or source_attack_event.get("display_action_type") or source_attack_event.get("action_type") or "" ).strip() if roll_category == "block": for key in ( "selected_block_die_index", "selected_block_die_value", "selected_block_die_outcome", "selected_block_die_source", "block_choice_source", ): if key in roll_resolution: roll_event[key] = roll_resolution.get(key) is_trivial_kickoff_followup = ( payload_type == "ResultRoll" and result_name == "ResultRoll" and roll_category == "action" and str(roll_root.get('Difficulty')) == "0" and str(roll_root.get('RollType')) in ("25", "26") and step_name in ("PlayerStep", "BallStep") and str(step_type) in ("3", "10") and probability_success == 1.0 and probability_fail == 0.0 ) if is_trivial_kickoff_followup: roll_event['exclude_from_detailed_report'] = True roll_event['exclude_from_surprise'] = True # KO recovery is modeled explicitly from EventRoll as synthetic # KO-RECOV rows. Keep those as the single source of truth in # reports by suppressing duplicate top-level ResultRoll rows. is_duplicate_ko_recovery_resultroll = ( step_name == "TopLevelEventRoll" and payload_type == "ResultRoll" and result_name == "ResultRoll" and str(roll_root.get("RollType")) == "13" ) if is_duplicate_ko_recovery_resultroll: roll_event['exclude_from_detailed_report'] = True roll_event['exclude_from_surprise'] = True # Ejection (Argue/Bribe) rolls are modeled explicitly from # QuestionBribeUsage + ResultRoll in synthetic extraction. # Suppress the raw top-level ResultRoll rows to avoid duplicate, # less-informative action rows in detailed report output. is_duplicate_ejection_resultroll = ( step_name == "TopLevelEventRoll" and payload_type == "ResultRoll" and result_name == "ResultRoll" and str(roll_root.get("RollType")) in ("30", "21", "29") ) if is_duplicate_ejection_resultroll: roll_event['exclude_from_detailed_report'] = True roll_event['exclude_from_surprise'] = True roll_events.append(roll_event) return roll_events, roll_ordinal_in_step, step_foul_emitted def _infer_blitz_players_in_step(self, step_data): """Infer player ids that are blitzing in this replay step. Heuristic used: - same player has a movement step (StepType=1) followed by a block step (StepType=6) in the same replay step. """ blitz_players = set() seen_move_by_player = set() exec_sequences = step_data.get("EventExecuteSequence") if not isinstance(exec_sequences, list): return blitz_players for seq_item in exec_sequences: if not isinstance(seq_item, dict): continue sequence = seq_item.get("Sequence") if not isinstance(sequence, dict): continue step_results = sequence.get("StepResult") if not isinstance(step_results, list): continue for step_result in step_results: if not isinstance(step_result, dict): continue step_node = step_result.get("Step") if not isinstance(step_node, dict) or not isinstance(step_node.get("MessageData"), str): continue step_message = self._parse_message_data_xml(step_node.get("MessageData")) if not isinstance(step_message, dict) or len(step_message) == 0: continue root_key = next(iter(step_message.keys())) root = step_message.get(root_key) if not isinstance(root, dict): continue player_id = root.get("PlayerId") step_type = str(root.get("StepType")) if root.get("StepType") is not None else "" if player_id is None: continue if step_type == "1": seen_move_by_player.add(str(player_id)) elif step_type == "6" and str(player_id) in seen_move_by_player: blitz_players.add(str(player_id)) return blitz_players def _build_ejection_roll_events(self, replay_steps): """Scan all replay steps for secret-weapon / foul ejection sequences. Trigger signal -------------- A top-level ``QuestionBribeUsage`` key in a step marks the start of an ejection decision. The fields we care about: GamerId – which team owns the threatened player PlayerId – the player at risk Reason – "1" = secret weapon (end-of-half), "2" = foul ArgueAvailable – "1" if the team coach can argue Resolution signal ----------------- The **same** step (e.g. an end-of-half step where multiple things happen at once) or the **immediately following** step carries the outcome as one or more top-level ``ResultRoll`` entries plus optional ``ResultPlayerSentOff``. Known RollTypes observed in the data ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - "30" (confirmed): Argue the Call - "21" (confirmed): Bribe - "29" (legacy/unknown): recognised ejection roll type (kept distinct, not mapped to Argue) A missing ``ResultPlayerSentOff`` with a present ``ResultRoll`` means the roll succeeded and the player was reprieved. If there is no ``ResultRoll`` at all after a ``QuestionBribeUsage`` the player was ejected automatically (coach was sent off – no option available). Each roll is emitted as a separate event with its own ``roll_category``, probability, and outcome so it integrates naturally with the existing roll reporting pipeline. """ return build_ejection_roll_events(self, replay_steps) def _build_officious_ref_roll_events(self, replay_steps): """Extract the follow-up send-off die from EventOfficiousRef. Event 11 has two distinct random stages: 1. Kickoff event selection + fan-factor roll-off (handled by kickoff reporting) 2. Independent D6 send-off check for the chosen player (handled here) This function emits stage (2) as a separate roll event so detailed roll reporting can show it alongside other independent rolls without folding it into the kickoff pass/fail classification. """ return build_officious_ref_roll_events(self, replay_steps) def _reset_processing_state(self): """Reset per-run processor state used during replay parsing.""" pass def _build_roll_event_category_index(self): """Build a per-category lookup over game_state.roll_events. Stores ``self._events_by_cat``: a dict mapping roll_category string → list of event dicts. The lists hold references (not copies), so in-place mutations to individual events are immediately visible. Must be rebuilt after any operation that replaces game_state.roll_events entirely (e.g. the _drop_* helpers in skill_transition.py). """ index: dict = {} for event in (getattr(self.game_state, "roll_events", None) or ()): if isinstance(event, dict): cat = event.get("roll_category") or "" bucket = index.get(cat) if bucket is None: index[cat] = [event] else: bucket.append(event) self._events_by_cat = index def _prepare_replay_steps(self, jsonData): """Extract and normalize replay step payloads from replay JSON.""" replay = jsonData.get("Replay", {}) replay_steps = replay.get("ReplayStep", []) if isinstance(replay_steps, dict): replay_steps = [replay_steps] self.normalize_replay_step_shapes(replay_steps) self.game_state.raw_replay_steps = replay_steps return replay_steps def _append_post_step_synthetic_rolls(self, replay_steps): """Append synthetic roll events that are derived after primary step parsing.""" synthetic_ko_rolls = self._build_ko_recovery_roll_events(replay_steps) for synthetic_roll in synthetic_ko_rolls: self.game_state.add_roll_event(synthetic_roll) ejection_rolls = self._build_ejection_roll_events(replay_steps) for ejection_roll in ejection_rolls: self.game_state.add_roll_event(ejection_roll) officious_ref_roll_events = self._build_officious_ref_roll_events(replay_steps) for officious_ref_roll in officious_ref_roll_events: self.game_state.add_roll_event(officious_ref_roll) synthetic_touchdown_rolls = self._build_touchdown_roll_events() for synthetic_touchdown_roll in synthetic_touchdown_rolls: self.game_state.add_roll_event(synthetic_touchdown_roll) self._assign_missing_roll_ordinals() def _assign_missing_roll_ordinals(self): """Assign in-step roll ordinals for events that do not already carry them.""" events = getattr(self.game_state, "roll_events", None) if not isinstance(events, list) or not events: return max_ordinal_by_step = defaultdict(int) max_type_ordinal_by_step = defaultdict(lambda: defaultdict(int)) for event in events: if not isinstance(event, dict): continue step_number = event.get("step_number") if step_number is None: continue try: step_key = int(step_number) except (TypeError, ValueError): continue existing_ordinal = event.get("roll_ordinal_in_step") if isinstance(existing_ordinal, int): if existing_ordinal > max_ordinal_by_step[step_key]: max_ordinal_by_step[step_key] = existing_ordinal roll_type_key = str(event.get("roll_type") or "") existing_type_ordinal = event.get("roll_type_ordinal_in_step") if isinstance(existing_type_ordinal, int) and roll_type_key: if existing_type_ordinal > max_type_ordinal_by_step[step_key][roll_type_key]: max_type_ordinal_by_step[step_key][roll_type_key] = existing_type_ordinal for event in events: if not isinstance(event, dict): continue if isinstance(event.get("roll_ordinal_in_step"), int): continue step_number = event.get("step_number") if step_number is None: continue try: step_key = int(step_number) except (TypeError, ValueError): continue max_ordinal_by_step[step_key] += 1 event["roll_ordinal_in_step"] = max_ordinal_by_step[step_key] roll_type_key = str(event.get("roll_type") or "") if roll_type_key: max_type_ordinal_by_step[step_key][roll_type_key] += 1 event["roll_type_ordinal_in_step"] = max_type_ordinal_by_step[step_key][roll_type_key] def _is_failed_pro_action_roll(self, event): """Return True for action rolls that represent a failed Pro test.""" if not isinstance(event, dict): return False if str(event.get("roll_category") or "") != "action": return False if str(event.get("roll_type") or "") != PRO_RULE.get("roll_type", ""): return False if str(event.get("result_classification") or "") != "fail": return False return self._event_player_has_skill_id(event, PRO_RULE.get("skill_id", "50")) def _is_successful_pro_action_roll(self, event): """Return True for action rolls that represent a successful Pro test.""" if not isinstance(event, dict): return False if str(event.get("roll_category") or "") != "action": return False if str(event.get("roll_type") or "") != PRO_RULE.get("roll_type", ""): return False if str(event.get("result_classification") or "") != "success": return False return self._event_player_has_skill_id(event, PRO_RULE.get("skill_id", "50")) def _is_any_pro_action_roll(self, event): """Return True for any Pro test roll (success or fail).""" return self._is_successful_pro_action_roll(event) or self._is_failed_pro_action_roll(event) def _extract_pro_reroll_index_from_block_event(self, event): """Return rerolled die index from explicit block reroll payload metadata.""" if not isinstance(event, dict): return None if str(event.get("roll_category") or "") != "block": return None raw = event.get("raw") if isinstance(event.get("raw"), dict) else {} candidates = [] direct_item = raw.get("RolledDiceIndexesItem") if direct_item is not None: candidates.append(direct_item) rolled_indexes = raw.get("RolledDiceIndexes") if isinstance(rolled_indexes, dict): nested_item = rolled_indexes.get("RolledDiceIndexesItem") if nested_item is None: nested_item = rolled_indexes.get("Item") if nested_item is not None: candidates.append(nested_item) for candidate in candidates: values = candidate if isinstance(candidate, list) else [candidate] for value in values: parsed = self._to_int(value) if parsed is not None: return parsed return None def _extract_rolled_dice_indexes_from_block_event(self, event): """Return all rerolled dice indexes declared in block payload metadata.""" if not isinstance(event, dict): return [] if str(event.get("roll_category") or "") != "block": return [] raw = event.get("raw") if isinstance(event.get("raw"), dict) else {} candidates = [] direct_item = raw.get("RolledDiceIndexesItem") if direct_item is not None: candidates.append(direct_item) rolled_indexes = raw.get("RolledDiceIndexes") if isinstance(rolled_indexes, dict): nested_item = rolled_indexes.get("RolledDiceIndexesItem") if nested_item is None: nested_item = rolled_indexes.get("Item") if nested_item is not None: candidates.append(nested_item) parsed_values = [] for candidate in candidates: values = candidate if isinstance(candidate, list) else [candidate] for value in values: parsed = self._to_int(value) if parsed is not None: parsed_values.append(parsed) # Preserve order while deduplicating. deduped = [] seen = set() for value in parsed_values: if value in seen: continue deduped.append(value) seen.add(value) return deduped def _extract_brawler_reroll_indexes_from_block_event(self, event): """Return Brawler-eligible die indexes from block payload metadata.""" if not isinstance(event, dict): return [] if str(event.get("roll_category") or "") != "block": return [] raw = event.get("raw") if isinstance(event.get("raw"), dict) else {} brawler_node = raw.get("BrawlerDiceIndexes") if not isinstance(brawler_node, dict): return [] candidate = brawler_node.get("BrawlerDiceIndexesItem") if candidate is None: candidate = brawler_node.get("Item") if candidate is None: return [] values = candidate if isinstance(candidate, list) else [candidate] indexes = [] for value in values: parsed = self._to_int(value) if parsed is not None: indexes.append(parsed) return indexes def _annotate_pro_not_tested(self): """Annotate Pro decision status for blocks by players with Pro.""" return self.skill_transition_backend.annotate_pro_not_tested() def _annotate_brawler_usage(self): """Annotate whether Brawler was offered/used on eligible block events.""" return self.skill_transition_backend.annotate_brawler_usage() def _annotate_pro_reroll_indexes(self): """Annotate successful Pro action rows with explicit rerolled die index.""" return self.skill_transition_backend.annotate_pro_reroll_indexes() def _get_attacker_choice_from_block(self, block_event): """Extract AttackerChoice from block event raw data (defaults to True/1).""" if not isinstance(block_event, dict): return True raw = block_event.get("raw") if isinstance(block_event.get("raw"), dict) else {} return str(raw.get("AttackerChoice", "1")) == "1" def _append_event_note(self, event, note_text): """Append a short note to event['notes'] without duplicates.""" if not isinstance(event, dict): return note = str(note_text or "").strip() if note == "": return existing = str(event.get("notes") or "").strip() if existing in ("", "-"): event["notes"] = note return parts = [part.strip() for part in existing.split(";") if part.strip()] if note in parts: return parts.append(note) event["notes"] = "; ".join(parts) def _calculate_block_pro_block_probs(self, block1, block2, pro_reroll_index): """Calculate and assign probabilities for block2 in a block-pro-block sequence. Logic: - Block1 outcome determined using non-pro-reroll dice - Block2 probabilities calculated using only the single pro-reroll die - Final probabilities combine these based on block1's outcome If block1 is fail: P(final = success/neutral/fail) = P(block2 result) If block1 is neutral: P(final = success) = P(block2 = success) P(final = neutral) = P(block2 = neutral) + P(block2 = fail) P(final = fail) = 0 If block1 is success: P(final = success) = 1.0 P(final = neutral/fail) = 0 """ # Get dice from both blocks block1_dice = block1.get("dice_values") block2_dice = block2.get("dice_values") if not isinstance(block1_dice, list) or not isinstance(block2_dice, list): return # Validate pro_reroll_index pro_reroll_index = self._to_int(pro_reroll_index) if pro_reroll_index is None or pro_reroll_index < 0 or pro_reroll_index >= len(block1_dice): return if pro_reroll_index >= len(block2_dice): return # Map replay block-face encoding (0..4) to calculator faces (1..6) # before evaluating synthetic block outcomes. block1_dice_normalized = self._normalize_block_dice_values(block1_dice) block2_dice_normalized = self._normalize_block_dice_values(block2_dice) if len(block1_dice_normalized) != len(block1_dice): return if len(block2_dice_normalized) != len(block2_dice): return # Get common parameters from block1 raw_attacker_skills = block1.get("player_skills", []) raw_defender_skills = block1.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(block1) defender_has_ball = block1.get("target_has_ball", False) attacker_is_blitzing = block1.get("attacker_is_blitzing", False) probs = self.calculator.calc_rerolled_block_followup_probabilities( block1_dice_normalized, block2_dice_normalized, pro_reroll_index, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: return p_final_success, p_final_neutral, p_final_fail = probs # Overwrite block2 probabilities with calculated conditional probabilities block2["probability_success"] = p_final_success block2["probability_neutral"] = p_final_neutral block2["probability_fail"] = p_final_fail block2["is_pro_adjusted_block"] = True self._append_event_note(block2, "pro update to block") def _calculate_block_brawler_block_probs(self, block1, block2, brawler_reroll_index): """Calculate and assign probabilities for block2 in a block-brawler(block) sequence. Same combination logic as successful Pro block reroll handling: - Evaluate block1 using all non-rerolled dice - Evaluate block2 as one-die block from the rerolled Brawler die - Combine based on block1 status (fail/neutral/success) """ block1_dice = block1.get("dice_values") block2_dice = block2.get("dice_values") if not isinstance(block1_dice, list) or not isinstance(block2_dice, list): return brawler_reroll_index = self._to_int(brawler_reroll_index) if brawler_reroll_index is None or brawler_reroll_index < 0 or brawler_reroll_index >= len(block1_dice): return if brawler_reroll_index >= len(block2_dice): return block1_dice_normalized = self._normalize_block_dice_values(block1_dice) block2_dice_normalized = self._normalize_block_dice_values(block2_dice) if len(block1_dice_normalized) != len(block1_dice): return if len(block2_dice_normalized) != len(block2_dice): return raw_attacker_skills = block1.get("player_skills", []) raw_defender_skills = block1.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(block1) defender_has_ball = block1.get("target_has_ball", False) attacker_is_blitzing = block1.get("attacker_is_blitzing", False) probs = self.calculator.calc_rerolled_block_followup_probabilities( block1_dice_normalized, block2_dice_normalized, brawler_reroll_index, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: return p_final_success, p_final_neutral, p_final_fail = probs block2["probability_success"] = p_final_success block2["probability_neutral"] = p_final_neutral block2["probability_fail"] = p_final_fail block2["is_brawler_adjusted_block"] = True self._append_event_note(block2, "brawler-adjusted block") def _calculate_block_pro_block_probs_ottd(self, block1, block2, pro_reroll_index): """OTTD-mode variant of block-pro-block conditional probability update.""" block1_dice = block1.get("dice_values") block2_dice = block2.get("dice_values") if not isinstance(block1_dice, list) or not isinstance(block2_dice, list): return pro_reroll_index = self._to_int(pro_reroll_index) if pro_reroll_index is None or pro_reroll_index < 0 or pro_reroll_index >= len(block1_dice): return if pro_reroll_index >= len(block2_dice): return block1_dice_normalized = self._normalize_block_dice_values(block1_dice) block2_dice_normalized = self._normalize_block_dice_values(block2_dice) if len(block1_dice_normalized) != len(block1_dice): return if len(block2_dice_normalized) != len(block2_dice): return raw_attacker_skills = block1.get("player_skills", []) raw_defender_skills = block1.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(block1) defender_has_ball = block1.get("target_has_ball", False) attacker_is_blitzing = block1.get("attacker_is_blitzing", False) probs = self.calculator.calc_rerolled_block_followup_probabilities_ottd( block1_dice_normalized, block2_dice_normalized, pro_reroll_index, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: return p_final_success, p_final_neutral, p_final_fail = probs block2["probability_success"] = p_final_success block2["probability_neutral"] = p_final_neutral block2["probability_fail"] = p_final_fail block2["is_pro_adjusted_block"] = True self._append_event_note(block2, "pro update to block") def _calculate_block_brawler_block_probs_ottd(self, block1, block2, brawler_reroll_index): """OTTD-mode variant of block-brawler(block) conditional probability update.""" block1_dice = block1.get("dice_values") block2_dice = block2.get("dice_values") if not isinstance(block1_dice, list) or not isinstance(block2_dice, list): return brawler_reroll_index = self._to_int(brawler_reroll_index) if brawler_reroll_index is None or brawler_reroll_index < 0 or brawler_reroll_index >= len(block1_dice): return if brawler_reroll_index >= len(block2_dice): return block1_dice_normalized = self._normalize_block_dice_values(block1_dice) block2_dice_normalized = self._normalize_block_dice_values(block2_dice) if len(block1_dice_normalized) != len(block1_dice): return if len(block2_dice_normalized) != len(block2_dice): return raw_attacker_skills = block1.get("player_skills", []) raw_defender_skills = block1.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(block1) defender_has_ball = block1.get("target_has_ball", False) attacker_is_blitzing = block1.get("attacker_is_blitzing", False) probs = self.calculator.calc_rerolled_block_followup_probabilities_ottd( block1_dice_normalized, block2_dice_normalized, brawler_reroll_index, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: return p_final_success, p_final_neutral, p_final_fail = probs block2["probability_success"] = p_final_success block2["probability_neutral"] = p_final_neutral block2["probability_fail"] = p_final_fail block2["is_brawler_adjusted_block"] = True self._append_event_note(block2, "brawler-adjusted block") def _get_ottd_mode_context_for_event(self, event): """Return (evaluation, mode) for an event when OTTD mode applies to its turn/team.""" if not isinstance(event, dict): return None turn_num = self._to_int(event.get("game_turn"), default=None) if turn_num is None: return None evaluations = getattr(self.game_state, "ottd_evaluations", {}) or {} evaluation = evaluations.get(turn_num) if evaluation is None: evaluation = evaluations.get(str(turn_num)) if not isinstance(evaluation, dict): return None if not bool(evaluation.get("final_decision")): return None mode = evaluation.get("ottd_mode") if isinstance(evaluation.get("ottd_mode"), dict) else {} if not bool(mode.get("enabled")): return None receiving_team_id = self._normalize_team_id(evaluation.get("receiving_team_id")) if self._normalize_team_id(event.get("team_id")) != receiving_team_id: return None step_number = self._to_int(event.get("step_number"), default=None) mode_on_step = self._to_int(mode.get("mode_on_step"), default=None) turned_off_step = self._to_int(mode.get("turned_off_step"), default=None) if step_number is None or mode_on_step is None: return None if step_number < mode_on_step: return None if turned_off_step is not None and step_number >= turned_off_step: return None return evaluation, mode def _is_ottd_mode_active_for_event(self, event): """Return True when an event falls in an active OTTD mode window.""" return self._get_ottd_mode_context_for_event(event) is not None def _is_ottd_mode_active_for_block_event(self, event): """Return True when a block event falls in an active OTTD mode window.""" if str(event.get("roll_category") or "").lower() != "block": return False return self._is_ottd_mode_active_for_event(event) def _append_ottd_mode_notes_for_event(self, event, mode): """Append OTTD mode notes for active-window rows.""" if not isinstance(event, dict): return self._append_event_note(event, "OTTD mode") def _append_ottd_mode_before_step_boundary_note(self, event): """Annotate the first non-OTTD step (the turned-off step itself) for all timing modes.""" if not isinstance(event, dict): return turn_num = self._to_int(event.get("game_turn"), default=None) if turn_num is None: return evaluations = getattr(self.game_state, "ottd_evaluations", {}) or {} evaluation = evaluations.get(turn_num) if evaluation is None: evaluation = evaluations.get(str(turn_num)) if not isinstance(evaluation, dict) or not bool(evaluation.get("final_decision")): return mode = evaluation.get("ottd_mode") if isinstance(evaluation.get("ottd_mode"), dict) else {} if not bool(mode.get("enabled")) or not bool(mode.get("turned_off")): return boundary_step = self._to_int(mode.get("turned_off_step"), default=None) step_number = self._to_int(event.get("step_number"), default=None) if boundary_step is None or step_number is None or step_number != boundary_step: return receiving_team_id = self._normalize_team_id(evaluation.get("receiving_team_id")) if self._normalize_team_id(event.get("team_id")) != receiving_team_id: return criterion = str(mode.get("turned_off_criterion") or "").strip() reason = str(mode.get("turned_off_reason") or "").strip() if criterion and reason: self._append_event_note(event, f"OTTD mode off from this step ({criterion}: {reason})") elif criterion: self._append_event_note(event, f"OTTD mode off from this step ({criterion})") elif reason: self._append_event_note(event, f"OTTD mode off from this step ({reason})") def _apply_ottd_mode_block_probability_overrides(self): """Apply OTTD-specific block result/probability model during active mode windows.""" events = getattr(self.game_state, "roll_events", None) if not isinstance(events, list) or len(events) == 0: return # Mark all events that occur while OTTD mode is active. # Block rows receive additional probability/classification overrides below. for event in events: if not isinstance(event, dict): continue event["ottd_mode_active"] = False self._append_ottd_mode_before_step_boundary_note(event) mode_context = self._get_ottd_mode_context_for_event(event) if mode_context is None: continue _, mode = mode_context event["ottd_mode_active"] = True self._append_ottd_mode_notes_for_event(event, mode) if str(event.get("roll_category") or "").lower() != "block": continue dice_values = event.get("dice_values") if not isinstance(dice_values, list) or len(dice_values) == 0: continue normalized_dice = self._normalize_block_dice_values(dice_values) if len(normalized_dice) != len(dice_values): continue raw_attacker_skills = event.get("player_skills", []) raw_defender_skills = event.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(event) defender_has_ball = bool(event.get("target_has_ball", False)) attacker_is_blitzing = bool(event.get("attacker_is_blitzing", False)) result, p_success, p_neutral, p_fail = self.calculator.calc_blocks_ottd( normalized_dice, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) event["result_value"] = result if result == 1: event["result_classification"] = "success" event["report_result_classification"] = "success" else: event["result_classification"] = "fail" event["report_result_classification"] = "fail" event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail # Build activation groups for OTTD-active rows only. activation_rows = defaultdict(list) for idx, event in enumerate(events): if not isinstance(event, dict): continue if not bool(event.get("ottd_mode_active")): continue category = str(event.get("roll_category") or "") if category not in ("block", "action", "other"): continue player_name = str(event.get("player_name") or "") if player_name in ("", "Unknown"): continue activation_key = ( str(event.get("game_turn")), str(event.get("team_id")), str(event.get("player_id")), player_name, ) activation_rows[activation_key].append(idx) # OTTD Pro sequence adjustment: block -> Pro(test) -> block. for row_indexes in activation_rows.values(): row_count = len(row_indexes) if row_count < 3: continue for pos, block1_index in enumerate(row_indexes): block1 = events[block1_index] if str(block1.get("roll_category") or "") != "block": continue scan_pos = pos + 1 pro_event_pos = None while scan_pos < row_count: candidate = events[row_indexes[scan_pos]] if self._is_successful_pro_action_roll(candidate): if pro_event_pos is None: pro_event_pos = scan_pos scan_pos += 1 else: break if pro_event_pos is None: continue pro_event = events[row_indexes[pro_event_pos]] pro_reroll_index = pro_event.get("pro_reroll_index") if pro_reroll_index is None: continue block2_pos = None while scan_pos < row_count: candidate = events[row_indexes[scan_pos]] if str(candidate.get("roll_category") or "") == "block": block2_pos = scan_pos break scan_pos += 1 if block2_pos is None: continue block2 = events[row_indexes[block2_pos]] self._calculate_block_pro_block_probs_ottd(block1, block2, pro_reroll_index) if bool(block2.get("ottd_mode_active")): self._append_event_note(block2, "OTTD mode") # OTTD Pro available but not tested. for event in events: if not isinstance(event, dict): continue if not bool(event.get("ottd_mode_active")): continue if str(event.get("roll_category") or "") != "block": continue if not bool(event.get("pro_not_tested")): continue dice_values = event.get("dice_values") if not isinstance(dice_values, list) or len(dice_values) == 0: continue normalized_dice = self._normalize_block_dice_values(dice_values) if len(normalized_dice) != len(dice_values): continue raw_attacker_skills = event.get("player_skills", []) raw_defender_skills = event.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(event) defender_has_ball = bool(event.get("target_has_ball", False)) attacker_is_blitzing = bool(event.get("attacker_is_blitzing", False)) probs = self.calculator.calc_blocks_with_pro_ottd( normalized_dice, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: continue p_success, p_neutral, p_fail = probs event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail event["is_pro_adjusted_block"] = True self._append_event_note(event, "pro-adjusted (skill available; not tested)") self._append_event_note(event, "OTTD mode") # OTTD Brawler sequence and offered-not-used adjustment. active_block_events = [ event for event in events if isinstance(event, dict) and bool(event.get("ottd_mode_active")) and str(event.get("roll_category") or "") == "block" ] source_by_step = {} for event in active_block_events: if str(event.get("brawler_status") or "") != "used": continue step_number = event.get("step_number") if isinstance(step_number, int): source_by_step[step_number] = event for event in active_block_events: if not bool(event.get("is_brawler_reroll_block")): continue source_step = event.get("brawler_source_step") source_block = source_by_step.get(source_step) if not isinstance(source_block, dict): continue brawler_reroll_index = source_block.get("brawler_used_index") self._calculate_block_brawler_block_probs_ottd(source_block, event, brawler_reroll_index) self._append_event_note(event, "OTTD mode") for event in active_block_events: if bool(event.get("is_brawler_reroll_block")): continue brawler_status = str(event.get("brawler_status") or "") if not brawler_status.startswith("offered_not_used"): continue dice_values = event.get("dice_values") if not isinstance(dice_values, list) or len(dice_values) == 0: continue normalized_dice = self._normalize_block_dice_values(dice_values) if len(normalized_dice) != len(dice_values): continue raw_attacker_skills = event.get("player_skills", []) raw_defender_skills = event.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(event) defender_has_ball = bool(event.get("target_has_ball", False)) attacker_is_blitzing = bool(event.get("attacker_is_blitzing", False)) probs = self.calculator.calc_blocks_with_brawler_ottd( normalized_dice, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: continue p_success, p_neutral, p_fail = probs event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail event["is_brawler_adjusted_block"] = True self._append_event_note(event, "brawler-adjusted block (not used)") self._append_event_note(event, "OTTD mode") def _enrich_block_pro_block_probabilities(self): """Enrich conditional probabilities for second block in block-pro-block sequences. Two patterns handled: 1. block-pro(test)-block sequences: Calculate conditional probabilities for second block 2. Blocks by Pro players where Pro was NOT tested: Apply Pro adjustment to first block """ events = getattr(self.game_state, "roll_events", None) if not isinstance(events, list) or len(events) == 0: return # Group events by player activation activation_rows = defaultdict(list) for idx, event in enumerate(events): if not isinstance(event, dict): continue category = str(event.get("roll_category") or "") if category not in ("block", "action", "other"): continue player_name = str(event.get("player_name") or "") if player_name in ("", "Unknown"): continue activation_key = ( str(event.get("game_turn")), str(event.get("team_id")), str(event.get("player_id")), player_name, ) activation_rows[activation_key].append(idx) # Process each activation for block-pro-block patterns for row_indexes in activation_rows.values(): row_count = len(row_indexes) if row_count < 3: # Need at least: block, pro, block continue for pos, block1_index in enumerate(row_indexes): block1 = events[block1_index] if str(block1.get("roll_category") or "") != "block": continue # Scan for successful Pro test(s) following this block scan_pos = pos + 1 pro_event_pos = None while scan_pos < row_count: candidate = events[row_indexes[scan_pos]] if self._is_successful_pro_action_roll(candidate): if pro_event_pos is None: pro_event_pos = scan_pos scan_pos += 1 else: break if pro_event_pos is None: continue # Get pro_reroll_index from Pro event pro_event = events[row_indexes[pro_event_pos]] pro_reroll_index = pro_event.get("pro_reroll_index") if pro_reroll_index is None: continue # Find second block following the Pro test(s) block2_pos = None while scan_pos < row_count: candidate = events[row_indexes[scan_pos]] if str(candidate.get("roll_category") or "") == "block": block2_pos = scan_pos break scan_pos += 1 if block2_pos is None: continue block2 = events[row_indexes[block2_pos]] # Calculate and assign conditional probabilities for block2 self._calculate_block_pro_block_probs(block1, block2, pro_reroll_index) # === PATTERN 2: Pro available but not tested === # Apply Pro adjustment to blocks where player has Pro but didn't test it for event in events: if not isinstance(event, dict): continue if str(event.get("roll_category") or "") != "block": continue # Only process blocks marked as pro_not_tested if not bool(event.get("pro_not_tested")): continue dice_values = event.get("dice_values") if not isinstance(dice_values, list) or len(dice_values) == 0: continue normalized_dice = self._normalize_block_dice_values(dice_values) if len(normalized_dice) != len(dice_values): continue raw_attacker_skills = event.get("player_skills", []) raw_defender_skills = event.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(event) defender_has_ball = bool(event.get("target_has_ball", False)) attacker_is_blitzing = bool(event.get("attacker_is_blitzing", False)) probs = self.calculator.calc_blocks_with_pro( normalized_dice, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=attacker_is_blitzing, ) if probs is None: continue p_success, p_neutral, p_fail = probs event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail event["is_pro_adjusted_block"] = True self._append_event_note(event, "pro-adjusted (skill available; not tested)") def _enrich_block_brawler_block_probabilities(self): """Enrich probabilities for Brawler-eligible blocks where Brawler was not used. Rules: - Only apply when brawler_status indicates offered-but-not-used. - Do not apply on blitz blocks. - Do not apply when both-down is already in success faces (handled by calculator guardrail). - If Brawler was used, leave probabilities unchanged. """ block_events = getattr(self, "_events_by_cat", {}).get("block", ()) if not block_events: return # First handle block->brawler(block) sequences where Brawler is used. source_by_step = {} for event in block_events: if str(event.get("brawler_status") or "") != "used": continue step_number = event.get("step_number") if isinstance(step_number, int): source_by_step[step_number] = event for event in block_events: if not bool(event.get("is_brawler_reroll_block")): continue source_step = event.get("brawler_source_step") source_block = source_by_step.get(source_step) if not isinstance(source_block, dict): continue brawler_reroll_index = source_block.get("brawler_used_index") self._calculate_block_brawler_block_probs(source_block, event, brawler_reroll_index) for event in block_events: # Rerolled block rows are handled by the Brawler-used branch above. if bool(event.get("is_brawler_reroll_block")): continue brawler_status = str(event.get("brawler_status") or "") if not brawler_status.startswith("offered_not_used"): continue # Guardrail: Brawler is not legal on blitz blocks. if bool(event.get("attacker_is_blitzing")): continue dice_values = event.get("dice_values") if not isinstance(dice_values, list) or len(dice_values) == 0: continue normalized_dice = self._normalize_block_dice_values(dice_values) if len(normalized_dice) != len(dice_values): continue raw_attacker_skills = event.get("player_skills", []) raw_defender_skills = event.get("target_player_skills", []) attacker_skills = raw_attacker_skills if isinstance(raw_attacker_skills, list) else [] defender_skills = raw_defender_skills if isinstance(raw_defender_skills, list) else [] attacker_choose = self._get_attacker_choice_from_block(event) defender_has_ball = bool(event.get("target_has_ball", False)) probs = self.calculator.calc_blocks_with_brawler( normalized_dice, attacker_skills, defender_skills, isAttackerChoose=attacker_choose, defenderHasBall=defender_has_ball, attackerIsBlitzing=False, ) if probs is None: continue p_success, p_neutral, p_fail = probs event["probability_success"] = p_success event["probability_neutral"] = p_neutral event["probability_fail"] = p_fail event["is_brawler_adjusted_block"] = True self._append_event_note(event, "brawler-adjusted block (not used)") def _block_repeat_signature(self, event): """Build a stable signature used to match repeated Pro-fail block restatements.""" if not isinstance(event, dict): return None raw = event.get("raw") if isinstance(event.get("raw"), dict) else {} return ( str(event.get("player_id")), str(event.get("target_player_id")), tuple(event.get("dice_values") if isinstance(event.get("dice_values"), list) else ()), str(raw.get("AttackerChoice") if raw.get("AttackerChoice") is not None else "1"), ) def _drop_repeat_block_after_failed_pro(self): """Drop duplicate block restatements after failed Pro test sequences.""" return self.skill_transition_backend.drop_repeat_block_after_failed_pro() def _drop_duplicate_block_after_successful_pro(self): """Drop duplicate block restatements after successful Pro reroll declarations.""" return self.skill_transition_backend.drop_duplicate_block_after_successful_pro() def process_replay(self, jsonData): """Main method to process the entire replay and build game state.""" # Pipeline contract (canonical processing order): # Stage 1: Parse replay payload into canonical step/roll event structures. # Stage 2: Annotate semantic markers on canonical events (no probability writes). # Stage 3: Enrich probability fields on already-annotated canonical events. # Stage 4: Finalize state for downstream report/stats consumers. # # Side-effect boundaries: # - Parse: may create/update game_state steps/events and metadata snapshots. # - Annotate: may add semantic status/notes/flags to existing events. # - Enrich: may write probability_* fields and adjusted-* flags. # - Finalize: no new semantic derivation; expose stable game_state to readers. # Stage 0: Reset per-run state and prepare replay inputs. self._reset_processing_state() # Build the player lookup first self.build_player_lookup(jsonData) replay_steps = self._prepare_replay_steps(jsonData) self._annotate_player_lookup_kickoff_los(replay_steps) # Stage 1: Parse replay steps into canonical step + roll-event streams. run_parse_stage(self, replay_steps) # Stage 1b: Append canonical synthetic rows derived from parsed replay context. run_post_parse_synthetic_stage(self, replay_steps) # Stage 2: Semantic annotation (status/flags/notes only; no generic probability writes). run_annotation_stage(self, replay_steps) # Stage 3: Probability enrichment on the canonical annotated event stream. run_probability_enrichment_stage(self) # Stage 4: Finalized canonical game_state is now available to report/stats readers. # Stage 5: Stamp step-turn counter game_turn onto all roll events. # After this, event["game_turn"] is the step-turn counter value for all # events with a step_number in the gt_map. Downstream report/stats code # no longer needs to apply any post-hoc game_turn correction. run_step_turn_stage(self) # Stage 6: Detect structural OTTD attempt turns. # Must run after step-turn stamping so kickoff event game_turn values # are the final canonical turn numbers. self._detect_ottd_attempt_turns() # Stage 7: Override block probabilities/classification in active OTTD mode windows. self._apply_ottd_mode_block_probability_overrides() def get_game_state(self): """Return the processed game state.""" return self.game_state