diff --git a/src/warchron/controller/app_controller.py b/src/warchron/controller/app_controller.py index 394d04b..7f0bf9b 100644 --- a/src/warchron/controller/app_controller.py +++ b/src/warchron/controller/app_controller.py @@ -96,7 +96,6 @@ 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: @@ -117,7 +116,6 @@ 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: @@ -215,13 +213,13 @@ class AppController: self.navigation.refresh_and_select( RefreshScope.WARS_TREE, item_type=ItemType.ROUND, item_id=rnd.id ) - self.is_dirty = True except DomainError as e: QMessageBox.warning( self.view, - "Deletion forbidden", + "Add forbidden", str(e), ) + return except RequiresConfirmation as e: reply = QMessageBox.question( self.view, @@ -231,7 +229,10 @@ class AppController: ) if reply == QMessageBox.StandardButton.Yes: e.action() - self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) + else: + return + self.is_dirty = True + self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) def edit_item(self, item_type: str, item_id: str) -> None: try: @@ -266,13 +267,13 @@ class AppController: elif item_type == ItemType.BATTLE: self.rounds.edit_round_battle(item_id) self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) - self.is_dirty = True except DomainError as e: QMessageBox.warning( self.view, - "Deletion forbidden", + "Update forbidden", str(e), ) + return except RequiresConfirmation as e: reply = QMessageBox.question( self.view, @@ -282,7 +283,10 @@ class AppController: ) if reply == QMessageBox.StandardButton.Yes: e.action() - self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) + else: + return + self.is_dirty = True + self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) def delete_item(self, item_type: str, item_id: str) -> None: reply = QMessageBox.question( @@ -326,13 +330,13 @@ class AppController: self.navigation.refresh_and_select( RefreshScope.WARS_TREE, item_type=ItemType.CAMPAIGN, item_id=camp_id ) - self.is_dirty = True except DomainError as e: QMessageBox.warning( self.view, "Deletion forbidden", str(e), ) + return except RequiresConfirmation as e: reply = QMessageBox.question( self.view, @@ -342,4 +346,7 @@ class AppController: ) if reply == QMessageBox.StandardButton.Yes: e.action() - self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) + else: + return + self.is_dirty = True + self.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) diff --git a/src/warchron/controller/campaign_controller.py b/src/warchron/controller/campaign_controller.py index d66154e..88199eb 100644 --- a/src/warchron/controller/campaign_controller.py +++ b/src/warchron/controller/campaign_controller.py @@ -1,15 +1,11 @@ -from typing import List, Dict, TYPE_CHECKING +from typing import List, Dict, Tuple, TYPE_CHECKING from PyQt6.QtWidgets import QMessageBox, QDialog -from PyQt6.QtGui import QIcon from warchron.constants import ( RefreshScope, ContextType, ItemType, - Icons, - IconName, - RANK_TO_ICON, ) if TYPE_CHECKING: @@ -28,8 +24,8 @@ from warchron.model.campaign_participant import CampaignParticipant from warchron.model.sector import Sector from warchron.model.tie_manager import TieContext, TieResolver from warchron.model.score_service import ScoreService -from warchron.model.result_checker import ResultChecker from warchron.controller.closure_workflow import CampaignClosureWorkflow +from warchron.controller.ranking_icon import RankingIcon from warchron.view.campaign_dialog import CampaignDialog from warchron.view.campaign_participant_dialog import CampaignParticipantDialog from warchron.view.sector_dialog import SectorDialog @@ -40,39 +36,6 @@ class CampaignController: def __init__(self, app: "AppController"): self.app = app - def _compute_campaign_ranking_icons( - self, war: War, campaign: Campaign - ) -> Dict[str, QIcon]: - scores = ScoreService.compute_scores( - war, - ContextType.CAMPAIGN, - campaign.id, - ) - ranking = ResultChecker.get_effective_ranking( - war, ContextType.CAMPAIGN, campaign.id, scores - ) - icon_map = {} - for rank, group, token_map in ranking: - base_icon = RANK_TO_ICON.get(rank, IconName.VPNTH) - tie_id = f"{campaign.id}:score:{scores[group[0]].victory_points}" - tie_resolved = TieResolver.is_tie_resolved( - war, ContextType.CAMPAIGN, tie_id - ) - for pid in group: - spent = token_map.get(pid, 0) - if not tie_resolved and spent == 0: - icon_name = getattr(IconName, f"{base_icon.name}DRAW") - elif tie_resolved and spent == 0 and len(group) > 1: - icon_name = getattr(IconName, f"{base_icon.name}DRAW") - elif tie_resolved and spent > 0 and len(group) == 1: - icon_name = getattr(IconName, f"{base_icon.name}BREAK") - elif tie_resolved and spent > 0 and len(group) > 1: - icon_name = getattr(IconName, f"{base_icon.name}TIEDRAW") - else: - icon_name = base_icon - icon_map[pid] = QIcon(Icons.get_pixmap(icon_name)) - return icon_map - def _fill_campaign_details(self, campaign_id: str) -> None: camp = self.app.model.get_campaign(campaign_id) self.app.view.show_campaign_details(name=camp.name, month=camp.month) @@ -96,7 +59,9 @@ class CampaignController: rows: List[CampaignParticipantScoreDTO] = [] icon_map = {} if camp.is_over: - icon_map = self._compute_campaign_ranking_icons(war, camp) + icon_map = RankingIcon.compute_icons( + war, ContextType.CAMPAIGN, campaign_id, scores + ) for camp_part in camp.get_all_campaign_participants(): war_part_id = camp_part.war_participant_id war_part = war.get_war_participant(war_part_id) @@ -177,9 +142,10 @@ class CampaignController: except DomainError as e: QMessageBox.warning( self.app.view, - "Deletion forbidden", + "Closure forbidden", str(e), ) + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) self.app.navigation.refresh_and_select( @@ -188,7 +154,7 @@ class CampaignController: def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[str, Dict[str, bool]]: + ) -> Dict[Tuple[ContextType, str, int | None], Dict[str, bool]]: bids_map = {} for ctx in contexts: active = TieResolver.get_active_participants( @@ -207,11 +173,17 @@ class CampaignController: context_id=ctx.context_id, ) if not dialog.exec(): - TieResolver.cancel_tie_break(war, ContextType.CAMPAIGN, ctx.context_id) + TieResolver.cancel_tie_break( + war, ContextType.CAMPAIGN, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.context_id] = dialog.get_bids() + bids_map[(ctx.context_type, ctx.context_id, ctx.score_value)] = ( + dialog.get_bids() + ) return bids_map + # Campaign participant methods + def create_campaign_participant(self) -> CampaignParticipant | None: if not self.app.navigation.selected_campaign_id: return None @@ -270,7 +242,6 @@ class CampaignController: self.app.view, "Invalid name", "Sector name cannot be empty." ) return False - # TODO allow same objectives in different fields? return True def create_sector(self) -> Sector | None: diff --git a/src/warchron/controller/closure_workflow.py b/src/warchron/controller/closure_workflow.py index d668338..fe368ac 100644 --- a/src/warchron/controller/closure_workflow.py +++ b/src/warchron/controller/closure_workflow.py @@ -24,11 +24,9 @@ class RoundClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.rounds.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.context_id] + bids = bids_map[(tie.context_type, tie.context_id, tie.score_value)] TieResolver.apply_bids(war, tie.context_type, tie.context_id, bids) - TieResolver.resolve_tie_state( - war, tie.context_type, tie.context_id, tie.participants, bids - ) + TieResolver.resolve_tie_state(war, tie, bids) ties = TieResolver.find_battle_ties(war, round.id) for battle in round.battles.values(): ClosureService.apply_battle_outcomes(war, campaign, battle) @@ -43,11 +41,9 @@ class CampaignClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.campaigns.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.context_id] + bids = bids_map[(tie.context_type, tie.context_id, tie.score_value)] TieResolver.apply_bids(war, tie.context_type, tie.context_id, bids) - TieResolver.resolve_tie_state( - war, tie.context_type, tie.context_id, tie.participants, bids - ) + TieResolver.resolve_tie_state(war, tie, bids) ties = TieResolver.find_campaign_ties(war, campaign.id) ClosureService.finalize_campaign(campaign) @@ -60,10 +56,8 @@ class WarClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.wars.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.context_id] + bids = bids_map[(tie.context_type, tie.context_id, tie.score_value)] TieResolver.apply_bids(war, tie.context_type, tie.context_id, bids) - TieResolver.resolve_tie_state( - war, tie.context_type, tie.context_id, tie.participants, bids - ) + TieResolver.resolve_tie_state(war, tie, bids) ties = TieResolver.find_war_ties(war) ClosureService.finalize_war(war) diff --git a/src/warchron/controller/navigation_controller.py b/src/warchron/controller/navigation_controller.py index fe84943..2a6aa47 100644 --- a/src/warchron/controller/navigation_controller.py +++ b/src/warchron/controller/navigation_controller.py @@ -31,7 +31,8 @@ class NavigationController: self.app.view.display_players(players_for_display) def refresh_wars_view(self) -> None: - wars: List[WarDTO] = [ + wars = self.app.model.get_all_wars() + wars_dto: List[WarDTO] = [ WarDTO( id=w.id, name=w.name, @@ -57,7 +58,21 @@ class NavigationController: ) for w in self.app.model.get_all_wars() ] - self.app.view.display_wars_tree(wars) + self.app.view.display_wars_tree(wars_dto) + if not wars: + self.clear_selection() + return + first_war = wars[0] + self.selected_war_id = first_war.id + self.selected_campaign_id = None + self.selected_round_id = None + self.app.view.select_tree_item( + item_type=ItemType.WAR, + item_id=first_war.id, + ) + self.app.view.show_details(ItemType.WAR) + self.app.wars._fill_war_details(first_war.id) + self.update_actions_state() def refresh(self, scope: RefreshScope) -> None: match scope: @@ -87,6 +102,13 @@ class NavigationController: # Commands methods + def clear_selection(self) -> None: + self.selected_war_id = None + self.selected_campaign_id = None + self.selected_round_id = None + self.app.view.clear_tree_selection() + self.app.view.show_details(None) + def on_tree_selection_changed(self, selection: TreeSelection | None) -> None: self.selected_war_id = None self.selected_campaign_id = None diff --git a/src/warchron/controller/ranking_icon.py b/src/warchron/controller/ranking_icon.py new file mode 100644 index 0000000..bb7d914 --- /dev/null +++ b/src/warchron/controller/ranking_icon.py @@ -0,0 +1,54 @@ +from typing import Dict + +from PyQt6.QtGui import QIcon + +from warchron.constants import ( + ContextType, + Icons, + IconName, + RANK_TO_ICON, +) +from warchron.model.war import War +from warchron.model.score_service import ParticipantScore +from warchron.model.result_checker import ResultChecker + + +class RankingIcon: + @staticmethod + def compute_icons( + war: War, + context_type: ContextType, + context_id: str, + scores: Dict[str, ParticipantScore], + ) -> Dict[str, QIcon]: + # scores = ScoreService.compute_scores( + # war, + # context_type, + # context_id, + # ) + ranking = ResultChecker.get_effective_ranking( + war, context_type, context_id, scores + ) + icon_map = {} + for rank, group, token_map in ranking: + base_icon = RANK_TO_ICON.get(rank, IconName.VPNTH) + vp = scores[group[0]].victory_points + original_group_size = sum( + 1 for s in scores.values() if s.victory_points == vp + ) + for pid in group: + spent = token_map.get(pid, 0) + if original_group_size == 1: + icon_name = base_icon + elif len(group) == 1: + if spent > 0: + icon_name = getattr(IconName, f"{base_icon.name}BREAK") + else: + icon_name = base_icon + else: + if spent > 0: + icon_name = getattr(IconName, f"{base_icon.name}TIEDRAW") + else: + icon_name = getattr(IconName, f"{base_icon.name}DRAW") + icon_map[pid] = QIcon(Icons.get_pixmap(icon_name)) + return icon_map diff --git a/src/warchron/controller/round_controller.py b/src/warchron/controller/round_controller.py index 8fdc4c9..5af114b 100644 --- a/src/warchron/controller/round_controller.py +++ b/src/warchron/controller/round_controller.py @@ -1,4 +1,4 @@ -from typing import List, Dict, TYPE_CHECKING +from typing import List, Dict, Tuple, TYPE_CHECKING from PyQt6.QtWidgets import QDialog from PyQt6.QtWidgets import QMessageBox @@ -167,9 +167,10 @@ class RoundController: except DomainError as e: QMessageBox.warning( self.app.view, - "Deletion forbidden", + "Closure forbidden", str(e), ) + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) self.app.navigation.refresh_and_select( @@ -178,7 +179,7 @@ class RoundController: def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[str, Dict[str, bool]]: + ) -> Dict[Tuple[ContextType, str, int | None], Dict[str, bool]]: bids_map = {} for ctx in contexts: players = [ @@ -197,8 +198,13 @@ class RoundController: context_id=ctx.context_id, ) if not dialog.exec(): + TieResolver.cancel_tie_break( + war, ContextType.BATTLE, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.context_id] = dialog.get_bids() + bids_map[(ctx.context_type, ctx.context_id, ctx.score_value)] = ( + dialog.get_bids() + ) return bids_map # Choice methods diff --git a/src/warchron/controller/war_controller.py b/src/warchron/controller/war_controller.py index 54f28f6..84977e9 100644 --- a/src/warchron/controller/war_controller.py +++ b/src/warchron/controller/war_controller.py @@ -1,17 +1,17 @@ -from typing import List, TYPE_CHECKING, Dict +from typing import List, Tuple, TYPE_CHECKING, Dict from PyQt6.QtWidgets import QMessageBox, QDialog -from PyQt6.QtGui import QIcon from warchron.constants import ( RefreshScope, ItemType, ContextType, - Icons, - IconName, - RANK_TO_ICON, ) -from warchron.model.exception import DomainError, ForbiddenOperation +from warchron.model.exception import ( + DomainError, + ForbiddenOperation, + RequiresConfirmation, +) if TYPE_CHECKING: from warchron.controller.app_controller import AppController @@ -25,8 +25,8 @@ from warchron.model.war_participant import WarParticipant from warchron.model.objective import Objective from warchron.model.tie_manager import TieContext, TieResolver from warchron.model.score_service import ScoreService -from warchron.model.result_checker import ResultChecker from warchron.controller.closure_workflow import WarClosureWorkflow +from warchron.controller.ranking_icon import RankingIcon from warchron.view.war_dialog import WarDialog from warchron.view.objective_dialog import ObjectiveDialog from warchron.view.war_participant_dialog import WarParticipantDialog @@ -37,35 +37,6 @@ class WarController: def __init__(self, app: "AppController"): self.app = app - def _compute_war_ranking_icons(self, war: War) -> Dict[str, QIcon]: - scores = ScoreService.compute_scores( - war, - ContextType.WAR, - war.id, - ) - ranking = ResultChecker.get_effective_ranking( - war, ContextType.WAR, war.id, scores - ) - icon_map = {} - for rank, group, token_map in ranking: - base_icon = RANK_TO_ICON.get(rank, IconName.VPNTH) - tie_id = f"{war.id}:score:{scores[group[0]].victory_points}" - tie_resolved = TieResolver.is_tie_resolved(war, ContextType.WAR, tie_id) - for pid in group: - spent = token_map.get(pid, 0) - if not tie_resolved and spent == 0: - icon_name = getattr(IconName, f"{base_icon.name}DRAW") - elif tie_resolved and spent == 0 and len(group) > 1: - icon_name = getattr(IconName, f"{base_icon.name}DRAW") - elif tie_resolved and spent > 0 and len(group) == 1: - icon_name = getattr(IconName, f"{base_icon.name}BREAK") - elif tie_resolved and spent > 0 and len(group) > 1: - icon_name = getattr(IconName, f"{base_icon.name}TIEDRAW") - else: - icon_name = base_icon - icon_map[pid] = QIcon(Icons.get_pixmap(icon_name)) - return icon_map - def _fill_war_details(self, war_id: str) -> None: war = self.app.model.get_war(war_id) self.app.view.show_war_details(name=war.name, year=war.year) @@ -84,7 +55,7 @@ class WarController: rows: List[WarParticipantScoreDTO] = [] icon_map = {} if war.is_over: - icon_map = self._compute_war_ranking_icons(war) + icon_map = RankingIcon.compute_icons(war, ContextType.WAR, war_id, scores) for war_part in war.get_all_war_participants(): player_name = self.app.model.get_player_name(war_part.player_id) score = scores[war_part.id] @@ -152,23 +123,26 @@ class WarController: except DomainError as e: QMessageBox.warning( self.app.view, - "Deletion forbidden", + "Closure forbidden", str(e), ) + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) self.app.navigation.refresh_and_select( RefreshScope.WARS_TREE, item_type=ItemType.WAR, item_id=war_id ) - # TODO fix ignored campaign tie-breaks def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[str, Dict[str, bool]]: + ) -> Dict[Tuple[ContextType, str, int | None], Dict[str, bool]]: bids_map = {} for ctx in contexts: active = TieResolver.get_active_participants( - war, ctx.context_type, ctx.context_id, ctx.participants + war, + ctx.context_type, + ctx.context_id, + ctx.participants, ) players = [ ParticipantOption(id=pid, name=self.app.model.get_participant_name(pid)) @@ -183,9 +157,13 @@ class WarController: context_id=ctx.context_id, ) if not dialog.exec(): - TieResolver.cancel_tie_break(war, ContextType.WAR, ctx.context_id) + TieResolver.cancel_tie_break( + war, ContextType.WAR, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.context_id] = dialog.get_bids() + bids_map[(ctx.context_type, ctx.context_id, ctx.score_value)] = ( + dialog.get_bids() + ) return bids_map def set_major_value(self, value: int) -> None: @@ -200,6 +178,7 @@ class WarController: "Setting forbidden", str(e), ) + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) @@ -215,6 +194,7 @@ class WarController: "Setting forbidden", str(e), ) + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) @@ -230,8 +210,23 @@ class WarController: "Setting forbidden", str(e), ) + return + except RequiresConfirmation as e: + reply = QMessageBox.question( + self.app.view, + "Confirm update", + str(e), + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + ) + if reply == QMessageBox.StandardButton.Yes: + e.action() + else: + return self.app.is_dirty = True self.app.navigation.refresh(RefreshScope.CURRENT_SELECTION_DETAILS) + self.app.navigation.refresh_and_select( + RefreshScope.WARS_TREE, item_type=ItemType.WAR, item_id=war_id + ) # Objective methods diff --git a/src/warchron/model/closure_service.py b/src/warchron/model/closure_service.py index 21a3b5f..0cc16f3 100644 --- a/src/warchron/model/closure_service.py +++ b/src/warchron/model/closure_service.py @@ -27,8 +27,7 @@ class ClosureService: from warchron.model.result_checker import ResultChecker already_granted = any( - isinstance(e, InfluenceGained) - and e.context_id == f"battle:{battle.sector_id}" + isinstance(e, InfluenceGained) and e.context_id == battle.sector_id for e in war.events ) if already_granted: @@ -51,7 +50,7 @@ class ClosureService: participant_id=effective_winner, amount=1, context_type=ContextType.BATTLE, - context_id=f"battle:{battle.sector_id}", + context_id=battle.sector_id, ) ) diff --git a/src/warchron/model/result_checker.py b/src/warchron/model/result_checker.py index d69003a..6430ef6 100644 --- a/src/warchron/model/result_checker.py +++ b/src/warchron/model/result_checker.py @@ -45,10 +45,42 @@ class ResultChecker: current_rank = 1 for vp in sorted_vps: participants = vp_buckets[vp] - tie_id = f"{context_id}:score:{vp}" + if context_type == ContextType.WAR and len(participants) > 1: + subgroups = ResultChecker._secondary_sorting_war(war, participants) + for subgroup in subgroups: + # no tie if campaigns' rank is enough to sort + if len(subgroup) == 1: + ranking.append( + (current_rank, subgroup, {pid: 0 for pid in subgroup}) + ) + current_rank += 1 + continue + # normal tie-break if tie persists + if not TieResolver.is_tie_resolved( + war, context_type, context_id, vp + ): + ranking.append( + (current_rank, subgroup, {pid: 0 for pid in subgroup}) + ) + current_rank += 1 + continue + groups = TieResolver.rank_by_tokens( + war, + context_type, + context_id, + subgroup, + ) + tokens_spent = TieResolver.tokens_spent_map( + war, context_type, context_id, subgroup + ) + for group in groups: + group_tokens = {pid: tokens_spent[pid] for pid in group} + ranking.append((current_rank, group, group_tokens)) + current_rank += 1 + continue # no tie if len(participants) == 1 or not TieResolver.is_tie_resolved( - war, context_type, tie_id + war, context_type, context_id, vp ): ranking.append( (current_rank, participants, {pid: 0 for pid in participants}) @@ -59,14 +91,46 @@ class ResultChecker: groups = TieResolver.rank_by_tokens( war, context_type, - tie_id, + context_id, participants, ) tokens_spent = TieResolver.tokens_spent_map( - war, context_type, tie_id, participants + war, context_type, context_id, participants ) for group in groups: group_tokens = {pid: tokens_spent[pid] for pid in group} ranking.append((current_rank, group, group_tokens)) current_rank += 1 return ranking + + @staticmethod + def _secondary_sorting_war( + war: War, + participants: List[str], + ) -> List[List[str]]: + from warchron.model.score_service import ScoreService + + rank_map: Dict[str, Tuple[int, ...]] = {} + for pid in participants: + ranks: List[int] = [] + for campaign in war.campaigns: + scores = ScoreService.compute_scores( + war, ContextType.CAMPAIGN, campaign.id + ) + ranking = ResultChecker.get_effective_ranking( + war, ContextType.CAMPAIGN, campaign.id, scores + ) + for rank, group, _ in ranking: + if pid in group: + ranks.append(rank) + break + rank_map[pid] = tuple(ranks) + sorted_items = sorted(rank_map.items(), key=lambda x: x[1]) + groups: List[List[str]] = [] + current_tuple = None + for pid, rank_tuple in sorted_items: + if rank_tuple != current_tuple: + groups.append([]) + current_tuple = rank_tuple + groups[-1].append(pid) + return groups diff --git a/src/warchron/model/tie_manager.py b/src/warchron/model/tie_manager.py index c398592..e174e8e 100644 --- a/src/warchron/model/tie_manager.py +++ b/src/warchron/model/tie_manager.py @@ -14,6 +14,7 @@ class TieContext: context_type: ContextType context_id: str participants: List[str] # war_participant_ids + score_value: int | None = None class TieResolver: @@ -26,9 +27,10 @@ class TieResolver: for battle in round.battles.values(): if not battle.is_draw(): continue - if TieResolver.is_tie_resolved(war, ContextType.BATTLE, battle.sector_id): + if TieResolver.is_tie_resolved( + war, ContextType.BATTLE, battle.sector_id, None + ): 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: @@ -47,14 +49,13 @@ class TieResolver: context_type=ContextType.BATTLE, context_id=battle.sector_id, participants=[p1, p2], + score_value=None, ) ) return ties @staticmethod def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]: - 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) for pid, score in scores.items(): @@ -63,48 +64,53 @@ class TieResolver: for score_value, participants in buckets.items(): if len(participants) <= 1: continue - tie_id = f"{campaign_id}:score:{score_value}" - if TieResolver.is_tie_resolved(war, ContextType.CAMPAIGN, tie_id): + if TieResolver.is_tie_resolved( + war, ContextType.CAMPAIGN, campaign_id, score_value + ): continue if not TieResolver.can_tie_be_resolved( - war, ContextType.CAMPAIGN, tie_id, participants + war, ContextType.CAMPAIGN, campaign_id, participants ): - war.events.append(TieResolved(None, ContextType.CAMPAIGN, tie_id)) + war.events.append( + TieResolved(None, ContextType.CAMPAIGN, campaign_id, score_value) + ) continue ties.append( TieContext( context_type=ContextType.CAMPAIGN, - context_id=tie_id, + context_id=campaign_id, participants=participants, + score_value=score_value, ) ) return ties @staticmethod def find_war_ties(war: War) -> List[TieContext]: - if TieResolver.is_tie_resolved(war, ContextType.WAR, war.id): - return [] + from warchron.model.result_checker import ResultChecker + scores = ScoreService.compute_scores(war, ContextType.WAR, war.id) - buckets: DefaultDict[int, List[str]] = defaultdict(list) - for pid, score in scores.items(): - buckets[score.victory_points].append(pid) + ranking = ResultChecker.get_effective_ranking( + war, ContextType.WAR, war.id, scores + ) ties: List[TieContext] = [] - for score_value, participants in buckets.items(): - if len(participants) <= 1: + for _, group, _ in ranking: + if len(group) <= 1: continue - tie_id = f"{war.id}:score:{score_value}" - if TieResolver.is_tie_resolved(war, ContextType.WAR, tie_id): + score_value = scores[group[0]].victory_points + if TieResolver.is_tie_resolved(war, ContextType.WAR, war.id, score_value): continue - if not TieResolver.can_tie_be_resolved( - war, ContextType.WAR, tie_id, participants - ): - war.events.append(TieResolved(None, ContextType.WAR, tie_id)) + if not TieResolver.can_tie_be_resolved(war, ContextType.WAR, war.id, group): + war.events.append( + TieResolved(None, ContextType.WAR, war.id, score_value) + ) continue ties.append( TieContext( context_type=ContextType.WAR, - context_id=tie_id, - participants=participants, + context_id=war.id, + participants=group, + score_value=score_value, ) ) return ties @@ -135,6 +141,7 @@ class TieResolver: war: War, context_type: ContextType, context_id: str, + score_value: int | None = None, ) -> None: war.events = [ ev @@ -149,6 +156,7 @@ class TieResolver: isinstance(ev, TieResolved) and ev.context_type == context_type and ev.context_id == context_id + and ev.score_value == score_value ) ) ] @@ -210,26 +218,37 @@ class TieResolver: @staticmethod def resolve_tie_state( war: War, - context_type: ContextType, - context_id: str, - participants: List[str], + ctx: TieContext, bids: dict[str, bool] | None = None, ) -> None: active = TieResolver.get_active_participants( - war, context_type, context_id, participants + war, + ctx.context_type, + ctx.context_id, + ctx.participants, ) # confirmed draw if non had bid if not active: - war.events.append(TieResolved(None, context_type, context_id)) + war.events.append( + TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value) + ) return # confirmed draw if current bids are 0 if bids is not None and not any(bids.values()): - war.events.append(TieResolved(None, context_type, context_id)) + war.events.append( + TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value) + ) return # else rank_by_tokens - groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants) + groups = TieResolver.rank_by_tokens( + war, ctx.context_type, ctx.context_id, ctx.participants + ) if len(groups[0]) == 1: - war.events.append(TieResolved(groups[0][0], context_type, context_id)) + war.events.append( + TieResolved( + groups[0][0], ctx.context_type, ctx.context_id, ctx.score_value + ) + ) return # if tie persists, do nothing, workflow will call again @@ -247,21 +266,29 @@ class TieResolver: war: War, context_type: ContextType, context_id: str, + score_value: int | None = None, ) -> bool: for ev in reversed(war.events): if ( isinstance(ev, TieResolved) and ev.context_type == context_type and ev.context_id == context_id + and ev.score_value == score_value ): return ev.participant_id is not None return False @staticmethod - def is_tie_resolved(war: War, context_type: ContextType, context_id: str) -> bool: + def is_tie_resolved( + war: War, + context_type: ContextType, + context_id: str, + score_value: int | None = None, + ) -> bool: return any( isinstance(ev, TieResolved) and ev.context_type == context_type and ev.context_id == context_id + and ev.score_value == score_value for ev in war.events ) diff --git a/src/warchron/model/war.py b/src/warchron/model/war.py index 8681d12..360d70e 100644 --- a/src/warchron/model/war.py +++ b/src/warchron/model/war.py @@ -3,8 +3,14 @@ from uuid import uuid4 from datetime import datetime from typing import Any, Dict, List -from warchron.model.war_event import WarEvent, InfluenceGained, InfluenceSpent -from warchron.model.exception import ForbiddenOperation +from warchron.constants import ContextType +from warchron.model.war_event import ( + WarEvent, + InfluenceGained, + InfluenceSpent, + TieResolved, +) +from warchron.model.exception import ForbiddenOperation, RequiresConfirmation from warchron.model.war_participant import WarParticipant from warchron.model.objective import Objective from warchron.model.campaign_participant import CampaignParticipant @@ -55,10 +61,66 @@ class War: def set_influence_token(self, new_state: bool) -> None: if self.is_over: raise ForbiddenOperation("Can't set influence token of a closed war.") - # TODO raise RequiresConfirmation - # * disable: cleanup if any token has already been gained/spent - # * enable: retrigger battle_outcomes and resolve tie again if any draw - self.influence_token = new_state + + def cleanup_token_and_tie() -> None: + new_events: List[WarEvent] = [] + for ev in self.events: + if isinstance(ev, (InfluenceSpent, InfluenceGained)): + continue + if isinstance(ev, TieResolved): + ev.set_participant(None) + new_events.append(ev) + self.events = new_events + self.influence_token = new_state + + def reset_tie_break() -> None: + new_events: List[WarEvent] = [] + for ev in self.events: + if isinstance(ev, (TieResolved)): + if ev.context_type == ContextType.BATTLE: + battle = self.get_battle(ev.context_id) + campaign = self.get_campaign_by_sector(battle.sector_id) + round = campaign.get_round_by_battle(ev.context_id) + round.is_over = False + elif ev.context_type == ContextType.CAMPAIGN: + campaign = self.get_campaign(ev.context_id) + campaign.is_over = False + else: + new_events.append(ev) + self.events = new_events + for camp in self.campaigns: + for sect in camp.get_all_sectors(): + if sect.influence_objective_id is not None: + round = camp.get_round_by_battle(sect.id) + round.is_over = False + self.influence_token = new_state + + if new_state is False: + if not self.events: + self.influence_token = new_state + return + raise RequiresConfirmation( + "Some influence tokens already exist in this war.\n" + "All tokens will be deleted and tie-breaks will become draw.\n" + "Do you want to continue?", + action=cleanup_token_and_tie, + ) + if new_state is True: + has_tie_resolved = any(isinstance(ev, TieResolved) for ev in self.events) + grant_influence = any( + (sect.influence_objective_id is not None) + for camp in self.campaigns + for sect in camp.get_all_sectors() + ) + if not has_tie_resolved and not grant_influence: + self.influence_token = new_state + return + raise RequiresConfirmation( + "Some influence tokens and draws exist in this war.\n" + "All influence outcomes and tie-breaks will be reset.\n" + "Do you want to continue?", + action=reset_tie_break, + ) def set_state(self, new_state: bool) -> None: self.is_over = new_state diff --git a/src/warchron/model/war_event.py b/src/warchron/model/war_event.py index 62209cd..3f3f4db 100644 --- a/src/warchron/model/war_event.py +++ b/src/warchron/model/war_event.py @@ -63,19 +63,32 @@ class WarEvent: class TieResolved(WarEvent): TYPE = "TieResolved" - def __init__(self, participant_id: str | None, context_type: str, context_id: str): + def __init__( + self, + participant_id: str | None, + context_type: str, + context_id: str, + score_value: int | None = None, + ): super().__init__(participant_id, context_type, context_id) + self.score_value = score_value def toDict(self) -> Dict[str, Any]: d = super().toDict() + d.update( + { + "score_value": self.score_value or None, + } + ) return d @classmethod def fromDict(cls, data: Dict[str, Any]) -> TieResolved: ev = cls( - data["participant_id"], + data["participant_id"] or None, data["context_type"], data["context_id"], + data["score_value"] or None, ) return cls._base_fromDict(ev, data) diff --git a/src/warchron/view/view.py b/src/warchron/view/view.py index 5e87ce3..31354a1 100644 --- a/src/warchron/view/view.py +++ b/src/warchron/view/view.py @@ -90,6 +90,7 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow): ) self.majorValue.valueChanged.connect(self._on_major_changed) self.minorValue.valueChanged.connect(self._on_minor_changed) + self.warsTree.currentItemChanged.connect(self._emit_selection_changed) self._apply_icons() def _apply_icons(self) -> None: @@ -143,6 +144,11 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow): return "wars" return "" + def clear_tree_selection(self) -> None: + self.warsTree.blockSignals(True) + self.warsTree.setCurrentItem(None) + self.warsTree.blockSignals(False) + # General popups def closeEvent(self, event: QCloseEvent | None = None) -> None: @@ -237,10 +243,7 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow): def display_wars_tree(self, wars: List[WarDTO]) -> None: tree = self.warsTree - try: - tree.currentItemChanged.disconnect() - except TypeError: - pass + tree.blockSignals(True) tree.clear() tree.setColumnCount(1) tree.setHeaderLabels(["Wars"]) @@ -264,8 +267,8 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow): rnd_item.setData(0, ROLE_TYPE, ItemType.ROUND) rnd_item.setData(0, ROLE_ID, rnd.id) camp_item.addChild(rnd_item) - tree.currentItemChanged.connect(self._emit_selection_changed) tree.expandAll() + tree.blockSignals(False) def select_tree_item(self, *, item_type: ItemType, item_id: str) -> None: def walk(item: QTreeWidgetItem) -> bool: