diff --git a/src/warchron/controller/app_controller.py b/src/warchron/controller/app_controller.py index 7ac6d7c..394d04b 100644 --- a/src/warchron/controller/app_controller.py +++ b/src/warchron/controller/app_controller.py @@ -96,6 +96,7 @@ class AppController: self.navigation.refresh_players_view() self.navigation.refresh_wars_view() self.update_window_title() + # TODO refresh details view if wars tab selected def open_file(self) -> None: if self.is_dirty: @@ -116,6 +117,7 @@ class AppController: self.navigation.refresh_players_view() self.navigation.refresh_wars_view() self.update_window_title() + # TODO refresh details view if wars tab selected def save(self) -> None: if not self.current_file: diff --git a/src/warchron/controller/closure_workflow.py b/src/warchron/controller/closure_workflow.py index 5699d67..16e0de3 100644 --- a/src/warchron/controller/closure_workflow.py +++ b/src/warchron/controller/closure_workflow.py @@ -3,7 +3,6 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from warchron.controller.app_controller import AppController -from warchron.model.war_event import TieResolved from warchron.model.war import War from warchron.model.campaign import Campaign from warchron.model.round import Round @@ -23,22 +22,8 @@ class RoundClosureWorkflow(ClosureWorkflow): ClosureService.check_round_closable(round) ties = TieResolver.find_battle_ties(war, round.id) while ties: - resolvable = [] + bids_map = self.app.rounds.resolve_ties(war, ties) for tie in ties: - if TieResolver.can_tie_be_resolved(war, tie.participants): - resolvable.append(tie) - else: - war.events.append( - TieResolved( - participant_id=None, # draw confirmed - context_type=tie.context_type, - context_id=tie.context_id, - ) - ) - if not resolvable: - break - bids_map = self.app.rounds.resolve_ties(war, resolvable) - for tie in resolvable: bids = bids_map[tie.context_id] TieResolver.apply_bids( war, @@ -46,11 +31,8 @@ class RoundClosureWorkflow(ClosureWorkflow): tie.context_id, bids, ) - TieResolver.try_tie_break( - war, - tie.context_type, - tie.context_id, - tie.participants, + TieResolver.resolve_tie_state( + war, tie.context_type, tie.context_id, tie.participants, bids ) ties = TieResolver.find_battle_ties(war, round.id) for battle in round.battles.values(): @@ -64,22 +46,8 @@ class CampaignClosureWorkflow(ClosureWorkflow): ClosureService.check_campaign_closable(campaign) ties = TieResolver.find_campaign_ties(war, campaign.id) while ties: - resolvable = [] + bids_map = self.app.campaigns.resolve_ties(war, ties) for tie in ties: - if TieResolver.can_tie_be_resolved(war, tie.participants): - resolvable.append(tie) - else: - war.events.append( - TieResolved( - participant_id=None, - context_type=tie.context_type, - context_id=tie.context_id, - ) - ) - if not resolvable: - break - bids_map = self.app.campaigns.resolve_ties(war, resolvable) - for tie in resolvable: bids = bids_map[tie.context_id] TieResolver.apply_bids( war, @@ -87,7 +55,7 @@ class CampaignClosureWorkflow(ClosureWorkflow): tie.context_id, bids, ) - TieResolver.try_tie_break( + TieResolver.resolve_tie_state( war, tie.context_type, tie.context_id, diff --git a/src/warchron/model/campaign.py b/src/warchron/model/campaign.py index e2fce92..eb55bd3 100644 --- a/src/warchron/model/campaign.py +++ b/src/warchron/model/campaign.py @@ -187,6 +187,7 @@ class Campaign: mission: str | None, description: str | None, ) -> None: + # TODO raise error if sector used in a closed round (potential tokens) if self.is_over: raise ForbiddenOperation("Can't update sector in a closed campaign.") sect = self.get_sector(sector_id) diff --git a/src/warchron/model/closure_service.py b/src/warchron/model/closure_service.py index 93ab166..3816e5c 100644 --- a/src/warchron/model/closure_service.py +++ b/src/warchron/model/closure_service.py @@ -27,7 +27,8 @@ class ClosureService: @staticmethod def apply_battle_outcomes(war: War, campaign: Campaign, battle: Battle) -> None: already_granted = any( - isinstance(e, InfluenceGained) and e.source == f"battle:{battle.sector_id}" + isinstance(e, InfluenceGained) + and e.context_id == f"battle:{battle.sector_id}" for e in war.events ) if already_granted: @@ -49,7 +50,8 @@ class ClosureService: InfluenceGained( participant_id=effective_winner, amount=1, - source=f"battle:{battle.sector_id}", + context_type=ContextType.BATTLE, + context_id=f"battle:{battle.sector_id}", ) ) diff --git a/src/warchron/model/tie_manager.py b/src/warchron/model/tie_manager.py index 36c08ab..f9c4054 100644 --- a/src/warchron/model/tie_manager.py +++ b/src/warchron/model/tie_manager.py @@ -16,6 +16,14 @@ class TieContext: participants: List[str] # war_participant_ids +@dataclass +class TieState: + winner: str | None + tied_players: List[str] + eliminated: List[str] + # is_final: bool + + class TieResolver: @staticmethod @@ -26,20 +34,20 @@ class TieResolver: 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: + if TieResolver.is_tie_resolved(war, ContextType.BATTLE, battle.sector_id): 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 + if not TieResolver.can_tie_be_resolved(war, [p1, p2]): + war.events.append( + TieResolved(None, ContextType.BATTLE, battle.sector_id) + ) + continue ties.append( TieContext( context_type=ContextType.BATTLE, @@ -51,13 +59,7 @@ class TieResolver: @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: + if TieResolver.is_tie_resolved(war, ContextType.CAMPAIGN, campaign_id): return [] scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id) buckets: DefaultDict[int, List[str]] = defaultdict(list) @@ -96,46 +98,54 @@ class TieResolver: participant_id=war_part_id, amount=1, context_type=context_type, + context_id=context_id, ) ) @staticmethod - def try_tie_break( + def rank_by_tokens( 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 + participants: List[str], + ) -> List[List[str]]: + spent = {pid: 0 for pid in participants} + for ev in war.events: + if ( + isinstance(ev, InfluenceSpent) + and ev.context_type == context_type + and ev.context_id == context_id + and ev.participant_id in spent + ): + spent[ev.participant_id] += ev.amount + sorted_items = sorted(spent.items(), key=lambda x: x[1], reverse=True) + groups: List[List[str]] = [] + current_score = None + for pid, score in sorted_items: + if score != current_score: + groups.append([]) + current_score = score + groups[-1].append(pid) + return groups + + @staticmethod + def resolve_tie_state( + war: War, + context_type: ContextType, + context_id: str, + participants: List[str], + bids: dict[str, bool] | None = None, + ) -> None: + # confirmed draw if bids are 0 + if bids is not None and not any(bids.values()): + war.events.append(TieResolved(None, context_type, context_id)) + return + # else rank_by_tokens + groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants) + if len(groups[0]) == 1: + war.events.append(TieResolved(groups[0][0], context_type, context_id)) + return + # if tie persists, do nothing, workflow will call again @staticmethod def can_tie_be_resolved(war: War, participants: List[str]) -> bool: @@ -155,3 +165,12 @@ class TieResolver: ): return ev.participant_id is not None return False + + @staticmethod + def is_tie_resolved(war: War, context_type: ContextType, context_id: str) -> bool: + return any( + isinstance(ev, TieResolved) + and ev.context_type == context_type + and ev.context_id == context_id + for ev in war.events + ) diff --git a/src/warchron/model/war_event.py b/src/warchron/model/war_event.py index 72c03f3..62209cd 100644 --- a/src/warchron/model/war_event.py +++ b/src/warchron/model/war_event.py @@ -15,9 +15,11 @@ def register_event(cls: Type[T]) -> Type[T]: class WarEvent: TYPE = "WarEvent" - def __init__(self, participant_id: str | None = None): + def __init__(self, participant_id: str | None, context_type: str, context_id: str): self.id: str = str(uuid4()) self.participant_id: str | None = participant_id + self.context_type = context_type # battle, round, campaign, war + self.context_id = context_id self.timestamp: datetime = datetime.now() def set_id(self, new_id: str) -> None: @@ -34,6 +36,8 @@ class WarEvent: "type": self.TYPE, "id": self.id, "participant_id": self.participant_id, + "context_type": self.context_type, + "context_id": self.context_id, "timestamp": self.timestamp.isoformat(), } @@ -41,6 +45,8 @@ class WarEvent: def _base_fromDict(cls: Type[T], ev: T, data: Dict[str, Any]) -> T: ev.id = data["id"] ev.participant_id = data["participant_id"] + ev.context_type = data["context_type"] + ev.context_id = data["context_id"] ev.timestamp = datetime.fromisoformat(data["timestamp"]) return ev @@ -58,21 +64,10 @@ class TieResolved(WarEvent): TYPE = "TieResolved" def __init__(self, participant_id: str | None, context_type: str, context_id: str): - super().__init__(participant_id) - self.participant_id: str | None = ( - participant_id # winner or None (confirmed tie) - ) - self.context_type = context_type # battle, round, campaign, war - self.context_id = context_id + super().__init__(participant_id, context_type, context_id) def toDict(self) -> Dict[str, Any]: d = super().toDict() - d.update( - { - "context_type": self.context_type, - "context_id": self.context_id, - } - ) return d @classmethod @@ -89,17 +84,17 @@ class TieResolved(WarEvent): class InfluenceGained(WarEvent): TYPE = "InfluenceGained" - def __init__(self, participant_id: str, amount: int, source: str): - super().__init__(participant_id) + def __init__( + self, participant_id: str, amount: int, context_type: str, context_id: str + ): + super().__init__(participant_id, context_type, context_id) self.amount = amount - self.source = source # "battle", "tie_resolution", etc. def toDict(self) -> Dict[str, Any]: d = super().toDict() d.update( { "amount": self.amount, - "source": self.source, } ) return d @@ -108,8 +103,9 @@ class InfluenceGained(WarEvent): def fromDict(cls, data: Dict[str, Any]) -> InfluenceGained: ev = cls( data["participant_id"], - data["amount"], - data["source"], + int(data["amount"]), + data["context_type"], + data["context_id"], ) return cls._base_fromDict(ev, data) @@ -118,17 +114,17 @@ class InfluenceGained(WarEvent): class InfluenceSpent(WarEvent): TYPE = "InfluenceSpent" - def __init__(self, participant_id: str, amount: int, context_type: str): - super().__init__(participant_id) + def __init__( + self, participant_id: str, amount: int, context_type: str, context_id: str + ): + super().__init__(participant_id, context_type, context_id) self.amount = amount - self.context_type = context_type # "battle_tie", "campaign_tie", etc. def toDict(self) -> Dict[str, Any]: d = super().toDict() d.update( { "amount": self.amount, - "context_type": self.context_type, } ) return d @@ -137,7 +133,8 @@ class InfluenceSpent(WarEvent): def fromDict(cls, data: Dict[str, Any]) -> InfluenceSpent: ev = cls( data["participant_id"], - data["amount"], + int(data["amount"]), data["context_type"], + data["context_id"], ) return cls._base_fromDict(ev, data)