diff --git a/src/warchron/constants.py b/src/warchron/constants.py index 5e2d857..5f1d037 100644 --- a/src/warchron/constants.py +++ b/src/warchron/constants.py @@ -308,8 +308,4 @@ class ContextType(StrEnum): CAMPAIGN = auto() CHOICE = auto() BATTLE = auto() - - -class ScoreKind(Enum): - VP = auto() - NP = auto() + OBJECTIVE = auto() diff --git a/src/warchron/controller/campaign_controller.py b/src/warchron/controller/campaign_controller.py index a67e989..00ffa3d 100644 --- a/src/warchron/controller/campaign_controller.py +++ b/src/warchron/controller/campaign_controller.py @@ -1,9 +1,13 @@ -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 +from warchron.constants import ( + RefreshScope, + ContextType, + ItemType, +) if TYPE_CHECKING: from warchron.controller.app_controller import AppController @@ -64,7 +68,7 @@ class CampaignController: objective_icon_maps[obj.id] = RankingIcon.compute_icons( war, ContextType.CAMPAIGN, - camp.id, + f"{camp.id}:{obj.id}", scores, objective_id=obj.id, ) @@ -166,10 +170,12 @@ class CampaignController: def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[tuple[str, str, int | None], 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, ctx.participants) + active = TieResolver.get_active_participants( + war, ctx.context_type, ctx.context_id, ctx.participants + ) players = [ ParticipantOption(id=pid, name=self.app.model.get_participant_name(pid)) for pid in active @@ -183,8 +189,9 @@ class CampaignController: context_id=ctx.context_id, context_name=None, ) - if ctx.objective_id: - objective = war.objectives[ctx.objective_id] + if ctx.context_type == ContextType.OBJECTIVE: + campaign_id, objective_id = ctx.context_id.split(":") + objective = war.objectives[objective_id] dialog = TieDialog( parent=self.app.view, players=players, @@ -194,10 +201,13 @@ class CampaignController: context_name=objective.name, ) if not dialog.exec(): - # FIXME lost tokens used for narrative tie-break (ContextType.OBJECTIVE) - TieResolver.cancel_tie_break(war, ctx) + TieResolver.cancel_tie_break( + war, ContextType.CAMPAIGN, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.key()] = dialog.get_bids() + bids_map[(ctx.context_type, ctx.context_id, ctx.score_value)] = ( + dialog.get_bids() + ) return bids_map # Campaign participant methods diff --git a/src/warchron/controller/closure_workflow.py b/src/warchron/controller/closure_workflow.py index 2e35568..e54f921 100644 --- a/src/warchron/controller/closure_workflow.py +++ b/src/warchron/controller/closure_workflow.py @@ -23,8 +23,8 @@ class RoundClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.rounds.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.key()] - TieResolver.apply_bids(war, tie, bids) + 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, bids) ties = TieResolver.find_battle_ties(war, round.id) for battle in round.battles.values(): @@ -40,8 +40,8 @@ class CampaignClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.campaigns.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.key()] - TieResolver.apply_bids(war, tie, bids) + 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, bids) ties = TieResolver.find_campaign_ties(war, campaign.id) for objective_id in war.objectives: @@ -53,8 +53,8 @@ class CampaignClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.campaigns.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.key()] - TieResolver.apply_bids(war, tie, bids) + 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, bids) ties = TieResolver.find_campaign_objective_ties( war, @@ -72,8 +72,8 @@ class WarClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.wars.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.key()] - TieResolver.apply_bids(war, tie, bids) + 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, bids) ties = TieResolver.find_war_ties(war) for objective_id in war.objectives: @@ -84,8 +84,8 @@ class WarClosureWorkflow(ClosureWorkflow): while ties: bids_map = self.app.wars.resolve_ties(war, ties) for tie in ties: - bids = bids_map[tie.key()] - TieResolver.apply_bids(war, tie, bids) + 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, bids) ties = TieResolver.find_war_objective_ties( war, diff --git a/src/warchron/controller/ranking_icon.py b/src/warchron/controller/ranking_icon.py index 1d9e131..27024d5 100644 --- a/src/warchron/controller/ranking_icon.py +++ b/src/warchron/controller/ranking_icon.py @@ -8,7 +8,6 @@ from warchron.constants import ( IconName, VP_RANK_TO_ICON, NP_RANK_TO_ICON, - ScoreKind, ) from warchron.model.war import War from warchron.model.score_service import ParticipantScore @@ -32,22 +31,14 @@ class RankingIcon: return score.victory_points icon_ranking = VP_RANK_TO_ICON - score_kind = ScoreKind.VP else: def value_getter(score: ParticipantScore) -> int: return score.narrative_points.get(objective_id, 0) icon_ranking = NP_RANK_TO_ICON - score_kind = ScoreKind.NP ranking = ResultChecker.get_effective_ranking( - war, - context_type, - context_id, - score_kind, - scores, - value_getter, - objective_id, + war, context_type, context_id, scores, value_getter=value_getter ) icon_map: Dict[str, QIcon] = {} for rank, group, token_map in ranking: diff --git a/src/warchron/controller/round_controller.py b/src/warchron/controller/round_controller.py index f49b416..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 @@ -105,8 +105,9 @@ class RoundController: if battle.is_draw(): p1_icon = Icons.get(IconName.DRAW) p2_icon = Icons.get(IconName.DRAW) - context = TieContext(ContextType.BATTLE, battle.sector_id) - if TieResolver.was_tie_broken_by_tokens(war, context): + if TieResolver.was_tie_broken_by_tokens( + war, ContextType.BATTLE, battle.sector_id + ): effective_winner = ResultChecker.get_effective_winner_id( war, ContextType.BATTLE, battle.sector_id, None ) @@ -178,7 +179,7 @@ class RoundController: def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[tuple[str, str, int | None], Dict[str, bool]]: + ) -> Dict[Tuple[ContextType, str, int | None], Dict[str, bool]]: bids_map = {} for ctx in contexts: players = [ @@ -197,9 +198,13 @@ class RoundController: context_id=ctx.context_id, ) if not dialog.exec(): - TieResolver.cancel_tie_break(war, ctx) + TieResolver.cancel_tie_break( + war, ContextType.BATTLE, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.key()] = 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 f687900..ddec9ff 100644 --- a/src/warchron/controller/war_controller.py +++ b/src/warchron/controller/war_controller.py @@ -1,4 +1,4 @@ -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 @@ -64,7 +64,7 @@ class WarController: objective_icon_maps[obj.id] = RankingIcon.compute_icons( war, ContextType.WAR, - war.id, + f"{war.id}:{obj.id}", scores, objective_id=obj.id, ) @@ -151,14 +151,16 @@ class WarController: RefreshScope.WARS_TREE, item_type=ItemType.WAR, item_id=war_id ) + # FIXME tie dialog with all participant even without tie def resolve_ties( self, war: War, contexts: List[TieContext] - ) -> Dict[tuple[str, str, int | None], 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, + ctx.context_type, + ctx.context_id, ctx.participants, ) players = [ @@ -174,21 +176,25 @@ class WarController: context_id=ctx.context_id, context_name=None, ) - if ctx.objective_id: - objective = war.objectives[ctx.objective_id] + if ctx.context_type == ContextType.OBJECTIVE: + _, objective_id = ctx.context_id.split(":") + objective = war.objectives[objective_id] dialog = TieDialog( parent=self.app.view, players=players, counters=counters, context_type=ctx.context_type, context_id=ctx.context_id, - context_name=f"Objective tie: {objective.name}", + context_name=objective.name, ) if not dialog.exec(): - # FIXME lost tokens used for narrative tie-break (ContextType.OBJECTIVE) - TieResolver.cancel_tie_break(war, ctx) + TieResolver.cancel_tie_break( + war, ContextType.WAR, ctx.context_id, ctx.score_value + ) raise ForbiddenOperation("Tie resolution cancelled") - bids_map[ctx.key()] = 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: diff --git a/src/warchron/model/battle.py b/src/warchron/model/battle.py index c63c1f4..3ff23a7 100644 --- a/src/warchron/model/battle.py +++ b/src/warchron/model/battle.py @@ -1,8 +1,6 @@ from __future__ import annotations from typing import Any, Dict -from warchron.model.json_helper import JsonHelper - class Battle: def __init__( @@ -73,13 +71,11 @@ class Battle: def fromDict(data: Dict[str, Any]) -> Battle: battle = Battle( data["sector_id"], - JsonHelper.none_if_empty(data.get("player_1_id")), - JsonHelper.none_if_empty(data.get("player_2_id")), + data.get("player_1_id") or None, + data.get("player_2_id") or None, ) - battle.winner_id = JsonHelper.none_if_empty(data.get("winner_id")) - battle.score = JsonHelper.none_if_empty(data.get("score")) - battle.victory_condition = JsonHelper.none_if_empty( - data.get("victory_condition") - ) - battle.comment = JsonHelper.none_if_empty(data.get("comment")) + battle.winner_id = data.get("winner_id") or None + battle.score = data.get("score") or None + battle.victory_condition = data.get("victory_condition") or None + battle.comment = data.get("comment") or None return battle diff --git a/src/warchron/model/choice.py b/src/warchron/model/choice.py index 9b88e21..0ed6aee 100644 --- a/src/warchron/model/choice.py +++ b/src/warchron/model/choice.py @@ -1,8 +1,6 @@ from __future__ import annotations from typing import Any, Dict -from warchron.model.json_helper import JsonHelper - class Choice: def __init__( @@ -44,8 +42,8 @@ class Choice: def fromDict(data: Dict[str, Any]) -> Choice: choice = Choice( data["participant_id"], - JsonHelper.none_if_empty(data.get("priority_sector_id")), - JsonHelper.none_if_empty(data.get("secondary_sector_id")), + data.get("priority_sector_id") or None, + data.get("secondary_sector_id") or None, ) - choice.comment = JsonHelper.none_if_empty(data.get("comment")) + choice.comment = data.get("comment") or None return choice diff --git a/src/warchron/model/json_helper.py b/src/warchron/model/json_helper.py deleted file mode 100644 index d12c0fd..0000000 --- a/src/warchron/model/json_helper.py +++ /dev/null @@ -1,11 +0,0 @@ -from typing import TypeVar - -T = TypeVar("T") - - -class JsonHelper: - @staticmethod - def none_if_empty(value: T | None) -> T | None: - if value == "": - return None - return value diff --git a/src/warchron/model/objective.py b/src/warchron/model/objective.py index 851a90f..67029f9 100644 --- a/src/warchron/model/objective.py +++ b/src/warchron/model/objective.py @@ -2,8 +2,6 @@ from __future__ import annotations from typing import Any, Dict from uuid import uuid4 -from warchron.model.json_helper import JsonHelper - class Objective: def __init__(self, name: str, description: str | None): @@ -29,6 +27,6 @@ class Objective: @staticmethod def fromDict(data: Dict[str, Any]) -> Objective: - obj = Objective(data["name"], JsonHelper.none_if_empty(data["description"])) + obj = Objective(data["name"], data["description"] or None) obj.set_id(data["id"]) return obj diff --git a/src/warchron/model/result_checker.py b/src/warchron/model/result_checker.py index bbfe190..b06ecb8 100644 --- a/src/warchron/model/result_checker.py +++ b/src/warchron/model/result_checker.py @@ -2,10 +2,10 @@ from __future__ import annotations from typing import List, Tuple, Dict, TYPE_CHECKING, Callable from collections import defaultdict -from warchron.constants import ContextType, ScoreKind +from warchron.constants import ContextType from warchron.model.war import War from warchron.model.war_event import TieResolved -from warchron.model.tie_manager import TieResolver, TieContext +from warchron.model.tie_manager import TieResolver if TYPE_CHECKING: from warchron.model.score_service import ParticipantScore @@ -35,37 +35,19 @@ class ResultChecker: war: War, context_type: ContextType, context_id: str, - score_kind: ScoreKind | None, scores: Dict[str, ParticipantScore], value_getter: Callable[[ParticipantScore], int], - objective_id: str | None = None, ) -> List[Tuple[int, List[str], Dict[str, int]]]: buckets: Dict[int, List[str]] = defaultdict(list) for pid, score in scores.items(): buckets[value_getter(score)].append(pid) - sorted_values = sorted(buckets.keys(), reverse=True) + sorted_vps = sorted(buckets.keys(), reverse=True) ranking: List[Tuple[int, List[str], Dict[str, int]]] = [] current_rank = 1 - assert score_kind is not None - for value in sorted_values: + for value in sorted_vps: participants = buckets[value] - context: TieContext = TieContext( - context_type, - context_id, - participants, - value, - score_kind, - objective_id, - ) if context_type == ContextType.WAR and len(participants) > 1: - subcontexts = ResultChecker._build_war_subcontexts( - war, - score_kind, - objective_id, - ) - subgroups = ResultChecker._secondary_sorting_war( - war, participants, value_getter, subcontexts - ) + subgroups = ResultChecker._secondary_sorting_war(war, participants) for subgroup in subgroups: # no tie if campaigns' rank is enough to sort if len(subgroup) == 1: @@ -75,29 +57,47 @@ class ResultChecker: current_rank += 1 continue # normal tie-break if tie persists - if not TieResolver.is_tie_resolved(war, context): + if not TieResolver.is_tie_resolved( + war, context_type, context_id, value + ): ranking.append( (current_rank, subgroup, {pid: 0 for pid in subgroup}) ) current_rank += 1 continue - groups = TieResolver.rank_by_tokens(war, context, subgroup) - tokens_spent = TieResolver.tokens_spent_map(war, context, subgroup) + 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): + if len(participants) == 1 or not TieResolver.is_tie_resolved( + war, context_type, context_id, value + ): ranking.append( (current_rank, participants, {pid: 0 for pid in participants}) ) current_rank += 1 continue # apply token ranking - groups = TieResolver.rank_by_tokens(war, context, participants) - tokens_spent = TieResolver.tokens_spent_map(war, context, participants) + groups = TieResolver.rank_by_tokens( + war, + context_type, + context_id, + participants, + ) + tokens_spent = TieResolver.tokens_spent_map( + 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)) @@ -108,35 +108,27 @@ class ResultChecker: def _secondary_sorting_war( war: War, participants: List[str], - value_getter: Callable[[ParticipantScore], int], - subcontexts: List[TieContext], ) -> List[List[str]]: from warchron.model.score_service import ScoreService rank_map: Dict[str, Tuple[int, ...]] = {} for pid in participants: ranks: List[int] = [] - for sub in subcontexts: + for campaign in war.campaigns: scores = ScoreService.compute_scores( - war, sub.context_type, sub.context_id + war, ContextType.CAMPAIGN, campaign.id ) ranking = ResultChecker.get_effective_ranking( war, - sub.context_type, - sub.context_id, - sub.score_kind, + ContextType.CAMPAIGN, + campaign.id, scores, - value_getter, - sub.objective_id, + lambda s: s.victory_points, ) - found = False for rank, group, _ in ranking: if pid in group: ranks.append(rank) - found = True break - if not found: - ranks.append(len(scores) + 1) rank_map[pid] = tuple(ranks) sorted_items = sorted(rank_map.items(), key=lambda x: x[1]) groups: List[List[str]] = [] @@ -147,23 +139,3 @@ class ResultChecker: current_tuple = rank_tuple groups[-1].append(pid) return groups - - @staticmethod - def _build_war_subcontexts( - war: War, - score_kind: ScoreKind, - objective_id: str | None, - ) -> List[TieContext]: - subcontexts = [] - for campaign in war.campaigns: - subcontexts.append( - TieContext( - ContextType.CAMPAIGN, - campaign.id, - [], - None, - score_kind, - objective_id, - ) - ) - return subcontexts diff --git a/src/warchron/model/sector.py b/src/warchron/model/sector.py index 57372c9..479dcf6 100644 --- a/src/warchron/model/sector.py +++ b/src/warchron/model/sector.py @@ -2,8 +2,6 @@ from __future__ import annotations from typing import Any, Dict from uuid import uuid4 -from warchron.model.json_helper import JsonHelper - class Sector: def __init__( @@ -66,11 +64,11 @@ class Sector: sec = Sector( data["name"], data["round_id"], - JsonHelper.none_if_empty(data.get("major_objective_id")), - JsonHelper.none_if_empty(data.get("minor_objective_id")), - JsonHelper.none_if_empty(data.get("influence_objective_id")), - JsonHelper.none_if_empty(data.get("mission")), - JsonHelper.none_if_empty(data.get("description")), + data.get("major_objective_id") or None, + data.get("minor_objective_id") or None, + data.get("influence_objective_id") or None, + data.get("mission") or None, + data.get("description") or None, ) sec.set_id(data["id"]) sec.mission = data.get("mission") or None diff --git a/src/warchron/model/tie_manager.py b/src/warchron/model/tie_manager.py index 9f78c13..13a220b 100644 --- a/src/warchron/model/tie_manager.py +++ b/src/warchron/model/tie_manager.py @@ -1,8 +1,8 @@ from typing import List, Dict, DefaultDict -from dataclasses import dataclass, field +from dataclasses import dataclass from collections import defaultdict -from warchron.constants import ContextType, ScoreKind +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 @@ -13,13 +13,8 @@ from warchron.model.score_service import ScoreService, ParticipantScore class TieContext: context_type: ContextType context_id: str - participants: List[str] = field(default_factory=list) # war_participant_ids + participants: List[str] # war_participant_ids score_value: int | None = None - score_kind: ScoreKind | None = None - objective_id: str | None = None - - def key(self) -> tuple[str, str, int | None]: - return (self.context_type, self.context_id, self.score_value) class TieResolver: @@ -32,11 +27,9 @@ class TieResolver: for battle in round.battles.values(): if not battle.is_draw(): continue - context: TieContext = TieContext( - ContextType.BATTLE, - battle.sector_id, - ) - if TieResolver.is_tie_resolved(war, context): + 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") @@ -44,7 +37,9 @@ class TieResolver: 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, context, [p1, p2]): + if not TieResolver.can_tie_be_resolved( + war, ContextType.BATTLE, battle.sector_id, [p1, p2] + ): war.events.append( TieResolved(None, ContextType.BATTLE, battle.sector_id) ) @@ -55,7 +50,6 @@ class TieResolver: context_id=battle.sector_id, participants=[p1, p2], score_value=None, - score_kind=None, ) ) return ties @@ -70,23 +64,15 @@ class TieResolver: for score_value, participants in buckets.items(): if len(participants) <= 1: continue - context: TieContext = TieContext( - ContextType.CAMPAIGN, - campaign_id, - [], - score_value, - ScoreKind.VP, - ) - if TieResolver.is_tie_resolved(war, context): + if TieResolver.is_tie_resolved( + war, ContextType.CAMPAIGN, campaign_id, score_value + ): continue - if not TieResolver.can_tie_be_resolved(war, context, participants): + if not TieResolver.can_tie_be_resolved( + war, ContextType.CAMPAIGN, campaign_id, participants + ): war.events.append( - TieResolved( - None, - ContextType.CAMPAIGN, - campaign_id, - score_value, - ) + TieResolved(None, ContextType.CAMPAIGN, campaign_id, score_value) ) continue ties.append( @@ -95,7 +81,6 @@ class TieResolver: context_id=campaign_id, participants=participants, score_value=score_value, - score_kind=ScoreKind.VP, ) ) return ties @@ -106,49 +91,51 @@ class TieResolver: campaign_id: str, objective_id: str, ) -> List[TieContext]: - scores = ScoreService.compute_scores( + base_scores = ScoreService.compute_scores( war, ContextType.CAMPAIGN, campaign_id, ) + scores = TieResolver._build_objective_scores( + base_scores, + objective_id, + ) buckets: DefaultDict[int, List[str]] = defaultdict(list) for pid, score in scores.items(): - np_value = score.narrative_points.get(objective_id, 0) - buckets[np_value].append(pid) + buckets[score.victory_points].append(pid) ties: List[TieContext] = [] - context_id = campaign_id - for np_value, participants in buckets.items(): + context_id = f"{campaign_id}:{objective_id}" + for score_value, participants in buckets.items(): if len(participants) <= 1: continue - context: TieContext = TieContext( - ContextType.CAMPAIGN, - campaign_id, - [], - np_value, - ScoreKind.NP, - objective_id, - ) - if TieResolver.is_tie_resolved(war, context): + if TieResolver.is_tie_resolved( + war, + ContextType.OBJECTIVE, + context_id, + score_value, + ): continue if not TieResolver.can_tie_be_resolved( war, - context, + ContextType.OBJECTIVE, + context_id, participants, ): war.events.append( TieResolved( - None, ContextType.CAMPAIGN, context_id, np_value, objective_id + None, + ContextType.OBJECTIVE, + context_id, + score_value, ) ) continue ties.append( TieContext( - context_type=ContextType.CAMPAIGN, + context_type=ContextType.OBJECTIVE, context_id=context_id, participants=participants, - score_value=np_value, - score_kind=ScoreKind.NP, - objective_id=objective_id, + score_value=score_value, ) ) return ties @@ -162,25 +149,17 @@ class TieResolver: war, ContextType.WAR, war.id, - ScoreKind.VP, scores, - lambda s: s.victory_points, + value_getter=lambda s: s.victory_points, ) ties: List[TieContext] = [] for _, group, _ in ranking: if len(group) <= 1: continue score_value = scores[group[0]].victory_points - context: TieContext = TieContext( - ContextType.WAR, - war.id, - [], - score_value, - ScoreKind.VP, - ) - if TieResolver.is_tie_resolved(war, context): + if TieResolver.is_tie_resolved(war, ContextType.WAR, war.id, score_value): continue - if not TieResolver.can_tie_be_resolved(war, context, group): + if not TieResolver.can_tie_be_resolved(war, ContextType.WAR, war.id, group): war.events.append( TieResolved(None, ContextType.WAR, war.id, score_value) ) @@ -191,7 +170,6 @@ class TieResolver: context_id=war.id, participants=group, score_value=score_value, - score_kind=ScoreKind.VP, ) ) return ties @@ -203,62 +181,78 @@ class TieResolver: ) -> List[TieContext]: from warchron.model.result_checker import ResultChecker - scores = ScoreService.compute_scores( + base_scores = ScoreService.compute_scores( war, ContextType.WAR, war.id, ) - - def value_getter(score: ParticipantScore) -> int: - return score.narrative_points.get(objective_id, 0) - + scores = TieResolver._build_objective_scores( + base_scores, + objective_id, + ) ranking = ResultChecker.get_effective_ranking( war, - ContextType.WAR, - war.id, - ScoreKind.NP, + ContextType.OBJECTIVE, + f"{war.id}:{objective_id}", scores, - value_getter, - objective_id, + value_getter=lambda s: s.narrative_points.get(objective_id, 0), ) ties: List[TieContext] = [] for _, group, _ in ranking: if len(group) <= 1: continue - np_value = value_getter(scores[group[0]]) - context: TieContext = TieContext( - ContextType.WAR, war.id, [], np_value, ScoreKind.NP, objective_id - ) + score_value = scores[group[0]].victory_points + context_id = f"{war.id}:{objective_id}" if TieResolver.is_tie_resolved( war, - context, + ContextType.OBJECTIVE, + context_id, + score_value, ): continue if not TieResolver.can_tie_be_resolved( war, - context, + ContextType.OBJECTIVE, + context_id, group, ): war.events.append( - TieResolved(None, ContextType.WAR, war.id, np_value, objective_id) + TieResolved( + None, + ContextType.OBJECTIVE, + context_id, + score_value, + ) ) continue ties.append( TieContext( - context_type=ContextType.WAR, - context_id=war.id, + context_type=ContextType.OBJECTIVE, + context_id=context_id, participants=group, - score_value=np_value, - score_kind=ScoreKind.NP, - objective_id=objective_id, + score_value=score_value, ) ) return ties + @staticmethod + def _build_objective_scores( + base_scores: Dict[str, ParticipantScore], + objective_id: str, + ) -> Dict[str, ParticipantScore]: + return { + pid: ParticipantScore( + victory_points=score.narrative_points.get(objective_id, 0), + narrative_points={}, + ) + for pid, score in base_scores.items() + } + @staticmethod def apply_bids( war: War, - context: TieContext, + context_type: ContextType, + context_id: str, bids: Dict[str, bool], # war_participant_id -> spend? ) -> None: for war_part_id, spend in bids.items(): @@ -270,17 +264,17 @@ class TieResolver: InfluenceSpent( participant_id=war_part_id, amount=1, - context_type=context.context_type, - context_id=context.context_id, - objective_id=context.objective_id, + context_type=context_type, + context_id=context_id, ) ) - # FIXME lost tokens used for narrative tie-break (ContextType.OBJECTIVE) @staticmethod def cancel_tie_break( war: War, - context: TieContext, + context_type: ContextType, + context_id: str, + score_value: int | None = None, ) -> None: war.events = [ ev @@ -288,15 +282,14 @@ class TieResolver: if not ( ( isinstance(ev, InfluenceSpent) - and ev.context_type == context.context_type - and ev.context_id == context.context_id + and ev.context_type == context_type + and ev.context_id == context_id ) or ( isinstance(ev, TieResolved) - and ev.context_type == context.context_type - and ev.context_id == context.context_id - and ev.score_value == context.score_value - and ev.objective_id == context.objective_id + and ev.context_type == context_type + and ev.context_id == context_id + and ev.score_value == score_value ) ) ] @@ -304,16 +297,16 @@ class TieResolver: @staticmethod def rank_by_tokens( war: War, - context: TieContext, + context_type: ContextType, + context_id: str, 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.context_type - and ev.context_id == context.context_id - and ev.objective_id == context.objective_id + and ev.context_type == context_type + and ev.context_id == context_id and ev.participant_id in spent ): spent[ev.participant_id] += ev.amount @@ -330,16 +323,16 @@ class TieResolver: @staticmethod def tokens_spent_map( war: War, - context: TieContext, + context_type: ContextType, + context_id: str, participants: List[str], ) -> Dict[str, int]: spent = {pid: 0 for pid in participants} for ev in war.events: if ( isinstance(ev, InfluenceSpent) - and ev.context_type == context.context_type - and ev.context_id == context.context_id - and ev.objective_id == context.objective_id + and ev.context_type == context_type + and ev.context_id == context_id and ev.participant_id in spent ): spent[ev.participant_id] += ev.amount @@ -348,57 +341,45 @@ class TieResolver: @staticmethod def get_active_participants( war: War, - context: TieContext, + context_type: ContextType, + context_id: str, participants: List[str], ) -> List[str]: - groups = TieResolver.rank_by_tokens(war, context, participants) + groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants) return groups[0] @staticmethod def resolve_tie_state( war: War, - context: TieContext, + ctx: TieContext, bids: dict[str, bool] | None = None, ) -> None: active = TieResolver.get_active_participants( war, - context, - context.participants, + ctx.context_type, + ctx.context_id, + ctx.participants, ) # confirmed draw if non had bid if not active: war.events.append( - TieResolved( - None, - context.context_type, - context.context_id, - context.score_value, - context.objective_id, - ) + 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.context_type, - context.context_id, - context.score_value, - context.objective_id, - ) + TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value) ) return # else rank_by_tokens - groups = TieResolver.rank_by_tokens(war, context, context.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.context_type, - context.context_id, - context.score_value, - context.objective_id, + groups[0][0], ctx.context_type, ctx.context_id, ctx.score_value ) ) return @@ -406,34 +387,41 @@ class TieResolver: @staticmethod def can_tie_be_resolved( - war: War, context: TieContext, participants: List[str] + war: War, context_type: ContextType, context_id: str, participants: List[str] ) -> bool: - active = TieResolver.get_active_participants(war, context, participants) + active = TieResolver.get_active_participants( + war, context_type, context_id, participants + ) return any(war.get_influence_tokens(pid) > 0 for pid in active) @staticmethod def was_tie_broken_by_tokens( war: War, - context: TieContext, + 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.context_type - and ev.context_id == context.context_id - and ev.score_value == context.score_value - and ev.objective_id == context.objective_id + 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: TieContext) -> 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.context_type - and ev.context_id == context.context_id - and ev.score_value == context.score_value - and ev.objective_id == context.objective_id + 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_event.py b/src/warchron/model/war_event.py index feb5842..3f3f4db 100644 --- a/src/warchron/model/war_event.py +++ b/src/warchron/model/war_event.py @@ -3,8 +3,6 @@ from typing import Dict, Any, TypeVar, Type, cast from datetime import datetime from uuid import uuid4 -from warchron.model.json_helper import JsonHelper - T = TypeVar("T", bound="WarEvent") EVENT_REGISTRY: Dict[str, Type["WarEvent"]] = {} @@ -17,12 +15,7 @@ def register_event(cls: Type[T]) -> Type[T]: class WarEvent: TYPE = "WarEvent" - 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): self.id: str = str(uuid4()) self.participant_id: str | None = participant_id self.context_type = context_type # battle, round, campaign, war @@ -76,18 +69,15 @@ class TieResolved(WarEvent): context_type: str, context_id: str, score_value: int | None = None, - objective_id: str | None = None, ): super().__init__(participant_id, context_type, context_id) self.score_value = score_value - self.objective_id = objective_id def toDict(self) -> Dict[str, Any]: d = super().toDict() d.update( { "score_value": self.score_value or None, - "objective_id": self.objective_id or None, } ) return d @@ -95,11 +85,10 @@ class TieResolved(WarEvent): @classmethod def fromDict(cls, data: Dict[str, Any]) -> TieResolved: ev = cls( - JsonHelper.none_if_empty(data["participant_id"]), + data["participant_id"] or None, data["context_type"], data["context_id"], - JsonHelper.none_if_empty(data["score_value"]), - JsonHelper.none_if_empty(data["objective_id"]), + data["score_value"] or None, ) return cls._base_fromDict(ev, data) @@ -109,17 +98,9 @@ class InfluenceGained(WarEvent): TYPE = "InfluenceGained" def __init__( - self, - participant_id: str, - amount: int, - context_type: str, - context_id: str, + self, participant_id: str, amount: int, context_type: str, context_id: str ): - super().__init__( - participant_id, - context_type, - context_id, - ) + super().__init__(participant_id, context_type, context_id) self.amount = amount def toDict(self) -> Dict[str, Any]: @@ -147,23 +128,16 @@ class InfluenceSpent(WarEvent): TYPE = "InfluenceSpent" def __init__( - self, - participant_id: str, - amount: int, - context_type: str, - context_id: str, - objective_id: str | None = None, + self, participant_id: str, amount: int, context_type: str, context_id: str ): super().__init__(participant_id, context_type, context_id) self.amount = amount - self.objective_id = objective_id def toDict(self) -> Dict[str, Any]: d = super().toDict() d.update( { "amount": self.amount, - "objective_id": self.objective_id, } ) return d @@ -175,6 +149,5 @@ class InfluenceSpent(WarEvent): int(data["amount"]), data["context_type"], data["context_id"], - JsonHelper.none_if_empty(data["objective_id"]), ) return cls._base_fromDict(ev, data) diff --git a/src/warchron/view/tie_dialog.py b/src/warchron/view/tie_dialog.py index fd08f34..41a17ff 100644 --- a/src/warchron/view/tie_dialog.py +++ b/src/warchron/view/tie_dialog.py @@ -76,12 +76,11 @@ class TieDialog(QDialog): def _get_context_title( context_type: ContextType, context_name: str | None = None ) -> str: - if context_name: - return f"{context_name} tie" titles = { ContextType.BATTLE: "Battle tie", ContextType.CAMPAIGN: "Campaign tie", ContextType.WAR: "War tie", ContextType.CHOICE: "Choice tie", + ContextType.OBJECTIVE: f"Objective tie: {context_name}", } return titles.get(context_type, "Tie") diff --git a/test_data/example.json b/test_data/example.json index 58cb4dc..9d3ae2f 100644 --- a/test_data/example.json +++ b/test_data/example.json @@ -339,8 +339,7 @@ "context_type": "battle", "context_id": "79accf7c-2d93-4ac3-b747-e7092bfe3feb", "timestamp": "2026-02-26T16:11:44.346337", - "score_value": null, - "objective_id": null + "score_value": null } ], "is_over": false