maximator/warchron_app
maximator/warchron_app
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix choice tiebreak loop and cancel tiebreak lost tokens

Browse source
This commit is contained in:
Maxime Réaux 2026-03-17 11:16:47 +01:00
parent a3b9f5a943
commit 42ad708e77
13 changed files with 333 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/warchron/controller/campaign_controller.py
View file

@ -167,7 +167,7 @@ class CampaignController:
def resolve_ties(
self, war: War, contexts: List[TieContext]
) -> Dict[tuple[str, str, int | None], Dict[str, bool]]:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {}
for ctx in contexts:
active = TieResolver.get_active_participants(war, ctx, ctx.participants)
@ -180,7 +180,7 @@ class CampaignController:
parent=self.app.view,
players=players,
counters=counters,
context_type=ContextType.CAMPAIGN,
context_type=ctx.context_type,
context_id=ctx.context_id,
context_name=None,
)
@ -192,7 +192,7 @@ class CampaignController:
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, ctx)

28
src/warchron/controller/closure_workflow.py
View file

@ -1,4 +1,5 @@
from typing import TYPE_CHECKING
from uuid import uuid4
if TYPE_CHECKING:
from warchron.controller.app_controller import AppController
@ -25,8 +26,9 @@ class RoundClosureWorkflow(ClosureWorkflow):
bids_map = self.app.rounds.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
tie_id = TieResolver.find_active_tie_id(war, tie) or str(uuid4())
TieResolver.apply_bids(war, tie, tie_id, bids)
TieResolver.resolve_tie_state(war, tie, tie_id, bids)
ties = TieResolver.find_battle_ties(war, round.id)
for battle in round.battles.values():
ClosureService.apply_battle_outcomes(war, campaign, battle)
@ -42,8 +44,9 @@ class CampaignClosureWorkflow(ClosureWorkflow):
bids_map = self.app.campaigns.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
tie_id = TieResolver.find_active_tie_id(war, tie) or str(uuid4())
TieResolver.apply_bids(war, tie, tie_id, bids)
TieResolver.resolve_tie_state(war, tie, tie_id, bids)
ties = TieResolver.find_campaign_ties(war, campaign.id)
for obj in war.get_objectives_used_as_maj_or_min():
objective_id = obj.id
@ -56,8 +59,9 @@ class CampaignClosureWorkflow(ClosureWorkflow):
bids_map = self.app.campaigns.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
tie_id = TieResolver.find_active_tie_id(war, tie) or str(uuid4())
TieResolver.apply_bids(war, tie, tie_id, bids)
TieResolver.resolve_tie_state(war, tie, tie_id, bids)
ties = TieResolver.find_campaign_objective_ties(
war,
campaign.id,
@ -75,8 +79,9 @@ class WarClosureWorkflow(ClosureWorkflow):
bids_map = self.app.wars.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
tie_id = TieResolver.find_active_tie_id(war, tie) or str(uuid4())
TieResolver.apply_bids(war, tie, tie_id, bids)
TieResolver.resolve_tie_state(war, tie, tie_id, bids)
ties = TieResolver.find_war_ties(war)
for obj in war.get_objectives_used_as_maj_or_min():
objective_id = obj.id
@ -88,8 +93,9 @@ class WarClosureWorkflow(ClosureWorkflow):
bids_map = self.app.wars.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
tie_id = TieResolver.find_active_tie_id(war, tie) or str(uuid4())
TieResolver.apply_bids(war, tie, tie_id, bids)
TieResolver.resolve_tie_state(war, tie, tie_id, bids)
ties = TieResolver.find_war_objective_ties(
war,
objective_id,
@ -103,7 +109,7 @@ class RoundPairingWorkflow:
self.app = controller
def start(self, war: War, round: Round) -> None:
Pairing.check_round_pairable(round)
Pairing.check_round_pairable(war, round)
Pairing.assign_battles_to_participants(
war,
round,

16
src/warchron/controller/round_controller.py
View file

@ -72,6 +72,7 @@ class RoundController:
comment=choice.comment,
)
)
# TODO display allocated sectors and used token
self.app.view.display_round_choices(choices_for_display)
battles_for_display: List[BattleDTO] = []
for sect in sectors:
@ -89,6 +90,7 @@ class RoundController:
player_1_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
p1_id = battle.player_1_id
else:
player_1_name = ""
if battle.player_2_id:
@ -96,6 +98,7 @@ class RoundController:
player_2_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
p2_id = battle.player_2_id
else:
player_2_name = ""
if battle.winner_id:
@ -112,7 +115,11 @@ 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)
context = TieContext(
ContextType.BATTLE,
battle.sector_id,
[p1_id, p2_id],
)
if TieResolver.was_tie_broken_by_tokens(war, context):
effective_winner = ResultChecker.get_effective_winner_id(
war, ContextType.BATTLE, battle.sector_id, None
@ -197,7 +204,7 @@ class RoundController:
str(e),
)
for bat in rnd.battles.values():
bat.cleanup_battle_players()
bat.clear_battle_players()
return
except DomainError as e:
QMessageBox.warning(
@ -225,7 +232,7 @@ class RoundController:
def resolve_ties(
self, war: War, contexts: List[TieContext]
) -> Dict[tuple[str, str, int | None], Dict[str, bool]]:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {}
for ctx in contexts:
players = [
@ -236,11 +243,12 @@ class RoundController:
for pid in ctx.participants
]
counters = [war.get_influence_tokens(pid) for pid in ctx.participants]
# TODO display sector name for BATTLE or CHOICE
dialog = TieDialog(
parent=self.app.view,
players=players,
counters=counters,
context_type=ContextType.BATTLE,
context_type=ctx.context_type,
context_id=ctx.context_id,
)
if not dialog.exec():

6
src/warchron/controller/war_controller.py
View file

@ -58,7 +58,7 @@ class WarController:
]
scores = ScoreService.compute_scores(war, ContextType.WAR, war.id)
rows: List[WarParticipantScoreDTO] = []
vp_icon_map: dict[str, QIcon] = {}
vp_icon_map: Dict[str, QIcon] = {}
objective_icon_maps: Dict[str, Dict[str, QIcon]] = {}
if war.is_over:
vp_icon_map = RankingIcon.compute_icons(
@ -157,7 +157,7 @@ class WarController:
def resolve_ties(
self, war: War, contexts: List[TieContext]
) -> Dict[tuple[str, str, int | None], Dict[str, bool]]:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {}
for ctx in contexts:
active = TieResolver.get_active_participants(
@ -174,7 +174,7 @@ class WarController:
parent=self.app.view,
players=players,
counters=counters,
context_type=ContextType.WAR,
context_type=ctx.context_type,
context_id=ctx.context_id,
context_name=None,
)

4
src/warchron/model/battle.py
View file

@ -57,7 +57,7 @@ class Battle:
return False
def get_available_places(self) -> List[str]:
places: list[str] = []
places: List[str] = []
if self.player_1_id is None:
places.append("player_1")
if self.player_2_id is None:
@ -73,7 +73,7 @@ class Battle:
return
raise DomainError("Battle has no available places")
def cleanup_battle_players(self) -> None:
def clear_battle_players(self) -> None:
self.player_1_id = None
self.player_2_id = None

10
src/warchron/model/campaign.py
View file

@ -1,7 +1,9 @@
from __future__ import annotations
from uuid import uuid4
from typing import Any, Dict, List, Set
from typing import Any, Dict, List, Set, TYPE_CHECKING
if TYPE_CHECKING:
from warchron.model.war import War
from warchron.model.exception import ForbiddenOperation, RequiresConfirmation
from warchron.model.campaign_participant import CampaignParticipant
from warchron.model.sector import Sector
@ -19,6 +21,7 @@ class Campaign:
self.sectors: Dict[str, Sector] = {}
self.rounds: List[Round] = []
self.is_over = False
self._war: War | None = None # private link
def set_id(self, new_id: str) -> None:
self.id = new_id
@ -60,8 +63,8 @@ class Campaign:
# Campaign participant methods
def get_all_campaign_participants_ids(self) -> set[str]:
return set(self.participants.keys())
def get_all_campaign_participants_ids(self) -> List[str]:
return list(self.participants.keys())
def has_participant(self, participant_id: str) -> bool:
return participant_id in self.participants
@ -351,6 +354,7 @@ class Campaign:
if self.is_over:
raise ForbiddenOperation("Can't add round in a closed campaign.")
round = Round()
round._campaign = self
self.rounds.append(round)
return round

4
src/warchron/model/model.py
View file

@ -360,8 +360,8 @@ class Model:
)
def remove_sector(self, sector_id: str) -> None:
camp = self.get_campaign_by_sector(sector_id)
camp.remove_sector(sector_id)
war = self.get_war_by_sector(sector_id)
war.remove_sector(sector_id)
# Campaign participant methods

90
src/warchron/model/pairing.py
View file

@ -1,5 +1,7 @@
from __future__ import annotations
from typing import Dict, List, Callable, Tuple
from uuid import uuid4
import random
from warchron.constants import ContextType, ScoreKind
@ -18,7 +20,7 @@ from warchron.model.score_service import ParticipantScore
ResolveTiesCallback = Callable[
["War", List["TieContext"]],
Dict[Tuple[str, str, int | None], Dict[str, bool]],
Dict[Tuple[str, str, int | None, str | None, str | None], Dict[str, bool]],
]
@ -26,6 +28,7 @@ class Pairing:
@staticmethod
def check_round_pairable(
war: War,
round: Round,
) -> None:
if round.is_over:
@ -48,8 +51,9 @@ class Pairing:
def cleanup() -> None:
for bat in round.battles.values():
bat.cleanup_battle_players()
# FIXME cancel TieResolved + TokenSpent
bat.clear_battle_players()
bat.set_winner(None)
war.revert_choice_ties(round.id)
if any(
bat.player_1_id is not None or bat.player_2_id is not None
@ -58,6 +62,7 @@ class Pairing:
raise RequiresConfirmation(
"Battle(s) already have player(s) assigned for this round.\n"
"Battle players will be cleared.\n"
"Choice tokens and tie-breaks will be deleted.\n"
"Do you want to continue?",
action=cleanup,
)
@ -90,20 +95,20 @@ class Pairing:
campaign.war_to_campaign_part_id(pid) for pid in group
]
Pairing._run_phase(
war,
round,
remaining,
sector_to_battle,
resolve_ties_callback,
war=war,
round=round,
remaining=remaining,
sector_to_battle=sector_to_battle,
resolve_ties_callback=resolve_ties_callback,
use_priority=True,
score_value=score_value,
)
Pairing._run_phase(
war,
round,
remaining,
sector_to_battle,
resolve_ties_callback,
war=war,
round=round,
remaining=remaining,
sector_to_battle=sector_to_battle,
resolve_ties_callback=resolve_ties_callback,
use_priority=False,
score_value=score_value,
)
@ -133,13 +138,13 @@ class Pairing:
if places <= 0:
continue
winners = Pairing._resolve_sector_allocation(
war,
round,
sector_id,
participants,
places,
resolve_ties_callback,
score_value,
war=war,
round=round,
sector_id=sector_id,
participants=participants,
places=places,
resolve_ties_callback=resolve_ties_callback,
score_value=score_value,
)
for pid in winners:
battle.assign_participant(pid)
@ -187,26 +192,59 @@ class Pairing:
participants=[
campaign.campaign_to_war_part_id(pid) for pid in participants
],
score_value=score_value,
score_kind=ScoreKind.VP,
sector_id=sector_id,
)
# ---- resolve tie loop ----
tie_id = TieResolver.find_active_tie_id(war, context) or str(uuid4())
while not TieResolver.is_tie_resolved(war, context):
if not TieResolver.can_tie_be_resolved(war, context, context.participants):
active = TieResolver.get_active_participants(
war, context, context.participants
)
if len(active) <= 1:
break
current_context = TieContext(
context_type=context.context_type,
context_id=context.context_id,
participants=active,
score_value=context.score_value,
score_kind=context.score_kind,
sector_id=context.sector_id,
)
if not TieResolver.can_tie_be_resolved(
war, context, current_context.participants
):
war.events.append(
TieResolved(
None,
context.context_type,
context.context_id,
participants,
tie_id=tie_id,
score_value=score_value,
sector_id=sector_id,
)
)
break
bids_map = resolve_ties_callback(war, [context])
bids = bids_map[context.key()]
TieResolver.apply_bids(war, context, bids)
TieResolver.resolve_tie_state(war, context, bids)
bids_map = resolve_ties_callback(war, [current_context])
bids = bids_map[current_context.key()]
# 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,
participants=context.participants,
tie_id=tie_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
break
TieResolver.apply_bids(war, context, tie_id, bids)
TieResolver.resolve_tie_state(war, context, tie_id, bids)
ranked_groups = TieResolver.rank_by_tokens(
war,
context,
@ -215,6 +253,8 @@ class Pairing:
ordered: List[str] = []
for group in ranked_groups:
shuffled_group = list(group)
# TODO improve tie break with history parsing
# TODO avoid rematch
random.shuffle(shuffled_group)
ordered.extend(
campaign.war_to_campaign_part_id(pid) for pid in shuffled_group

34
src/warchron/model/round.py
View file

@ -1,7 +1,9 @@
from __future__ import annotations
from uuid import uuid4
from typing import Any, Dict, List
from typing import Any, Dict, List, TYPE_CHECKING
if TYPE_CHECKING:
from warchron.model.campaign import Campaign
from warchron.model.exception import ForbiddenOperation
from warchron.model.choice import Choice
from warchron.model.battle import Battle
@ -13,6 +15,7 @@ class Round:
self.choices: Dict[str, Choice] = {}
self.battles: Dict[str, Battle] = {}
self.is_over: bool = False
self._campaign: Campaign | None = None # private link
def set_id(self, new_id: str) -> None:
self.id = new_id
@ -60,6 +63,7 @@ class Round:
def create_choice(self, participant_id: str) -> Choice:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create choice in a closed round.")
if participant_id not in self.choices:
choice = Choice(
@ -78,23 +82,38 @@ class Round:
comment: str | None,
) -> None:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't update choice in a closed round.")
# TODO prevent if battles already assigned
choice = self.get_choice(participant_id)
if choice:
choice.set_priority(priority_sector_id)
choice.set_secondary(secondary_sector_id)
choice.set_comment(comment)
# FIXME remove corresponding InfluenceSpent and TieResolved
def clear_sector_references(self, sector_id: str) -> None:
for choice in self.choices.values():
trigger_revert_ties = False
if choice.priority_sector_id == sector_id:
choice.priority_sector_id = None
trigger_revert_ties = True
if choice.secondary_sector_id == sector_id:
choice.secondary_sector_id = None
trigger_revert_ties = True
if trigger_revert_ties:
if self._campaign and self._campaign._war:
self._campaign._war.revert_choice_ties(self.id, sector_id=sector_id)
def remove_choice(self, participant_id: str) -> None:
if self.is_over:
# TODO catch me if you can (inner)
raise ForbiddenOperation("Can't remove choice in a closed round.")
# TODO prevent if battles already assigned
if self._campaign and self._campaign._war:
self._campaign._war.revert_choice_ties(
self.id, participants=[participant_id]
)
del self.choices[participant_id]
# Battle methods
@ -126,6 +145,7 @@ class Round:
def create_battle(self, sector_id: str) -> Battle:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create battle in a closed round.")
if sector_id not in self.battles:
battle = Battle(sector_id=sector_id, player_1_id=None, player_2_id=None)
@ -143,8 +163,10 @@ class Round:
comment: str | None,
) -> None:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't update battle in a closed round.")
bat = self.get_battle(sector_id)
# TODO require confirmation if there was choice tie to clear it
if bat:
bat.set_player_1(player_1_id)
bat.set_player_2(player_2_id)
@ -155,17 +177,27 @@ class Round:
def clear_participant_references(self, participant_id: str) -> None:
for battle in self.battles.values():
trigger_revert_ties = False
if battle.player_1_id == participant_id:
battle.player_1_id = None
trigger_revert_ties = True
if battle.player_2_id == participant_id:
battle.player_2_id = None
trigger_revert_ties = True
if battle.winner_id == participant_id:
battle.winner_id = None
if trigger_revert_ties:
if self._campaign and self._campaign._war:
self._campaign._war.revert_battle_ties(battle.sector_id)
def remove_battle(self, sector_id: str) -> None:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't remove battle in a closed round.")
bat = self.battles[sector_id]
if bat and bat.is_finished():
# TODO catch me if you can
raise ForbiddenOperation("Can't remove finished battle.")
if self._campaign and self._campaign._war:
self._campaign._war.revert_battle_ties(sector_id)
del self.battles[sector_id]

174
src/warchron/model/tie_manager.py
View file

@ -1,5 +1,5 @@
from typing import List, Dict, DefaultDict
from dataclasses import dataclass, field
from typing import List, Dict, DefaultDict, Tuple
from dataclasses import dataclass
from collections import defaultdict
from warchron.constants import ContextType, ScoreKind
@ -13,41 +13,70 @@ 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]
score_value: int | None = None
score_kind: ScoreKind | None = None
objective_id: str | None = None
sector_id: str | None = None
def key(self) -> tuple[str, str, int | None]:
return (self.context_type, self.context_id, self.score_value)
def key(self) -> Tuple[str, str, int | None, str | None, str | None]:
return (
self.context_type,
self.context_id,
self.score_value,
self.objective_id,
self.sector_id,
)
class TieResolver:
@staticmethod
def find_active_tie_id(
war: War,
context: TieContext,
) -> str | None:
for ev in reversed(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.sector_id == context.sector_id
and ev.participant_id in context.participants
):
return ev.tie_id
return None
@staticmethod
def find_battle_ties(war: War, round_id: str) -> List[TieContext]:
round = war.get_round(round_id)
campaign = war.get_campaign_by_round(round_id)
ties = []
for battle in round.battles.values():
if not battle.is_draw():
continue
context: TieContext = TieContext(
ContextType.BATTLE,
battle.sector_id,
)
if TieResolver.is_tie_resolved(war, context):
continue
if campaign is None:
raise DomainError("No campaign for this battle tie")
if battle.player_1_id is None or battle.player_2_id is None:
raise DomainError("Missing player(s) in this battle context.")
p1_id = campaign.campaign_to_war_part_id(battle.player_1_id)
p2_id = campaign.campaign_to_war_part_id(battle.player_2_id)
if not battle.is_draw():
continue
context: TieContext = TieContext(
ContextType.BATTLE, battle.sector_id, [p1_id, p2_id]
)
if TieResolver.is_tie_resolved(war, context):
continue
tie_id = TieResolver.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, [p1_id, p2_id]):
war.events.append(
TieResolved(None, ContextType.BATTLE, battle.sector_id)
TieResolved(
participant_id=None,
context_type=ContextType.BATTLE,
context_id=battle.sector_id,
participants=[p1_id, p2_id],
tie_id=tie_id,
)
)
continue
ties.append(
@ -80,13 +109,16 @@ class TieResolver:
)
if TieResolver.is_tie_resolved(war, context):
continue
tie_id = TieResolver.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, participants):
war.events.append(
TieResolved(
None,
ContextType.CAMPAIGN,
campaign_id,
score_value,
participant_id=None,
context_type=ContextType.CAMPAIGN,
context_id=campaign_id,
participants=participants,
tie_id=tie_id,
score_value=score_value,
)
)
continue
@ -103,9 +135,7 @@ class TieResolver:
@staticmethod
def find_campaign_objective_ties(
war: War,
campaign_id: str,
objective_id: str,
war: War, campaign_id: str, objective_id: str
) -> List[TieContext]:
scores = ScoreService.compute_scores(
war,
@ -131,6 +161,7 @@ class TieResolver:
)
if TieResolver.is_tie_resolved(war, context):
continue
tie_id = TieResolver.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(
war,
context,
@ -138,7 +169,13 @@ class TieResolver:
):
war.events.append(
TieResolved(
None, ContextType.CAMPAIGN, context_id, np_value, objective_id
participant_id=None,
context_type=ContextType.CAMPAIGN,
context_id=context_id,
participants=participants,
tie_id=tie_id,
score_value=np_value,
objective_id=objective_id,
)
)
continue
@ -181,9 +218,17 @@ class TieResolver:
)
if TieResolver.is_tie_resolved(war, context):
continue
tie_id = TieResolver.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, group):
war.events.append(
TieResolved(None, ContextType.WAR, war.id, score_value)
TieResolved(
participant_id=None,
context_type=ContextType.WAR,
context_id=war.id,
participants=group,
tie_id=tie_id,
score_value=score_value,
)
)
continue
ties.append(
@ -198,10 +243,7 @@ class TieResolver:
return ties
@staticmethod
def find_war_objective_ties(
war: War,
objective_id: str,
) -> List[TieContext]:
def find_war_objective_ties(war: War, objective_id: str) -> List[TieContext]:
from warchron.model.result_checker import ResultChecker
scores = ScoreService.compute_scores(
@ -235,13 +277,22 @@ class TieResolver:
context,
):
continue
tie_id = TieResolver.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(
war,
context,
group,
):
war.events.append(
TieResolved(None, ContextType.WAR, war.id, np_value, objective_id)
TieResolved(
participant_id=None,
context_type=ContextType.WAR,
context_id=war.id,
participants=group,
tie_id=tie_id,
score_value=np_value,
objective_id=objective_id,
)
)
continue
ties.append(
@ -260,6 +311,7 @@ class TieResolver:
def apply_bids(
war: War,
context: TieContext,
tie_id: str,
bids: Dict[str, bool], # war_participant_id -> spend?
) -> None:
for war_part_id, spend in bids.items():
@ -273,6 +325,7 @@ class TieResolver:
amount=1,
context_type=context.context_type,
context_id=context.context_id,
tie_id=tie_id,
objective_id=context.objective_id,
)
)
@ -287,12 +340,7 @@ class TieResolver:
for ev in war.events
if not (
(
isinstance(ev, InfluenceSpent)
and ev.context_type == context.context_type
and ev.context_id == context.context_id
)
or (
isinstance(ev, TieResolved)
(isinstance(ev, InfluenceSpent) or isinstance(ev, TieResolved))
and ev.context_type == context.context_type
and ev.context_id == context.context_id
)
@ -356,25 +404,9 @@ class TieResolver:
def resolve_tie_state(
war: War,
context: TieContext,
bids: dict[str, bool] | None = None,
tie_id: str,
bids: Dict[str, bool] | None = None,
) -> None:
active = TieResolver.get_active_participants(
war,
context,
context.participants,
)
# confirmed draw if none had bid
if not active:
war.events.append(
TieResolved(
None,
context.context_type,
context.context_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
return
# confirmed draw if current bids are 0
if bids is not None and not any(bids.values()):
war.events.append(
@ -382,12 +414,13 @@ class TieResolver:
None,
context.context_type,
context.context_id,
participants=context.participants,
tie_id=tie_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
return
# else rank_by_tokens
groups = TieResolver.rank_by_tokens(war, context, context.participants)
if len(groups[0]) == 1:
war.events.append(
@ -395,8 +428,10 @@ class TieResolver:
groups[0][0],
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
participants=context.participants,
tie_id=tie_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
return
@ -427,11 +462,24 @@ class TieResolver:
@staticmethod
def is_tie_resolved(war: War, context: TieContext) -> 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
for ev in war.events
)
for ev in war.events:
if not isinstance(ev, TieResolved):
continue
if ev.context_type != context.context_type:
continue
if ev.context_id != context.context_id:
continue
if (
context.score_value is not None
and ev.score_value != context.score_value
):
continue
if (
context.objective_id is not None
and ev.objective_id != context.objective_id
):
continue
if context.sector_id is not None and ev.sector_id != context.sector_id:
continue
return True
return False

61
src/warchron/model/war.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from uuid import uuid4
from datetime import datetime
from typing import Any, Dict, List
from typing import Any, Dict, List, Set
from warchron.constants import ContextType
from warchron.model.war_event import (
@ -62,7 +62,7 @@ class War:
if self.is_over:
raise ForbiddenOperation("Can't set influence token of a closed war.")
def cleanup_token_and_tie() -> None:
def remove_token_and_draw_tie() -> None:
new_events: List[WarEvent] = []
for ev in self.events:
if isinstance(ev, (InfluenceSpent, InfluenceGained)):
@ -73,11 +73,10 @@ class War:
self.events = new_events
self.influence_token = new_state
def reset_tie_break() -> None:
def remove_tiebreak_and_not_over() -> None:
new_events: List[WarEvent] = []
for ev in self.events:
if isinstance(ev, (TieResolved)):
# FIXME cancel TieResolved + TokenSpent
if ev.context_type == ContextType.BATTLE:
battle = self.get_battle(ev.context_id)
campaign = self.get_campaign_by_sector(battle.sector_id)
@ -86,6 +85,7 @@ class War:
elif ev.context_type == ContextType.CAMPAIGN:
campaign = self.get_campaign(ev.context_id)
campaign.is_over = False
# nothing specific to do with CHOICE (can retrigger pairing)
else:
new_events.append(ev)
self.events = new_events
@ -104,7 +104,7 @@ class War:
"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,
action=remove_token_and_draw_tie,
)
if new_state is True:
has_tie_resolved = any(isinstance(ev, TieResolved) for ev in self.events)
@ -120,7 +120,7 @@ class War:
"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,
action=remove_tiebreak_and_not_over,
)
def set_state(self, new_state: bool) -> None:
@ -216,8 +216,8 @@ class War:
# War participant methods
def get_all_war_participants_ids(self) -> set[str]:
return set(self.participants.keys())
def get_all_war_participants_ids(self) -> List[str]:
return list(self.participants.keys())
def has_participant(self, participant_id: str) -> bool:
return participant_id in self.participants
@ -288,6 +288,7 @@ class War:
if month is None:
month = self.get_default_campaign_values()["month"]
campaign = Campaign(name, month)
campaign._war = self
self.campaigns.append(campaign)
return campaign
@ -550,3 +551,47 @@ class War:
if isinstance(e, InfluenceSpent) and e.participant_id == participant_id
)
return gained - spent
def get_events_by_ties_session(self, tie_id: str) -> List[WarEvent]:
return [ev for ev in self.events if ev.tie_id == tie_id]
def remove_ties_session(self, tie_id: str) -> None:
self.events = [ev for ev in self.events if ev.tie_id != tie_id]
def revert_choice_ties(
self,
round_id: str,
*,
sector_id: str | None = None,
participants: List[str] | None = None,
) -> None:
removed_ties: Set[str] = set()
for ev in self.events:
if (
isinstance(ev, TieResolved)
and ev.context_type == ContextType.CHOICE
and ev.context_id == round_id
):
if (
sector_id is None
or ev.sector_id == sector_id
or participants is None
or any(p in ev.participants for p in participants)
):
if ev.tie_id:
removed_ties.add(ev.tie_id)
self.events = [ev for ev in self.events if ev.tie_id not in removed_ties]
def revert_battle_ties(self, sector_id: str) -> None:
removed_ties = {
ev.tie_id
for ev in self.events
if isinstance(ev, TieResolved)
and ev.context_type == ContextType.BATTLE
and ev.context_id == sector_id
and ev.tie_id
}
self.events = [ev for ev in self.events if ev.tie_id not in removed_ties]
def revert_tie(self, tie_id: str) -> None:
self.events = [ev for ev in self.events if ev.tie_id != tie_id]

24
src/warchron/model/war_event.py
View file

@ -1,5 +1,5 @@
from __future__ import annotations
from typing import Dict, Any, TypeVar, Type, cast
from typing import Dict, Any, TypeVar, Type, cast, List
from datetime import datetime
from uuid import uuid4
@ -22,12 +22,14 @@ class WarEvent:
participant_id: str | None,
context_type: str,
context_id: str,
tie_id: str | None = None,
):
self.id: str = str(uuid4())
self.participant_id: str | None = participant_id
self.context_type = context_type # battle, round, campaign, war
self.context_id = context_id
self.timestamp: datetime = datetime.now()
self.tie_id = tie_id
def set_id(self, new_id: str) -> None:
self.id = new_id
@ -46,6 +48,7 @@ class WarEvent:
"context_type": self.context_type,
"context_id": self.context_id,
"timestamp": self.timestamp.isoformat(),
"tie_id": self.tie_id,
}
@classmethod
@ -55,6 +58,7 @@ class WarEvent:
ev.context_type = data["context_type"]
ev.context_id = data["context_id"]
ev.timestamp = datetime.fromisoformat(data["timestamp"])
ev.tie_id = data.get("tie_id")
return ev
@staticmethod
@ -72,14 +76,17 @@ class TieResolved(WarEvent):
def __init__(
self,
participant_id: str | None,
participant_id: str | None, # winner_id or None if confirmed draw
context_type: str,
context_id: str,
participants: List[str],
tie_id: str | None = None, # None if draw without tie-break
score_value: int | None = None,
objective_id: str | None = None,
sector_id: str | None = None,
):
super().__init__(participant_id, context_type, context_id)
super().__init__(participant_id, context_type, context_id, tie_id)
self.participants = participants
self.score_value = score_value
self.objective_id = objective_id
self.sector_id = sector_id
@ -88,6 +95,7 @@ class TieResolved(WarEvent):
d = super().toDict()
d.update(
{
"participants": self.participants,
"score_value": self.score_value or None,
"objective_id": self.objective_id or None,
"sector_id": self.sector_id or None,
@ -101,6 +109,8 @@ class TieResolved(WarEvent):
JsonHelper.none_if_empty(data["participant_id"]),
data["context_type"],
data["context_id"],
data["participants"],
data["tie_id"],
JsonHelper.none_if_empty(data["score_value"]),
JsonHelper.none_if_empty(data["objective_id"]),
JsonHelper.none_if_empty(data["sector_id"]),
@ -156,11 +166,14 @@ class InfluenceSpent(WarEvent):
amount: int,
context_type: str,
context_id: str,
tie_id: str,
objective_id: str | None = None,
sector_id: str | None = None,
):
super().__init__(participant_id, context_type, context_id)
super().__init__(participant_id, context_type, context_id, tie_id)
self.amount = amount
self.objective_id = objective_id
self.sector_id = sector_id
def toDict(self) -> Dict[str, Any]:
d = super().toDict()
@ -168,6 +181,7 @@ class InfluenceSpent(WarEvent):
{
"amount": self.amount,
"objective_id": self.objective_id,
"sector_id": self.sector_id or None,
}
)
return d
@ -179,6 +193,8 @@ class InfluenceSpent(WarEvent):
int(data["amount"]),
data["context_type"],
data["context_id"],
data["tie_id"],
JsonHelper.none_if_empty(data["objective_id"]),
JsonHelper.none_if_empty(data["sector_id"]),
)
return cls._base_fromDict(ev, data)

5
test_data/example.json
View file

@ -338,7 +338,12 @@
"participant_id": null,
"context_type": "battle",
"context_id": "79accf7c-2d93-4ac3-b747-e7092bfe3feb",
"participants": [
"602e2eaf-297e-490b-b0e9-efec818e466a",
"1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de"
],
"timestamp": "2026-02-26T16:11:44.346337",
"tie_id": null,
"score_value": null,
"objective_id": null,
"sector_id": null