494 lines
17 KiB
Python
494 lines
17 KiB
Python
from typing import Any, Dict, List
|
|
from pathlib import Path
|
|
import json
|
|
import shutil
|
|
from datetime import datetime
|
|
|
|
from warchron.model.exception import ForbiddenOperation
|
|
from warchron.model.player import Player
|
|
from warchron.model.war import War
|
|
from warchron.model.war_participant import WarParticipant
|
|
from warchron.model.objective import Objective
|
|
from warchron.model.campaign import Campaign
|
|
from warchron.model.campaign_participant import CampaignParticipant
|
|
from warchron.model.sector import Sector
|
|
from warchron.model.round import Round
|
|
from warchron.model.choice import Choice
|
|
from warchron.model.battle import Battle
|
|
|
|
|
|
class Model:
|
|
def __init__(self) -> None:
|
|
self.players: Dict[str, Player] = {}
|
|
self.wars: Dict[str, War] = {}
|
|
|
|
# File management methods
|
|
|
|
def new(self) -> None:
|
|
self.players.clear()
|
|
self.wars.clear()
|
|
|
|
def load(self, path: Path) -> None:
|
|
self.players.clear()
|
|
self.wars.clear()
|
|
self._load_data(path)
|
|
|
|
def save(self, path: Path) -> None:
|
|
self._save_data(path)
|
|
|
|
def _load_data(self, path: Path) -> None:
|
|
if not path.exists() or path.stat().st_size == 0:
|
|
return # Start empty
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
self.players.clear()
|
|
self.wars.clear()
|
|
for p in data.get("players", []):
|
|
player = Player.fromDict(p)
|
|
self.players[player.id] = player
|
|
for w in data.get("wars", []):
|
|
war = War.fromDict(w)
|
|
self.wars[war.id] = war
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError("Data file is corrupted")
|
|
|
|
def _save_data(self, path: Path) -> None:
|
|
if path.exists():
|
|
shutil.copy(path, path.with_suffix(".json.bak"))
|
|
data = {
|
|
"version": "1.0",
|
|
"players": [p.toDict() for p in self.players.values()],
|
|
"wars": [w.toDict() for w in self.wars.values()],
|
|
}
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
# Player methods
|
|
|
|
def add_player(self, name: str) -> Player:
|
|
player = Player(name)
|
|
self.players[player.id] = player
|
|
return player
|
|
|
|
def get_player(self, id: str) -> Player:
|
|
return self.players[id]
|
|
|
|
def get_player_name(self, player_id: str) -> str:
|
|
return self.players[player_id].name
|
|
|
|
def update_player(self, player_id: str, *, name: str) -> None:
|
|
player = self.get_player(player_id)
|
|
player.set_name(name)
|
|
|
|
def get_all_players(self) -> List[Player]:
|
|
return list(self.players.values())
|
|
|
|
def remove_player(self, player_id: str) -> None:
|
|
wars_using_player: List[str] = []
|
|
for war in self.wars.values():
|
|
if war.has_player(player_id):
|
|
wars_using_player.append(war.name)
|
|
if wars_using_player:
|
|
wars_str = ", ".join(wars_using_player)
|
|
raise ForbiddenOperation(
|
|
f"This player is participating in war(s): {wars_str}.\n"
|
|
"Remove it from participants first."
|
|
)
|
|
del self.players[player_id]
|
|
|
|
# War methods
|
|
|
|
def get_default_war_values(self) -> Dict[str, Any]:
|
|
return {"year": datetime.now().year}
|
|
|
|
def add_war(self, name: str, year: int) -> War:
|
|
war = War(name, year)
|
|
self.wars[war.id] = war
|
|
return war
|
|
|
|
def get_war(self, id: str) -> War:
|
|
return self.wars[id]
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_war_by_campaign(self, campaign_id: str) -> War:
|
|
for war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
if camp.id == campaign_id:
|
|
return war
|
|
raise KeyError(f"Campaign {campaign_id} not found in any War")
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_war_by_sector(self, sector_id: str) -> War:
|
|
for war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
for sect in camp.sectors.values():
|
|
if sect.id == sector_id:
|
|
return war
|
|
raise KeyError(f"Sector {sector_id} not found in any War")
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_war_by_round(self, round_id: str) -> War:
|
|
for war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
for rnd in camp.rounds:
|
|
if rnd.id == round_id:
|
|
return war
|
|
raise KeyError(f"Round {round_id} not found in any War")
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_war_by_objective(self, objective_id: str) -> War:
|
|
for war in self.wars.values():
|
|
for obj in war.objectives.values():
|
|
if obj.id == objective_id:
|
|
return war
|
|
raise KeyError(f"Objective {objective_id} not found in any War")
|
|
|
|
def get_war_by_war_participant(self, participant_id: str) -> War:
|
|
for war in self.wars.values():
|
|
if war.has_participant(participant_id):
|
|
return war
|
|
raise KeyError(f"Participant {participant_id} not found in any War")
|
|
|
|
def get_war_by_campaign_participant(self, participant_id: str) -> War:
|
|
for war in self.wars.values():
|
|
camp = war.get_campaign_by_campaign_participant(participant_id)
|
|
if camp is not None:
|
|
return war
|
|
raise KeyError(f"Participant {participant_id} not found")
|
|
|
|
def update_war(self, war_id: str, *, name: str, year: int) -> None:
|
|
war = self.get_war(war_id)
|
|
if war.is_over:
|
|
raise ForbiddenOperation("Can't update a closed war.")
|
|
war.set_name(name)
|
|
war.set_year(year)
|
|
|
|
def set_major_value(self, war_id: str, value: int) -> None:
|
|
war = self.get_war(war_id)
|
|
war.set_major(value)
|
|
|
|
def set_minor_value(self, war_id: str, value: int) -> None:
|
|
war = self.get_war(war_id)
|
|
war.set_minor(value)
|
|
|
|
def set_influence_token(self, war_id: str, value: bool) -> None:
|
|
war = self.get_war(war_id)
|
|
war.set_influence(value)
|
|
|
|
def get_all_wars(self) -> List[War]:
|
|
return list(self.wars.values())
|
|
|
|
def remove_war(self, war_id: str) -> None:
|
|
war = self.wars[war_id]
|
|
if war:
|
|
if war.is_over:
|
|
raise ForbiddenOperation("Can't remove closed war.")
|
|
if war.has_finished_campaign():
|
|
raise ForbiddenOperation("Can't remove war with finished campaign(s).")
|
|
if war.has_finished_round():
|
|
raise ForbiddenOperation(
|
|
"Can't remove war with finished round(s) in campaign(s)."
|
|
)
|
|
if war.has_finished_battle():
|
|
raise ForbiddenOperation(
|
|
"Can't remove war with finished battle(s) in round(s) "
|
|
"in campaign(s)."
|
|
)
|
|
del self.wars[war_id]
|
|
|
|
# Objective methods
|
|
|
|
def add_objective(
|
|
self, war_id: str, name: str, description: str | None
|
|
) -> Objective:
|
|
war = self.get_war(war_id)
|
|
return war.add_objective(name, description)
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_objective(self, objective_id: str) -> Objective:
|
|
for war in self.wars.values():
|
|
for obj in war.objectives.values():
|
|
if obj.id == objective_id:
|
|
return obj
|
|
raise KeyError("Objective not found")
|
|
|
|
def update_objective(
|
|
self, objective_id: str, *, name: str, description: str | None
|
|
) -> None:
|
|
war = self.get_war_by_objective(objective_id)
|
|
war.update_objective(objective_id, name=name, description=description)
|
|
|
|
def remove_objective(self, objective_id: str) -> None:
|
|
war = self.get_war_by_objective(objective_id)
|
|
war.remove_objective(objective_id)
|
|
|
|
# War participant methods
|
|
|
|
def get_available_players(self, war_id: str) -> List[Player]:
|
|
war = self.get_war(war_id)
|
|
return [
|
|
player for player in self.players.values() if not war.has_player(player.id)
|
|
]
|
|
|
|
def add_war_participant(
|
|
self, war_id: str, player_id: str, faction: str
|
|
) -> WarParticipant:
|
|
war = self.get_war(war_id)
|
|
return war.add_war_participant(player_id, faction)
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_war_participant(self, participant_id: str) -> WarParticipant:
|
|
for war in self.wars.values():
|
|
for part in war.participants.values():
|
|
if part.id == participant_id:
|
|
return part
|
|
raise KeyError("Participant not found")
|
|
|
|
def get_player_from_war_participant(self, war_part: WarParticipant) -> Player:
|
|
return self.get_player(war_part.player_id)
|
|
|
|
def get_participant_name(self, participant_id: str) -> str:
|
|
war = self.get_war_by_war_participant(participant_id)
|
|
war_part = war.get_war_participant(participant_id)
|
|
return self.players[war_part.player_id].name
|
|
|
|
def update_war_participant(self, participant_id: str, *, faction: str) -> None:
|
|
war = self.get_war_by_war_participant(participant_id)
|
|
war.update_war_participant(participant_id, faction=faction)
|
|
|
|
def remove_war_participant(self, participant_id: str) -> None:
|
|
war = self.get_war_by_war_participant(participant_id)
|
|
war.remove_war_participant(participant_id)
|
|
|
|
# Campaign methods
|
|
|
|
def get_default_campaign_values(self, war_id: str) -> Dict[str, Any]:
|
|
war = self.get_war(war_id)
|
|
return war.get_default_campaign_values()
|
|
|
|
def add_campaign(self, war_id: str, name: str, month: int) -> Campaign:
|
|
war = self.get_war(war_id)
|
|
return war.add_campaign(name, month)
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_campaign(self, campaign_id: str) -> Campaign:
|
|
for war in self.wars.values():
|
|
for campaign in war.campaigns:
|
|
if campaign.id == campaign_id:
|
|
return campaign
|
|
raise KeyError("Campaign not found")
|
|
|
|
def get_campaign_by_round(self, round_id: str) -> Campaign:
|
|
for war in self.wars.values():
|
|
camp = war.get_campaign_by_round(round_id)
|
|
if camp is not None:
|
|
return camp
|
|
raise KeyError(f"Round {round_id} not found")
|
|
|
|
def get_campaign_by_campaign_participant(self, participant_id: str) -> Campaign:
|
|
for war in self.wars.values():
|
|
camp = war.get_campaign_by_campaign_participant(participant_id)
|
|
if camp is not None:
|
|
return camp
|
|
raise KeyError(f"Participant {participant_id} not found")
|
|
|
|
def get_campaign_by_sector(self, sector_id: str) -> Campaign:
|
|
for war in self.wars.values():
|
|
camp = war.get_campaign_by_sector(sector_id)
|
|
if camp is not None:
|
|
return camp
|
|
raise KeyError(f"Sector {sector_id} not found")
|
|
|
|
def update_campaign(self, campaign_id: str, *, name: str, month: int) -> None:
|
|
war = self.get_war_by_campaign(campaign_id)
|
|
war.update_campaign(campaign_id, name=name, month=month)
|
|
|
|
def remove_campaign(self, campaign_id: str) -> None:
|
|
war = self.get_war_by_campaign(campaign_id)
|
|
war.remove_campaign(campaign_id)
|
|
|
|
# Sector methods
|
|
|
|
def add_sector(
|
|
self,
|
|
campaign_id: str,
|
|
name: str,
|
|
round_id: str | None,
|
|
major_id: str | None,
|
|
minor_id: str | None,
|
|
influence_id: str | None,
|
|
mission: str | None,
|
|
description: str | None,
|
|
) -> Sector:
|
|
camp = self.get_campaign(campaign_id)
|
|
return camp.add_sector(
|
|
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 war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
for sect in camp.sectors.values():
|
|
if sect.id == sector_id:
|
|
return sect
|
|
raise KeyError("Sector not found")
|
|
|
|
def update_sector(
|
|
self,
|
|
sector_id: str,
|
|
*,
|
|
name: str,
|
|
round_id: str | None,
|
|
major_id: str | None,
|
|
minor_id: str | None,
|
|
influence_id: str | None,
|
|
mission: str | None,
|
|
description: str | None,
|
|
) -> None:
|
|
war = self.get_war_by_sector(sector_id)
|
|
war.update_sector(
|
|
sector_id,
|
|
name=name,
|
|
round_id=round_id,
|
|
major_id=major_id,
|
|
minor_id=minor_id,
|
|
influence_id=influence_id,
|
|
mission=mission,
|
|
description=description,
|
|
)
|
|
|
|
def remove_sector(self, sector_id: str) -> None:
|
|
camp = self.get_campaign_by_sector(sector_id)
|
|
camp.remove_sector(sector_id)
|
|
|
|
# Campaign participant methods
|
|
|
|
def get_available_war_participants(self, campaign_id: str) -> List[WarParticipant]:
|
|
war = self.get_war_by_campaign(campaign_id)
|
|
return war.get_available_war_participants(campaign_id)
|
|
|
|
def add_campaign_participant(
|
|
self, camp_id: str, player_id: str, leader: str, theme: str
|
|
) -> CampaignParticipant:
|
|
camp = self.get_campaign(camp_id)
|
|
return camp.add_campaign_participant(player_id, leader, theme)
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_campaign_participant(self, participant_id: str) -> CampaignParticipant:
|
|
for war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
for part in camp.participants.values():
|
|
if part.id == participant_id:
|
|
return part
|
|
raise KeyError("Participant not found")
|
|
|
|
def get_player_from_campaign_participant(
|
|
self, camp_part: CampaignParticipant
|
|
) -> Player:
|
|
war_part = self.get_war_participant(camp_part.war_participant_id)
|
|
return self.get_player(war_part.player_id)
|
|
|
|
def update_campaign_participant(
|
|
self,
|
|
participant_id: str,
|
|
*,
|
|
leader: str,
|
|
theme: str,
|
|
) -> None:
|
|
war = self.get_war_by_campaign_participant(participant_id)
|
|
war.update_campaign_participant(participant_id, leader=leader, theme=theme)
|
|
|
|
def remove_campaign_participant(self, participant_id: str) -> None:
|
|
war = self.get_war_by_campaign_participant(participant_id)
|
|
war.remove_campaign_participant(participant_id)
|
|
|
|
# Round methods
|
|
|
|
def add_round(self, campaign_id: str) -> Round:
|
|
war = self.get_war_by_campaign(campaign_id)
|
|
return war.add_round(campaign_id)
|
|
|
|
# TODO replace multiloops by internal has_* method
|
|
def get_round(self, round_id: str) -> Round:
|
|
for war in self.wars.values():
|
|
for camp in war.campaigns:
|
|
for rnd in camp.rounds:
|
|
if rnd.id == round_id:
|
|
return rnd
|
|
raise KeyError("Round not found")
|
|
|
|
def get_round_index(self, round_id: str | None) -> int | None:
|
|
if round_id is None:
|
|
return None
|
|
camp = self.get_campaign_by_round(round_id)
|
|
return camp.get_round_index(round_id)
|
|
|
|
def get_round_sectors(self, round_id: str) -> List[Sector]:
|
|
camp = self.get_campaign_by_round(round_id)
|
|
return [s for s in camp.sectors.values() if s.round_id == round_id]
|
|
|
|
def get_round_participants(self, round_id: str) -> List[CampaignParticipant]:
|
|
camp = self.get_campaign_by_round(round_id)
|
|
return list(camp.participants.values())
|
|
|
|
def remove_round(self, round_id: str) -> None:
|
|
war = self.get_war_by_round(round_id)
|
|
war.remove_round(round_id)
|
|
|
|
# Choice methods
|
|
|
|
def create_choice(self, round_id: str, participant_id: str) -> Choice:
|
|
war = self.get_war_by_round(round_id)
|
|
return war.create_choice(round_id, participant_id)
|
|
|
|
def update_choice(
|
|
self,
|
|
round_id: str,
|
|
participant_id: str,
|
|
priority_sector_id: str | None,
|
|
secondary_sector_id: str | None,
|
|
comment: str | None,
|
|
) -> None:
|
|
war = self.get_war_by_round(round_id)
|
|
war.update_choice(
|
|
round_id, participant_id, priority_sector_id, secondary_sector_id, comment
|
|
)
|
|
|
|
def remove_choice(self, round_id: str, participant_id: str) -> None:
|
|
war = self.get_war_by_round(round_id)
|
|
war.remove_choice(round_id, participant_id)
|
|
|
|
# Battle methods
|
|
|
|
def create_battle(self, round_id: str, sector_id: str) -> Battle:
|
|
war = self.get_war_by_round(round_id)
|
|
return war.create_battle(round_id, sector_id)
|
|
|
|
def update_battle(
|
|
self,
|
|
round_id: str,
|
|
sector_id: str,
|
|
player_1_id: str | None,
|
|
player_2_id: str | None,
|
|
winner_id: str | None,
|
|
score: str | None,
|
|
victory_condition: str | None,
|
|
comment: str | None,
|
|
) -> None:
|
|
war = self.get_war_by_round(round_id)
|
|
war.update_battle(
|
|
round_id,
|
|
sector_id,
|
|
player_1_id,
|
|
player_2_id,
|
|
winner_id,
|
|
score,
|
|
victory_condition,
|
|
comment,
|
|
)
|
|
|
|
def remove_battle(self, round_id: str, sector_id: str) -> None:
|
|
war = self.get_war_by_round(round_id)
|
|
war.remove_battle(round_id, sector_id)
|