Compare commits
10 commits
db78c6dacc
...
b7a35f6712
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b7a35f6712 | ||
|
|
719b0128ed | ||
|
|
e7e2de9d0a | ||
|
|
956fa63c0b | ||
|
|
3d0d7874e3 | ||
|
|
0081e52e9a | ||
|
|
f5ad45f671 | ||
|
|
4396b15c3a | ||
|
|
9e602e8ca4 | ||
|
|
aa75a5b84f |
23 changed files with 932 additions and 536 deletions
|
|
@ -22,6 +22,7 @@ class IconName(StrEnum):
|
|||
PAIRING = auto()
|
||||
DRAW = auto()
|
||||
TIEBREAK = auto()
|
||||
DRAWTOKEN = auto()
|
||||
DELETE = auto()
|
||||
SAVE_AS = auto()
|
||||
SAVE = auto()
|
||||
|
|
@ -49,7 +50,8 @@ class IconName(StrEnum):
|
|||
NP1ST = auto()
|
||||
NP2ND = auto()
|
||||
NP3RD = auto()
|
||||
TIEBREAK_TOKEN = auto()
|
||||
WINTOKEN = auto()
|
||||
TIEBREAKTOKEN = auto()
|
||||
VP1STDRAW = auto()
|
||||
VP1STBREAK = auto()
|
||||
VP1STTIEDRAW = auto()
|
||||
|
|
@ -71,6 +73,9 @@ class IconName(StrEnum):
|
|||
NP3RDDRAW = auto()
|
||||
NP3RDBREAK = auto()
|
||||
NP3RDTIEDRAW = auto()
|
||||
ALLOCATED = auto()
|
||||
ALLOCATEDTOKEN = auto()
|
||||
FALLBACK = auto()
|
||||
|
||||
|
||||
VP_RANK_TO_ICON = {
|
||||
|
|
@ -125,6 +130,8 @@ class Icons:
|
|||
IconName.NP2ND: "medal-silver.png",
|
||||
IconName.NP3RD: "medal-bronze.png",
|
||||
IconName.WARCHRONBACK: "warchron_background.png",
|
||||
IconName.ALLOCATED: "map.png",
|
||||
IconName.FALLBACK: "cross-script.png",
|
||||
}
|
||||
|
||||
@classmethod
|
||||
|
|
@ -138,11 +145,21 @@ class Icons:
|
|||
def get_pixmap(cls, name: IconName) -> QPixmap:
|
||||
if name in cls._pixmap_cache:
|
||||
return cls._pixmap_cache[name]
|
||||
if name == IconName.TIEBREAK_TOKEN:
|
||||
if name == IconName.TIEBREAKTOKEN:
|
||||
pix = cls._compose(
|
||||
cls.get_pixmap(IconName.TIEBREAK),
|
||||
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:
|
||||
pix = cls._compose(
|
||||
cls.get_pixmap(IconName.VP1ST),
|
||||
|
|
@ -255,6 +272,11 @@ class Icons:
|
|||
cls.get_pixmap(IconName.DRAW),
|
||||
cls.get_pixmap(IconName.TOKEN),
|
||||
)
|
||||
elif name == IconName.ALLOCATEDTOKEN:
|
||||
pix = cls._compose(
|
||||
cls.get_pixmap(IconName.ALLOCATED),
|
||||
cls.get_pixmap(IconName.TOKEN),
|
||||
)
|
||||
else:
|
||||
path = RESOURCES_DIR / cls._paths[name]
|
||||
pix = QPixmap(path.as_posix())
|
||||
|
|
@ -313,3 +335,16 @@ class ContextType(StrEnum):
|
|||
class ScoreKind(Enum):
|
||||
VP = auto()
|
||||
NP = auto()
|
||||
|
||||
|
||||
class ChoiceStatus(StrEnum):
|
||||
NONE = auto()
|
||||
TOKEN = auto()
|
||||
ALLOCATED = auto()
|
||||
ALLOCATEDTOKEN = auto()
|
||||
|
||||
|
||||
class AllocationType(Enum):
|
||||
PRIORITY = auto()
|
||||
SECONDARY = auto()
|
||||
FALLBACK = auto()
|
||||
|
|
|
|||
|
|
@ -357,12 +357,20 @@ class AppController:
|
|||
except RequiresConfirmation as e:
|
||||
reply = QMessageBox.question(
|
||||
self.view,
|
||||
"Confirm deletion",
|
||||
"Confirm update",
|
||||
str(e),
|
||||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||||
)
|
||||
if reply == QMessageBox.StandardButton.Yes:
|
||||
try:
|
||||
e.action()
|
||||
except DomainError as inner:
|
||||
QMessageBox.warning(
|
||||
self.view,
|
||||
"Update forbidden",
|
||||
str(inner),
|
||||
)
|
||||
return
|
||||
else:
|
||||
return
|
||||
self.is_dirty = True
|
||||
|
|
|
|||
|
|
@ -19,10 +19,11 @@ from warchron.model.war import War
|
|||
from warchron.model.campaign import Campaign
|
||||
from warchron.model.campaign_participant import CampaignParticipant
|
||||
from warchron.model.sector import Sector
|
||||
from warchron.model.tie_manager import TieContext, TieResolver
|
||||
from warchron.model.score_service import ScoreService
|
||||
from warchron.controller.closure_workflow import CampaignClosureWorkflow
|
||||
from warchron.controller.ranking_icon import RankingIcon
|
||||
from warchron.model.tiebreaking import TieContext, TieBreaker
|
||||
from warchron.model.scoring import ScoreComputer
|
||||
from warchron.controller.workflows import CampaignClosureWorkflow
|
||||
|
||||
from warchron.controller.presenter import Presenter
|
||||
from warchron.view.campaign_dialog import CampaignDialog
|
||||
from warchron.view.campaign_participant_dialog import CampaignParticipantDialog
|
||||
from warchron.view.sector_dialog import SectorDialog
|
||||
|
|
@ -52,16 +53,16 @@ class CampaignController:
|
|||
for sect in sectors
|
||||
]
|
||||
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] = []
|
||||
vp_icon_map: Dict[str, QIcon] = {}
|
||||
objective_icon_maps: Dict[str, Dict[str, QIcon]] = {}
|
||||
if camp.is_over:
|
||||
vp_icon_map = RankingIcon.compute_icons(
|
||||
vp_icon_map = Presenter.compute_ranking_icons(
|
||||
war, ContextType.CAMPAIGN, campaign_id, scores
|
||||
)
|
||||
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,
|
||||
ContextType.CAMPAIGN,
|
||||
camp.id,
|
||||
|
|
@ -170,32 +171,25 @@ class CampaignController:
|
|||
) -> 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)
|
||||
active = TieBreaker.get_active_participants(war, ctx, ctx.participants)
|
||||
players = [
|
||||
ParticipantOption(id=pid, name=self.app.model.get_participant_name(pid))
|
||||
for pid in active
|
||||
]
|
||||
counters = [war.get_influence_tokens(pid) for pid in active]
|
||||
dialog = TieDialog(
|
||||
parent=self.app.view,
|
||||
players=players,
|
||||
counters=counters,
|
||||
context_type=ctx.context_type,
|
||||
context_id=ctx.context_id,
|
||||
context_name=None,
|
||||
data = Presenter.build_dialog_data(
|
||||
war, ctx, campaign=war.get_campaign(ctx.context_id)
|
||||
)
|
||||
if ctx.objective_id:
|
||||
objective = war.objectives[ctx.objective_id]
|
||||
dialog = TieDialog(
|
||||
parent=self.app.view,
|
||||
players=players,
|
||||
counters=counters,
|
||||
context_type=ctx.context_type,
|
||||
context_id=ctx.context_id,
|
||||
context_name=f"Objective tie: {objective.name}",
|
||||
context_name=data.title,
|
||||
)
|
||||
if not dialog.exec():
|
||||
TieResolver.cancel_tie_break(war, ctx)
|
||||
TieBreaker.cancel_tie_break(war, ctx)
|
||||
raise AbortedOperation("Tie resolution cancelled")
|
||||
bids_map[ctx.key()] = dialog.get_bids()
|
||||
return bids_map
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -88,6 +88,9 @@ class ChoiceDTO:
|
|||
priority_sector: str | None
|
||||
secondary_sector: 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)
|
||||
|
|
@ -103,8 +106,6 @@ class BattleDTO:
|
|||
state_icon: QIcon | None
|
||||
player1_icon: QIcon | None
|
||||
player2_icon: QIcon | None
|
||||
player1_tooltip: str | None = None
|
||||
player2_tooltip: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -132,3 +133,8 @@ class WarParticipantScoreDTO:
|
|||
tokens: int
|
||||
rank_icon: QIcon | None = None
|
||||
objective_icons: Dict[str, QIcon] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TieDialogData:
|
||||
title: str
|
||||
|
|
|
|||
178
src/warchron/controller/presenter.py
Normal file
178
src/warchron/controller/presenter.py
Normal 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),
|
||||
)
|
||||
|
|
@ -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
|
||||
|
|
@ -4,19 +4,27 @@ from PyQt6.QtWidgets import QDialog
|
|||
from PyQt6.QtWidgets import QMessageBox
|
||||
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 (
|
||||
AbortedOperation,
|
||||
DomainError,
|
||||
RequiresConfirmation,
|
||||
)
|
||||
from warchron.model.tie_manager import TieResolver, TieContext
|
||||
from warchron.model.result_checker import ResultChecker
|
||||
from warchron.model.tiebreaking import TieBreaker, TieContext
|
||||
from warchron.model.pairing import Pairing
|
||||
from warchron.model.round import Round
|
||||
from warchron.model.war import War
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from warchron.controller.app_controller import AppController
|
||||
from warchron.model.campaign import Campaign
|
||||
|
||||
from warchron.controller.dtos import (
|
||||
ParticipantOption,
|
||||
|
|
@ -24,10 +32,11 @@ from warchron.controller.dtos import (
|
|||
ChoiceDTO,
|
||||
BattleDTO,
|
||||
)
|
||||
from warchron.controller.closure_workflow import (
|
||||
from warchron.controller.workflows import (
|
||||
RoundClosureWorkflow,
|
||||
RoundPairingWorkflow,
|
||||
)
|
||||
from warchron.controller.presenter import Presenter
|
||||
from warchron.view.choice_dialog import ChoiceDialog
|
||||
from warchron.view.battle_dialog import BattleDialog
|
||||
from warchron.view.tie_dialog import TieDialog
|
||||
|
|
@ -48,9 +57,17 @@ class RoundController:
|
|||
for part in participants:
|
||||
choice = rnd.get_choice(part.id)
|
||||
if not choice:
|
||||
try:
|
||||
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
|
||||
|
|
@ -61,6 +78,24 @@ class RoundController:
|
|||
if choice.secondary_sector_id is not None
|
||||
else ""
|
||||
)
|
||||
priority_icon = None
|
||||
secondary_icon = None
|
||||
fallback_icon = None
|
||||
alloc = Pairing.get_round_allocation(
|
||||
war,
|
||||
rnd,
|
||||
part.id,
|
||||
)
|
||||
if alloc.priority != ChoiceStatus.NONE:
|
||||
priority_icon = QIcon(
|
||||
Icons.get_pixmap(IconName[alloc.priority.name])
|
||||
)
|
||||
if alloc.secondary != ChoiceStatus.NONE:
|
||||
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,
|
||||
|
|
@ -70,18 +105,27 @@ class RoundController:
|
|||
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)
|
||||
battles_for_display: List[BattleDTO] = []
|
||||
for sect in sectors:
|
||||
battle = rnd.get_battle(sect.id)
|
||||
|
||||
if not battle:
|
||||
try:
|
||||
battle = self.app.model.create_battle(
|
||||
round_id=rnd.id, sector_id=sect.id
|
||||
)
|
||||
except DomainError as e:
|
||||
QMessageBox.warning(
|
||||
self.app.view,
|
||||
"Create forbidden",
|
||||
str(e),
|
||||
)
|
||||
if battle:
|
||||
state_icon = Icons.get(IconName.ONGOING)
|
||||
if battle.is_finished():
|
||||
state_icon = Icons.get(IconName.DONE)
|
||||
|
|
@ -90,7 +134,6 @@ 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:
|
||||
|
|
@ -98,7 +141,6 @@ 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:
|
||||
|
|
@ -108,37 +150,9 @@ class RoundController:
|
|||
)
|
||||
else:
|
||||
winner_name = ""
|
||||
p1_icon = None
|
||||
p2_icon = None
|
||||
p1_tooltip = None
|
||||
p2_tooltip = None
|
||||
if battle.is_draw():
|
||||
p1_icon = Icons.get(IconName.DRAW)
|
||||
p2_icon = Icons.get(IconName.DRAW)
|
||||
context = TieContext(
|
||||
ContextType.BATTLE,
|
||||
battle.sector_id,
|
||||
[p1_id, p2_id],
|
||||
p1_icon, p2_icon = Presenter.compute_battle_icons(
|
||||
war, round_id, battle.sector_id
|
||||
)
|
||||
if TieResolver.was_tie_broken_by_tokens(war, context):
|
||||
effective_winner = ResultChecker.get_effective_winner_id(
|
||||
war, ContextType.BATTLE, battle.sector_id, None
|
||||
)
|
||||
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,
|
||||
|
|
@ -152,8 +166,6 @@ class RoundController:
|
|||
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)
|
||||
|
|
@ -243,16 +255,29 @@ 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
|
||||
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(
|
||||
parent=self.app.view,
|
||||
players=players,
|
||||
counters=counters,
|
||||
context_type=ctx.context_type,
|
||||
context_id=ctx.context_id,
|
||||
context_name=data.title,
|
||||
)
|
||||
if not dialog.exec():
|
||||
TieResolver.cancel_tie_break(war, ctx)
|
||||
TieBreaker.cancel_tie_break(war, ctx)
|
||||
raise AbortedOperation("Tie resolution cancelled")
|
||||
bids_map[ctx.key()] = dialog.get_bids()
|
||||
return bids_map
|
||||
|
|
|
|||
|
|
@ -24,10 +24,11 @@ from warchron.controller.dtos import (
|
|||
from warchron.model.war import War
|
||||
from warchron.model.war_participant import WarParticipant
|
||||
from warchron.model.objective import Objective
|
||||
from warchron.model.tie_manager import TieContext, TieResolver
|
||||
from warchron.model.score_service import ScoreService
|
||||
from warchron.controller.closure_workflow import WarClosureWorkflow
|
||||
from warchron.controller.ranking_icon import RankingIcon
|
||||
from warchron.model.tiebreaking import TieContext, TieBreaker
|
||||
from warchron.model.scoring import ScoreComputer
|
||||
from warchron.controller.workflows import WarClosureWorkflow
|
||||
|
||||
from warchron.controller.presenter import Presenter
|
||||
from warchron.view.war_dialog import WarDialog
|
||||
from warchron.view.objective_dialog import ObjectiveDialog
|
||||
from warchron.view.war_participant_dialog import WarParticipantDialog
|
||||
|
|
@ -56,16 +57,16 @@ class WarController:
|
|||
ObjectiveDTO(id=obj.id, name=obj.name, description=obj.description)
|
||||
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] = []
|
||||
vp_icon_map: Dict[str, QIcon] = {}
|
||||
objective_icon_maps: Dict[str, Dict[str, QIcon]] = {}
|
||||
if war.is_over:
|
||||
vp_icon_map = RankingIcon.compute_icons(
|
||||
vp_icon_map = Presenter.compute_ranking_icons(
|
||||
war, ContextType.WAR, war_id, scores
|
||||
)
|
||||
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,
|
||||
ContextType.WAR,
|
||||
war.id,
|
||||
|
|
@ -160,7 +161,7 @@ class WarController:
|
|||
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
|
||||
bids_map = {}
|
||||
for ctx in contexts:
|
||||
active = TieResolver.get_active_participants(
|
||||
active = TieBreaker.get_active_participants(
|
||||
war,
|
||||
ctx,
|
||||
ctx.participants,
|
||||
|
|
@ -170,26 +171,20 @@ class WarController:
|
|||
for pid in active
|
||||
]
|
||||
counters = [war.get_influence_tokens(pid) for pid in active]
|
||||
dialog = TieDialog(
|
||||
parent=self.app.view,
|
||||
players=players,
|
||||
counters=counters,
|
||||
context_type=ctx.context_type,
|
||||
context_id=ctx.context_id,
|
||||
context_name=None,
|
||||
data = Presenter.build_dialog_data(
|
||||
war,
|
||||
ctx,
|
||||
)
|
||||
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}",
|
||||
context_name=data.title,
|
||||
)
|
||||
if not dialog.exec():
|
||||
TieResolver.cancel_tie_break(war, ctx)
|
||||
TieBreaker.cancel_tie_break(war, ctx)
|
||||
raise AbortedOperation("Tie resolution cancelled")
|
||||
bids_map[ctx.key()] = dialog.get_bids()
|
||||
return bids_map
|
||||
|
|
|
|||
116
src/warchron/controller/workflows.py
Normal file
116
src/warchron/controller/workflows.py
Normal 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,
|
||||
)
|
||||
|
|
@ -144,8 +144,8 @@ class Campaign:
|
|||
|
||||
def cleanup() -> None:
|
||||
for rnd in rounds_blocking:
|
||||
rnd.clear_participant_references(participant_id)
|
||||
rnd.remove_choice(participant_id)
|
||||
rnd.clear_participant_references(participant_id)
|
||||
del self.participants[participant_id]
|
||||
|
||||
rounds_str = ", ".join(
|
||||
|
|
@ -263,8 +263,8 @@ class Campaign:
|
|||
|
||||
def cleanup_and_update() -> None:
|
||||
for rnd in affected_rounds:
|
||||
rnd.clear_sector_references(sector_id)
|
||||
rnd.remove_battle(sector_id)
|
||||
rnd.clear_sector_references(sector_id)
|
||||
apply_update()
|
||||
|
||||
rounds_str = ", ".join(
|
||||
|
|
@ -299,8 +299,8 @@ class Campaign:
|
|||
|
||||
def cleanup() -> None:
|
||||
for rnd in rounds_blocking:
|
||||
rnd.clear_sector_references(sector_id)
|
||||
rnd.remove_battle(sector_id)
|
||||
rnd.clear_sector_references(sector_id)
|
||||
del self.sectors[sector_id]
|
||||
|
||||
rounds_str = ", ".join(
|
||||
|
|
@ -345,11 +345,9 @@ class Campaign:
|
|||
return index
|
||||
raise KeyError("Round not found in campaign")
|
||||
|
||||
# TODO replace multiloops by internal has_* method
|
||||
def get_round_by_battle(self, sector_id: str) -> Round:
|
||||
for rnd in self.rounds:
|
||||
for bat in rnd.battles.values():
|
||||
if bat.sector_id == sector_id:
|
||||
if rnd.has_battle_with_sector(sector_id):
|
||||
return rnd
|
||||
raise KeyError(f"Battle {sector_id} not found in any Round")
|
||||
|
||||
|
|
@ -408,6 +406,12 @@ class Campaign:
|
|||
rnd = self.get_round(round_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(
|
||||
self,
|
||||
round_id: str,
|
||||
|
|
|
|||
|
|
@ -5,10 +5,10 @@ from collections import defaultdict
|
|||
from warchron.constants import ContextType, ScoreKind
|
||||
from warchron.model.war import War
|
||||
from warchron.model.war_event import TieResolved
|
||||
from warchron.model.tie_manager import TieResolver, TieContext
|
||||
from warchron.model.tiebreaking import TieBreaker, TieContext
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from warchron.model.score_service import ParticipantScore
|
||||
from warchron.model.scoring import ParticipantScore
|
||||
|
||||
|
||||
class ResultChecker:
|
||||
|
|
@ -75,29 +75,29 @@ class ResultChecker:
|
|||
current_rank += 1
|
||||
continue
|
||||
# normal tie-break if tie persists
|
||||
if not TieResolver.is_tie_resolved(war, context):
|
||||
if not TieBreaker.is_tie_resolved(war, context):
|
||||
ranking.append(
|
||||
(current_rank, subgroup, {pid: 0 for pid in subgroup})
|
||||
)
|
||||
current_rank += 1
|
||||
continue
|
||||
groups = TieResolver.rank_by_tokens(war, context, subgroup)
|
||||
tokens_spent = TieResolver.tokens_spent_map(war, context, subgroup)
|
||||
groups = TieBreaker.rank_by_tokens(war, context, subgroup)
|
||||
tokens_spent = TieBreaker.tokens_spent_map(war, context, subgroup)
|
||||
for group in groups:
|
||||
group_tokens = {pid: tokens_spent[pid] for pid in group}
|
||||
ranking.append((current_rank, group, group_tokens))
|
||||
current_rank += 1
|
||||
continue
|
||||
# no tie
|
||||
if len(participants) == 1 or not TieResolver.is_tie_resolved(war, context):
|
||||
if len(participants) == 1 or not TieBreaker.is_tie_resolved(war, context):
|
||||
ranking.append(
|
||||
(current_rank, participants, {pid: 0 for pid in participants})
|
||||
)
|
||||
current_rank += 1
|
||||
continue
|
||||
# apply token ranking
|
||||
groups = TieResolver.rank_by_tokens(war, context, participants)
|
||||
tokens_spent = TieResolver.tokens_spent_map(war, context, participants)
|
||||
groups = TieBreaker.rank_by_tokens(war, context, participants)
|
||||
tokens_spent = TieBreaker.tokens_spent_map(war, context, participants)
|
||||
for group in groups:
|
||||
group_tokens = {pid: tokens_spent[pid] for pid in group}
|
||||
ranking.append((current_rank, group, group_tokens))
|
||||
|
|
@ -111,13 +111,13 @@ class ResultChecker:
|
|||
value_getter: Callable[[ParticipantScore], int],
|
||||
subcontexts: List[TieContext],
|
||||
) -> List[List[str]]:
|
||||
from warchron.model.score_service import ScoreService
|
||||
from warchron.model.scoring import ScoreComputer
|
||||
|
||||
rank_map: Dict[str, Tuple[int, ...]] = {}
|
||||
for pid in participants:
|
||||
ranks: List[int] = []
|
||||
for sub in subcontexts:
|
||||
scores = ScoreService.compute_scores(
|
||||
scores = ScoreComputer.compute_scores(
|
||||
war, sub.context_type, sub.context_id
|
||||
)
|
||||
ranking = ResultChecker.get_effective_ranking(
|
||||
|
|
@ -9,7 +9,7 @@ from warchron.model.round import Round
|
|||
from warchron.model.battle import Battle
|
||||
|
||||
|
||||
class ClosureService:
|
||||
class Closer:
|
||||
|
||||
# Round methods
|
||||
|
||||
|
|
@ -24,7 +24,7 @@ class ClosureService:
|
|||
|
||||
@staticmethod
|
||||
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(
|
||||
isinstance(e, InfluenceGained) and e.context_id == battle.sector_id
|
||||
|
|
@ -1,10 +1,11 @@
|
|||
from __future__ import annotations
|
||||
from typing import Dict, List, Callable, Tuple
|
||||
from typing import Dict, List
|
||||
from dataclasses import dataclass
|
||||
from uuid import uuid4
|
||||
|
||||
import random
|
||||
|
||||
from warchron.constants import ContextType, ScoreKind
|
||||
from warchron.constants import ContextType, ScoreKind, ChoiceStatus, AllocationType
|
||||
from warchron.model.exception import (
|
||||
DomainError,
|
||||
ForbiddenOperation,
|
||||
|
|
@ -13,15 +14,17 @@ from warchron.model.exception import (
|
|||
from warchron.model.war import War
|
||||
from warchron.model.round import Round
|
||||
from warchron.model.battle import Battle
|
||||
from warchron.model.score_service import ScoreService
|
||||
from warchron.model.tie_manager import TieResolver, TieContext
|
||||
from warchron.model.scoring import ScoreComputer
|
||||
from warchron.model.tiebreaking import TieBreaker, TieContext, ResolveTiesCallback
|
||||
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"]],
|
||||
Dict[Tuple[str, str, int | None, str | None, str | None], Dict[str, bool]],
|
||||
]
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AllocationResult:
|
||||
priority: ChoiceStatus
|
||||
secondary: ChoiceStatus
|
||||
fallback: bool
|
||||
|
||||
|
||||
class Pairing:
|
||||
|
|
@ -39,15 +42,6 @@ class Pairing:
|
|||
raise DomainError(
|
||||
"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:
|
||||
for bat in round.battles.values():
|
||||
|
|
@ -66,6 +60,15 @@ class Pairing:
|
|||
"Do you want to continue?",
|
||||
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
|
||||
def assign_battles_to_participants(
|
||||
|
|
@ -76,7 +79,7 @@ class Pairing:
|
|||
campaign = war.get_campaign_by_round(round.id)
|
||||
if campaign is None:
|
||||
raise DomainError(f"Campaign for round {round.id} doesn't exist")
|
||||
scores = ScoreService.compute_scores(
|
||||
scores = ScoreComputer.compute_scores(
|
||||
war,
|
||||
ContextType.CAMPAIGN,
|
||||
campaign.id,
|
||||
|
|
@ -85,7 +88,7 @@ class Pairing:
|
|||
def value_getter(score: ParticipantScore) -> int:
|
||||
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] = {
|
||||
b.sector_id: b for b in round.battles.values()
|
||||
}
|
||||
|
|
@ -198,13 +201,11 @@ class Pairing:
|
|||
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):
|
||||
active = TieResolver.get_active_participants(
|
||||
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
|
||||
)
|
||||
if len(active) <= 1:
|
||||
break
|
||||
current_context = TieContext(
|
||||
context_type=context.context_type,
|
||||
context_id=context.context_id,
|
||||
|
|
@ -213,9 +214,12 @@ class Pairing:
|
|||
score_kind=context.score_kind,
|
||||
sector_id=context.sector_id,
|
||||
)
|
||||
# natural or unbreakable draw
|
||||
if not TieResolver.can_tie_be_resolved(
|
||||
# natural, unbreakable or acceptable (enough places) draw
|
||||
if (
|
||||
not TieBreaker.can_tie_be_resolved(
|
||||
war, context, current_context.participants
|
||||
)
|
||||
or len(active) <= places
|
||||
):
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
|
|
@ -245,9 +249,9 @@ class Pairing:
|
|||
)
|
||||
)
|
||||
break
|
||||
TieResolver.apply_bids(war, context, tie_id, bids)
|
||||
TieResolver.resolve_tie_state(war, context, tie_id, bids)
|
||||
ranked_groups = TieResolver.rank_by_tokens(
|
||||
TieBreaker.apply_bids(war, context, tie_id, bids)
|
||||
TieBreaker.resolve_tie_state(war, context, tie_id, bids)
|
||||
ranked_groups = TieBreaker.rank_by_tokens(
|
||||
war,
|
||||
context,
|
||||
context.participants,
|
||||
|
|
@ -277,3 +281,80 @@ class Pairing:
|
|||
remaining.remove(pid)
|
||||
continue
|
||||
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,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -5,7 +5,12 @@ from typing import Any, Dict, List, TYPE_CHECKING
|
|||
if TYPE_CHECKING:
|
||||
from warchron.model.campaign import Campaign
|
||||
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.battle import Battle
|
||||
|
||||
|
|
@ -74,7 +79,6 @@ 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(
|
||||
|
|
@ -93,11 +97,16 @@ 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:
|
||||
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_secondary(secondary_sector_id)
|
||||
choice.set_comment(comment)
|
||||
|
|
@ -118,9 +127,9 @@ class Round:
|
|||
if participant_id not in self.choices:
|
||||
return
|
||||
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.has_battle_with_participant(participant_id):
|
||||
raise ForbiddenOperation("Can't remove choice already assigned to battle.")
|
||||
self.war.revert_choice_ties(
|
||||
self.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:
|
||||
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:
|
||||
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:
|
||||
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)
|
||||
|
|
@ -173,12 +193,15 @@ class Round:
|
|||
victory_condition: str | None,
|
||||
comment: str | None,
|
||||
) -> None:
|
||||
from warchron.model.tiebreaking import TieBreaker
|
||||
|
||||
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:
|
||||
if not 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_2(player_2_id)
|
||||
bat.set_winner(winner_id)
|
||||
|
|
@ -186,6 +209,48 @@ class Round:
|
|||
bat.set_victory_condition(victory_condition)
|
||||
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:
|
||||
for battle in self.battles.values():
|
||||
trigger_revert_ties = False
|
||||
|
|
@ -209,11 +274,9 @@ class Round:
|
|||
if sector_id not in self.battles:
|
||||
return
|
||||
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.")
|
||||
self.war.revert_battle_ties(self.id, sector_id=sector_id)
|
||||
del self.battles[sector_id]
|
||||
|
|
|
|||
|
|
@ -4,7 +4,9 @@ from collections import defaultdict
|
|||
|
||||
from warchron.constants import ContextType
|
||||
from warchron.model.war import War
|
||||
from warchron.model.campaign import Campaign
|
||||
from warchron.model.battle import Battle
|
||||
from warchron.model.exception import DomainError
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
|
|
@ -13,7 +15,7 @@ class ParticipantScore:
|
|||
narrative_points: Dict[str, int] = field(default_factory=dict)
|
||||
|
||||
|
||||
class ScoreService:
|
||||
class ScoreComputer:
|
||||
|
||||
@staticmethod
|
||||
def _get_battles_for_context(
|
||||
|
|
@ -26,7 +28,8 @@ class ScoreService:
|
|||
continue
|
||||
yield from rnd.battles.values()
|
||||
elif context_type == ContextType.CAMPAIGN:
|
||||
campaign = war.get_campaign(context_id)
|
||||
campaign: Campaign | None = war.get_campaign(context_id)
|
||||
if campaign:
|
||||
for rnd in campaign.rounds:
|
||||
if not rnd.is_over:
|
||||
continue
|
||||
|
|
@ -34,6 +37,8 @@ class ScoreService:
|
|||
elif context_type == ContextType.BATTLE:
|
||||
battle = war.get_battle(context_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)
|
||||
if rnd and rnd.is_over:
|
||||
yield battle
|
||||
|
|
@ -42,7 +47,7 @@ class ScoreService:
|
|||
def compute_scores(
|
||||
war: War, context_type: ContextType, context_id: str
|
||||
) -> Dict[str, ParticipantScore]:
|
||||
from warchron.model.result_checker import ResultChecker
|
||||
from warchron.model.checking import ResultChecker
|
||||
|
||||
if context_type == ContextType.CAMPAIGN:
|
||||
camp = war.get_campaign(context_id)
|
||||
|
|
@ -56,7 +61,7 @@ class ScoreService:
|
|||
)
|
||||
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:
|
||||
base_winner = None
|
||||
if battle.winner_id is not None:
|
||||
|
|
@ -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 collections import defaultdict
|
||||
from uuid import uuid4
|
||||
|
||||
from warchron.constants import ContextType, ScoreKind
|
||||
from warchron.model.exception import ForbiddenOperation, DomainError
|
||||
from warchron.model.war import War
|
||||
from warchron.model.war_event import InfluenceSpent, TieResolved
|
||||
from warchron.model.score_service import ScoreService, ParticipantScore
|
||||
from warchron.model.scoring import ScoreComputer, ParticipantScore
|
||||
|
||||
|
||||
@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
|
||||
def find_active_tie_id(
|
||||
|
|
@ -65,10 +72,10 @@ class TieResolver:
|
|||
context: TieContext = TieContext(
|
||||
ContextType.BATTLE, battle.sector_id, [p1_id, p2_id]
|
||||
)
|
||||
if TieResolver.is_tie_resolved(war, context):
|
||||
if TieBreaker.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]):
|
||||
tie_id = TieBreaker.find_active_tie_id(war, context)
|
||||
if not TieBreaker.can_tie_be_resolved(war, context, [p1_id, p2_id]):
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
participant_id=None,
|
||||
|
|
@ -92,14 +99,22 @@ class TieResolver:
|
|||
|
||||
@staticmethod
|
||||
def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]:
|
||||
scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
|
||||
buckets: DefaultDict[int, List[str]] = defaultdict(list)
|
||||
for pid, score in scores.items():
|
||||
buckets[score.victory_points].append(pid)
|
||||
from warchron.model.checking import ResultChecker
|
||||
|
||||
scores = ScoreComputer.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
|
||||
ranking = ResultChecker.get_effective_ranking(
|
||||
war,
|
||||
ContextType.CAMPAIGN,
|
||||
campaign_id,
|
||||
ScoreKind.VP,
|
||||
scores,
|
||||
lambda s: s.victory_points,
|
||||
)
|
||||
ties: List[TieContext] = []
|
||||
for score_value, participants in buckets.items():
|
||||
if len(participants) <= 1:
|
||||
for _, group, _ in ranking:
|
||||
if len(group) <= 1:
|
||||
continue
|
||||
score_value = scores[group[0]].victory_points
|
||||
context: TieContext = TieContext(
|
||||
ContextType.CAMPAIGN,
|
||||
campaign_id,
|
||||
|
|
@ -107,16 +122,16 @@ class TieResolver:
|
|||
score_value,
|
||||
ScoreKind.VP,
|
||||
)
|
||||
if TieResolver.is_tie_resolved(war, context):
|
||||
if TieBreaker.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):
|
||||
tie_id = TieBreaker.find_active_tie_id(war, context)
|
||||
if not TieBreaker.can_tie_be_resolved(war, context, group):
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
participant_id=None,
|
||||
context_type=ContextType.CAMPAIGN,
|
||||
context_id=campaign_id,
|
||||
participants=participants,
|
||||
participants=group,
|
||||
tie_id=tie_id,
|
||||
score_value=score_value,
|
||||
)
|
||||
|
|
@ -126,7 +141,7 @@ class TieResolver:
|
|||
TieContext(
|
||||
context_type=ContextType.CAMPAIGN,
|
||||
context_id=campaign_id,
|
||||
participants=participants,
|
||||
participants=group,
|
||||
score_value=score_value,
|
||||
score_kind=ScoreKind.VP,
|
||||
)
|
||||
|
|
@ -137,20 +152,32 @@ class TieResolver:
|
|||
def find_campaign_objective_ties(
|
||||
war: War, campaign_id: str, objective_id: str
|
||||
) -> List[TieContext]:
|
||||
scores = ScoreService.compute_scores(
|
||||
from warchron.model.checking import ResultChecker
|
||||
|
||||
scores = ScoreComputer.compute_scores(
|
||||
war,
|
||||
ContextType.CAMPAIGN,
|
||||
campaign_id,
|
||||
)
|
||||
buckets: DefaultDict[int, List[str]] = defaultdict(list)
|
||||
for pid, score in scores.items():
|
||||
np_value = score.narrative_points.get(objective_id, 0)
|
||||
buckets[np_value].append(pid)
|
||||
|
||||
def value_getter(score: ParticipantScore) -> int:
|
||||
return score.narrative_points.get(objective_id, 0)
|
||||
|
||||
ranking = ResultChecker.get_effective_ranking(
|
||||
war,
|
||||
ContextType.CAMPAIGN,
|
||||
campaign_id,
|
||||
ScoreKind.NP,
|
||||
scores,
|
||||
value_getter,
|
||||
objective_id,
|
||||
)
|
||||
ties: List[TieContext] = []
|
||||
context_id = campaign_id
|
||||
for np_value, participants in buckets.items():
|
||||
if len(participants) <= 1:
|
||||
for _, group, _ in ranking:
|
||||
if len(group) <= 1:
|
||||
continue
|
||||
np_value = value_getter(scores[group[0]])
|
||||
context: TieContext = TieContext(
|
||||
ContextType.CAMPAIGN,
|
||||
campaign_id,
|
||||
|
|
@ -159,20 +186,20 @@ class TieResolver:
|
|||
ScoreKind.NP,
|
||||
objective_id,
|
||||
)
|
||||
if TieResolver.is_tie_resolved(war, context):
|
||||
if TieBreaker.is_tie_resolved(war, context):
|
||||
continue
|
||||
tie_id = TieResolver.find_active_tie_id(war, context)
|
||||
if not TieResolver.can_tie_be_resolved(
|
||||
tie_id = TieBreaker.find_active_tie_id(war, context)
|
||||
if not TieBreaker.can_tie_be_resolved(
|
||||
war,
|
||||
context,
|
||||
participants,
|
||||
group,
|
||||
):
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
participant_id=None,
|
||||
context_type=ContextType.CAMPAIGN,
|
||||
context_id=context_id,
|
||||
participants=participants,
|
||||
participants=group,
|
||||
tie_id=tie_id,
|
||||
score_value=np_value,
|
||||
objective_id=objective_id,
|
||||
|
|
@ -183,7 +210,7 @@ class TieResolver:
|
|||
TieContext(
|
||||
context_type=ContextType.CAMPAIGN,
|
||||
context_id=context_id,
|
||||
participants=participants,
|
||||
participants=group,
|
||||
score_value=np_value,
|
||||
score_kind=ScoreKind.NP,
|
||||
objective_id=objective_id,
|
||||
|
|
@ -193,9 +220,9 @@ class TieResolver:
|
|||
|
||||
@staticmethod
|
||||
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(
|
||||
war,
|
||||
ContextType.WAR,
|
||||
|
|
@ -216,10 +243,10 @@ class TieResolver:
|
|||
score_value=score_value,
|
||||
score_kind=ScoreKind.VP,
|
||||
)
|
||||
if TieResolver.is_tie_resolved(war, context):
|
||||
if TieBreaker.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):
|
||||
tie_id = TieBreaker.find_active_tie_id(war, context)
|
||||
if not TieBreaker.can_tie_be_resolved(war, context, group):
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
participant_id=None,
|
||||
|
|
@ -244,9 +271,9 @@ class TieResolver:
|
|||
|
||||
@staticmethod
|
||||
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,
|
||||
ContextType.WAR,
|
||||
war.id,
|
||||
|
|
@ -272,13 +299,13 @@ class TieResolver:
|
|||
context: TieContext = TieContext(
|
||||
ContextType.WAR, war.id, [], np_value, ScoreKind.NP, objective_id
|
||||
)
|
||||
if TieResolver.is_tie_resolved(
|
||||
if TieBreaker.is_tie_resolved(
|
||||
war,
|
||||
context,
|
||||
):
|
||||
continue
|
||||
tie_id = TieResolver.find_active_tie_id(war, context)
|
||||
if not TieResolver.can_tie_be_resolved(
|
||||
tie_id = TieBreaker.find_active_tie_id(war, context)
|
||||
if not TieBreaker.can_tie_be_resolved(
|
||||
war,
|
||||
context,
|
||||
group,
|
||||
|
|
@ -307,6 +334,51 @@ class TieResolver:
|
|||
)
|
||||
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
|
||||
def apply_bids(
|
||||
war: War,
|
||||
|
|
@ -327,6 +399,7 @@ class TieResolver:
|
|||
context_id=context.context_id,
|
||||
tie_id=tie_id,
|
||||
objective_id=context.objective_id,
|
||||
sector_id=context.sector_id,
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -397,7 +470,7 @@ class TieResolver:
|
|||
context: TieContext,
|
||||
participants: List[str],
|
||||
) -> List[str]:
|
||||
groups = TieResolver.rank_by_tokens(war, context, participants)
|
||||
groups = TieBreaker.rank_by_tokens(war, context, participants)
|
||||
return groups[0]
|
||||
|
||||
@staticmethod
|
||||
|
|
@ -418,10 +491,11 @@ class TieResolver:
|
|||
tie_id=tie_id,
|
||||
score_value=context.score_value,
|
||||
objective_id=context.objective_id,
|
||||
sector_id=context.sector_id,
|
||||
)
|
||||
)
|
||||
return
|
||||
groups = TieResolver.rank_by_tokens(war, context, context.participants)
|
||||
groups = TieBreaker.rank_by_tokens(war, context, context.participants)
|
||||
if len(groups[0]) == 1:
|
||||
war.events.append(
|
||||
TieResolved(
|
||||
|
|
@ -432,6 +506,7 @@ class TieResolver:
|
|||
tie_id=tie_id,
|
||||
score_value=context.score_value,
|
||||
objective_id=context.objective_id,
|
||||
sector_id=context.sector_id,
|
||||
)
|
||||
)
|
||||
return
|
||||
|
|
@ -441,25 +516,9 @@ class TieResolver:
|
|||
def can_tie_be_resolved(
|
||||
war: War, context: TieContext, participants: List[str]
|
||||
) -> 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)
|
||||
|
||||
@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
|
||||
def is_tie_resolved(war: War, context: TieContext) -> bool:
|
||||
for ev in war.events:
|
||||
|
|
@ -483,3 +542,26 @@ class TieResolver:
|
|||
continue
|
||||
return True
|
||||
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
|
||||
|
|
@ -10,7 +10,11 @@ from warchron.model.war_event import (
|
|||
InfluenceSpent,
|
||||
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.objective import Objective
|
||||
from warchron.model.campaign_participant import CampaignParticipant
|
||||
|
|
@ -80,6 +84,10 @@ class War:
|
|||
if ev.context_type == ContextType.BATTLE:
|
||||
battle = self.get_battle(ev.context_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.is_over = False
|
||||
elif ev.context_type == ContextType.CAMPAIGN:
|
||||
|
|
@ -304,21 +312,17 @@ class War:
|
|||
return camp
|
||||
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:
|
||||
for camp in self.campaigns:
|
||||
for rnd in camp.rounds:
|
||||
if rnd.id == round_id:
|
||||
if camp.has_round(round_id):
|
||||
return camp
|
||||
return None
|
||||
|
||||
# TODO replace multiloops by internal has_* method
|
||||
def get_campaign_by_sector(self, sector_id: str) -> Campaign:
|
||||
def get_campaign_by_sector(self, sector_id: str) -> Campaign | None:
|
||||
for camp in self.campaigns:
|
||||
for sect in camp.sectors.values():
|
||||
if sect.id == sector_id:
|
||||
if camp.has_sector(sector_id):
|
||||
return camp
|
||||
raise KeyError(f"Sector {sector_id} not found in any Campaign")
|
||||
return None
|
||||
|
||||
def get_campaign_by_campaign_participant(
|
||||
self, participant_id: str
|
||||
|
|
@ -375,12 +379,10 @@ class War:
|
|||
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:
|
||||
for camp in self.campaigns:
|
||||
for sect in camp.sectors.values():
|
||||
if sect.id == sector_id:
|
||||
return sect
|
||||
if camp.has_sector(sector_id):
|
||||
return camp.get_sector(sector_id)
|
||||
raise KeyError("Sector not found")
|
||||
|
||||
def update_sector(
|
||||
|
|
@ -396,6 +398,8 @@ class War:
|
|||
description: str | None,
|
||||
) -> None:
|
||||
camp = self.get_campaign_by_sector(sector_id)
|
||||
if not camp:
|
||||
raise DomainError(f"No campaign found for sector {sector_id}")
|
||||
camp.update_sector(
|
||||
sector_id,
|
||||
name=name,
|
||||
|
|
@ -409,6 +413,8 @@ class War:
|
|||
|
||||
def remove_sector(self, sector_id: str) -> None:
|
||||
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)
|
||||
|
||||
# Campaign participant methods
|
||||
|
|
@ -427,12 +433,10 @@ class War:
|
|||
camp = self.get_campaign(campaign_id)
|
||||
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:
|
||||
for camp in self.campaigns:
|
||||
for part in camp.participants.values():
|
||||
if part.id == participant_id:
|
||||
return part
|
||||
if camp.has_participant(participant_id):
|
||||
return camp.get_campaign_participant(participant_id)
|
||||
raise KeyError("Participant not found")
|
||||
|
||||
def update_campaign_participant(
|
||||
|
|
@ -449,12 +453,10 @@ class War:
|
|||
|
||||
# Round methods
|
||||
|
||||
# TODO replace multiloops by internal has_* method
|
||||
def get_round(self, round_id: str) -> Round:
|
||||
for camp in self.campaigns:
|
||||
for rnd in camp.rounds:
|
||||
if rnd.id == round_id:
|
||||
return rnd
|
||||
if camp.has_round(round_id):
|
||||
return camp.get_round(round_id)
|
||||
raise KeyError("Round not found")
|
||||
|
||||
def add_round(self, campaign_id: str) -> Round:
|
||||
|
|
@ -472,7 +474,7 @@ class War:
|
|||
camp = self.get_campaign_by_round(round_id)
|
||||
if camp is not None:
|
||||
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(
|
||||
self,
|
||||
|
|
@ -499,20 +501,18 @@ class War:
|
|||
|
||||
# Battle methods
|
||||
|
||||
# TODO replace multiloops by internal has_* method
|
||||
def get_battle(self, battle_id: str) -> Battle:
|
||||
for camp in self.campaigns:
|
||||
for rnd in camp.rounds:
|
||||
for bat in rnd.battles.values():
|
||||
if bat.sector_id == battle_id:
|
||||
return bat
|
||||
raise KeyError("Battle not found")
|
||||
battle = camp.get_battle(battle_id)
|
||||
if battle is not None:
|
||||
return battle
|
||||
raise KeyError(f"War did not find Battle {battle_id}")
|
||||
|
||||
def create_battle(self, round_id: str, sector_id: str) -> Battle:
|
||||
camp = self.get_campaign_by_round(round_id)
|
||||
if camp is not None:
|
||||
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(
|
||||
self,
|
||||
|
|
|
|||
BIN
src/warchron/view/resources/cross-script.png
Normal file
BIN
src/warchron/view/resources/cross-script.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 663 B |
BIN
src/warchron/view/resources/map.png
Normal file
BIN
src/warchron/view/resources/map.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 1.5 KiB |
|
|
@ -34,7 +34,7 @@ class TieDialog(QDialog):
|
|||
self.ui: Ui_tieDialog = Ui_tieDialog()
|
||||
self.ui.setupUi(self) # type: ignore
|
||||
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
|
||||
icon_path = (RESOURCES_DIR / Icons._paths[IconName.TOKENS]).as_posix()
|
||||
token_html = (
|
||||
|
|
@ -71,17 +71,3 @@ class TieDialog(QDialog):
|
|||
|
||||
def get_bids(self) -> Dict[str, bool]:
|
||||
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")
|
||||
|
|
|
|||
|
|
@ -603,15 +603,30 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow):
|
|||
table = self.choicesTable
|
||||
table.setSortingEnabled(False)
|
||||
table.clearContents()
|
||||
table.setColumnCount(5)
|
||||
table.setHorizontalHeaderLabels(
|
||||
["Participant", "Priority", "Secondary", "", "Comment"]
|
||||
)
|
||||
table.setRowCount(len(participants))
|
||||
table.setIconSize(QSize(32, 16))
|
||||
for row, choice in enumerate(participants):
|
||||
participant_item = QtWidgets.QTableWidgetItem(choice.participant_name)
|
||||
priority_item = QtWidgets.QTableWidgetItem(choice.priority_sector)
|
||||
if choice.priority_icon:
|
||||
priority_item.setIcon(choice.priority_icon)
|
||||
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)
|
||||
table.setItem(row, 0, participant_item)
|
||||
table.setItem(row, 1, priority_item)
|
||||
table.setItem(row, 2, secondary_item)
|
||||
table.setItem(row, 3, status_item)
|
||||
table.setItem(row, 4, comment_item)
|
||||
table.setSortingEnabled(True)
|
||||
table.resizeColumnsToContents()
|
||||
|
||||
|
|
|
|||
|
|
@ -273,44 +273,44 @@
|
|||
"choices": [
|
||||
{
|
||||
"participant_id": "602e2eaf-297e-490b-b0e9-efec818e466a",
|
||||
"priority_sector_id": null,
|
||||
"secondary_sector_id": null,
|
||||
"priority_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
|
||||
"secondary_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
|
||||
"comment": null
|
||||
},
|
||||
{
|
||||
"participant_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de",
|
||||
"priority_sector_id": null,
|
||||
"secondary_sector_id": null,
|
||||
"priority_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
|
||||
"secondary_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
|
||||
"comment": null
|
||||
},
|
||||
{
|
||||
"participant_id": "237c1291-4331-4242-bd70-bf648185a627",
|
||||
"priority_sector_id": null,
|
||||
"secondary_sector_id": null,
|
||||
"priority_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
|
||||
"secondary_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
|
||||
"comment": null
|
||||
},
|
||||
{
|
||||
"participant_id": "876a1684-7fe2-4d8d-a50a-a4d7e354c5b0",
|
||||
"priority_sector_id": null,
|
||||
"secondary_sector_id": null,
|
||||
"priority_sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
|
||||
"secondary_sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
|
||||
"comment": null
|
||||
}
|
||||
],
|
||||
"battles": [
|
||||
{
|
||||
"sector_id": "d1f7c6cf-40ff-42b8-b1b6-05576017de1f",
|
||||
"player_1_id": "602e2eaf-297e-490b-b0e9-efec818e466a",
|
||||
"player_2_id": "237c1291-4331-4242-bd70-bf648185a627",
|
||||
"player_1_id": "237c1291-4331-4242-bd70-bf648185a627",
|
||||
"player_2_id": "602e2eaf-297e-490b-b0e9-efec818e466a",
|
||||
"winner_id": null,
|
||||
"score": null,
|
||||
"victory_condition": "tie",
|
||||
"victory_condition": "Tie",
|
||||
"comment": "Never finished..."
|
||||
},
|
||||
{
|
||||
"sector_id": "096da98e-d8b4-4c71-946a-3cee2cc38499",
|
||||
"player_1_id": "1f6b4e7c-b1e4-4a2e-9aea-7bb75b20b4de",
|
||||
"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",
|
||||
"victory_condition": "Mission",
|
||||
"comment": "Decisive fast attack impossible to resist."
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue