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

choice tie-break and random fallback + exception cleanup + war<->camp pid

Browse source
This commit is contained in:
Maxime Réaux 2026-03-12 16:28:20 +01:00
parent 241d7f10f5
commit 241e76c937
15 changed files with 241 additions and 157 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -111,7 +111,15 @@ class AppController:
path = self.view.ask_open_file()
if not path:
return
self.model.load(path)
try:
self.model.load(path)
except RuntimeError as e:
QMessageBox.warning(
self.view,
"Add forbidden",
str(e),
)
return
self.current_file = path
self.is_dirty = False
self.navigation.refresh_players_view()

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

@ -14,7 +14,7 @@ from warchron.controller.dtos import (
RoundDTO,
CampaignParticipantScoreDTO,
)
from warchron.model.exception import ForbiddenOperation, DomainError
from warchron.model.exception import AbortedOperation, DomainError
from warchron.model.war import War
from warchron.model.campaign import Campaign
from warchron.model.campaign_participant import CampaignParticipant
@ -196,7 +196,7 @@ class CampaignController:
)
if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx)
raise ForbiddenOperation("Tie resolution cancelled")
raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids()
return bids_map

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

@ -104,12 +104,8 @@ class RoundPairingWorkflow:
def start(self, war: War, round: Round) -> None:
Pairing.check_round_pairable(round)
ties = TieResolver.find_choice_ties(war, round.id)
while ties:
bids_map = self.app.rounds.resolve_ties(war, ties)
for tie in ties:
bids = bids_map[tie.key()]
TieResolver.apply_bids(war, tie, bids)
TieResolver.resolve_tie_state(war, tie, bids)
ties = TieResolver.find_choice_ties(war, round.id)
Pairing.assign_battles_to_participants(war, round)
Pairing.assign_battles_to_participants(
war,
round,
resolve_ties_callback=self.app.rounds.resolve_ties,
)

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

@ -6,7 +6,7 @@ from PyQt6.QtGui import QIcon
from warchron.constants import ItemType, RefreshScope, Icons, IconName, ContextType
from warchron.model.exception import (
ForbiddenOperation,
AbortedOperation,
DomainError,
RequiresConfirmation,
)
@ -119,9 +119,7 @@ class RoundController:
)
p1_war = None
if battle.player_1_id is not None:
p1_war = camp.participants[
battle.player_1_id
].war_participant_id
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)
@ -192,10 +190,19 @@ class RoundController:
workflow = RoundPairingWorkflow(self.app)
try:
workflow.start(war, rnd)
except AbortedOperation as e:
QMessageBox.warning(
self.app.view,
"Canceled pairing",
str(e),
)
for bat in rnd.battles.values():
bat.cleanup_battle_players()
return
except DomainError as e:
QMessageBox.warning(
self.app.view,
"Closure forbidden",
"Pairing impossible",
str(e),
)
return
@ -238,7 +245,7 @@ class RoundController:
)
if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx)
raise ForbiddenOperation("Tie resolution cancelled")
raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids()
return bids_map

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

@ -10,7 +10,7 @@ from warchron.constants import (
)
from warchron.model.exception import (
DomainError,
ForbiddenOperation,
AbortedOperation,
RequiresConfirmation,
)
@ -190,7 +190,7 @@ class WarController:
)
if not dialog.exec():
TieResolver.cancel_tie_break(war, ctx)
raise ForbiddenOperation("Tie resolution cancelled")
raise AbortedOperation("Tie resolution cancelled")
bids_map[ctx.key()] = dialog.get_bids()
return bids_map

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

@ -1,6 +1,7 @@
from __future__ import annotations
from typing import Any, Dict, List
from warchron.model.exception import DomainError
from warchron.model.json_helper import JsonHelper
@ -70,7 +71,7 @@ class Battle:
if self.player_2_id is None:
self.player_2_id = participant_id
return
raise RuntimeError("Battle has no available places")
raise DomainError("Battle has no available places")
def cleanup_battle_players(self) -> None:
self.player_1_id = None

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

@ -91,14 +91,21 @@ class Campaign:
except KeyError:
raise KeyError(f"Participant {participant_id} not in campaign {self.id}")
def get_campaign_participant_by_war_participant_id(
def campaign_to_war_part_id(
self,
war_participant_id: str,
) -> CampaignParticipant | None:
campaign_pid: str,
) -> str:
cp = self.get_campaign_participant(campaign_pid)
return cp.war_participant_id
def war_to_campaign_part_id(
self,
war_pid: str,
) -> str:
for cp in self.participants.values():
if cp.war_participant_id == war_participant_id:
return cp
return None
if cp.war_participant_id == war_pid:
return cp.id
raise KeyError(f"War participant {war_pid} not within campaign participants")
def get_all_campaign_participants(self) -> List[CampaignParticipant]:
return list(self.participants.values())

2
src/warchron/model/closure_service.py
View file

@ -34,7 +34,7 @@ class ClosureService:
return
base_winner = None
if battle.winner_id is not None:
base_winner = campaign.participants[battle.winner_id].war_participant_id
base_winner = campaign.campaign_to_war_part_id(battle.winner_id)
effective_winner = ResultChecker.get_effective_winner_id(
war,
ContextType.BATTLE,

6
src/warchron/model/exception.py
View file

@ -13,6 +13,12 @@ class ForbiddenOperation(DomainError):
pass
class AbortedOperation(DomainError):
"""Generic 'you canceled this' rule."""
pass
class DomainDecision(Exception):
"""Base class for domain actions requiring user decision."""

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

@ -1,8 +1,8 @@
from __future__ import annotations
from typing import Dict, List
from typing import Dict, List, Callable, Tuple
import random
from warchron.constants import ContextType
from warchron.constants import ContextType, ScoreKind
from warchron.model.exception import (
DomainError,
ForbiddenOperation,
@ -12,6 +12,14 @@ 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.war_event import TieResolved
from warchron.model.score_service import ParticipantScore
ResolveTiesCallback = Callable[
["War", List["TieContext"]],
Dict[Tuple[str, str, int | None], Dict[str, bool]],
]
class Pairing:
@ -30,13 +38,18 @@ class Pairing:
)
for pid, choice in round.choices.items():
if choice is not None and not choice.priority_sector_id:
raise DomainError(f"Missing priority choice for participant {pid}")
raise ForbiddenOperation(
f"Missing priority choice for participant {pid}"
)
if choice is not None and not choice.secondary_sector_id:
raise DomainError(f"Missing secondary choice for participant {pid}")
raise ForbiddenOperation(
f"Missing secondary choice for participant {pid}"
)
def cleanup() -> None:
for bat in round.battles.values():
bat.cleanup_battle_players()
# FIXME cancel TieResolved + TokenSpent
if any(
bat.player_1_id is not None or bat.player_2_id is not None
@ -53,6 +66,7 @@ class Pairing:
def assign_battles_to_participants(
war: War,
round: Round,
resolve_ties_callback: ResolveTiesCallback,
) -> None:
campaign = war.get_campaign_by_round(round.id)
if campaign is None:
@ -62,57 +76,162 @@ class Pairing:
ContextType.CAMPAIGN,
campaign.id,
)
score_groups = ScoreService.group_participants_by_score(
scores, lambda score: score.victory_points
)
def value_getter(score: ParticipantScore) -> int:
return score.victory_points
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()
}
for group in score_groups:
# persistent equality → random order
ordered_group = list(group)
random.shuffle(ordered_group)
for participant_id in ordered_group:
camp_part = campaign.get_campaign_participant_by_war_participant_id(
participant_id
)
if camp_part:
Pairing._assign_single_participant(
round,
camp_part.id,
sector_to_battle,
)
score_value = value_getter(scores[group[0]])
remaining: List[str] = [
campaign.war_to_campaign_part_id(pid) for pid in group
]
Pairing._run_phase(
war,
round,
remaining,
sector_to_battle,
resolve_ties_callback,
use_priority=True,
score_value=score_value,
)
Pairing._run_phase(
war,
round,
remaining,
sector_to_battle,
resolve_ties_callback,
use_priority=False,
score_value=score_value,
)
Pairing._assign_fallback(round, remaining)
@staticmethod
def _assign_single_participant(
def _run_phase(
war: War,
round: Round,
participant_id: str,
remaining: List[str],
sector_to_battle: Dict[str, Battle],
resolve_ties_callback: ResolveTiesCallback,
*,
use_priority: bool,
score_value: int,
) -> None:
choice = round.choices.get(participant_id)
preferred_sectors: List[str] = []
if choice:
if choice.priority_sector_id:
preferred_sectors.append(choice.priority_sector_id)
if choice.secondary_sector_id:
preferred_sectors.append(choice.secondary_sector_id)
# --- try preferred sectors ---
for sect_id in preferred_sectors:
battle = sector_to_battle.get(sect_id)
demand = Pairing._build_sector_demand(
round,
remaining,
use_priority,
)
for sector_id, participants in demand.items():
battle = sector_to_battle.get(sector_id)
if not battle:
continue
if battle.get_available_places():
battle.assign_participant(participant_id)
return
# --- fallback rules ---
available_battles = round.get_battles_with_places()
if not available_battles:
raise RuntimeError("No available battle remaining")
if len(available_battles) == 1:
available_battles[0].assign_participant(participant_id)
return
# multiple remaining battles → warning
raise RuntimeError(
f"Ambiguous fallback for participant {participant_id}: "
"multiple battles still available"
places = len(battle.get_available_places())
if places <= 0:
continue
winners = Pairing._resolve_sector_allocation(
war,
round,
sector_id,
participants,
places,
resolve_ties_callback,
score_value,
)
for pid in winners:
battle.assign_participant(pid)
remaining.remove(pid)
@staticmethod
def _build_sector_demand(
round: Round,
participants: List[str],
use_priority: bool,
) -> Dict[str, List[str]]:
demand: Dict[str, List[str]] = {}
for pid in participants:
choice = round.choices.get(pid)
if not choice:
continue
sector_id = (
choice.priority_sector_id
if use_priority
else choice.secondary_sector_id
)
if not sector_id:
continue
demand.setdefault(sector_id, []).append(pid)
return demand
@staticmethod
def _resolve_sector_allocation(
war: War,
round: Round,
sector_id: str,
participants: List[str],
places: int,
resolve_ties_callback: ResolveTiesCallback,
score_value: int,
) -> List[str]:
if len(participants) <= places:
return participants
campaign = war.get_campaign_by_round(round.id)
if campaign is None:
raise DomainError("Campaign not found for round {round.id}")
context = TieContext(
context_type=ContextType.CHOICE,
context_id=round.id,
participants=[
campaign.campaign_to_war_part_id(pid) for pid in participants
],
score_kind=ScoreKind.VP,
sector_id=sector_id,
)
# ---- resolve tie loop ----
while not TieResolver.is_tie_resolved(war, context):
if not TieResolver.can_tie_be_resolved(war, context, context.participants):
war.events.append(
TieResolved(
None,
context.context_type,
context.context_id,
score_value=score_value,
sector_id=sector_id,
)
)
break
bids_map = resolve_ties_callback(war, [context])
bids = bids_map[context.key()]
TieResolver.apply_bids(war, context, bids)
TieResolver.resolve_tie_state(war, context, bids)
ranked_groups = TieResolver.rank_by_tokens(
war,
context,
context.participants,
)
ordered: List[str] = []
for group in ranked_groups:
shuffled_group = list(group)
random.shuffle(shuffled_group)
ordered.extend(
campaign.war_to_campaign_part_id(pid) for pid in shuffled_group
)
return ordered[:places]
@staticmethod
def _assign_fallback(
round: Round,
remaining: List[str],
) -> None:
for pid in list(remaining):
available = round.get_battles_with_places()
if not available:
raise DomainError("No available battle remaining")
if len(available) == 1:
available[0].assign_participant(pid)
remaining.remove(pid)
continue
raise DomainError(f"Ambiguous fallback for participant {pid}")

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

@ -56,9 +56,7 @@ class ScoreService:
if battle.winner_id is not None:
campaign = war.get_campaign_by_campaign_participant(battle.winner_id)
if campaign is not None:
base_winner = campaign.participants[
battle.winner_id
].war_participant_id
base_winner = campaign.campaign_to_war_part_id(battle.winner_id)
winner = ResultChecker.get_effective_winner_id(
war, ContextType.BATTLE, battle.sector_id, base_winner
)

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

@ -3,7 +3,7 @@ from dataclasses import dataclass, field
from collections import defaultdict
from warchron.constants import ContextType, ScoreKind
from warchron.model.exception import ForbiddenOperation
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
@ -17,6 +17,7 @@ class TieContext:
score_value: int | None = None
score_kind: ScoreKind | None = None
objective_id: str | None = None
sector_id: str | None = None
def key(self) -> tuple[str, str, int | None]:
return (self.context_type, self.context_id, self.score_value)
@ -24,71 +25,6 @@ class TieContext:
class TieResolver:
@staticmethod
def find_choice_ties(
war: War,
round_id: str,
) -> List[TieContext]:
round = war.get_round(round_id)
campaign = war.get_campaign_by_round(round_id)
if campaign is None:
raise RuntimeError("Round without campaign")
ties: List[TieContext] = []
scores = ScoreService.compute_scores(
war,
ContextType.CAMPAIGN,
campaign.id,
)
score_groups = ScoreService.group_participants_by_score(
scores, lambda score: score.victory_points
)
sector_to_battle = {b.sector_id: b for b in round.battles.values()}
for group in score_groups:
if len(group) <= 1:
continue
demand: Dict[str, List[str]] = {}
for pid in group:
choice = round.choices.get(pid)
if not choice:
continue
for sec_id in (
choice.priority_sector_id,
choice.secondary_sector_id,
):
if sec_id:
demand.setdefault(sec_id, []).append(pid)
for sector_id, demanders in demand.items():
battle = sector_to_battle.get(sector_id)
if battle is None:
continue
places = len(battle.get_available_places())
if len(demanders) <= places:
continue
context = TieContext(
ContextType.CHOICE,
round_id,
demanders,
score_value=None,
score_kind=ScoreKind.VP,
)
if TieResolver.is_tie_resolved(war, context):
continue
if not TieResolver.can_tie_be_resolved(
war,
context,
demanders,
):
war.events.append(
TieResolved(
None,
ContextType.CHOICE,
round_id,
)
)
continue
ties.append(context)
return ties
@staticmethod
def find_battle_ties(war: War, round_id: str) -> List[TieContext]:
round = war.get_round(round_id)
@ -104,12 +40,12 @@ class TieResolver:
if TieResolver.is_tie_resolved(war, context):
continue
if campaign is None:
raise RuntimeError("No campaign for this battle tie")
raise DomainError("No campaign for this battle tie")
if battle.player_1_id is None or battle.player_2_id is None:
raise RuntimeError("Missing player(s) in this battle context.")
p1 = campaign.participants[battle.player_1_id].war_participant_id
p2 = campaign.participants[battle.player_2_id].war_participant_id
if not TieResolver.can_tie_be_resolved(war, context, [p1, p2]):
raise DomainError("Missing player(s) in this battle context.")
p1_id = campaign.campaign_to_war_part_id(battle.player_1_id)
p2_id = campaign.campaign_to_war_part_id(battle.player_2_id)
if not TieResolver.can_tie_be_resolved(war, context, [p1_id, p2_id]):
war.events.append(
TieResolved(None, ContextType.BATTLE, battle.sector_id)
)
@ -118,7 +54,7 @@ class TieResolver:
TieContext(
context_type=ContextType.BATTLE,
context_id=battle.sector_id,
participants=[p1, p2],
participants=[p1_id, p2_id],
score_value=None,
score_kind=None,
)
@ -240,8 +176,8 @@ class TieResolver:
ContextType.WAR,
war.id,
[],
score_value,
ScoreKind.VP,
score_value=score_value,
score_kind=ScoreKind.VP,
)
if TieResolver.is_tie_resolved(war, context):
continue
@ -427,15 +363,15 @@ class TieResolver:
context,
context.participants,
)
# confirmed draw if non had bid
# confirmed draw if none had bid
if not active:
war.events.append(
TieResolved(
None,
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
return
@ -446,8 +382,8 @@ class TieResolver:
None,
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
score_value=context.score_value,
objective_id=context.objective_id,
)
)
return

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

@ -77,6 +77,7 @@ class War:
new_events: List[WarEvent] = []
for ev in self.events:
if isinstance(ev, (TieResolved)):
# FIXME cancel TieResolved + TokenSpent
if ev.context_type == ContextType.BATTLE:
battle = self.get_battle(ev.context_id)
campaign = self.get_campaign_by_sector(battle.sector_id)

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

@ -77,10 +77,12 @@ class TieResolved(WarEvent):
context_id: str,
score_value: int | None = None,
objective_id: str | None = None,
sector_id: str | None = None,
):
super().__init__(participant_id, context_type, context_id)
self.score_value = score_value
self.objective_id = objective_id
self.sector_id = sector_id
def toDict(self) -> Dict[str, Any]:
d = super().toDict()
@ -88,6 +90,7 @@ class TieResolved(WarEvent):
{
"score_value": self.score_value or None,
"objective_id": self.objective_id or None,
"sector_id": self.sector_id or None,
}
)
return d
@ -100,6 +103,7 @@ class TieResolved(WarEvent):
data["context_id"],
JsonHelper.none_if_empty(data["score_value"]),
JsonHelper.none_if_empty(data["objective_id"]),
JsonHelper.none_if_empty(data["sector_id"]),
)
return cls._base_fromDict(ev, data)