Compare commits

...

10 commits

Author SHA1 Message Date
Maxime Réaux
b7a35f6712 fix tie resolve group order + battle draw token icon 2026-03-20 10:37:35 +01:00
Maxime Réaux
719b0128ed fix refresh choice & battle tables + refactor loops 2026-03-20 08:54:08 +01:00
Maxime Réaux
e7e2de9d0a display choice comment 2026-03-19 15:28:30 +01:00
Maxime Réaux
956fa63c0b refacto file and class names 2026-03-19 15:23:50 +01:00
Maxime Réaux
3d0d7874e3 unifomise tiebreak icons + refacto presenter 2026-03-19 15:10:55 +01:00
Maxime Réaux
0081e52e9a fix uncaucht choice/battle exceptions 2026-03-19 12:05:27 +01:00
Maxime Réaux
f5ad45f671 raise exception on choice/battle constrained update/delete 2026-03-19 11:25:40 +01:00
Maxime Réaux
4396b15c3a display pairing results in choice table 2026-03-19 09:02:22 +01:00
Maxime Réaux
9e602e8ca4 tie dialog title with context details 2026-03-18 14:30:57 +01:00
Maxime Réaux
aa75a5b84f avoid choic tiebreak when sector places are enough for remaining active tied participants 2026-03-18 10:48:29 +01:00
23 changed files with 932 additions and 536 deletions

View file

@ -22,6 +22,7 @@ class IconName(StrEnum):
PAIRING = auto() PAIRING = auto()
DRAW = auto() DRAW = auto()
TIEBREAK = auto() TIEBREAK = auto()
DRAWTOKEN = auto()
DELETE = auto() DELETE = auto()
SAVE_AS = auto() SAVE_AS = auto()
SAVE = auto() SAVE = auto()
@ -49,7 +50,8 @@ class IconName(StrEnum):
NP1ST = auto() NP1ST = auto()
NP2ND = auto() NP2ND = auto()
NP3RD = auto() NP3RD = auto()
TIEBREAK_TOKEN = auto() WINTOKEN = auto()
TIEBREAKTOKEN = auto()
VP1STDRAW = auto() VP1STDRAW = auto()
VP1STBREAK = auto() VP1STBREAK = auto()
VP1STTIEDRAW = auto() VP1STTIEDRAW = auto()
@ -71,6 +73,9 @@ class IconName(StrEnum):
NP3RDDRAW = auto() NP3RDDRAW = auto()
NP3RDBREAK = auto() NP3RDBREAK = auto()
NP3RDTIEDRAW = auto() NP3RDTIEDRAW = auto()
ALLOCATED = auto()
ALLOCATEDTOKEN = auto()
FALLBACK = auto()
VP_RANK_TO_ICON = { VP_RANK_TO_ICON = {
@ -125,6 +130,8 @@ class Icons:
IconName.NP2ND: "medal-silver.png", IconName.NP2ND: "medal-silver.png",
IconName.NP3RD: "medal-bronze.png", IconName.NP3RD: "medal-bronze.png",
IconName.WARCHRONBACK: "warchron_background.png", IconName.WARCHRONBACK: "warchron_background.png",
IconName.ALLOCATED: "map.png",
IconName.FALLBACK: "cross-script.png",
} }
@classmethod @classmethod
@ -138,11 +145,21 @@ class Icons:
def get_pixmap(cls, name: IconName) -> QPixmap: def get_pixmap(cls, name: IconName) -> QPixmap:
if name in cls._pixmap_cache: if name in cls._pixmap_cache:
return cls._pixmap_cache[name] return cls._pixmap_cache[name]
if name == IconName.TIEBREAK_TOKEN: if name == IconName.TIEBREAKTOKEN:
pix = cls._compose( pix = cls._compose(
cls.get_pixmap(IconName.TIEBREAK), cls.get_pixmap(IconName.TIEBREAK),
cls.get_pixmap(IconName.TOKEN), cls.get_pixmap(IconName.TOKEN),
) )
elif name == IconName.DRAWTOKEN:
pix = cls._compose(
cls.get_pixmap(IconName.DRAW),
cls.get_pixmap(IconName.TOKEN),
)
elif name == IconName.WINTOKEN:
pix = cls._compose(
cls.get_pixmap(IconName.WIN),
cls.get_pixmap(IconName.TOKEN),
)
elif name == IconName.VP1STDRAW: elif name == IconName.VP1STDRAW:
pix = cls._compose( pix = cls._compose(
cls.get_pixmap(IconName.VP1ST), cls.get_pixmap(IconName.VP1ST),
@ -255,6 +272,11 @@ class Icons:
cls.get_pixmap(IconName.DRAW), cls.get_pixmap(IconName.DRAW),
cls.get_pixmap(IconName.TOKEN), cls.get_pixmap(IconName.TOKEN),
) )
elif name == IconName.ALLOCATEDTOKEN:
pix = cls._compose(
cls.get_pixmap(IconName.ALLOCATED),
cls.get_pixmap(IconName.TOKEN),
)
else: else:
path = RESOURCES_DIR / cls._paths[name] path = RESOURCES_DIR / cls._paths[name]
pix = QPixmap(path.as_posix()) pix = QPixmap(path.as_posix())
@ -313,3 +335,16 @@ class ContextType(StrEnum):
class ScoreKind(Enum): class ScoreKind(Enum):
VP = auto() VP = auto()
NP = auto() NP = auto()
class ChoiceStatus(StrEnum):
NONE = auto()
TOKEN = auto()
ALLOCATED = auto()
ALLOCATEDTOKEN = auto()
class AllocationType(Enum):
PRIORITY = auto()
SECONDARY = auto()
FALLBACK = auto()

View file

@ -357,12 +357,20 @@ class AppController:
except RequiresConfirmation as e: except RequiresConfirmation as e:
reply = QMessageBox.question( reply = QMessageBox.question(
self.view, self.view,
"Confirm deletion", "Confirm update",
str(e), str(e),
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
) )
if reply == QMessageBox.StandardButton.Yes: if reply == QMessageBox.StandardButton.Yes:
e.action() try:
e.action()
except DomainError as inner:
QMessageBox.warning(
self.view,
"Update forbidden",
str(inner),
)
return
else: else:
return return
self.is_dirty = True self.is_dirty = True

View file

@ -19,10 +19,11 @@ from warchron.model.war import War
from warchron.model.campaign import Campaign from warchron.model.campaign import Campaign
from warchron.model.campaign_participant import CampaignParticipant from warchron.model.campaign_participant import CampaignParticipant
from warchron.model.sector import Sector from warchron.model.sector import Sector
from warchron.model.tie_manager import TieContext, TieResolver from warchron.model.tiebreaking import TieContext, TieBreaker
from warchron.model.score_service import ScoreService from warchron.model.scoring import ScoreComputer
from warchron.controller.closure_workflow import CampaignClosureWorkflow from warchron.controller.workflows import CampaignClosureWorkflow
from warchron.controller.ranking_icon import RankingIcon
from warchron.controller.presenter import Presenter
from warchron.view.campaign_dialog import CampaignDialog from warchron.view.campaign_dialog import CampaignDialog
from warchron.view.campaign_participant_dialog import CampaignParticipantDialog from warchron.view.campaign_participant_dialog import CampaignParticipantDialog
from warchron.view.sector_dialog import SectorDialog from warchron.view.sector_dialog import SectorDialog
@ -52,16 +53,16 @@ class CampaignController:
for sect in sectors for sect in sectors
] ]
self.app.view.display_campaign_sectors(sectors_for_display) self.app.view.display_campaign_sectors(sectors_for_display)
scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id) scores = ScoreComputer.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
rows: List[CampaignParticipantScoreDTO] = [] rows: List[CampaignParticipantScoreDTO] = []
vp_icon_map: Dict[str, QIcon] = {} vp_icon_map: Dict[str, QIcon] = {}
objective_icon_maps: Dict[str, Dict[str, QIcon]] = {} objective_icon_maps: Dict[str, Dict[str, QIcon]] = {}
if camp.is_over: if camp.is_over:
vp_icon_map = RankingIcon.compute_icons( vp_icon_map = Presenter.compute_ranking_icons(
war, ContextType.CAMPAIGN, campaign_id, scores war, ContextType.CAMPAIGN, campaign_id, scores
) )
for obj in war.get_objectives_used_as_maj_or_min(): for obj in war.get_objectives_used_as_maj_or_min():
objective_icon_maps[obj.id] = RankingIcon.compute_icons( objective_icon_maps[obj.id] = Presenter.compute_ranking_icons(
war, war,
ContextType.CAMPAIGN, ContextType.CAMPAIGN,
camp.id, camp.id,
@ -170,32 +171,25 @@ class CampaignController:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]: ) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {} bids_map = {}
for ctx in contexts: for ctx in contexts:
active = TieResolver.get_active_participants(war, ctx, ctx.participants) active = TieBreaker.get_active_participants(war, ctx, ctx.participants)
players = [ players = [
ParticipantOption(id=pid, name=self.app.model.get_participant_name(pid)) ParticipantOption(id=pid, name=self.app.model.get_participant_name(pid))
for pid in active for pid in active
] ]
counters = [war.get_influence_tokens(pid) for pid in active] counters = [war.get_influence_tokens(pid) for pid in active]
data = Presenter.build_dialog_data(
war, ctx, campaign=war.get_campaign(ctx.context_id)
)
dialog = TieDialog( dialog = TieDialog(
parent=self.app.view, parent=self.app.view,
players=players, players=players,
counters=counters, counters=counters,
context_type=ctx.context_type, context_type=ctx.context_type,
context_id=ctx.context_id, context_id=ctx.context_id,
context_name=None, context_name=data.title,
) )
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=f"Objective tie: {objective.name}",
)
if not dialog.exec(): if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx) TieBreaker.cancel_tie_break(war, ctx)
raise AbortedOperation("Tie resolution cancelled") raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids() bids_map[ctx.key()] = dialog.get_bids()
return bids_map return bids_map

View file

@ -1,117 +0,0 @@
from typing import TYPE_CHECKING
from uuid import uuid4
if TYPE_CHECKING:
from warchron.controller.app_controller import AppController
from warchron.model.war import War
from warchron.model.campaign import Campaign
from warchron.model.round import Round
from warchron.model.closure_service import ClosureService
from warchron.model.tie_manager import TieResolver
from warchron.model.pairing import Pairing
class ClosureWorkflow:
def __init__(self, controller: "AppController"):
self.app = controller
class RoundClosureWorkflow(ClosureWorkflow):
def start(self, war: War, campaign: Campaign, round: Round) -> None:
ClosureService.check_round_closable(round)
ties = TieResolver.find_battle_ties(war, round.id)
while ties:
bids_map = self.app.rounds.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
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)
ClosureService.finalize_round(round)
class CampaignClosureWorkflow(ClosureWorkflow):
def start(self, war: War, campaign: Campaign) -> None:
ClosureService.check_campaign_closable(campaign)
ties = TieResolver.find_campaign_ties(war, campaign.id)
while ties:
bids_map = self.app.campaigns.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
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
ties = TieResolver.find_campaign_objective_ties(
war,
campaign.id,
objective_id,
)
while ties:
bids_map = self.app.campaigns.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
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,
objective_id,
)
ClosureService.finalize_campaign(campaign)
class WarClosureWorkflow(ClosureWorkflow):
def start(self, war: War) -> None:
ClosureService.check_war_closable(war)
ties = TieResolver.find_war_ties(war)
while ties:
bids_map = self.app.wars.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
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
ties = TieResolver.find_war_objective_ties(
war,
objective_id,
)
while ties:
bids_map = self.app.wars.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
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,
)
ClosureService.finalize_war(war)
class RoundPairingWorkflow:
def __init__(self, controller: "AppController"):
self.app = controller
def start(self, war: War, round: Round) -> None:
Pairing.check_round_pairable(war, round)
Pairing.assign_battles_to_participants(
war,
round,
resolve_ties_callback=self.app.rounds.resolve_ties,
)

View file

@ -88,6 +88,9 @@ class ChoiceDTO:
priority_sector: str | None priority_sector: str | None
secondary_sector: str | None secondary_sector: str | None
comment: str | None comment: str | None
priority_icon: QIcon | None = None
secondary_icon: QIcon | None = None
fallback_icon: QIcon | None = None
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@ -103,8 +106,6 @@ class BattleDTO:
state_icon: QIcon | None state_icon: QIcon | None
player1_icon: QIcon | None player1_icon: QIcon | None
player2_icon: QIcon | None player2_icon: QIcon | None
player1_tooltip: str | None = None
player2_tooltip: str | None = None
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@ -132,3 +133,8 @@ class WarParticipantScoreDTO:
tokens: int tokens: int
rank_icon: QIcon | None = None rank_icon: QIcon | None = None
objective_icons: Dict[str, QIcon] = field(default_factory=dict) objective_icons: Dict[str, QIcon] = field(default_factory=dict)
@dataclass
class TieDialogData:
title: str

View file

@ -0,0 +1,178 @@
from typing import Dict
from PyQt6.QtGui import QIcon
from warchron.constants import (
ContextType,
Icons,
IconName,
VP_RANK_TO_ICON,
NP_RANK_TO_ICON,
ScoreKind,
)
from warchron.controller.dtos import TieDialogData
from warchron.model.tiebreaking import TieContext, TieBreaker
from warchron.model.war import War
from warchron.model.campaign import Campaign
from warchron.model.round import Round
from warchron.model.scoring import ParticipantScore
from warchron.model.checking import ResultChecker
from warchron.model.exception import DomainError
class Presenter:
@staticmethod
def build_dialog_data(
war: War,
ctx: TieContext,
campaign: Campaign | None = None,
round: Round | None = None,
) -> TieDialogData:
if ctx.context_type == ContextType.WAR:
if ctx.objective_id:
obj = war.objectives[ctx.objective_id]
return TieDialogData(f"War objective tie — {obj.name}")
return TieDialogData("War tie")
if ctx.context_type == ContextType.CAMPAIGN:
if ctx.objective_id:
obj = war.objectives[ctx.objective_id]
return TieDialogData(f"Campaign objective tie — {obj.name}")
return TieDialogData("Campaign tie")
if ctx.context_type == ContextType.BATTLE:
if campaign:
sector = campaign.sectors[ctx.context_id]
return TieDialogData(f"Battle tie — {sector.name}")
return TieDialogData("Battle tie")
if ctx.context_type == ContextType.CHOICE:
if ctx.sector_id and campaign and round:
sector = campaign.sectors[ctx.sector_id]
kind = Presenter._choice_kind(round, ctx)
return TieDialogData(f"Choice tie — {sector.name} ({kind})")
return TieDialogData("Choice tie")
return TieDialogData("Tie")
@staticmethod
def _choice_kind(round: Round, ctx: TieContext) -> str:
for pid in ctx.participants:
camp_pid = round.campaign.war_to_campaign_part_id(pid)
choice = round.choices.get(camp_pid)
if not choice:
continue
if choice.priority_sector_id == ctx.sector_id:
return "priority"
if choice.secondary_sector_id == ctx.sector_id:
return "secondary"
return "choice"
@staticmethod
def compute_ranking_icons(
war: War,
context_type: ContextType,
context_id: str,
scores: Dict[str, ParticipantScore],
*,
objective_id: str | None = None,
) -> Dict[str, QIcon]:
if objective_id is None:
def value_getter(score: ParticipantScore) -> int:
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,
)
icon_map: Dict[str, QIcon] = {}
for rank, group, token_map in ranking:
if objective_id and rank not in icon_ranking:
continue
base_icon = icon_ranking.get(
rank, IconName.VPNTH if objective_id is None else None
)
if base_icon is None:
continue
value = value_getter(scores[group[0]])
original_group_size = sum(
1 for s in scores.values() if value_getter(s) == value
)
for pid in group:
spent = token_map.get(pid, 0)
if original_group_size == 1:
icon_name = base_icon
elif len(group) == 1:
if spent > 0:
icon_name = getattr(IconName, f"{base_icon.name}BREAK")
else:
icon_name = base_icon
else:
if spent > 0:
icon_name = getattr(IconName, f"{base_icon.name}TIEDRAW")
else:
icon_name = getattr(IconName, f"{base_icon.name}DRAW")
icon_map[pid] = QIcon(Icons.get_pixmap(icon_name))
return icon_map
@staticmethod
def compute_battle_icons(
war: War,
round_id: str,
battle_id: str,
) -> tuple[QIcon | None, QIcon | None]:
battle = war.get_battle(battle_id)
if not (battle.player_1_id and battle.player_2_id and battle.is_finished()):
return None, None
campaign = war.get_campaign_by_sector(battle.sector_id)
if not campaign:
raise DomainError("No campaign found for this battle")
base_winner = None
if battle.winner_id is not None:
base_winner = campaign.campaign_to_war_part_id(battle.winner_id)
winner_id = ResultChecker.get_effective_winner_id(
war,
ContextType.BATTLE,
battle.sector_id,
base_winner,
)
def compute_icon(player: str) -> QIcon | None:
base_icon: IconName | None = None
if winner_id is None:
base_icon = IconName.DRAW
elif campaign.war_to_campaign_part_id(winner_id) == player:
base_icon = IconName.WIN
elif battle.is_draw():
base_icon = IconName.TIEBREAK
if base_icon is None:
return None
spent = TieBreaker.participant_spent_token(
war,
ContextType.BATTLE,
battle.sector_id,
None,
campaign.campaign_to_war_part_id(player),
)
icon_name = (
getattr(IconName, f"{base_icon.name}TOKEN") if spent else base_icon
)
return QIcon(Icons.get_pixmap(icon_name))
return (
compute_icon(battle.player_1_id),
compute_icon(battle.player_2_id),
)

View file

@ -1,80 +0,0 @@
from typing import Dict
from PyQt6.QtGui import QIcon
from warchron.constants import (
ContextType,
Icons,
IconName,
VP_RANK_TO_ICON,
NP_RANK_TO_ICON,
ScoreKind,
)
from warchron.model.war import War
from warchron.model.score_service import ParticipantScore
from warchron.model.result_checker import ResultChecker
class RankingIcon:
@staticmethod
def compute_icons(
war: War,
context_type: ContextType,
context_id: str,
scores: Dict[str, ParticipantScore],
*,
objective_id: str | None = None,
) -> Dict[str, QIcon]:
if objective_id is None:
def value_getter(score: ParticipantScore) -> int:
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,
)
icon_map: Dict[str, QIcon] = {}
for rank, group, token_map in ranking:
if objective_id and rank not in icon_ranking:
continue
base_icon = icon_ranking.get(
rank, IconName.VPNTH if objective_id is None else None
)
if base_icon is None:
continue
value = value_getter(scores[group[0]])
original_group_size = sum(
1 for s in scores.values() if value_getter(s) == value
)
for pid in group:
spent = token_map.get(pid, 0)
if original_group_size == 1:
icon_name = base_icon
elif len(group) == 1:
if spent > 0:
icon_name = getattr(IconName, f"{base_icon.name}BREAK")
else:
icon_name = base_icon
else:
if spent > 0:
icon_name = getattr(IconName, f"{base_icon.name}TIEDRAW")
else:
icon_name = getattr(IconName, f"{base_icon.name}DRAW")
icon_map[pid] = QIcon(Icons.get_pixmap(icon_name))
return icon_map

View file

@ -4,19 +4,27 @@ from PyQt6.QtWidgets import QDialog
from PyQt6.QtWidgets import QMessageBox from PyQt6.QtWidgets import QMessageBox
from PyQt6.QtGui import QIcon from PyQt6.QtGui import QIcon
from warchron.constants import ItemType, RefreshScope, Icons, IconName, ContextType from warchron.constants import (
ItemType,
RefreshScope,
Icons,
IconName,
ContextType,
ChoiceStatus,
)
from warchron.model.exception import ( from warchron.model.exception import (
AbortedOperation, AbortedOperation,
DomainError, DomainError,
RequiresConfirmation, RequiresConfirmation,
) )
from warchron.model.tie_manager import TieResolver, TieContext from warchron.model.tiebreaking import TieBreaker, TieContext
from warchron.model.result_checker import ResultChecker from warchron.model.pairing import Pairing
from warchron.model.round import Round from warchron.model.round import Round
from warchron.model.war import War from warchron.model.war import War
if TYPE_CHECKING: if TYPE_CHECKING:
from warchron.controller.app_controller import AppController from warchron.controller.app_controller import AppController
from warchron.model.campaign import Campaign
from warchron.controller.dtos import ( from warchron.controller.dtos import (
ParticipantOption, ParticipantOption,
@ -24,10 +32,11 @@ from warchron.controller.dtos import (
ChoiceDTO, ChoiceDTO,
BattleDTO, BattleDTO,
) )
from warchron.controller.closure_workflow import ( from warchron.controller.workflows import (
RoundClosureWorkflow, RoundClosureWorkflow,
RoundPairingWorkflow, RoundPairingWorkflow,
) )
from warchron.controller.presenter import Presenter
from warchron.view.choice_dialog import ChoiceDialog from warchron.view.choice_dialog import ChoiceDialog
from warchron.view.battle_dialog import BattleDialog from warchron.view.battle_dialog import BattleDialog
from warchron.view.tie_dialog import TieDialog from warchron.view.tie_dialog import TieDialog
@ -48,114 +57,117 @@ class RoundController:
for part in participants: for part in participants:
choice = rnd.get_choice(part.id) choice = rnd.get_choice(part.id)
if not choice: if not choice:
choice = self.app.model.create_choice( try:
round_id=rnd.id, participant_id=part.id choice = self.app.model.create_choice(
round_id=rnd.id, participant_id=part.id
)
except DomainError as e:
QMessageBox.warning(
self.app.view,
"Create forbidden",
str(e),
)
if choice:
priority_name = (
camp.get_sector_name(choice.priority_sector_id)
if choice.priority_sector_id is not None
else ""
) )
priority_name = ( secondary_name = (
camp.get_sector_name(choice.priority_sector_id) camp.get_sector_name(choice.secondary_sector_id)
if choice.priority_sector_id is not None if choice.secondary_sector_id is not None
else "" else ""
) )
secondary_name = ( priority_icon = None
camp.get_sector_name(choice.secondary_sector_id) secondary_icon = None
if choice.secondary_sector_id is not None fallback_icon = None
else "" alloc = Pairing.get_round_allocation(
) war,
choices_for_display.append( rnd,
ChoiceDTO( part.id,
id=choice.participant_id, )
participant_name=self.app.model.get_participant_name( if alloc.priority != ChoiceStatus.NONE:
part.war_participant_id priority_icon = QIcon(
), Icons.get_pixmap(IconName[alloc.priority.name])
priority_sector=priority_name, )
secondary_sector=secondary_name, if alloc.secondary != ChoiceStatus.NONE:
comment=choice.comment, secondary_icon = QIcon(
Icons.get_pixmap(IconName[alloc.secondary.name])
)
if alloc.fallback:
fallback_icon = QIcon(Icons.get_pixmap(IconName.FALLBACK))
choices_for_display.append(
ChoiceDTO(
id=choice.participant_id,
participant_name=self.app.model.get_participant_name(
part.war_participant_id
),
priority_sector=priority_name,
secondary_sector=secondary_name,
comment=choice.comment,
priority_icon=priority_icon,
secondary_icon=secondary_icon,
fallback_icon=fallback_icon,
)
) )
)
# TODO display allocated sectors and used token
self.app.view.display_round_choices(choices_for_display) self.app.view.display_round_choices(choices_for_display)
battles_for_display: List[BattleDTO] = [] battles_for_display: List[BattleDTO] = []
for sect in sectors: for sect in sectors:
battle = rnd.get_battle(sect.id) battle = rnd.get_battle(sect.id)
if not battle: if not battle:
battle = self.app.model.create_battle( try:
round_id=rnd.id, sector_id=sect.id battle = self.app.model.create_battle(
) round_id=rnd.id, sector_id=sect.id
state_icon = Icons.get(IconName.ONGOING) )
if battle.is_finished(): except DomainError as e:
state_icon = Icons.get(IconName.DONE) QMessageBox.warning(
if battle.player_1_id: self.app.view,
camp_part = camp.participants[battle.player_1_id] "Create forbidden",
player_1_name = self.app.model.get_participant_name( str(e),
camp_part.war_participant_id )
) if battle:
p1_id = battle.player_1_id state_icon = Icons.get(IconName.ONGOING)
else: if battle.is_finished():
player_1_name = "" state_icon = Icons.get(IconName.DONE)
if battle.player_2_id: if battle.player_1_id:
camp_part = camp.participants[battle.player_2_id] camp_part = camp.participants[battle.player_1_id]
player_2_name = self.app.model.get_participant_name( player_1_name = self.app.model.get_participant_name(
camp_part.war_participant_id camp_part.war_participant_id
) )
p2_id = battle.player_2_id else:
else: player_1_name = ""
player_2_name = "" if battle.player_2_id:
if battle.winner_id: camp_part = camp.participants[battle.player_2_id]
camp_part = camp.participants[battle.winner_id] player_2_name = self.app.model.get_participant_name(
winner_name = self.app.model.get_participant_name( camp_part.war_participant_id
camp_part.war_participant_id )
) else:
else: player_2_name = ""
winner_name = "" if battle.winner_id:
p1_icon = None camp_part = camp.participants[battle.winner_id]
p2_icon = None winner_name = self.app.model.get_participant_name(
p1_tooltip = None camp_part.war_participant_id
p2_tooltip = None )
if battle.is_draw(): else:
p1_icon = Icons.get(IconName.DRAW) winner_name = ""
p2_icon = Icons.get(IconName.DRAW) p1_icon, p2_icon = Presenter.compute_battle_icons(
context = TieContext( war, round_id, battle.sector_id
ContextType.BATTLE, )
battle.sector_id, battles_for_display.append(
[p1_id, p2_id], BattleDTO(
) id=battle.sector_id,
if TieResolver.was_tie_broken_by_tokens(war, context): sector_name=camp.get_sector_name(battle.sector_id),
effective_winner = ResultChecker.get_effective_winner_id( player_1=player_1_name,
war, ContextType.BATTLE, battle.sector_id, None player_2=player_2_name,
winner=winner_name,
score=battle.score,
victory_condition=battle.victory_condition,
comment=battle.comment,
state_icon=state_icon,
player1_icon=p1_icon,
player2_icon=p2_icon,
) )
p1_war = None
if battle.player_1_id is not None:
p1_war = camp.campaign_to_war_part_id(battle.player_1_id)
pixmap = Icons.get_pixmap(IconName.TIEBREAK_TOKEN)
if effective_winner == p1_war:
p1_icon = QIcon(pixmap)
p1_tooltip = "Won by tie-break"
else:
p2_icon = QIcon(pixmap)
p2_tooltip = "Won by tie-break"
elif battle.winner_id:
if battle.winner_id == battle.player_1_id:
p1_icon = Icons.get(IconName.WIN)
elif battle.winner_id == battle.player_2_id:
p2_icon = Icons.get(IconName.WIN)
battles_for_display.append(
BattleDTO(
id=battle.sector_id,
sector_name=camp.get_sector_name(battle.sector_id),
player_1=player_1_name,
player_2=player_2_name,
winner=winner_name,
score=battle.score,
victory_condition=battle.victory_condition,
comment=battle.comment,
state_icon=state_icon,
player1_icon=p1_icon,
player2_icon=p2_icon,
player1_tooltip=p1_tooltip,
player2_tooltip=p2_tooltip,
) )
)
self.app.view.display_round_battles(battles_for_display) self.app.view.display_round_battles(battles_for_display)
self.app.view.endRoundBtn.setEnabled(not rnd.is_over) self.app.view.endRoundBtn.setEnabled(not rnd.is_over)
@ -243,16 +255,29 @@ class RoundController:
for pid in ctx.participants for pid in ctx.participants
] ]
counters = [war.get_influence_tokens(pid) for pid in ctx.participants] counters = [war.get_influence_tokens(pid) for pid in ctx.participants]
# TODO display sector name for BATTLE or CHOICE round: Round | None = None
campaign: Campaign | None = None
if ctx.context_type == ContextType.BATTLE:
# context_id corresponds to battle.sector_id
campaign = war.get_campaign_by_sector(ctx.context_id)
if campaign:
round = campaign.get_round_by_battle(ctx.context_id)
if ctx.context_type == ContextType.CHOICE:
# context_id corresponds to round.id
campaign = war.get_campaign_by_round(ctx.context_id)
if campaign:
round = war.get_round(ctx.context_id)
data = Presenter.build_dialog_data(war, ctx, round=round, campaign=campaign)
dialog = TieDialog( dialog = TieDialog(
parent=self.app.view, parent=self.app.view,
players=players, players=players,
counters=counters, counters=counters,
context_type=ctx.context_type, context_type=ctx.context_type,
context_id=ctx.context_id, context_id=ctx.context_id,
context_name=data.title,
) )
if not dialog.exec(): if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx) TieBreaker.cancel_tie_break(war, ctx)
raise AbortedOperation("Tie resolution cancelled") raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids() bids_map[ctx.key()] = dialog.get_bids()
return bids_map return bids_map

View file

@ -24,10 +24,11 @@ from warchron.controller.dtos import (
from warchron.model.war import War from warchron.model.war import War
from warchron.model.war_participant import WarParticipant from warchron.model.war_participant import WarParticipant
from warchron.model.objective import Objective from warchron.model.objective import Objective
from warchron.model.tie_manager import TieContext, TieResolver from warchron.model.tiebreaking import TieContext, TieBreaker
from warchron.model.score_service import ScoreService from warchron.model.scoring import ScoreComputer
from warchron.controller.closure_workflow import WarClosureWorkflow from warchron.controller.workflows import WarClosureWorkflow
from warchron.controller.ranking_icon import RankingIcon
from warchron.controller.presenter import Presenter
from warchron.view.war_dialog import WarDialog from warchron.view.war_dialog import WarDialog
from warchron.view.objective_dialog import ObjectiveDialog from warchron.view.objective_dialog import ObjectiveDialog
from warchron.view.war_participant_dialog import WarParticipantDialog from warchron.view.war_participant_dialog import WarParticipantDialog
@ -56,16 +57,16 @@ class WarController:
ObjectiveDTO(id=obj.id, name=obj.name, description=obj.description) ObjectiveDTO(id=obj.id, name=obj.name, description=obj.description)
for obj in war.get_objectives_used_as_maj_or_min() for obj in war.get_objectives_used_as_maj_or_min()
] ]
scores = ScoreService.compute_scores(war, ContextType.WAR, war.id) scores = ScoreComputer.compute_scores(war, ContextType.WAR, war.id)
rows: List[WarParticipantScoreDTO] = [] rows: List[WarParticipantScoreDTO] = []
vp_icon_map: Dict[str, QIcon] = {} vp_icon_map: Dict[str, QIcon] = {}
objective_icon_maps: Dict[str, Dict[str, QIcon]] = {} objective_icon_maps: Dict[str, Dict[str, QIcon]] = {}
if war.is_over: if war.is_over:
vp_icon_map = RankingIcon.compute_icons( vp_icon_map = Presenter.compute_ranking_icons(
war, ContextType.WAR, war_id, scores war, ContextType.WAR, war_id, scores
) )
for obj in war.get_objectives_used_as_maj_or_min(): for obj in war.get_objectives_used_as_maj_or_min():
objective_icon_maps[obj.id] = RankingIcon.compute_icons( objective_icon_maps[obj.id] = Presenter.compute_ranking_icons(
war, war,
ContextType.WAR, ContextType.WAR,
war.id, war.id,
@ -160,7 +161,7 @@ class WarController:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]: ) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {} bids_map = {}
for ctx in contexts: for ctx in contexts:
active = TieResolver.get_active_participants( active = TieBreaker.get_active_participants(
war, war,
ctx, ctx,
ctx.participants, ctx.participants,
@ -170,26 +171,20 @@ class WarController:
for pid in active for pid in active
] ]
counters = [war.get_influence_tokens(pid) for pid in active] counters = [war.get_influence_tokens(pid) for pid in active]
data = Presenter.build_dialog_data(
war,
ctx,
)
dialog = TieDialog( dialog = TieDialog(
parent=self.app.view, parent=self.app.view,
players=players, players=players,
counters=counters, counters=counters,
context_type=ctx.context_type, context_type=ctx.context_type,
context_id=ctx.context_id, context_id=ctx.context_id,
context_name=None, context_name=data.title,
) )
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=f"Objective tie: {objective.name}",
)
if not dialog.exec(): if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx) TieBreaker.cancel_tie_break(war, ctx)
raise AbortedOperation("Tie resolution cancelled") raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids() bids_map[ctx.key()] = dialog.get_bids()
return bids_map return bids_map

View file

@ -0,0 +1,116 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from warchron.controller.app_controller import AppController
from warchron.model.war import War
from warchron.model.campaign import Campaign
from warchron.model.round import Round
from warchron.model.closing import Closer
from warchron.model.tiebreaking import TieBreaker
from warchron.model.pairing import Pairing
class Workflow:
def __init__(self, controller: "AppController"):
self.app = controller
class RoundClosureWorkflow(Workflow):
def start(self, war: War, campaign: Campaign, round: Round) -> None:
Closer.check_round_closable(round)
ties = TieBreaker.find_battle_ties(war, round.id)
while ties:
for tie in ties:
TieBreaker.resolve_group(
war,
tie,
self.app.rounds.resolve_ties,
)
ties = TieBreaker.find_battle_ties(war, round.id)
for battle in round.battles.values():
Closer.apply_battle_outcomes(war, campaign, battle)
Closer.finalize_round(round)
class CampaignClosureWorkflow(Workflow):
def start(self, war: War, campaign: Campaign) -> None:
Closer.check_campaign_closable(campaign)
ties = TieBreaker.find_campaign_ties(war, campaign.id)
while ties:
for tie in ties:
TieBreaker.resolve_group(
war,
tie,
self.app.rounds.resolve_ties,
)
ties = TieBreaker.find_campaign_ties(war, campaign.id)
for obj in war.get_objectives_used_as_maj_or_min():
objective_id = obj.id
ties = TieBreaker.find_campaign_objective_ties(
war,
campaign.id,
objective_id,
)
while ties:
for tie in ties:
TieBreaker.resolve_group(
war,
tie,
self.app.rounds.resolve_ties,
)
ties = TieBreaker.find_campaign_objective_ties(
war,
campaign.id,
objective_id,
)
Closer.finalize_campaign(campaign)
class WarClosureWorkflow(Workflow):
def start(self, war: War) -> None:
Closer.check_war_closable(war)
ties = TieBreaker.find_war_ties(war)
while ties:
for tie in ties:
TieBreaker.resolve_group(
war,
tie,
self.app.rounds.resolve_ties,
)
ties = TieBreaker.find_war_ties(war)
for obj in war.get_objectives_used_as_maj_or_min():
objective_id = obj.id
ties = TieBreaker.find_war_objective_ties(
war,
objective_id,
)
while ties:
for tie in ties:
TieBreaker.resolve_group(
war,
tie,
self.app.rounds.resolve_ties,
)
ties = TieBreaker.find_war_objective_ties(
war,
objective_id,
)
Closer.finalize_war(war)
class RoundPairingWorkflow:
def __init__(self, controller: "AppController"):
self.app = controller
def start(self, war: War, round: Round) -> None:
Pairing.check_round_pairable(war, round)
Pairing.assign_battles_to_participants(
war,
round,
resolve_ties_callback=self.app.rounds.resolve_ties,
)

View file

@ -144,8 +144,8 @@ class Campaign:
def cleanup() -> None: def cleanup() -> None:
for rnd in rounds_blocking: for rnd in rounds_blocking:
rnd.clear_participant_references(participant_id)
rnd.remove_choice(participant_id) rnd.remove_choice(participant_id)
rnd.clear_participant_references(participant_id)
del self.participants[participant_id] del self.participants[participant_id]
rounds_str = ", ".join( rounds_str = ", ".join(
@ -263,8 +263,8 @@ class Campaign:
def cleanup_and_update() -> None: def cleanup_and_update() -> None:
for rnd in affected_rounds: for rnd in affected_rounds:
rnd.clear_sector_references(sector_id)
rnd.remove_battle(sector_id) rnd.remove_battle(sector_id)
rnd.clear_sector_references(sector_id)
apply_update() apply_update()
rounds_str = ", ".join( rounds_str = ", ".join(
@ -299,8 +299,8 @@ class Campaign:
def cleanup() -> None: def cleanup() -> None:
for rnd in rounds_blocking: for rnd in rounds_blocking:
rnd.clear_sector_references(sector_id)
rnd.remove_battle(sector_id) rnd.remove_battle(sector_id)
rnd.clear_sector_references(sector_id)
del self.sectors[sector_id] del self.sectors[sector_id]
rounds_str = ", ".join( rounds_str = ", ".join(
@ -345,12 +345,10 @@ class Campaign:
return index return index
raise KeyError("Round not found in campaign") raise KeyError("Round not found in campaign")
# TODO replace multiloops by internal has_* method
def get_round_by_battle(self, sector_id: str) -> Round: def get_round_by_battle(self, sector_id: str) -> Round:
for rnd in self.rounds: for rnd in self.rounds:
for bat in rnd.battles.values(): if rnd.has_battle_with_sector(sector_id):
if bat.sector_id == sector_id: return rnd
return rnd
raise KeyError(f"Battle {sector_id} not found in any Round") raise KeyError(f"Battle {sector_id} not found in any Round")
def get_all_rounds(self) -> List[Round]: def get_all_rounds(self) -> List[Round]:
@ -408,6 +406,12 @@ class Campaign:
rnd = self.get_round(round_id) rnd = self.get_round(round_id)
return rnd.create_battle(sector_id) return rnd.create_battle(sector_id)
def get_battle(self, battle_id: str) -> Battle | None:
for rnd in self.rounds:
if rnd.has_battle_with_sector(battle_id):
return rnd.get_battle(battle_id)
return None
def update_battle( def update_battle(
self, self,
round_id: str, round_id: str,

View file

@ -5,10 +5,10 @@ from collections import defaultdict
from warchron.constants import ContextType, ScoreKind from warchron.constants import ContextType, ScoreKind
from warchron.model.war import War from warchron.model.war import War
from warchron.model.war_event import TieResolved from warchron.model.war_event import TieResolved
from warchron.model.tie_manager import TieResolver, TieContext from warchron.model.tiebreaking import TieBreaker, TieContext
if TYPE_CHECKING: if TYPE_CHECKING:
from warchron.model.score_service import ParticipantScore from warchron.model.scoring import ParticipantScore
class ResultChecker: class ResultChecker:
@ -75,29 +75,29 @@ class ResultChecker:
current_rank += 1 current_rank += 1
continue continue
# normal tie-break if tie persists # normal tie-break if tie persists
if not TieResolver.is_tie_resolved(war, context): if not TieBreaker.is_tie_resolved(war, context):
ranking.append( ranking.append(
(current_rank, subgroup, {pid: 0 for pid in subgroup}) (current_rank, subgroup, {pid: 0 for pid in subgroup})
) )
current_rank += 1 current_rank += 1
continue continue
groups = TieResolver.rank_by_tokens(war, context, subgroup) groups = TieBreaker.rank_by_tokens(war, context, subgroup)
tokens_spent = TieResolver.tokens_spent_map(war, context, subgroup) tokens_spent = TieBreaker.tokens_spent_map(war, context, subgroup)
for group in groups: for group in groups:
group_tokens = {pid: tokens_spent[pid] for pid in group} group_tokens = {pid: tokens_spent[pid] for pid in group}
ranking.append((current_rank, group, group_tokens)) ranking.append((current_rank, group, group_tokens))
current_rank += 1 current_rank += 1
continue continue
# no tie # no tie
if len(participants) == 1 or not TieResolver.is_tie_resolved(war, context): if len(participants) == 1 or not TieBreaker.is_tie_resolved(war, context):
ranking.append( ranking.append(
(current_rank, participants, {pid: 0 for pid in participants}) (current_rank, participants, {pid: 0 for pid in participants})
) )
current_rank += 1 current_rank += 1
continue continue
# apply token ranking # apply token ranking
groups = TieResolver.rank_by_tokens(war, context, participants) groups = TieBreaker.rank_by_tokens(war, context, participants)
tokens_spent = TieResolver.tokens_spent_map(war, context, participants) tokens_spent = TieBreaker.tokens_spent_map(war, context, participants)
for group in groups: for group in groups:
group_tokens = {pid: tokens_spent[pid] for pid in group} group_tokens = {pid: tokens_spent[pid] for pid in group}
ranking.append((current_rank, group, group_tokens)) ranking.append((current_rank, group, group_tokens))
@ -111,13 +111,13 @@ class ResultChecker:
value_getter: Callable[[ParticipantScore], int], value_getter: Callable[[ParticipantScore], int],
subcontexts: List[TieContext], subcontexts: List[TieContext],
) -> List[List[str]]: ) -> List[List[str]]:
from warchron.model.score_service import ScoreService from warchron.model.scoring import ScoreComputer
rank_map: Dict[str, Tuple[int, ...]] = {} rank_map: Dict[str, Tuple[int, ...]] = {}
for pid in participants: for pid in participants:
ranks: List[int] = [] ranks: List[int] = []
for sub in subcontexts: for sub in subcontexts:
scores = ScoreService.compute_scores( scores = ScoreComputer.compute_scores(
war, sub.context_type, sub.context_id war, sub.context_type, sub.context_id
) )
ranking = ResultChecker.get_effective_ranking( ranking = ResultChecker.get_effective_ranking(

View file

@ -9,7 +9,7 @@ from warchron.model.round import Round
from warchron.model.battle import Battle from warchron.model.battle import Battle
class ClosureService: class Closer:
# Round methods # Round methods
@ -24,7 +24,7 @@ class ClosureService:
@staticmethod @staticmethod
def apply_battle_outcomes(war: War, campaign: Campaign, battle: Battle) -> None: def apply_battle_outcomes(war: War, campaign: Campaign, battle: Battle) -> None:
from warchron.model.result_checker import ResultChecker from warchron.model.checking import ResultChecker
already_granted = any( already_granted = any(
isinstance(e, InfluenceGained) and e.context_id == battle.sector_id isinstance(e, InfluenceGained) and e.context_id == battle.sector_id

View file

@ -1,10 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import Dict, List, Callable, Tuple from typing import Dict, List
from dataclasses import dataclass
from uuid import uuid4 from uuid import uuid4
import random import random
from warchron.constants import ContextType, ScoreKind from warchron.constants import ContextType, ScoreKind, ChoiceStatus, AllocationType
from warchron.model.exception import ( from warchron.model.exception import (
DomainError, DomainError,
ForbiddenOperation, ForbiddenOperation,
@ -13,15 +14,17 @@ from warchron.model.exception import (
from warchron.model.war import War from warchron.model.war import War
from warchron.model.round import Round from warchron.model.round import Round
from warchron.model.battle import Battle from warchron.model.battle import Battle
from warchron.model.score_service import ScoreService from warchron.model.scoring import ScoreComputer
from warchron.model.tie_manager import TieResolver, TieContext from warchron.model.tiebreaking import TieBreaker, TieContext, ResolveTiesCallback
from warchron.model.war_event import TieResolved from warchron.model.war_event import TieResolved
from warchron.model.score_service import ParticipantScore from warchron.model.scoring import ParticipantScore
ResolveTiesCallback = Callable[
["War", List["TieContext"]], @dataclass(frozen=True, slots=True)
Dict[Tuple[str, str, int | None, str | None, str | None], Dict[str, bool]], class AllocationResult:
] priority: ChoiceStatus
secondary: ChoiceStatus
fallback: bool
class Pairing: class Pairing:
@ -39,15 +42,6 @@ class Pairing:
raise DomainError( raise DomainError(
"There are not enough sectors for all participants to battle" "There are not enough sectors for all participants to battle"
) )
for pid, choice in round.choices.items():
if choice is not None and not choice.priority_sector_id:
raise ForbiddenOperation(
f"Missing priority choice for participant {pid}"
)
if choice is not None and not choice.secondary_sector_id:
raise ForbiddenOperation(
f"Missing secondary choice for participant {pid}"
)
def cleanup() -> None: def cleanup() -> None:
for bat in round.battles.values(): for bat in round.battles.values():
@ -66,6 +60,15 @@ class Pairing:
"Do you want to continue?", "Do you want to continue?",
action=cleanup, action=cleanup,
) )
for pid, choice in round.choices.items():
if choice is not None and not choice.priority_sector_id:
raise ForbiddenOperation(
f"Missing priority choice for participant {pid}"
)
if choice is not None and not choice.secondary_sector_id:
raise ForbiddenOperation(
f"Missing secondary choice for participant {pid}"
)
@staticmethod @staticmethod
def assign_battles_to_participants( def assign_battles_to_participants(
@ -76,7 +79,7 @@ class Pairing:
campaign = war.get_campaign_by_round(round.id) campaign = war.get_campaign_by_round(round.id)
if campaign is None: if campaign is None:
raise DomainError(f"Campaign for round {round.id} doesn't exist") raise DomainError(f"Campaign for round {round.id} doesn't exist")
scores = ScoreService.compute_scores( scores = ScoreComputer.compute_scores(
war, war,
ContextType.CAMPAIGN, ContextType.CAMPAIGN,
campaign.id, campaign.id,
@ -85,7 +88,7 @@ class Pairing:
def value_getter(score: ParticipantScore) -> int: def value_getter(score: ParticipantScore) -> int:
return score.victory_points return score.victory_points
score_groups = ScoreService.group_participants_by_score(scores, value_getter) score_groups = ScoreComputer.group_participants_by_score(scores, value_getter)
sector_to_battle: Dict[str, Battle] = { sector_to_battle: Dict[str, Battle] = {
b.sector_id: b for b in round.battles.values() b.sector_id: b for b in round.battles.values()
} }
@ -198,13 +201,11 @@ class Pairing:
sector_id=sector_id, sector_id=sector_id,
) )
# ---- resolve tie loop ---- # ---- resolve tie loop ----
tie_id = TieResolver.find_active_tie_id(war, context) or str(uuid4()) tie_id = TieBreaker.find_active_tie_id(war, context) or str(uuid4())
while not TieResolver.is_tie_resolved(war, context): while not TieBreaker.is_tie_resolved(war, context):
active = TieResolver.get_active_participants( active = TieBreaker.get_active_participants(
war, context, context.participants war, context, context.participants
) )
if len(active) <= 1:
break
current_context = TieContext( current_context = TieContext(
context_type=context.context_type, context_type=context.context_type,
context_id=context.context_id, context_id=context.context_id,
@ -213,9 +214,12 @@ class Pairing:
score_kind=context.score_kind, score_kind=context.score_kind,
sector_id=context.sector_id, sector_id=context.sector_id,
) )
# natural or unbreakable draw # natural, unbreakable or acceptable (enough places) draw
if not TieResolver.can_tie_be_resolved( if (
war, context, current_context.participants not TieBreaker.can_tie_be_resolved(
war, context, current_context.participants
)
or len(active) <= places
): ):
war.events.append( war.events.append(
TieResolved( TieResolved(
@ -245,9 +249,9 @@ class Pairing:
) )
) )
break break
TieResolver.apply_bids(war, context, tie_id, bids) TieBreaker.apply_bids(war, context, tie_id, bids)
TieResolver.resolve_tie_state(war, context, tie_id, bids) TieBreaker.resolve_tie_state(war, context, tie_id, bids)
ranked_groups = TieResolver.rank_by_tokens( ranked_groups = TieBreaker.rank_by_tokens(
war, war,
context, context,
context.participants, context.participants,
@ -277,3 +281,80 @@ class Pairing:
remaining.remove(pid) remaining.remove(pid)
continue continue
raise DomainError(f"Ambiguous fallback for participant {pid}") raise DomainError(f"Ambiguous fallback for participant {pid}")
@staticmethod
def get_allocation_kind(
war: War,
round_id: str,
participant_id: str,
sector_id: str,
) -> AllocationType:
round = war.get_round(round_id)
choice = round.choices.get(participant_id)
if not choice:
raise DomainError(f"No choice found for participant {participant_id}")
if choice.priority_sector_id == sector_id:
return AllocationType.PRIORITY
if choice.secondary_sector_id == sector_id:
return AllocationType.SECONDARY
return AllocationType.FALLBACK
@staticmethod
def get_round_allocation(
war: War,
round: Round,
campaign_participant_id: str,
) -> AllocationResult:
choice = round.choices[campaign_participant_id]
campaign = war.get_campaign_by_round(round.id)
if campaign is None:
raise DomainError(f"No campaign found for round {round.id}")
war_pid = campaign.campaign_to_war_part_id(campaign_participant_id)
token_priority = TieBreaker.participant_spent_token(
war,
ContextType.CHOICE,
round.id,
choice.priority_sector_id,
war_pid,
)
token_secondary = TieBreaker.participant_spent_token(
war,
ContextType.CHOICE,
round.id,
choice.secondary_sector_id,
war_pid,
)
battle = round.get_battle_for_participant(campaign_participant_id)
allocation = AllocationType.FALLBACK
if battle:
allocation = Pairing.get_allocation_kind(
war,
round.id,
campaign_participant_id,
battle.sector_id,
)
priority_status = ChoiceStatus.NONE
secondary_status = ChoiceStatus.NONE
fallback = allocation == AllocationType.FALLBACK
if allocation == AllocationType.PRIORITY:
priority_status = (
ChoiceStatus.ALLOCATEDTOKEN
if token_priority
else ChoiceStatus.ALLOCATED
)
elif token_priority:
priority_status = ChoiceStatus.TOKEN
if allocation == AllocationType.SECONDARY:
secondary_status = (
ChoiceStatus.ALLOCATEDTOKEN
if token_secondary
else ChoiceStatus.ALLOCATED
)
elif token_secondary:
secondary_status = ChoiceStatus.TOKEN
return AllocationResult(
priority=priority_status,
secondary=secondary_status,
fallback=fallback,
)

View file

@ -5,7 +5,12 @@ from typing import Any, Dict, List, TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from warchron.model.campaign import Campaign from warchron.model.campaign import Campaign
from warchron.model.war import War from warchron.model.war import War
from warchron.model.exception import ForbiddenOperation from warchron.constants import ContextType
from warchron.model.exception import (
ForbiddenOperation,
DomainError,
RequiresConfirmation,
)
from warchron.model.choice import Choice from warchron.model.choice import Choice
from warchron.model.battle import Battle from warchron.model.battle import Battle
@ -74,7 +79,6 @@ class Round:
def create_choice(self, participant_id: str) -> Choice: def create_choice(self, participant_id: str) -> Choice:
if self.is_over: if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create choice in a closed round.") raise ForbiddenOperation("Can't create choice in a closed round.")
if participant_id not in self.choices: if participant_id not in self.choices:
choice = Choice( choice = Choice(
@ -93,11 +97,16 @@ class Round:
comment: str | None, comment: str | None,
) -> None: ) -> None:
if self.is_over: if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't update choice in a closed round.") raise ForbiddenOperation("Can't update choice in a closed round.")
# TODO prevent if battles already assigned
choice = self.get_choice(participant_id) choice = self.get_choice(participant_id)
if choice: if choice:
if self.has_battle_with_participant(participant_id) and (
priority_sector_id != choice.priority_sector_id
or secondary_sector_id != choice.secondary_sector_id
):
raise ForbiddenOperation(
"Can't update choice already assigned to battle."
)
choice.set_priority(priority_sector_id) choice.set_priority(priority_sector_id)
choice.set_secondary(secondary_sector_id) choice.set_secondary(secondary_sector_id)
choice.set_comment(comment) choice.set_comment(comment)
@ -118,9 +127,9 @@ class Round:
if participant_id not in self.choices: if participant_id not in self.choices:
return return
if self.is_over: if self.is_over:
# TODO catch me if you can (inner)
raise ForbiddenOperation("Can't remove choice in a closed round.") raise ForbiddenOperation("Can't remove choice in a closed round.")
# TODO prevent if battles already assigned if self.has_battle_with_participant(participant_id):
raise ForbiddenOperation("Can't remove choice already assigned to battle.")
self.war.revert_choice_ties( self.war.revert_choice_ties(
self.id, self.id,
participants=[self.campaign.campaign_to_war_part_id(participant_id)], participants=[self.campaign.campaign_to_war_part_id(participant_id)],
@ -132,6 +141,18 @@ class Round:
def get_battle(self, sector_id: str) -> Battle | None: def get_battle(self, sector_id: str) -> Battle | None:
return self.battles.get(sector_id) return self.battles.get(sector_id)
def get_battle_for_participant(
self,
campaign_participant_id: str,
) -> Battle | None:
for battle in self.battles.values():
if (
battle.player_1_id == campaign_participant_id
or battle.player_2_id == campaign_participant_id
):
return battle
return None
def has_battle_with_sector(self, sector_id: str) -> bool: def has_battle_with_sector(self, sector_id: str) -> bool:
return any(bat.sector_id == sector_id for bat in self.battles.values()) return any(bat.sector_id == sector_id for bat in self.battles.values())
@ -156,7 +177,6 @@ class Round:
def create_battle(self, sector_id: str) -> Battle: def create_battle(self, sector_id: str) -> Battle:
if self.is_over: if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create battle in a closed round.") raise ForbiddenOperation("Can't create battle in a closed round.")
if sector_id not in self.battles: if sector_id not in self.battles:
battle = Battle(sector_id=sector_id, player_1_id=None, player_2_id=None) battle = Battle(sector_id=sector_id, player_1_id=None, player_2_id=None)
@ -173,12 +193,15 @@ class Round:
victory_condition: str | None, victory_condition: str | None,
comment: str | None, comment: str | None,
) -> None: ) -> None:
from warchron.model.tiebreaking import TieBreaker
if self.is_over: if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't update battle in a closed round.") raise ForbiddenOperation("Can't update battle in a closed round.")
bat = self.get_battle(sector_id) bat = self.get_battle(sector_id)
# TODO require confirmation if there was choice tie to clear it if not bat:
if bat: raise DomainError(f"No battle found for sector {sector_id}")
def apply_update() -> None:
bat.set_player_1(player_1_id) bat.set_player_1(player_1_id)
bat.set_player_2(player_2_id) bat.set_player_2(player_2_id)
bat.set_winner(winner_id) bat.set_winner(winner_id)
@ -186,6 +209,48 @@ class Round:
bat.set_victory_condition(victory_condition) bat.set_victory_condition(victory_condition)
bat.set_comment(comment) bat.set_comment(comment)
if bat.player_1_id == player_1_id and bat.player_2_id == player_2_id:
apply_update()
return
affected_choices: List[Choice] = []
affected_players: List[str | None] = list(
{
player_1_id,
player_2_id,
bat.player_1_id,
bat.player_2_id,
}
- {None}
)
for choice in self.choices.values():
for player in affected_players:
if (
player
and self.has_choice_with_participant(player)
and TieBreaker.participant_spent_token(
self.war,
ContextType.CHOICE,
self.id,
sector_id,
self.campaign.campaign_to_war_part_id(player),
)
):
affected_choices.append(choice)
if not affected_choices:
apply_update()
return
def cleanup_and_update() -> None:
self.clear_sector_references(sector_id)
apply_update()
raise RequiresConfirmation(
"Changing the player(s) of this sector will affect choices.\n"
"Choices will be cleared and their tokens and tie-breaks will be deleted.\n"
"Do you want to continue?",
action=cleanup_and_update,
)
def clear_participant_references(self, participant_id: str) -> None: def clear_participant_references(self, participant_id: str) -> None:
for battle in self.battles.values(): for battle in self.battles.values():
trigger_revert_ties = False trigger_revert_ties = False
@ -209,11 +274,9 @@ class Round:
if sector_id not in self.battles: if sector_id not in self.battles:
return return
if self.is_over: if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't remove battle in a closed round.") raise ForbiddenOperation("Can't remove battle in a closed round.")
bat = self.battles[sector_id] bat = self.battles[sector_id]
if bat and bat.is_finished(): if bat and bat.is_finished():
# TODO catch me if you can
raise ForbiddenOperation("Can't remove finished battle.") raise ForbiddenOperation("Can't remove finished battle.")
self.war.revert_battle_ties(self.id, sector_id=sector_id) self.war.revert_battle_ties(self.id, sector_id=sector_id)
del self.battles[sector_id] del self.battles[sector_id]

View file

@ -4,7 +4,9 @@ from collections import defaultdict
from warchron.constants import ContextType from warchron.constants import ContextType
from warchron.model.war import War from warchron.model.war import War
from warchron.model.campaign import Campaign
from warchron.model.battle import Battle from warchron.model.battle import Battle
from warchron.model.exception import DomainError
@dataclass(slots=True) @dataclass(slots=True)
@ -13,7 +15,7 @@ class ParticipantScore:
narrative_points: Dict[str, int] = field(default_factory=dict) narrative_points: Dict[str, int] = field(default_factory=dict)
class ScoreService: class ScoreComputer:
@staticmethod @staticmethod
def _get_battles_for_context( def _get_battles_for_context(
@ -26,14 +28,17 @@ class ScoreService:
continue continue
yield from rnd.battles.values() yield from rnd.battles.values()
elif context_type == ContextType.CAMPAIGN: elif context_type == ContextType.CAMPAIGN:
campaign = war.get_campaign(context_id) campaign: Campaign | None = war.get_campaign(context_id)
for rnd in campaign.rounds: if campaign:
if not rnd.is_over: for rnd in campaign.rounds:
continue if not rnd.is_over:
yield from rnd.battles.values() continue
yield from rnd.battles.values()
elif context_type == ContextType.BATTLE: elif context_type == ContextType.BATTLE:
battle = war.get_battle(context_id) battle = war.get_battle(context_id)
campaign = war.get_campaign_by_sector(battle.sector_id) campaign = war.get_campaign_by_sector(battle.sector_id)
if not campaign:
raise DomainError(f"No campaign found for secor {battle.sector_id}")
rnd = campaign.get_round_by_battle(context_id) rnd = campaign.get_round_by_battle(context_id)
if rnd and rnd.is_over: if rnd and rnd.is_over:
yield battle yield battle
@ -42,7 +47,7 @@ class ScoreService:
def compute_scores( def compute_scores(
war: War, context_type: ContextType, context_id: str war: War, context_type: ContextType, context_id: str
) -> Dict[str, ParticipantScore]: ) -> Dict[str, ParticipantScore]:
from warchron.model.result_checker import ResultChecker from warchron.model.checking import ResultChecker
if context_type == ContextType.CAMPAIGN: if context_type == ContextType.CAMPAIGN:
camp = war.get_campaign(context_id) camp = war.get_campaign(context_id)
@ -56,7 +61,7 @@ class ScoreService:
) )
for pid in participant_ids for pid in participant_ids
} }
battles = ScoreService._get_battles_for_context(war, context_type, context_id) battles = ScoreComputer._get_battles_for_context(war, context_type, context_id)
for battle in battles: for battle in battles:
base_winner = None base_winner = None
if battle.winner_id is not None: if battle.winner_id is not None:

View file

@ -1,12 +1,13 @@
from typing import List, Dict, DefaultDict, Tuple from __future__ import annotations
from typing import List, Dict, Tuple, Callable, TypeAlias
from dataclasses import dataclass from dataclasses import dataclass
from collections import defaultdict from uuid import uuid4
from warchron.constants import ContextType, ScoreKind from warchron.constants import ContextType, ScoreKind
from warchron.model.exception import ForbiddenOperation, DomainError from warchron.model.exception import ForbiddenOperation, DomainError
from warchron.model.war import War from warchron.model.war import War
from warchron.model.war_event import InfluenceSpent, TieResolved from warchron.model.war_event import InfluenceSpent, TieResolved
from warchron.model.score_service import ScoreService, ParticipantScore from warchron.model.scoring import ScoreComputer, ParticipantScore
@dataclass @dataclass
@ -29,7 +30,13 @@ class TieContext:
) )
class TieResolver: ResolveTiesCallback: TypeAlias = Callable[
[War, List[TieContext]],
Dict[Tuple[str, str, int | None, str | None, str | None], Dict[str, bool]],
]
class TieBreaker:
@staticmethod @staticmethod
def find_active_tie_id( def find_active_tie_id(
@ -65,10 +72,10 @@ class TieResolver:
context: TieContext = TieContext( context: TieContext = TieContext(
ContextType.BATTLE, battle.sector_id, [p1_id, p2_id] ContextType.BATTLE, battle.sector_id, [p1_id, p2_id]
) )
if TieResolver.is_tie_resolved(war, context): if TieBreaker.is_tie_resolved(war, context):
continue continue
tie_id = TieResolver.find_active_tie_id(war, context) tie_id = TieBreaker.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, [p1_id, p2_id]): if not TieBreaker.can_tie_be_resolved(war, context, [p1_id, p2_id]):
war.events.append( war.events.append(
TieResolved( TieResolved(
participant_id=None, participant_id=None,
@ -92,14 +99,22 @@ class TieResolver:
@staticmethod @staticmethod
def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]: def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]:
scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id) from warchron.model.checking import ResultChecker
buckets: DefaultDict[int, List[str]] = defaultdict(list)
for pid, score in scores.items(): scores = ScoreComputer.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
buckets[score.victory_points].append(pid) ranking = ResultChecker.get_effective_ranking(
war,
ContextType.CAMPAIGN,
campaign_id,
ScoreKind.VP,
scores,
lambda s: s.victory_points,
)
ties: List[TieContext] = [] ties: List[TieContext] = []
for score_value, participants in buckets.items(): for _, group, _ in ranking:
if len(participants) <= 1: if len(group) <= 1:
continue continue
score_value = scores[group[0]].victory_points
context: TieContext = TieContext( context: TieContext = TieContext(
ContextType.CAMPAIGN, ContextType.CAMPAIGN,
campaign_id, campaign_id,
@ -107,16 +122,16 @@ class TieResolver:
score_value, score_value,
ScoreKind.VP, ScoreKind.VP,
) )
if TieResolver.is_tie_resolved(war, context): if TieBreaker.is_tie_resolved(war, context):
continue continue
tie_id = TieResolver.find_active_tie_id(war, context) tie_id = TieBreaker.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, participants): if not TieBreaker.can_tie_be_resolved(war, context, group):
war.events.append( war.events.append(
TieResolved( TieResolved(
participant_id=None, participant_id=None,
context_type=ContextType.CAMPAIGN, context_type=ContextType.CAMPAIGN,
context_id=campaign_id, context_id=campaign_id,
participants=participants, participants=group,
tie_id=tie_id, tie_id=tie_id,
score_value=score_value, score_value=score_value,
) )
@ -126,7 +141,7 @@ class TieResolver:
TieContext( TieContext(
context_type=ContextType.CAMPAIGN, context_type=ContextType.CAMPAIGN,
context_id=campaign_id, context_id=campaign_id,
participants=participants, participants=group,
score_value=score_value, score_value=score_value,
score_kind=ScoreKind.VP, score_kind=ScoreKind.VP,
) )
@ -137,20 +152,32 @@ class TieResolver:
def find_campaign_objective_ties( def find_campaign_objective_ties(
war: War, campaign_id: str, objective_id: str war: War, campaign_id: str, objective_id: str
) -> List[TieContext]: ) -> List[TieContext]:
scores = ScoreService.compute_scores( from warchron.model.checking import ResultChecker
scores = ScoreComputer.compute_scores(
war, war,
ContextType.CAMPAIGN, ContextType.CAMPAIGN,
campaign_id, campaign_id,
) )
buckets: DefaultDict[int, List[str]] = defaultdict(list)
for pid, score in scores.items(): def value_getter(score: ParticipantScore) -> int:
np_value = score.narrative_points.get(objective_id, 0) return score.narrative_points.get(objective_id, 0)
buckets[np_value].append(pid)
ranking = ResultChecker.get_effective_ranking(
war,
ContextType.CAMPAIGN,
campaign_id,
ScoreKind.NP,
scores,
value_getter,
objective_id,
)
ties: List[TieContext] = [] ties: List[TieContext] = []
context_id = campaign_id context_id = campaign_id
for np_value, participants in buckets.items(): for _, group, _ in ranking:
if len(participants) <= 1: if len(group) <= 1:
continue continue
np_value = value_getter(scores[group[0]])
context: TieContext = TieContext( context: TieContext = TieContext(
ContextType.CAMPAIGN, ContextType.CAMPAIGN,
campaign_id, campaign_id,
@ -159,20 +186,20 @@ class TieResolver:
ScoreKind.NP, ScoreKind.NP,
objective_id, objective_id,
) )
if TieResolver.is_tie_resolved(war, context): if TieBreaker.is_tie_resolved(war, context):
continue continue
tie_id = TieResolver.find_active_tie_id(war, context) tie_id = TieBreaker.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved( if not TieBreaker.can_tie_be_resolved(
war, war,
context, context,
participants, group,
): ):
war.events.append( war.events.append(
TieResolved( TieResolved(
participant_id=None, participant_id=None,
context_type=ContextType.CAMPAIGN, context_type=ContextType.CAMPAIGN,
context_id=context_id, context_id=context_id,
participants=participants, participants=group,
tie_id=tie_id, tie_id=tie_id,
score_value=np_value, score_value=np_value,
objective_id=objective_id, objective_id=objective_id,
@ -183,7 +210,7 @@ class TieResolver:
TieContext( TieContext(
context_type=ContextType.CAMPAIGN, context_type=ContextType.CAMPAIGN,
context_id=context_id, context_id=context_id,
participants=participants, participants=group,
score_value=np_value, score_value=np_value,
score_kind=ScoreKind.NP, score_kind=ScoreKind.NP,
objective_id=objective_id, objective_id=objective_id,
@ -193,9 +220,9 @@ class TieResolver:
@staticmethod @staticmethod
def find_war_ties(war: War) -> List[TieContext]: def find_war_ties(war: War) -> List[TieContext]:
from warchron.model.result_checker import ResultChecker from warchron.model.checking import ResultChecker
scores = ScoreService.compute_scores(war, ContextType.WAR, war.id) scores = ScoreComputer.compute_scores(war, ContextType.WAR, war.id)
ranking = ResultChecker.get_effective_ranking( ranking = ResultChecker.get_effective_ranking(
war, war,
ContextType.WAR, ContextType.WAR,
@ -216,10 +243,10 @@ class TieResolver:
score_value=score_value, score_value=score_value,
score_kind=ScoreKind.VP, score_kind=ScoreKind.VP,
) )
if TieResolver.is_tie_resolved(war, context): if TieBreaker.is_tie_resolved(war, context):
continue continue
tie_id = TieResolver.find_active_tie_id(war, context) tie_id = TieBreaker.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved(war, context, group): if not TieBreaker.can_tie_be_resolved(war, context, group):
war.events.append( war.events.append(
TieResolved( TieResolved(
participant_id=None, participant_id=None,
@ -244,9 +271,9 @@ class TieResolver:
@staticmethod @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 from warchron.model.checking import ResultChecker
scores = ScoreService.compute_scores( scores = ScoreComputer.compute_scores(
war, war,
ContextType.WAR, ContextType.WAR,
war.id, war.id,
@ -272,13 +299,13 @@ class TieResolver:
context: TieContext = TieContext( context: TieContext = TieContext(
ContextType.WAR, war.id, [], np_value, ScoreKind.NP, objective_id ContextType.WAR, war.id, [], np_value, ScoreKind.NP, objective_id
) )
if TieResolver.is_tie_resolved( if TieBreaker.is_tie_resolved(
war, war,
context, context,
): ):
continue continue
tie_id = TieResolver.find_active_tie_id(war, context) tie_id = TieBreaker.find_active_tie_id(war, context)
if not TieResolver.can_tie_be_resolved( if not TieBreaker.can_tie_be_resolved(
war, war,
context, context,
group, group,
@ -307,6 +334,51 @@ class TieResolver:
) )
return ties return ties
@staticmethod
def resolve_group(
war: War,
context: TieContext,
resolve_ties_callback: ResolveTiesCallback,
) -> None:
tie_id = TieBreaker.find_active_tie_id(war, context) or str(uuid4())
while not TieBreaker.is_tie_resolved(war, context):
active = TieBreaker.get_active_participants(
war,
context,
context.participants,
)
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,
objective_id=context.objective_id,
sector_id=context.sector_id,
)
if not TieBreaker.can_tie_be_resolved(
war,
context,
current_context.participants,
):
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,
sector_id=context.sector_id,
)
)
return
bids_map = resolve_ties_callback(war, [current_context])
bids = bids_map[current_context.key()]
TieBreaker.apply_bids(war, context, tie_id, bids)
TieBreaker.resolve_tie_state(war, context, tie_id, bids)
@staticmethod @staticmethod
def apply_bids( def apply_bids(
war: War, war: War,
@ -327,6 +399,7 @@ class TieResolver:
context_id=context.context_id, context_id=context.context_id,
tie_id=tie_id, tie_id=tie_id,
objective_id=context.objective_id, objective_id=context.objective_id,
sector_id=context.sector_id,
) )
) )
@ -397,7 +470,7 @@ class TieResolver:
context: TieContext, context: TieContext,
participants: List[str], participants: List[str],
) -> List[str]: ) -> List[str]:
groups = TieResolver.rank_by_tokens(war, context, participants) groups = TieBreaker.rank_by_tokens(war, context, participants)
return groups[0] return groups[0]
@staticmethod @staticmethod
@ -418,10 +491,11 @@ class TieResolver:
tie_id=tie_id, tie_id=tie_id,
score_value=context.score_value, score_value=context.score_value,
objective_id=context.objective_id, objective_id=context.objective_id,
sector_id=context.sector_id,
) )
) )
return return
groups = TieResolver.rank_by_tokens(war, context, context.participants) groups = TieBreaker.rank_by_tokens(war, context, context.participants)
if len(groups[0]) == 1: if len(groups[0]) == 1:
war.events.append( war.events.append(
TieResolved( TieResolved(
@ -432,6 +506,7 @@ class TieResolver:
tie_id=tie_id, tie_id=tie_id,
score_value=context.score_value, score_value=context.score_value,
objective_id=context.objective_id, objective_id=context.objective_id,
sector_id=context.sector_id,
) )
) )
return return
@ -441,25 +516,9 @@ class TieResolver:
def can_tie_be_resolved( def can_tie_be_resolved(
war: War, context: TieContext, participants: List[str] war: War, context: TieContext, participants: List[str]
) -> bool: ) -> bool:
active = TieResolver.get_active_participants(war, context, participants) active = TieBreaker.get_active_participants(war, context, participants)
return any(war.get_influence_tokens(pid) > 0 for pid in active) return any(war.get_influence_tokens(pid) > 0 for pid in active)
@staticmethod
def was_tie_broken_by_tokens(
war: War,
context: TieContext,
) -> 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
):
return ev.participant_id is not None
return False
@staticmethod @staticmethod
def is_tie_resolved(war: War, context: TieContext) -> bool: def is_tie_resolved(war: War, context: TieContext) -> bool:
for ev in war.events: for ev in war.events:
@ -483,3 +542,26 @@ class TieResolver:
continue continue
return True return True
return False return False
@staticmethod
def participant_spent_token(
war: War,
context_type: ContextType,
context_id: str,
sector_id: str | None,
war_participant_id: str,
) -> bool:
if context_type == ContextType.CHOICE and sector_id is None:
return False
for ev in war.events:
if not isinstance(ev, InfluenceSpent):
continue
if ev.context_type != context_type:
continue
if ev.context_id != context_id:
continue
if ev.sector_id != sector_id:
continue
if ev.participant_id == war_participant_id:
return True
return False

View file

@ -10,7 +10,11 @@ from warchron.model.war_event import (
InfluenceSpent, InfluenceSpent,
TieResolved, TieResolved,
) )
from warchron.model.exception import ForbiddenOperation, RequiresConfirmation from warchron.model.exception import (
ForbiddenOperation,
RequiresConfirmation,
DomainError,
)
from warchron.model.war_participant import WarParticipant from warchron.model.war_participant import WarParticipant
from warchron.model.objective import Objective from warchron.model.objective import Objective
from warchron.model.campaign_participant import CampaignParticipant from warchron.model.campaign_participant import CampaignParticipant
@ -80,6 +84,10 @@ class War:
if ev.context_type == ContextType.BATTLE: if ev.context_type == ContextType.BATTLE:
battle = self.get_battle(ev.context_id) battle = self.get_battle(ev.context_id)
campaign = self.get_campaign_by_sector(battle.sector_id) campaign = self.get_campaign_by_sector(battle.sector_id)
if not campaign:
raise DomainError(
f"No campaign found for sector {battle.sector_id}"
)
round = campaign.get_round_by_battle(ev.context_id) round = campaign.get_round_by_battle(ev.context_id)
round.is_over = False round.is_over = False
elif ev.context_type == ContextType.CAMPAIGN: elif ev.context_type == ContextType.CAMPAIGN:
@ -304,21 +312,17 @@ class War:
return camp return camp
raise KeyError(f"Campaign {campaign_id} not found in War {self.id}") raise KeyError(f"Campaign {campaign_id} not found in War {self.id}")
# TODO replace multiloops by internal has_* method
def get_campaign_by_round(self, round_id: str) -> Campaign | None: def get_campaign_by_round(self, round_id: str) -> Campaign | None:
for camp in self.campaigns: for camp in self.campaigns:
for rnd in camp.rounds: if camp.has_round(round_id):
if rnd.id == round_id: return camp
return camp
return None return None
# TODO replace multiloops by internal has_* method def get_campaign_by_sector(self, sector_id: str) -> Campaign | None:
def get_campaign_by_sector(self, sector_id: str) -> Campaign:
for camp in self.campaigns: for camp in self.campaigns:
for sect in camp.sectors.values(): if camp.has_sector(sector_id):
if sect.id == sector_id: return camp
return camp return None
raise KeyError(f"Sector {sector_id} not found in any Campaign")
def get_campaign_by_campaign_participant( def get_campaign_by_campaign_participant(
self, participant_id: str self, participant_id: str
@ -375,12 +379,10 @@ class War:
name, round_id, major_id, minor_id, influence_id, mission, description name, round_id, major_id, minor_id, influence_id, mission, description
) )
# TODO replace multiloops by internal has_* method
def get_sector(self, sector_id: str) -> Sector: def get_sector(self, sector_id: str) -> Sector:
for camp in self.campaigns: for camp in self.campaigns:
for sect in camp.sectors.values(): if camp.has_sector(sector_id):
if sect.id == sector_id: return camp.get_sector(sector_id)
return sect
raise KeyError("Sector not found") raise KeyError("Sector not found")
def update_sector( def update_sector(
@ -396,6 +398,8 @@ class War:
description: str | None, description: str | None,
) -> None: ) -> None:
camp = self.get_campaign_by_sector(sector_id) camp = self.get_campaign_by_sector(sector_id)
if not camp:
raise DomainError(f"No campaign found for sector {sector_id}")
camp.update_sector( camp.update_sector(
sector_id, sector_id,
name=name, name=name,
@ -409,6 +413,8 @@ class War:
def remove_sector(self, sector_id: str) -> None: def remove_sector(self, sector_id: str) -> None:
camp = self.get_campaign_by_sector(sector_id) camp = self.get_campaign_by_sector(sector_id)
if not camp:
raise DomainError(f"No campaign found for sector {sector_id}")
camp.remove_sector(sector_id) camp.remove_sector(sector_id)
# Campaign participant methods # Campaign participant methods
@ -427,12 +433,10 @@ class War:
camp = self.get_campaign(campaign_id) camp = self.get_campaign(campaign_id)
return camp.add_campaign_participant(participant_id, leader, theme) return camp.add_campaign_participant(participant_id, leader, theme)
# TODO replace multiloops by internal has_* method
def get_campaign_participant(self, participant_id: str) -> CampaignParticipant: def get_campaign_participant(self, participant_id: str) -> CampaignParticipant:
for camp in self.campaigns: for camp in self.campaigns:
for part in camp.participants.values(): if camp.has_participant(participant_id):
if part.id == participant_id: return camp.get_campaign_participant(participant_id)
return part
raise KeyError("Participant not found") raise KeyError("Participant not found")
def update_campaign_participant( def update_campaign_participant(
@ -449,12 +453,10 @@ class War:
# Round methods # Round methods
# TODO replace multiloops by internal has_* method
def get_round(self, round_id: str) -> Round: def get_round(self, round_id: str) -> Round:
for camp in self.campaigns: for camp in self.campaigns:
for rnd in camp.rounds: if camp.has_round(round_id):
if rnd.id == round_id: return camp.get_round(round_id)
return rnd
raise KeyError("Round not found") raise KeyError("Round not found")
def add_round(self, campaign_id: str) -> Round: def add_round(self, campaign_id: str) -> Round:
@ -472,7 +474,7 @@ class War:
camp = self.get_campaign_by_round(round_id) camp = self.get_campaign_by_round(round_id)
if camp is not None: if camp is not None:
return camp.create_choice(round_id, participant_id) return camp.create_choice(round_id, participant_id)
raise KeyError("Campaign with round {round_id} doesn't exist") raise KeyError(f"Campaign with round {round_id} doesn't exist")
def update_choice( def update_choice(
self, self,
@ -499,20 +501,18 @@ class War:
# Battle methods # Battle methods
# TODO replace multiloops by internal has_* method
def get_battle(self, battle_id: str) -> Battle: def get_battle(self, battle_id: str) -> Battle:
for camp in self.campaigns: for camp in self.campaigns:
for rnd in camp.rounds: battle = camp.get_battle(battle_id)
for bat in rnd.battles.values(): if battle is not None:
if bat.sector_id == battle_id: return battle
return bat raise KeyError(f"War did not find Battle {battle_id}")
raise KeyError("Battle not found")
def create_battle(self, round_id: str, sector_id: str) -> Battle: def create_battle(self, round_id: str, sector_id: str) -> Battle:
camp = self.get_campaign_by_round(round_id) camp = self.get_campaign_by_round(round_id)
if camp is not None: if camp is not None:
return camp.create_battle(round_id, sector_id) return camp.create_battle(round_id, sector_id)
raise KeyError("Campaign with round {round_id} doesn't exist") raise KeyError(f"Campaign with round {round_id} doesn't exist")
def update_battle( def update_battle(
self, self,

Binary file not shown.

After

Width:  |  Height:  |  Size: 663 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 KiB

View file

@ -34,7 +34,7 @@ class TieDialog(QDialog):
self.ui: Ui_tieDialog = Ui_tieDialog() self.ui: Ui_tieDialog = Ui_tieDialog()
self.ui.setupUi(self) # type: ignore self.ui.setupUi(self) # type: ignore
self.setWindowIcon(Icons.get(IconName.WARCHRONICO)) self.setWindowIcon(Icons.get(IconName.WARCHRONICO))
self.ui.tieContext.setText(self._get_context_title(context_type, context_name)) self.ui.tieContext.setText(context_name)
grid = self.ui.playersGridLayout grid = self.ui.playersGridLayout
icon_path = (RESOURCES_DIR / Icons._paths[IconName.TOKENS]).as_posix() icon_path = (RESOURCES_DIR / Icons._paths[IconName.TOKENS]).as_posix()
token_html = ( token_html = (
@ -71,17 +71,3 @@ class TieDialog(QDialog):
def get_bids(self) -> Dict[str, bool]: def get_bids(self) -> Dict[str, bool]:
return {pid: checkbox.isChecked() for pid, checkbox in self._checkboxes.items()} return {pid: checkbox.isChecked() for pid, checkbox in self._checkboxes.items()}
@staticmethod
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",
}
return titles.get(context_type, "Tie")

View file

@ -603,15 +603,30 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow):
table = self.choicesTable table = self.choicesTable
table.setSortingEnabled(False) table.setSortingEnabled(False)
table.clearContents() table.clearContents()
table.setColumnCount(5)
table.setHorizontalHeaderLabels(
["Participant", "Priority", "Secondary", "", "Comment"]
)
table.setRowCount(len(participants)) table.setRowCount(len(participants))
table.setIconSize(QSize(32, 16))
for row, choice in enumerate(participants): for row, choice in enumerate(participants):
participant_item = QtWidgets.QTableWidgetItem(choice.participant_name) participant_item = QtWidgets.QTableWidgetItem(choice.participant_name)
priority_item = QtWidgets.QTableWidgetItem(choice.priority_sector) priority_item = QtWidgets.QTableWidgetItem(choice.priority_sector)
if choice.priority_icon:
priority_item.setIcon(choice.priority_icon)
secondary_item = QtWidgets.QTableWidgetItem(choice.secondary_sector) secondary_item = QtWidgets.QTableWidgetItem(choice.secondary_sector)
if choice.secondary_icon:
secondary_item.setIcon(choice.secondary_icon)
status_item = QtWidgets.QTableWidgetItem()
if choice.fallback_icon:
status_item.setIcon(choice.fallback_icon)
comment_item = QtWidgets.QTableWidgetItem(choice.comment)
participant_item.setData(Qt.ItemDataRole.UserRole, choice.id) participant_item.setData(Qt.ItemDataRole.UserRole, choice.id)
table.setItem(row, 0, participant_item) table.setItem(row, 0, participant_item)
table.setItem(row, 1, priority_item) table.setItem(row, 1, priority_item)
table.setItem(row, 2, secondary_item) table.setItem(row, 2, secondary_item)
table.setItem(row, 3, status_item)
table.setItem(row, 4, comment_item)
table.setSortingEnabled(True) table.setSortingEnabled(True)
table.resizeColumnsToContents() table.resizeColumnsToContents()

View file

@ -273,44 +273,44 @@
"choices": [ "choices": [
{ {
"participant_id": "602e2eaf-297e-490b-b0e9-efec818e466a", "participant_id": "602e2eaf-297e-490b-b0e9-efec818e466a",
"priority_sector_id": null, "priority_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
"secondary_sector_id": null, "secondary_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
"comment": null "comment": null
}, },
{ {
"participant_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de", "participant_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de",
"priority_sector_id": null, "priority_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
"secondary_sector_id": null, "secondary_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
"comment": null "comment": null
}, },
{ {
"participant_id": "237c1291-4331-4242-bd70-bf648185a627", "participant_id": "237c1291-4331-4242-bd70-bf648185a627",
"priority_sector_id": null, "priority_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
"secondary_sector_id": null, "secondary_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
"comment": null "comment": null
}, },
{ {
"participant_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0", "participant_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0",
"priority_sector_id": null, "priority_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
"secondary_sector_id": null, "secondary_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
"comment": null "comment": null
} }
], ],
"battles": [ "battles": [
{ {
"sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f", "sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
"player_1_id": "602e2eaf-297e-490b-b0e9-efec818e466a", "player_1_id": "237c1291-4331-4242-bd70-bf648185a627",
"player_2_id": "237c1291-4331-4242-bd70-bf648185a627", "player_2_id": "602e2eaf-297e-490b-b0e9-efec818e466a",
"winner_id": null, "winner_id": null,
"score": null, "score": null,
"victory_condition": "tie", "victory_condition": "Tie",
"comment": "Never finished..." "comment": "Never finished..."
}, },
{ {
"sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499", "sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
"player_1_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de", "player_1_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de",
"player_2_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0", "player_2_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0",
"winner_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0", "winner_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de",
"score": "4/2", "score": "4/2",
"victory_condition": "Mission", "victory_condition": "Mission",
"comment": "Decisive fast attack impossible to resist." "comment": "Decisive fast attack impossible to resist."