293 lines
9.8 KiB
Python
293 lines
9.8 KiB
Python
from typing import List, Dict, DefaultDict
|
|
from dataclasses import dataclass
|
|
from collections import defaultdict
|
|
|
|
from warchron.constants import ContextType
|
|
from warchron.model.exception import ForbiddenOperation
|
|
from warchron.model.war import War
|
|
from warchron.model.war_event import InfluenceSpent, TieResolved
|
|
from warchron.model.score_service import ScoreService
|
|
|
|
|
|
@dataclass
|
|
class TieContext:
|
|
context_type: ContextType
|
|
context_id: str
|
|
participants: List[str] # war_participant_ids
|
|
score_value: int | None = None
|
|
|
|
|
|
class TieResolver:
|
|
|
|
@staticmethod
|
|
def find_battle_ties(war: War, round_id: str) -> List[TieContext]:
|
|
round = war.get_round(round_id)
|
|
campaign = war.get_campaign_by_round(round_id)
|
|
ties = []
|
|
for battle in round.battles.values():
|
|
if not battle.is_draw():
|
|
continue
|
|
if TieResolver.is_tie_resolved(
|
|
war, ContextType.BATTLE, battle.sector_id, None
|
|
):
|
|
continue
|
|
if campaign is None:
|
|
raise RuntimeError("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, ContextType.BATTLE, battle.sector_id, [p1, p2]
|
|
):
|
|
war.events.append(
|
|
TieResolved(None, ContextType.BATTLE, battle.sector_id)
|
|
)
|
|
continue
|
|
ties.append(
|
|
TieContext(
|
|
context_type=ContextType.BATTLE,
|
|
context_id=battle.sector_id,
|
|
participants=[p1, p2],
|
|
score_value=None,
|
|
)
|
|
)
|
|
return ties
|
|
|
|
@staticmethod
|
|
def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]:
|
|
scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
|
|
buckets: DefaultDict[int, List[str]] = defaultdict(list)
|
|
for pid, score in scores.items():
|
|
buckets[score.victory_points].append(pid)
|
|
ties: List[TieContext] = []
|
|
for score_value, participants in buckets.items():
|
|
if len(participants) <= 1:
|
|
continue
|
|
if TieResolver.is_tie_resolved(
|
|
war, ContextType.CAMPAIGN, campaign_id, score_value
|
|
):
|
|
continue
|
|
if not TieResolver.can_tie_be_resolved(
|
|
war, ContextType.CAMPAIGN, campaign_id, participants
|
|
):
|
|
war.events.append(
|
|
TieResolved(None, ContextType.CAMPAIGN, campaign_id, score_value)
|
|
)
|
|
continue
|
|
ties.append(
|
|
TieContext(
|
|
context_type=ContextType.CAMPAIGN,
|
|
context_id=campaign_id,
|
|
participants=participants,
|
|
score_value=score_value,
|
|
)
|
|
)
|
|
return ties
|
|
|
|
@staticmethod
|
|
def find_war_ties(war: War) -> List[TieContext]:
|
|
scores = ScoreService.compute_scores(war, ContextType.WAR, war.id)
|
|
buckets: DefaultDict[int, List[str]] = defaultdict(list)
|
|
for pid, score in scores.items():
|
|
buckets[score.victory_points].append(pid)
|
|
ties: List[TieContext] = []
|
|
for score_value, participants in buckets.items():
|
|
if len(participants) <= 1:
|
|
continue
|
|
if TieResolver.is_tie_resolved(war, ContextType.WAR, war.id, score_value):
|
|
continue
|
|
if not TieResolver.can_tie_be_resolved(
|
|
war, ContextType.WAR, war.id, participants
|
|
):
|
|
war.events.append(
|
|
TieResolved(None, ContextType.WAR, war.id, score_value)
|
|
)
|
|
continue
|
|
ties.append(
|
|
TieContext(
|
|
context_type=ContextType.WAR,
|
|
context_id=war.id,
|
|
participants=participants,
|
|
score_value=score_value,
|
|
)
|
|
)
|
|
return ties
|
|
|
|
@staticmethod
|
|
def apply_bids(
|
|
war: War,
|
|
context_type: ContextType,
|
|
context_id: str,
|
|
bids: Dict[str, bool], # war_participant_id -> spend?
|
|
) -> None:
|
|
for war_part_id, spend in bids.items():
|
|
if not spend:
|
|
continue
|
|
if war.get_influence_tokens(war_part_id) < 1:
|
|
raise ForbiddenOperation("Not enough tokens")
|
|
war.events.append(
|
|
InfluenceSpent(
|
|
participant_id=war_part_id,
|
|
amount=1,
|
|
context_type=context_type,
|
|
context_id=context_id,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def cancel_tie_break(
|
|
war: War,
|
|
context_type: ContextType,
|
|
context_id: str,
|
|
score_value: int | None = None,
|
|
) -> None:
|
|
war.events = [
|
|
ev
|
|
for ev in war.events
|
|
if not (
|
|
(
|
|
isinstance(ev, InfluenceSpent)
|
|
and ev.context_type == context_type
|
|
and ev.context_id == context_id
|
|
)
|
|
or (
|
|
isinstance(ev, TieResolved)
|
|
and ev.context_type == context_type
|
|
and ev.context_id == context_id
|
|
and ev.score_value == score_value
|
|
)
|
|
)
|
|
]
|
|
|
|
@staticmethod
|
|
def rank_by_tokens(
|
|
war: War,
|
|
context_type: ContextType,
|
|
context_id: str,
|
|
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.participant_id in spent
|
|
):
|
|
spent[ev.participant_id] += ev.amount
|
|
sorted_items = sorted(spent.items(), key=lambda x: x[1], reverse=True)
|
|
groups: List[List[str]] = []
|
|
current_score = None
|
|
for pid, score in sorted_items:
|
|
if score != current_score:
|
|
groups.append([])
|
|
current_score = score
|
|
groups[-1].append(pid)
|
|
return groups
|
|
|
|
@staticmethod
|
|
def tokens_spent_map(
|
|
war: War,
|
|
context_type: ContextType,
|
|
context_id: str,
|
|
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.participant_id in spent
|
|
):
|
|
spent[ev.participant_id] += ev.amount
|
|
return spent
|
|
|
|
@staticmethod
|
|
def get_active_participants(
|
|
war: War,
|
|
context_type: ContextType,
|
|
context_id: str,
|
|
participants: List[str],
|
|
) -> List[str]:
|
|
groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants)
|
|
return groups[0]
|
|
|
|
@staticmethod
|
|
def resolve_tie_state(
|
|
war: War,
|
|
ctx: TieContext,
|
|
bids: dict[str, bool] | None = None,
|
|
) -> None:
|
|
active = TieResolver.get_active_participants(
|
|
war,
|
|
ctx.context_type,
|
|
ctx.context_id,
|
|
ctx.participants,
|
|
)
|
|
# confirmed draw if non had bid
|
|
if not active:
|
|
war.events.append(
|
|
TieResolved(None, ctx.context_type, ctx.context_id, ctx.score_value)
|
|
)
|
|
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)
|
|
)
|
|
return
|
|
# else rank_by_tokens
|
|
groups = TieResolver.rank_by_tokens(
|
|
war, ctx.context_type, ctx.context_id, ctx.participants
|
|
)
|
|
if len(groups[0]) == 1:
|
|
war.events.append(
|
|
TieResolved(
|
|
groups[0][0], ctx.context_type, ctx.context_id, ctx.score_value
|
|
)
|
|
)
|
|
return
|
|
# if tie persists, do nothing, workflow will call again
|
|
|
|
@staticmethod
|
|
def can_tie_be_resolved(
|
|
war: War, context_type: ContextType, context_id: str, participants: List[str]
|
|
) -> bool:
|
|
active = TieResolver.get_active_participants(
|
|
war, context_type, context_id, 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,
|
|
) -> 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
|
|
):
|
|
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:
|
|
return any(
|
|
isinstance(ev, TieResolved)
|
|
and ev.context_type == context_type
|
|
and ev.context_id == context_id
|
|
and ev.score_value == score_value
|
|
for ev in war.events
|
|
)
|