from collections import defaultdict from modelling.roll_family import infer_action_attempt_family from core.ruleset import RuleSet class ActionProbabilityEnricher: """Own canonical action probability enrichment.""" def __init__(self, processor, ruleset=None): self.processor = processor rs = ruleset if ruleset is not None else RuleSet() self.pro_rule = rs.pro_rule self.action_skill_reroll_policies = rs.action_skill_reroll_policies self.action_skill_by_label = { str(action_label): str(skill_name) for action_label, skill_name in self.action_skill_reroll_policies } self.pro_status_blocked_category = rs.pro_status_blocked_category self.pro_status_blocked_skill = rs.pro_status_blocked_skill self.pro_status_no_pro = rs.pro_status_no_pro self.pro_status_eligible = rs.pro_status_eligible self.pro_status_used = rs.pro_status_used def _get_event_relevant_skills(self, event): """Return deduplicated relevant skills recorded on an event.""" if not isinstance(event, dict): return [] skills = [] value = event.get("relevant_skills") if isinstance(value, list): for item in value: skill = str(item or "").strip() if skill and skill not in skills: skills.append(skill) legacy_value = str(event.get("relevant_skill") or "").strip() if legacy_value not in ("", "-"): for item in legacy_value.split(","): skill = str(item or "").strip() if skill and skill not in skills: skills.append(skill) return skills def _add_event_relevant_skill(self, event, skill_name): """Add one relevant skill to the event without overwriting prior annotations.""" if not isinstance(event, dict): return skill = str(skill_name or "").strip() if skill in ("", "-"): return skills = self._get_event_relevant_skills(event) if skill not in skills: skills.append(skill) event["relevant_skills"] = skills event["relevant_skill"] = ", ".join(skills) def _annotate_relevant_action_skills(self, events, section_team_id=None): """Record all relevant action skills, including Pro, without overwriting.""" if not isinstance(events, list): return for event in events: if not isinstance(event, dict): continue ctx = self._action_event_context(event, section_team_id=section_team_id) if not ctx["is_friendly_action"]: continue action_label = str(event.get("action_attempt_label") or "").strip() if action_label: mapped_skill = self.action_skill_by_label.get(action_label) if mapped_skill and ( self.processor._event_player_has_skill(event, mapped_skill) or bool(event.get("relevant_skill_confirmed")) ): self._add_event_relevant_skill(event, mapped_skill) if self.processor._event_player_has_skill(event, "Pro"): self._add_event_relevant_skill(event, "Pro") def _attempt_has_non_pro_reroll(self, events, current_activation_key, player_name, action_family, action_ordinal): """Return True when the same action attempt already has a non-Pro reroll row.""" if not isinstance(events, list): return False for candidate in events: if not isinstance(candidate, dict): continue if not self._is_event_in_activation(candidate, current_activation_key): continue if str(candidate.get("player_name") or "").strip() != player_name: continue if str(candidate.get("roll_category") or "") != "action": continue if str(candidate.get("action_attempt_label") or "") != action_family: continue candidate_ordinal = candidate.get("action_attempt_ordinal_in_activation") if not (isinstance(candidate_ordinal, int) and isinstance(action_ordinal, int) and candidate_ordinal == action_ordinal): continue raw_value = candidate.get("raw") raw = raw_value if isinstance(raw_value, dict) else {} if str(raw.get("Status") or "") != "2": continue roll_type = str(candidate.get("roll_type") or raw.get("RollType") or "") if roll_type == self.pro_rule.get("roll_type", ""): continue return True return False def _event_base_context(self, event, section_team_id=None): """Return the shared event fields used by all context methods.""" category = str(event.get("roll_category") or "") if isinstance(event, dict) else "" event_team_id = self.processor._normalize_team_id(event.get("team_id")) if isinstance(event, dict) else None player_name = str(event.get("player_name") or "").strip() if isinstance(event, dict) else "" has_named_player = player_name not in ("", "Unknown") normalized_section_team = self.processor._normalize_team_id(section_team_id) return { "category": category, "event_team_id": event_team_id, "player_name": player_name, "has_named_player": has_named_player, "normalized_section_team": normalized_section_team, } def _action_event_context(self, event, section_team_id=None): """Return normalized action-context fields used by reroll processors.""" base = self._event_base_context(event, section_team_id) is_friendly_action = ( isinstance(event, dict) and base["category"] == "action" and base["has_named_player"] and base["event_team_id"] is not None and ( section_team_id is None or base["normalized_section_team"] is None or base["event_team_id"] == base["normalized_section_team"] ) ) return { "category": base["category"], "event_team_id": base["event_team_id"], "player_name": base["player_name"], "has_named_player": base["has_named_player"], "is_friendly_action": is_friendly_action, } def _action_activation_key(self, event, event_team_id, player_name): """Return activation key for action-processing state machines.""" return (str(event.get("game_turn")), event_team_id, player_name) def _player_activation_context(self, event, section_team_id=None): """Return normalized player-activation fields across action/block/other rows.""" base = self._event_base_context(event, section_team_id) is_friendly_player_event = ( isinstance(event, dict) and base["category"] in ("action", "block", "other") and base["has_named_player"] and base["event_team_id"] is not None and ( section_team_id is None or base["normalized_section_team"] is None or base["event_team_id"] == base["normalized_section_team"] ) ) return { "category": base["category"], "event_team_id": base["event_team_id"], "player_name": base["player_name"], "has_named_player": base["has_named_player"], "is_friendly_player_event": is_friendly_player_event, } def _event_explicitly_uses_pro(self, event): """Return True when replay payload explicitly represents Pro usage.""" if not isinstance(event, dict): return False raw_value = event.get("raw") raw = raw_value if isinstance(raw_value, dict) else {} payload_type = str(event.get("payload_type") or "") if payload_type == "QuestionChooseDice": skill_raw = str(raw.get("Skill") or "").strip().lower() if skill_raw == str(self.pro_rule.get("skill_id") or "").lower() or isinstance(raw.get("ProRoll"), dict): return True roll_type = str(event.get("roll_type") or raw.get("RollType") or "") negative_trait_skill_ids = set(self.pro_rule.get("negative_trait_skill_ids") or ()) has_negative_trait = any(self.processor._event_player_has_skill_id(event, sid) for sid in negative_trait_skill_ids) if ( payload_type == "QuestionTeamRerollUsage" and roll_type == self.pro_rule.get("roll_type", "") and self.processor._event_player_has_skill_id(event, self.pro_rule.get("skill_id", "50")) and not has_negative_trait ): return True return False def _annotate_activation_pro_state(self, events, section_team_id=None): """Annotate Pro availability/usage once per activation across categories.""" if not isinstance(events, list): return current_activation_key = None has_used_pro_in_activation = False for event in events: if not isinstance(event, dict): continue ctx = self._player_activation_context(event, section_team_id=section_team_id) if not ctx["is_friendly_player_event"]: continue activation_key = self._action_activation_key( event, ctx["event_team_id"], ctx["player_name"], ) if activation_key != current_activation_key: current_activation_key = activation_key has_used_pro_in_activation = False has_pro_skill = self.processor._event_player_has_skill_id(event, self.pro_rule.get("skill_id", "50")) or bool(event.get("pro_skill_available")) if has_pro_skill: event["pro_available_at_event"] = not has_used_pro_in_activation event["pro_used_in_activation"] = has_used_pro_in_activation explicit_pro_use = self._event_explicitly_uses_pro(event) # Treat any recognized Pro test roll as consuming the one allowed Pro # usage for this activation (including failed tests). any_pro_test_roll = bool(getattr(self.processor, "_is_any_pro_action_roll", lambda _e: False)(event)) if explicit_pro_use or any_pro_test_roll: event["pro_explicit_usage"] = True has_used_pro_in_activation = True if has_pro_skill: event["pro_used_in_activation"] = True def _annotate_action_attempt_ordinals(self, events, section_team_id=None): """Annotate action-attempt labels/ordinals in replay order.""" if not isinstance(events, list): return current_activator_key = None attempt_counts = defaultdict(int) pending_attempt = None last_attempt = None attempt_rows = defaultdict(list) def mark_attempt_skill_confirmed(family, attempt_index): key = (family, attempt_index) for row in attempt_rows.get(key, []): row["relevant_skill_confirmed"] = True for event in events: if not isinstance(event, dict): continue ctx = self._action_event_context(event, section_team_id=section_team_id) category = ctx["category"] event_team_id = ctx["event_team_id"] player_name = ctx["player_name"] has_named_player = ctx["has_named_player"] is_friendly_action = ctx["is_friendly_action"] if is_friendly_action: activator_key = self._action_activation_key(event, event_team_id, player_name) if activator_key != current_activator_key: current_activator_key = activator_key attempt_counts = defaultdict(int) pending_attempt = None last_attempt = None if not ( category == "action" and has_named_player and current_activator_key is not None and self._action_activation_key(event, event_team_id, player_name) == current_activator_key ): continue family = infer_action_attempt_family(event) if not family: continue payload_type = str(event.get("payload_type") or "") is_question_payload = payload_type.startswith("Question") raw_value = event.get("raw") raw = raw_value if isinstance(raw_value, dict) else {} is_explicit_reroll = str(raw.get("Status")) == "2" if isinstance(pending_attempt, tuple) and len(pending_attempt) == 2: pending_family, pending_index = pending_attempt else: pending_family, pending_index = None, None if isinstance(last_attempt, tuple) and len(last_attempt) == 2: last_family, last_index = last_attempt else: last_family, last_index = None, None if pending_family == family and isinstance(pending_index, int): event["action_attempt_label"] = family event["action_attempt_ordinal_in_activation"] = pending_index last_attempt = (family, pending_index) attempt_rows[(family, pending_index)].append(event) if payload_type == "QuestionSkillUsage": mark_attempt_skill_confirmed(family, pending_index) if not is_question_payload: pending_attempt = None continue if is_explicit_reroll and last_family == family and isinstance(last_index, int): event["action_attempt_label"] = family event["action_attempt_ordinal_in_activation"] = last_index last_attempt = (family, last_index) attempt_rows[(family, last_index)].append(event) continue attempt_counts[family] += 1 attempt_index = attempt_counts[family] event["action_attempt_label"] = family event["action_attempt_ordinal_in_activation"] = attempt_index last_attempt = (family, attempt_index) attempt_rows[(family, attempt_index)].append(event) if is_question_payload: pending_attempt = (family, attempt_index) if payload_type == "QuestionSkillUsage": mark_attempt_skill_confirmed(family, attempt_index) def _apply_skill_reroll_probabilities(self, events, action_label, skill_name, section_team_id=None): """Apply one-free-reroll action probabilities while relevant skill is unused.""" if not isinstance(events, list): return current_activator_key = None has_seen_fail_in_activation = False has_used_skill_in_activation = False for event in events: if not isinstance(event, dict): continue ctx = self._action_event_context(event, section_team_id=section_team_id) is_friendly_action = ctx["is_friendly_action"] event_team_id = ctx["event_team_id"] player_name = ctx["player_name"] if is_friendly_action: activator_key = self._action_activation_key(event, event_team_id, player_name) if activator_key != current_activator_key: current_activator_key = activator_key has_seen_fail_in_activation = False has_used_skill_in_activation = False if not is_friendly_action or event.get("action_attempt_label") != action_label: continue has_required_skill = self.processor._event_player_has_skill(event, skill_name) or bool(event.get("relevant_skill_confirmed")) if not has_required_skill: continue classification = str(event.get("report_result_classification") or event.get("result_classification") or "") if classification == "fail": has_seen_fail_in_activation = True if classification == "success" and not has_used_skill_in_activation: if has_seen_fail_in_activation: has_used_skill_in_activation = True else: try: difficulty_int = int(str(event.get("difficulty"))) except (TypeError, ValueError): difficulty_int = None dice_values = event.get("dice_values") if isinstance(event.get("dice_values"), list) else None if difficulty_int is not None and dice_values: _is_success, p_success_rr, p_fail_rr = self.processor.calculator.calc_action_with_reroll(dice_values, difficulty_int) event["probability_success"] = p_success_rr event["probability_neutral"] = 0.0 event["probability_fail"] = p_fail_rr def _find_pro_test_after_action(self, events, idx, current_activator_key, player_name, action_family, action_ordinal): """Return (tested, result_classification, test_index) for Pro test after action.""" for j in range(idx + 1, len(events)): fut = events[j] if not isinstance(fut, dict): continue if not self._is_event_in_activation(fut, current_activator_key): break fut_player = str(fut.get("player_name") or "").strip() if str(fut.get("roll_category") or "") == "block" and fut_player == player_name: break fut_family = str(fut.get("action_attempt_label") or "") fut_ordinal = fut.get("action_attempt_ordinal_in_activation") if ( fut_family == action_family and action_family != "" and isinstance(fut_ordinal, int) and isinstance(action_ordinal, int) and fut_ordinal > action_ordinal ): break if str(fut.get("roll_type") or "") == self.pro_rule.get("roll_type", "") and fut_player == player_name: result = str(fut.get("report_result_classification") or fut.get("result_classification") or "") return True, result, j return False, None, None def _is_event_in_activation(self, event, activation_key): """Return True if event belongs to the same activation key tuple.""" if not isinstance(event, dict): return False fut_player = str(event.get("player_name") or "").strip() fut_turn = str(event.get("game_turn") or "") fut_team = self.processor._normalize_team_id(event.get("team_id")) if fut_player and fut_turn and (fut_turn, fut_team, fut_player) != activation_key: return False return True def _exclude_followup_action_after_failed_pro( self, events, *, start_idx, current_activator_key, player_name, action_family, action_ordinal, ): """Exclude the replayed action echo that follows a failed Pro test. On failed Pro, BB3 can emit a repeated action row for the same attempt even though the original failure already stands; exclude that echo from detailed report and surprise accounting. """ for j in range(start_idx, len(events)): fut = events[j] if not isinstance(fut, dict): continue if not self._is_event_in_activation(fut, current_activator_key): break fut_player = str(fut.get("player_name") or "").strip() if str(fut.get("roll_type") or "") == self.pro_rule.get("roll_type", ""): continue if str(fut.get("roll_category") or "") != "action" or fut_player != player_name: continue fut_family = str(fut.get("action_attempt_label") or "") fut_ordinal = fut.get("action_attempt_ordinal_in_activation") same_attempt = ( fut_family == action_family and action_family != "" and isinstance(fut_ordinal, int) and isinstance(action_ordinal, int) and fut_ordinal == action_ordinal ) if not same_attempt: continue fut["exclude_from_detailed_report"] = True fut["exclude_from_surprise"] = True fut["pro_post_fail_echo_excluded"] = True break def _apply_pro_reroll_probabilities(self, events, section_team_id=None): """Apply Pro-based action probabilities when Pro is not explicitly tested after the action.""" if not isinstance(events, list): return current_activator_key = None has_used_pro_in_activation = False for idx, event in enumerate(events): if not isinstance(event, dict): continue if bool(event.get("exclude_from_detailed_report")): continue ctx = self._action_event_context(event, section_team_id=section_team_id) category = ctx["category"] is_friendly_action = ctx["is_friendly_action"] event_team_id = ctx["event_team_id"] player_name = ctx["player_name"] if is_friendly_action: activator_key = self._action_activation_key(event, event_team_id, player_name) if activator_key != current_activator_key: current_activator_key = activator_key has_used_pro_in_activation = False if bool(event.get("pro_used_in_activation")): has_used_pro_in_activation = True if category != "action": event["pro_assumption_status"] = self.pro_status_blocked_category continue if not is_friendly_action: continue roll_type = str(event.get("roll_type") or "") if roll_type == self.pro_rule.get("roll_type", ""): if bool(event.get("pro_explicit_usage")) or bool(event.get("pro_used_in_activation")): event["pro_assumption_status"] = self.pro_status_used else: event["pro_assumption_status"] = self.pro_status_blocked_skill continue relevant_skills = self._get_event_relevant_skills(event) non_pro_relevant_skills = [skill for skill in relevant_skills if skill != "Pro"] if non_pro_relevant_skills: event["pro_assumption_status"] = self.pro_status_blocked_skill event["pro_precedence_blocked"] = True event["pro_precedence_reason"] = "other_reroll_skill" event["pro_precedence_skills"] = list(non_pro_relevant_skills) continue has_pro_skill = self.processor._event_player_has_skill_id(event, self.pro_rule.get("skill_id", "50")) or bool(event.get("pro_skill_available")) if not has_pro_skill: event["pro_assumption_status"] = self.pro_status_no_pro continue if has_used_pro_in_activation: event["pro_assumption_status"] = self.pro_status_used event["pro_decision_status"] = "pro_already_used_in_activation_no_modification" continue if bool(event.get("pro_explicit_usage")): has_used_pro_in_activation = True event["pro_assumption_status"] = self.pro_status_used continue # Explicit forward scan: check whether a Pro test roll follows this # action attempt before the next attempt of the same family starts. action_family = str(event.get("action_attempt_label") or "") action_ordinal = event.get("action_attempt_ordinal_in_activation") if self._attempt_has_non_pro_reroll( events, current_activator_key, player_name, action_family, action_ordinal, ): event["pro_assumption_status"] = self.pro_status_blocked_skill event["pro_precedence_blocked"] = True event["pro_precedence_reason"] = "reroll_already_used" continue pro_tested, pro_test_result, pro_test_idx = self._find_pro_test_after_action( events, idx, current_activator_key, player_name, action_family, action_ordinal ) if pro_tested: has_used_pro_in_activation = True event["pro_assumption_status"] = self.pro_status_used event["pro_decision_status"] = "pro_tested_no_modification" if pro_test_result == "fail" and isinstance(pro_test_idx, int): self._exclude_followup_action_after_failed_pro( events, start_idx=pro_test_idx + 1, current_activator_key=current_activator_key, player_name=player_name, action_family=action_family, action_ordinal=action_ordinal, ) continue # Pro was available but not tested — replace probabilities. event["pro_assumption_status"] = self.pro_status_eligible event["pro_decision_status"] = "pro_not_tested" try: difficulty_int = int(str(event.get("difficulty"))) except (TypeError, ValueError): difficulty_int = None dice_values = event.get("dice_values") if isinstance(event.get("dice_values"), list) else None if difficulty_int is not None and dice_values: _is_success, p_success_pro, p_fail_pro = self.processor.calculator.calc_action_with_pro(dice_values, difficulty_int) event["probability_success"] = p_success_pro event["probability_neutral"] = 0.0 event["probability_fail"] = p_fail_pro def enrich_action_probabilities(self): events = getattr(self.processor.game_state, "roll_events", None) if not isinstance(events, list) or len(events) == 0: return canonical_events = [ event for event in events if isinstance(event, dict) and not event.get("exclude_from_detailed_report") ] self._annotate_activation_pro_state(canonical_events, section_team_id=None) self._annotate_action_attempt_ordinals(canonical_events, section_team_id=None) self._annotate_relevant_action_skills(canonical_events, section_team_id=None) for action_label, skill_name in self.action_skill_reroll_policies: self._apply_skill_reroll_probabilities( canonical_events, action_label, skill_name, section_team_id=None, ) self._apply_pro_reroll_probabilities(canonical_events, section_team_id=None)