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

fix ignored campaign NP tie-break when closing war

Browse source
This commit is contained in:
Maxime Réaux 2026-03-06 15:02:53 +01:00
parent b1bde76319
commit 72f80563f1
16 changed files with 314 additions and 219 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,6 +1,8 @@
from __future__ import annotations
from typing import Any, Dict
from warchron.model.json_helper import JsonHelper
class Battle:
def __init__(
@ -71,11 +73,13 @@ class Battle:
def fromDict(data: Dict[str, Any]) -> Battle:
battle = Battle(
data["sector_id"],
data.get("player_1_id") or None,
data.get("player_2_id") or None,
JsonHelper.none_if_empty(data.get("player_1_id")),
JsonHelper.none_if_empty(data.get("player_2_id")),
)
battle.winner_id = data.get("winner_id") or None
battle.score = data.get("score") or None
battle.victory_condition = data.get("victory_condition") or None
battle.comment = data.get("comment") or None
battle.winner_id = JsonHelper.none_if_empty(data.get("winner_id"))
battle.score = JsonHelper.none_if_empty(data.get("score"))
battle.victory_condition = JsonHelper.none_if_empty(
data.get("victory_condition")
)
battle.comment = JsonHelper.none_if_empty(data.get("comment"))
return battle

8
src/warchron/model/choice.py
View file

@ -1,6 +1,8 @@
from __future__ import annotations
from typing import Any, Dict
from warchron.model.json_helper import JsonHelper
class Choice:
def __init__(
@ -42,8 +44,8 @@ class Choice:
def fromDict(data: Dict[str, Any]) -> Choice:
choice = Choice(
data["participant_id"],
data.get("priority_sector_id") or None,
data.get("secondary_sector_id") or None,
JsonHelper.none_if_empty(data.get("priority_sector_id")),
JsonHelper.none_if_empty(data.get("secondary_sector_id")),
)
choice.comment = data.get("comment") or None
choice.comment = JsonHelper.none_if_empty(data.get("comment"))
return choice

11
src/warchron/model/json_helper.py Normal file
View file

@ -0,0 +1,11 @@
from typing import TypeVar
T = TypeVar("T")
class JsonHelper:
@staticmethod
def none_if_empty(value: T | None) -> T | None:
if value == "":
return None
return value

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

@ -2,6 +2,8 @@ from __future__ import annotations
from typing import Any, Dict
from uuid import uuid4
from warchron.model.json_helper import JsonHelper
class Objective:
def __init__(self, name: str, description: str | None):
@ -27,6 +29,6 @@ class Objective:
@staticmethod
def fromDict(data: Dict[str, Any]) -> Objective:
obj = Objective(data["name"], data["description"] or None)
obj = Objective(data["name"], JsonHelper.none_if_empty(data["description"]))
obj.set_id(data["id"])
return obj

96
src/warchron/model/result_checker.py
View file

@ -2,10 +2,10 @@ from __future__ import annotations
from typing import List, Tuple, Dict, TYPE_CHECKING, Callable
from collections import defaultdict
from warchron.constants import ContextType
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
from warchron.model.tie_manager import TieResolver, TieContext
if TYPE_CHECKING:
from warchron.model.score_service import ParticipantScore
@ -35,19 +35,37 @@ class ResultChecker:
war: War,
context_type: ContextType,
context_id: str,
score_kind: ScoreKind | None,
scores: Dict[str, ParticipantScore],
value_getter: Callable[[ParticipantScore], int],
objective_id: str | None = None,
) -> List[Tuple[int, List[str], Dict[str, int]]]:
buckets: Dict[int, List[str]] = defaultdict(list)
for pid, score in scores.items():
buckets[value_getter(score)].append(pid)
sorted_vps = sorted(buckets.keys(), reverse=True)
sorted_values = sorted(buckets.keys(), reverse=True)
ranking: List[Tuple[int, List[str], Dict[str, int]]] = []
current_rank = 1
for value in sorted_vps:
assert score_kind is not None
for value in sorted_values:
participants = buckets[value]
context: TieContext = TieContext(
context_type,
context_id,
participants,
value,
score_kind,
objective_id,
)
if context_type == ContextType.WAR and len(participants) > 1:
subgroups = ResultChecker._secondary_sorting_war(war, participants)
subcontexts = ResultChecker._build_war_subcontexts(
war,
score_kind,
objective_id,
)
subgroups = ResultChecker._secondary_sorting_war(
war, participants, value_getter, subcontexts
)
for subgroup in subgroups:
# no tie if campaigns' rank is enough to sort
if len(subgroup) == 1:
@ -57,47 +75,29 @@ class ResultChecker:
current_rank += 1
continue
# normal tie-break if tie persists
if not TieResolver.is_tie_resolved(
war, context_type, context_id, value
):
if not TieResolver.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_type,
context_id,
subgroup,
)
tokens_spent = TieResolver.tokens_spent_map(
war, context_type, context_id, 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 TieResolver.is_tie_resolved(
war, context_type, context_id, value
):
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 = TieResolver.rank_by_tokens(
war,
context_type,
context_id,
participants,
)
tokens_spent = TieResolver.tokens_spent_map(
war, context_type, context_id, 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))
@ -108,27 +108,35 @@ class ResultChecker:
def _secondary_sorting_war(
war: War,
participants: List[str],
value_getter: Callable[[ParticipantScore], int],
subcontexts: List[TieContext],
) -> List[List[str]]:
from warchron.model.score_service import ScoreService
rank_map: Dict[str, Tuple[int, ...]] = {}
for pid in participants:
ranks: List[int] = []
for campaign in war.campaigns:
for sub in subcontexts:
scores = ScoreService.compute_scores(
war, ContextType.CAMPAIGN, campaign.id
war, sub.context_type, sub.context_id
)
ranking = ResultChecker.get_effective_ranking(
war,
ContextType.CAMPAIGN,
campaign.id,
sub.context_type,
sub.context_id,
sub.score_kind,
scores,
lambda s: s.victory_points,
value_getter,
sub.objective_id,
)
found = False
for rank, group, _ in ranking:
if pid in group:
ranks.append(rank)
found = True
break
if not found:
ranks.append(len(scores) + 1)
rank_map[pid] = tuple(ranks)
sorted_items = sorted(rank_map.items(), key=lambda x: x[1])
groups: List[List[str]] = []
@ -139,3 +147,23 @@ class ResultChecker:
current_tuple = rank_tuple
groups[-1].append(pid)
return groups
@staticmethod
def _build_war_subcontexts(
war: War,
score_kind: ScoreKind,
objective_id: str | None,
) -> List[TieContext]:
subcontexts = []
for campaign in war.campaigns:
subcontexts.append(
TieContext(
ContextType.CAMPAIGN,
campaign.id,
[],
None,
score_kind,
objective_id,
)
)
return subcontexts

12
src/warchron/model/sector.py
View file

@ -2,6 +2,8 @@ from __future__ import annotations
from typing import Any, Dict
from uuid import uuid4
from warchron.model.json_helper import JsonHelper
class Sector:
def __init__(
@ -64,11 +66,11 @@ class Sector:
sec = Sector(
data["name"],
data["round_id"],
data.get("major_objective_id") or None,
data.get("minor_objective_id") or None,
data.get("influence_objective_id") or None,
data.get("mission") or None,
data.get("description") or None,
JsonHelper.none_if_empty(data.get("major_objective_id")),
JsonHelper.none_if_empty(data.get("minor_objective_id")),
JsonHelper.none_if_empty(data.get("influence_objective_id")),
JsonHelper.none_if_empty(data.get("mission")),
JsonHelper.none_if_empty(data.get("description")),
)
sec.set_id(data["id"])
sec.mission = data.get("mission") or None

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

@ -1,8 +1,8 @@
from typing import List, Dict, DefaultDict
from dataclasses import dataclass
from dataclasses import dataclass, field
from collections import defaultdict
from warchron.constants import ContextType
from warchron.constants import ContextType, ScoreKind
from warchron.model.exception import ForbiddenOperation
from warchron.model.war import War
from warchron.model.war_event import InfluenceSpent, TieResolved
@ -13,8 +13,13 @@ from warchron.model.score_service import ScoreService, ParticipantScore
class TieContext:
context_type: ContextType
context_id: str
participants: List[str] # war_participant_ids
participants: List[str] = field(default_factory=list) # war_participant_ids
score_value: int | None = None
score_kind: ScoreKind | None = None
objective_id: str | None = None
def key(self) -> tuple[str, str, int | None]:
return (self.context_type, self.context_id, self.score_value)
class TieResolver:
@ -27,9 +32,11 @@ class TieResolver:
for battle in round.battles.values():
if not battle.is_draw():
continue
if TieResolver.is_tie_resolved(
war, ContextType.BATTLE, battle.sector_id, None
):
context: TieContext = TieContext(
ContextType.BATTLE,
battle.sector_id,
)
if TieResolver.is_tie_resolved(war, context):
continue
if campaign is None:
raise RuntimeError("No campaign for this battle tie")
@ -37,9 +44,7 @@ class TieResolver:
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, ContextType.BATTLE, battle.sector_id, [p1, p2]
):
if not TieResolver.can_tie_be_resolved(war, context, [p1, p2]):
war.events.append(
TieResolved(None, ContextType.BATTLE, battle.sector_id)
)
@ -50,6 +55,7 @@ class TieResolver:
context_id=battle.sector_id,
participants=[p1, p2],
score_value=None,
score_kind=None,
)
)
return ties
@ -64,15 +70,23 @@ class TieResolver:
for score_value, participants in buckets.items():
if len(participants) <= 1:
continue
if TieResolver.is_tie_resolved(
war, ContextType.CAMPAIGN, campaign_id, score_value
):
context: TieContext = TieContext(
ContextType.CAMPAIGN,
campaign_id,
[],
score_value,
ScoreKind.VP,
)
if TieResolver.is_tie_resolved(war, context):
continue
if not TieResolver.can_tie_be_resolved(
war, ContextType.CAMPAIGN, campaign_id, participants
):
if not TieResolver.can_tie_be_resolved(war, context, participants):
war.events.append(
TieResolved(None, ContextType.CAMPAIGN, campaign_id, score_value)
TieResolved(
None,
ContextType.CAMPAIGN,
campaign_id,
score_value,
)
)
continue
ties.append(
@ -81,6 +95,7 @@ class TieResolver:
context_id=campaign_id,
participants=participants,
score_value=score_value,
score_kind=ScoreKind.VP,
)
)
return ties
@ -101,38 +116,39 @@ class TieResolver:
np_value = score.narrative_points.get(objective_id, 0)
buckets[np_value].append(pid)
ties: List[TieContext] = []
context_id = f"{campaign_id}:{objective_id}"
context_id = campaign_id
for np_value, participants in buckets.items():
if len(participants) <= 1:
continue
if TieResolver.is_tie_resolved(
war,
ContextType.OBJECTIVE,
context_id,
context: TieContext = TieContext(
ContextType.CAMPAIGN,
campaign_id,
[],
np_value,
):
ScoreKind.NP,
objective_id,
)
if TieResolver.is_tie_resolved(war, context):
continue
if not TieResolver.can_tie_be_resolved(
war,
ContextType.OBJECTIVE,
context_id,
context,
participants,
):
war.events.append(
TieResolved(
None,
ContextType.OBJECTIVE,
context_id,
np_value,
None, ContextType.CAMPAIGN, context_id, np_value, objective_id
)
)
continue
ties.append(
TieContext(
context_type=ContextType.OBJECTIVE,
context_type=ContextType.CAMPAIGN,
context_id=context_id,
participants=participants,
score_value=np_value,
score_kind=ScoreKind.NP,
objective_id=objective_id,
)
)
return ties
@ -146,17 +162,25 @@ class TieResolver:
war,
ContextType.WAR,
war.id,
ScoreKind.VP,
scores,
value_getter=lambda s: s.victory_points,
lambda s: s.victory_points,
)
ties: List[TieContext] = []
for _, group, _ in ranking:
if len(group) <= 1:
continue
score_value = scores[group[0]].victory_points
if TieResolver.is_tie_resolved(war, ContextType.WAR, war.id, score_value):
context: TieContext = TieContext(
ContextType.WAR,
war.id,
[],
score_value,
ScoreKind.VP,
)
if TieResolver.is_tie_resolved(war, context):
continue
if not TieResolver.can_tie_be_resolved(war, ContextType.WAR, war.id, group):
if not TieResolver.can_tie_be_resolved(war, context, group):
war.events.append(
TieResolved(None, ContextType.WAR, war.id, score_value)
)
@ -167,6 +191,7 @@ class TieResolver:
context_id=war.id,
participants=group,
score_value=score_value,
score_kind=ScoreKind.VP,
)
)
return ties
@ -189,45 +214,43 @@ class TieResolver:
ranking = ResultChecker.get_effective_ranking(
war,
ContextType.OBJECTIVE,
f"{war.id}:{objective_id}",
ContextType.WAR,
war.id,
ScoreKind.NP,
scores,
value_getter=value_getter,
value_getter,
objective_id,
)
ties: List[TieContext] = []
for _, group, _ in ranking:
if len(group) <= 1:
continue
np_value = value_getter(scores[group[0]])
context_id = f"{war.id}:{objective_id}"
context: TieContext = TieContext(
ContextType.WAR, war.id, [], np_value, ScoreKind.NP, objective_id
)
if TieResolver.is_tie_resolved(
war,
ContextType.OBJECTIVE,
context_id,
np_value,
context,
):
continue
if not TieResolver.can_tie_be_resolved(
war,
ContextType.OBJECTIVE,
context_id,
context,
group,
):
war.events.append(
TieResolved(
None,
ContextType.OBJECTIVE,
context_id,
np_value,
)
TieResolved(None, ContextType.WAR, war.id, np_value, objective_id)
)
continue
ties.append(
TieContext(
context_type=ContextType.OBJECTIVE,
context_id=context_id,
context_type=ContextType.WAR,
context_id=war.id,
participants=group,
score_value=np_value,
score_kind=ScoreKind.NP,
objective_id=objective_id,
)
)
return ties
@ -235,8 +258,7 @@ class TieResolver:
@staticmethod
def apply_bids(
war: War,
context_type: ContextType,
context_id: str,
context: TieContext,
bids: Dict[str, bool], # war_participant_id -> spend?
) -> None:
for war_part_id, spend in bids.items():
@ -248,8 +270,9 @@ class TieResolver:
InfluenceSpent(
participant_id=war_part_id,
amount=1,
context_type=context_type,
context_id=context_id,
context_type=context.context_type,
context_id=context.context_id,
objective_id=context.objective_id,
)
)
@ -257,9 +280,7 @@ class TieResolver:
@staticmethod
def cancel_tie_break(
war: War,
context_type: ContextType,
context_id: str,
score_value: int | None = None,
context: TieContext,
) -> None:
war.events = [
ev
@ -267,14 +288,15 @@ class TieResolver:
if not (
(
isinstance(ev, InfluenceSpent)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.context_type == context.context_type
and ev.context_id == context.context_id
)
or (
isinstance(ev, TieResolved)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.score_value == score_value
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
)
)
]
@ -282,16 +304,16 @@ class TieResolver:
@staticmethod
def rank_by_tokens(
war: War,
context_type: ContextType,
context_id: str,
context: TieContext,
participants: List[str],
) -> List[List[str]]:
spent = {pid: 0 for pid in participants}
for ev in war.events:
if (
isinstance(ev, InfluenceSpent)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.context_type == context.context_type
and ev.context_id == context.context_id
and ev.objective_id == context.objective_id
and ev.participant_id in spent
):
spent[ev.participant_id] += ev.amount
@ -308,16 +330,16 @@ class TieResolver:
@staticmethod
def tokens_spent_map(
war: War,
context_type: ContextType,
context_id: str,
context: TieContext,
participants: List[str],
) -> Dict[str, int]:
spent = {pid: 0 for pid in participants}
for ev in war.events:
if (
isinstance(ev, InfluenceSpent)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.context_type == context.context_type
and ev.context_id == context.context_id
and ev.objective_id == context.objective_id
and ev.participant_id in spent
):
spent[ev.participant_id] += ev.amount
@ -326,45 +348,57 @@ class TieResolver:
@staticmethod
def get_active_participants(
war: War,
context_type: ContextType,
context_id: str,
context: TieContext,
participants: List[str],
) -> List[str]:
groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants)
groups = TieResolver.rank_by_tokens(war, context, participants)
return groups[0]
@staticmethod
def resolve_tie_state(
war: War,
ctx: TieContext,
context: TieContext,
bids: dict[str, bool] | None = None,
) -> None:
active = TieResolver.get_active_participants(
war,
ctx.context_type,
ctx.context_id,
ctx.participants,
context,
context.participants,
)
# confirmed draw if non had bid
if not active:
war.events.append(
TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value)
TieResolved(
None,
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
)
)
return
# confirmed draw if current bids are 0
if bids is not None and not any(bids.values()):
war.events.append(
TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value)
TieResolved(
None,
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
)
)
return
# else rank_by_tokens
groups = TieResolver.rank_by_tokens(
war, ctx.context_type, ctx.context_id, ctx.participants
)
groups = TieResolver.rank_by_tokens(war, context, context.participants)
if len(groups[0]) == 1:
war.events.append(
TieResolved(
groups[0][0], ctx.context_type, ctx.context_id, ctx.score_value
groups[0][0],
context.context_type,
context.context_id,
context.score_value,
context.objective_id,
)
)
return
@ -372,41 +406,34 @@ class TieResolver:
@staticmethod
def can_tie_be_resolved(
war: War, context_type: ContextType, context_id: str, participants: List[str]
war: War, context: TieContext, participants: List[str]
) -> bool:
active = TieResolver.get_active_participants(
war, context_type, context_id, 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_type: ContextType,
context_id: str,
score_value: int | None = None,
context: TieContext,
) -> bool:
for ev in reversed(war.events):
if (
isinstance(ev, TieResolved)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.score_value == score_value
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_type: ContextType,
context_id: str,
score_value: int | None = None,
) -> bool:
def is_tie_resolved(war: War, context: TieContext) -> bool:
return any(
isinstance(ev, TieResolved)
and ev.context_type == context_type
and ev.context_id == context_id
and ev.score_value == score_value
and ev.context_type == context.context_type
and ev.context_id == context.context_id
and ev.score_value == context.score_value
and ev.objective_id == context.objective_id
for ev in war.events
)

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

@ -3,6 +3,8 @@ from typing import Dict, Any, TypeVar, Type, cast
from datetime import datetime
from uuid import uuid4
from warchron.model.json_helper import JsonHelper
T = TypeVar("T", bound="WarEvent")
EVENT_REGISTRY: Dict[str, Type["WarEvent"]] = {}
@ -15,7 +17,12 @@ def register_event(cls: Type[T]) -> Type[T]:
class WarEvent:
TYPE = "WarEvent"
def __init__(self, participant_id: str | None, context_type: str, context_id: str):
def __init__(
self,
participant_id: str | None,
context_type: str,
context_id: str,
):
self.id: str = str(uuid4())
self.participant_id: str | None = participant_id
self.context_type = context_type # battle, round, campaign, war
@ -69,15 +76,18 @@ class TieResolved(WarEvent):
context_type: str,
context_id: str,
score_value: int | None = None,
objective_id: str | None = None,
):
super().__init__(participant_id, context_type, context_id)
self.score_value = score_value
self.objective_id = objective_id
def toDict(self) -> Dict[str, Any]:
d = super().toDict()
d.update(
{
"score_value": self.score_value or None,
"objective_id": self.objective_id or None,
}
)
return d
@ -85,10 +95,11 @@ class TieResolved(WarEvent):
@classmethod
def fromDict(cls, data: Dict[str, Any]) -> TieResolved:
ev = cls(
data["participant_id"] or None,
JsonHelper.none_if_empty(data["participant_id"]),
data["context_type"],
data["context_id"],
data["score_value"] or None,
JsonHelper.none_if_empty(data["score_value"]),
JsonHelper.none_if_empty(data["objective_id"]),
)
return cls._base_fromDict(ev, data)
@ -98,9 +109,17 @@ class InfluenceGained(WarEvent):
TYPE = "InfluenceGained"
def __init__(
self, participant_id: str, amount: int, context_type: str, context_id: str
self,
participant_id: str,
amount: int,
context_type: str,
context_id: str,
):
super().__init__(participant_id, context_type, context_id)
super().__init__(
participant_id,
context_type,
context_id,
)
self.amount = amount
def toDict(self) -> Dict[str, Any]:
@ -128,16 +147,23 @@ class InfluenceSpent(WarEvent):
TYPE = "InfluenceSpent"
def __init__(
self, participant_id: str, amount: int, context_type: str, context_id: str
self,
participant_id: str,
amount: int,
context_type: str,
context_id: str,
objective_id: str | None = None,
):
super().__init__(participant_id, context_type, context_id)
self.amount = amount
self.objective_id = objective_id
def toDict(self) -> Dict[str, Any]:
d = super().toDict()
d.update(
{
"amount": self.amount,
"objective_id": self.objective_id,
}
)
return d
@ -149,5 +175,6 @@ class InfluenceSpent(WarEvent):
int(data["amount"]),
data["context_type"],
data["context_id"],
JsonHelper.none_if_empty(data["objective_id"]),
)
return cls._base_fromDict(ev, data)