Spaces:
Running
Running
| 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) |