from typing import List, Dict, DefaultDict from dataclasses import dataclass from collections import defaultdict from warchron.constants import ContextType from warchron.model.exception import ForbiddenOperation from warchron.model.war import War from warchron.model.war_event import InfluenceSpent, TieResolved from warchron.model.score_service import ScoreService @dataclass class TieContext: context_type: ContextType context_id: str participants: List[str] # war_participant_ids class TieResolver: @staticmethod def find_battle_ties(war: War, round_id: str) -> List[TieContext]: round = war.get_round(round_id) campaign = war.get_campaign_by_round(round_id) ties = [] for battle in round.battles.values(): if not battle.is_draw(): continue resolved = any( isinstance(e, TieResolved) and e.context_type == ContextType.BATTLE and e.context_id == battle.sector_id for e in war.events ) if resolved: continue if campaign is None: raise RuntimeError("No campaign for this battle tie") if battle.player_1_id is None or battle.player_2_id is None: raise RuntimeError("Missing player(s) in this battle context.") p1 = campaign.participants[battle.player_1_id].war_participant_id p2 = campaign.participants[battle.player_2_id].war_participant_id ties.append( TieContext( context_type=ContextType.BATTLE, context_id=battle.sector_id, participants=[p1, p2], ) ) return ties @staticmethod def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]: resolved = any( isinstance(e, TieResolved) and e.context_type == ContextType.CAMPAIGN and e.context_id == campaign_id for e in war.events ) if resolved: return [] scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id) buckets: DefaultDict[int, List[str]] = defaultdict(list) for pid, score in scores.items(): buckets[score.victory_points].append(pid) ties = [] for participants in buckets.values(): if len(participants) > 1: ties.append( TieContext( context_type=ContextType.CAMPAIGN, context_id=campaign_id, participants=participants, ) ) return ties @staticmethod def find_war_ties(war: War) -> List[TieContext]: return [] # TODO @staticmethod def apply_bids( war: War, context_type: ContextType, context_id: str, bids: Dict[str, bool], # war_participant_id -> spend? ) -> None: for war_part_id, spend in bids.items(): if not spend: continue if war.get_influence_tokens(war_part_id) < 1: raise ForbiddenOperation("Not enough tokens") war.events.append( InfluenceSpent( participant_id=war_part_id, amount=1, context_type=context_type, ) ) @staticmethod def try_tie_break( war: War, context_type: ContextType, context_id: str, participants: List[str], # war_participant_ids ) -> bool: spent: Dict[str, int] = {} for war_part_id in participants: spent[war_part_id] = sum( e.amount for e in war.events if isinstance(e, InfluenceSpent) and e.participant_id == war_part_id and e.context_type == context_type ) values = set(spent.values()) if values == {0}: # no bid = confirmed draw war.events.append( TieResolved( participant_id=None, context_type=context_type, context_id=context_id, ) ) return True if len(values) == 1: # tie again, continue return False winner = max(spent.items(), key=lambda item: item[1])[0] war.events.append( TieResolved( participant_id=winner, context_type=context_type, context_id=context_id, ) ) return True @staticmethod def can_tie_be_resolved(war: War, participants: List[str]) -> bool: return any(war.get_influence_tokens(pid) > 0 for pid in participants) @staticmethod def was_tie_broken_by_tokens( war: War, context_type: ContextType, context_id: str, ) -> bool: for ev in reversed(war.events): if ( isinstance(ev, TieResolved) and ev.context_type == context_type and ev.context_id == context_id ): return ev.participant_id is not None return False