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