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

Compare commits

base: maximator:b7a35f671218276a47bb307069d8bc3afe62d2a0
Branches Tags
maximator:main
..
compare: maximator:db78c6dacc98a475c1a330041a6f9e9306c674c8
Branches Tags
maximator:main

No commits in common. "b7a35f671218276a47bb307069d8bc3afe62d2a0" and "db78c6dacc98a475c1a330041a6f9e9306c674c8" have entirely different histories.
b7a35f6712 ... db78c6dacc

23 changed files with 536 additions and 932 deletions
Download patch file Download diff file Expand all files Collapse all files

39
src/warchron/constants.py
View file

@ -22,7 +22,6 @@ class IconName(StrEnum):
PAIRING = auto()
DRAW = auto()
TIEBREAK = auto()
DRAWTOKEN = auto()
DELETE = auto()
SAVE_AS = auto()
SAVE = auto()
@ -50,8 +49,7 @@ class IconName(StrEnum):
NP1ST = auto()
NP2ND = auto()
NP3RD = auto()
WINTOKEN = auto()
TIEBREAKTOKEN = auto()
TIEBREAK_TOKEN = auto()
VP1STDRAW = auto()
VP1STBREAK = auto()
VP1STTIEDRAW = auto()
@ -73,9 +71,6 @@ class IconName(StrEnum):
NP3RDDRAW = auto()
NP3RDBREAK = auto()
NP3RDTIEDRAW = auto()
ALLOCATED = auto()
ALLOCATEDTOKEN = auto()
FALLBACK = auto()
VP_RANK_TO_ICON = {
@ -130,8 +125,6 @@ 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
@ -145,21 +138,11 @@ class Icons:
def get_pixmap(cls, name: IconName) -> QPixmap:
if name in cls._pixmap_cache:
return cls._pixmap_cache[name]
if name == IconName.TIEBREAKTOKEN:
if name == IconName.TIEBREAK_TOKEN:
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),
@ -272,11 +255,6 @@ 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())
@ -335,16 +313,3 @@ 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()

12
src/warchron/controller/app_controller.py
View file

@ -357,20 +357,12 @@ class AppController:
except RequiresConfirmation as e:
reply = QMessageBox.question(
self.view,
"Confirm update",
"Confirm deletion",
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
e.action()
else:
return
self.is_dirty = True

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

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

117
src/warchron/controller/closure_workflow.py Normal file
View file

@ -0,0 +1,117 @@
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,
)

10
src/warchron/controller/dtos.py
View file

@ -88,9 +88,6 @@ 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)
@ -106,6 +103,8 @@ 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)
@ -133,8 +132,3 @@ 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
View file

@ -1,178 +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.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),
)

80
src/warchron/controller/ranking_icon.py Normal file
View file

@ -0,0 +1,80 @@
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

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

@ -4,27 +4,19 @@ from PyQt6.QtWidgets import QDialog
from PyQt6.QtWidgets import QMessageBox
from PyQt6.QtGui import QIcon
from warchron.constants import (
ItemType,
RefreshScope,
Icons,
IconName,
ContextType,
ChoiceStatus,
)
from warchron.constants import ItemType, RefreshScope, Icons, IconName, ContextType
from warchron.model.exception import (
AbortedOperation,
DomainError,
RequiresConfirmation,
)
from warchron.model.tiebreaking import TieBreaker, TieContext
from warchron.model.pairing import Pairing
from warchron.model.tie_manager import TieResolver, TieContext
from warchron.model.result_checker import ResultChecker
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,
@ -32,11 +24,10 @@ from warchron.controller.dtos import (
ChoiceDTO,
BattleDTO,
)
from warchron.controller.workflows import (
from warchron.controller.closure_workflow 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
@ -57,117 +48,114 @@ 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
else ""
choice = self.app.model.create_choice(
round_id=rnd.id, participant_id=part.id
)
secondary_name = (
camp.get_sector_name(choice.secondary_sector_id)
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,
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,
)
priority_name = (
camp.get_sector_name(choice.priority_sector_id)
if choice.priority_sector_id is not None
else ""
)
secondary_name = (
camp.get_sector_name(choice.secondary_sector_id)
if choice.secondary_sector_id is not None
else ""
)
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,
)
)
# 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)
if battle.player_1_id:
camp_part = camp.participants[battle.player_1_id]
player_1_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
else:
player_1_name = ""
if battle.player_2_id:
camp_part = camp.participants[battle.player_2_id]
player_2_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
else:
player_2_name = ""
if battle.winner_id:
camp_part = camp.participants[battle.winner_id]
winner_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
else:
winner_name = ""
p1_icon, p2_icon = Presenter.compute_battle_icons(
war, round_id, battle.sector_id
battle = self.app.model.create_battle(
round_id=rnd.id, sector_id=sect.id
)
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,
)
state_icon = Icons.get(IconName.ONGOING)
if battle.is_finished():
state_icon = Icons.get(IconName.DONE)
if battle.player_1_id:
camp_part = camp.participants[battle.player_1_id]
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:
camp_part = camp.participants[battle.player_2_id]
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:
camp_part = camp.participants[battle.winner_id]
winner_name = self.app.model.get_participant_name(
camp_part.war_participant_id
)
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],
)
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,
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.endRoundBtn.setEnabled(not rnd.is_over)
@ -255,29 +243,16 @@ class RoundController:
for pid in ctx.participants
]
counters = [war.get_influence_tokens(pid) for pid in ctx.participants]
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)
# TODO display sector name for BATTLE or CHOICE
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():
TieBreaker.cancel_tie_break(war, ctx)
TieResolver.cancel_tie_break(war, ctx)
raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids()
return bids_map

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

@ -24,11 +24,10 @@ 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.tiebreaking import TieContext, TieBreaker
from warchron.model.scoring import ScoreComputer
from warchron.controller.workflows import WarClosureWorkflow
from warchron.controller.presenter import Presenter
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.view.war_dialog import WarDialog
from warchron.view.objective_dialog import ObjectiveDialog
from warchron.view.war_participant_dialog import WarParticipantDialog
@ -57,16 +56,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 = ScoreComputer.compute_scores(war, ContextType.WAR, war.id)
scores = ScoreService.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 = Presenter.compute_ranking_icons(
vp_icon_map = RankingIcon.compute_icons(
war, ContextType.WAR, war_id, scores
)
for obj in war.get_objectives_used_as_maj_or_min():
objective_icon_maps[obj.id] = Presenter.compute_ranking_icons(
objective_icon_maps[obj.id] = RankingIcon.compute_icons(
war,
ContextType.WAR,
war.id,
@ -161,7 +160,7 @@ class WarController:
) -> Dict[tuple[str, str, int | None, str | None, str | None], Dict[str, bool]]:
bids_map = {}
for ctx in contexts:
active = TieBreaker.get_active_participants(
active = TieResolver.get_active_participants(
war,
ctx,
ctx.participants,
@ -171,20 +170,26 @@ class WarController:
for pid in active
]
counters = [war.get_influence_tokens(pid) for pid in active]
data = Presenter.build_dialog_data(
war,
ctx,
)
dialog = TieDialog(
parent=self.app.view,
players=players,
counters=counters,
context_type=ctx.context_type,
context_id=ctx.context_id,
context_name=data.title,
context_name=None,
)
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():
TieBreaker.cancel_tie_break(war, ctx)
TieResolver.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
View file

@ -1,116 +0,0 @@
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,
)

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

@ -144,8 +144,8 @@ class Campaign:
def cleanup() -> None:
for rnd in rounds_blocking:
rnd.remove_choice(participant_id)
rnd.clear_participant_references(participant_id)
rnd.remove_choice(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.remove_battle(sector_id)
rnd.clear_sector_references(sector_id)
rnd.remove_battle(sector_id)
apply_update()
rounds_str = ", ".join(
@ -299,8 +299,8 @@ class Campaign:
def cleanup() -> None:
for rnd in rounds_blocking:
rnd.remove_battle(sector_id)
rnd.clear_sector_references(sector_id)
rnd.remove_battle(sector_id)
del self.sectors[sector_id]
rounds_str = ", ".join(
@ -345,10 +345,12 @@ 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:
if rnd.has_battle_with_sector(sector_id):
return rnd
for bat in rnd.battles.values():
if bat.sector_id == sector_id:
return rnd
raise KeyError(f"Battle {sector_id} not found in any Round")
def get_all_rounds(self) -> List[Round]:
@ -406,12 +408,6 @@ 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,

4
src/warchron/model/closing.py → src/warchron/model/closure_service.py
View file

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

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

@ -1,11 +1,10 @@
from __future__ import annotations
from typing import Dict, List
from dataclasses import dataclass
from typing import Dict, List, Callable, Tuple
from uuid import uuid4
import random
from warchron.constants import ContextType, ScoreKind, ChoiceStatus, AllocationType
from warchron.constants import ContextType, ScoreKind
from warchron.model.exception import (
DomainError,
ForbiddenOperation,
@ -14,17 +13,15 @@ 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.scoring import ScoreComputer
from warchron.model.tiebreaking import TieBreaker, TieContext, ResolveTiesCallback
from warchron.model.score_service import ScoreService
from warchron.model.tie_manager import TieResolver, TieContext
from warchron.model.war_event import TieResolved
from warchron.model.scoring import ParticipantScore
from warchron.model.score_service import ParticipantScore
@dataclass(frozen=True, slots=True)
class AllocationResult:
priority: ChoiceStatus
secondary: ChoiceStatus
fallback: bool
ResolveTiesCallback = Callable[
["War", List["TieContext"]],
Dict[Tuple[str, str, int | None, str | None, str | None], Dict[str, bool]],
]
class Pairing:
@ -42,6 +39,15 @@ 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():
@ -60,15 +66,6 @@ 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(
@ -79,7 +76,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 = ScoreComputer.compute_scores(
scores = ScoreService.compute_scores(
war,
ContextType.CAMPAIGN,
campaign.id,
@ -88,7 +85,7 @@ class Pairing:
def value_getter(score: ParticipantScore) -> int:
return score.victory_points
score_groups = ScoreComputer.group_participants_by_score(scores, value_getter)
score_groups = ScoreService.group_participants_by_score(scores, value_getter)
sector_to_battle: Dict[str, Battle] = {
b.sector_id: b for b in round.battles.values()
}
@ -201,11 +198,13 @@ class Pairing:
sector_id=sector_id,
)
# ---- resolve tie loop ----
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(
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(
war, context, context.participants
)
if len(active) <= 1:
break
current_context = TieContext(
context_type=context.context_type,
context_id=context.context_id,
@ -214,12 +213,9 @@ class Pairing:
score_kind=context.score_kind,
sector_id=context.sector_id,
)
# natural, unbreakable or acceptable (enough places) draw
if (
not TieBreaker.can_tie_be_resolved(
war, context, current_context.participants
)
or len(active) <= places
# natural or unbreakable draw
if not TieResolver.can_tie_be_resolved(
war, context, current_context.participants
):
war.events.append(
TieResolved(
@ -249,9 +245,9 @@ class Pairing:
)
)
break
TieBreaker.apply_bids(war, context, tie_id, bids)
TieBreaker.resolve_tie_state(war, context, tie_id, bids)
ranked_groups = TieBreaker.rank_by_tokens(
TieResolver.apply_bids(war, context, tie_id, bids)
TieResolver.resolve_tie_state(war, context, tie_id, bids)
ranked_groups = TieResolver.rank_by_tokens(
war,
context,
context.participants,
@ -281,80 +277,3 @@ 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,
)

20
src/warchron/model/checking.py → src/warchron/model/result_checker.py
View file

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

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

@ -5,12 +5,7 @@ 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.constants import ContextType
from warchron.model.exception import (
ForbiddenOperation,
DomainError,
RequiresConfirmation,
)
from warchron.model.exception import ForbiddenOperation
from warchron.model.choice import Choice
from warchron.model.battle import Battle
@ -79,6 +74,7 @@ class Round:
def create_choice(self, participant_id: str) -> Choice:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create choice in a closed round.")
if participant_id not in self.choices:
choice = Choice(
@ -97,16 +93,11 @@ 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)
@ -127,9 +118,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.")
if self.has_battle_with_participant(participant_id):
raise ForbiddenOperation("Can't remove choice already assigned to battle.")
# TODO prevent if battles already assigned
self.war.revert_choice_ties(
self.id,
participants=[self.campaign.campaign_to_war_part_id(participant_id)],
@ -141,18 +132,6 @@ 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())
@ -177,6 +156,7 @@ class Round:
def create_battle(self, sector_id: str) -> Battle:
if self.is_over:
# TODO catch me if you can
raise ForbiddenOperation("Can't create battle in a closed round.")
if sector_id not in self.battles:
battle = Battle(sector_id=sector_id, player_1_id=None, player_2_id=None)
@ -193,15 +173,12 @@ 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)
if not bat:
raise DomainError(f"No battle found for sector {sector_id}")
def apply_update() -> None:
# TODO require confirmation if there was choice tie to clear it
if bat:
bat.set_player_1(player_1_id)
bat.set_player_2(player_2_id)
bat.set_winner(winner_id)
@ -209,48 +186,6 @@ 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
@ -274,9 +209,11 @@ 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]

21
src/warchron/model/scoring.py → src/warchron/model/score_service.py
View file

@ -4,9 +4,7 @@ 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)
@ -15,7 +13,7 @@ class ParticipantScore:
narrative_points: Dict[str, int] = field(default_factory=dict)
class ScoreComputer:
class ScoreService:
@staticmethod
def _get_battles_for_context(
@ -28,17 +26,14 @@ class ScoreComputer:
continue
yield from rnd.battles.values()
elif context_type == ContextType.CAMPAIGN:
campaign: Campaign | None = war.get_campaign(context_id)
if campaign:
for rnd in campaign.rounds:
if not rnd.is_over:
continue
yield from rnd.battles.values()
campaign = war.get_campaign(context_id)
for rnd in campaign.rounds:
if not rnd.is_over:
continue
yield from rnd.battles.values()
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
@ -47,7 +42,7 @@ class ScoreComputer:
def compute_scores(
war: War, context_type: ContextType, context_id: str
) -> Dict[str, ParticipantScore]:
from warchron.model.checking import ResultChecker
from warchron.model.result_checker import ResultChecker
if context_type == ContextType.CAMPAIGN:
camp = war.get_campaign(context_id)
@ -61,7 +56,7 @@ class ScoreComputer:
)
for pid in participant_ids
}
battles = ScoreComputer._get_battles_for_context(war, context_type, context_id)
battles = ScoreService._get_battles_for_context(war, context_type, context_id)
for battle in battles:
base_winner = None
if battle.winner_id is not None:

202
src/warchron/model/tiebreaking.py → src/warchron/model/tie_manager.py
View file

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

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

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

BIN
src/warchron/view/resources/cross-script.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 663 B

BIN
src/warchron/view/resources/map.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 1.5 KiB

16
src/warchron/view/tie_dialog.py
View file

@ -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(context_name)
self.ui.tieContext.setText(self._get_context_title(context_type, context_name))
grid = self.ui.playersGridLayout
icon_path = (RESOURCES_DIR / Icons._paths[IconName.TOKENS]).as_posix()
token_html = (
@ -71,3 +71,17 @@ 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")

15
src/warchron/view/view.py
View file

@ -603,30 +603,15 @@ 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()

24
test_data/example.json
View file

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