weyne/dcs_liberation
1
0
Fork 0
mirror of https://github.com/dcs-liberation/dcs_liberation.git synced 2025-11-10 14:22:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean up mission result handling.

Browse Source
This commit is contained in:
Dan Albert 2021-10-22 13:32:58 -07:00
parent 74291271e3
commit b728fcc2d6
11 changed files with 109 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
game/debriefing.py
View File

@ -8,6 +8,7 @@ import threading
import time
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import (
Any,
Callable,
@ -35,6 +36,7 @@ from game.ato.flight import Flight
if TYPE_CHECKING:
from game import Game
from game.sim import MissionSimulation
DEBRIEFING_LOG_EXTENSION = "log"
@ -361,13 +363,14 @@ class PollDebriefingFileThread(threading.Thread):
regularly for the stopped() condition."""
def __init__(
self, callback: Callable[[Debriefing], None], game: Game, unit_map: UnitMap
self,
callback: Callable[[Debriefing], None],
mission_simulation: MissionSimulation,
) -> None:
super().__init__()
self._stop_event = threading.Event()
self.callback = callback
self.game = game
self.unit_map = unit_map
self.mission_sim = mission_simulation
def stop(self) -> None:
self._stop_event.set()
@ -386,10 +389,9 @@ class PollDebriefingFileThread(threading.Thread):
os.path.isfile("state.json")
and os.path.getmtime("state.json") > last_modified
):
with open("state.json", "r", encoding="utf-8") as json_file:
json_data = json.load(json_file)
debriefing = Debriefing(json_data, self.game, self.unit_map)
self.callback(debriefing)
self.callback(
self.mission_sim.debrief_current_state(Path("state.json"))
)
break
except json.JSONDecodeError:
logging.exception(
@ -400,8 +402,8 @@ class PollDebriefingFileThread(threading.Thread):
def wait_for_debriefing(
callback: Callable[[Debriefing], None], game: Game, unit_map: UnitMap
callback: Callable[[Debriefing], None], mission_simulation: MissionSimulation
) -> PollDebriefingFileThread:
thread = PollDebriefingFileThread(callback, game, unit_map)
thread = PollDebriefingFileThread(callback, mission_simulation)
thread.start()
return thread

2
game/event/__init__.py
View File

@ -1,2 +0,0 @@
from .event import *
from .frontlineattack import *

10
game/event/airwar.py
View File

@ -1,10 +0,0 @@
from __future__ import annotations
from .event import Event
class AirWarEvent(Event):
"""Event handler for the air battle"""
def __str__(self) -> str:
return "AirWar"

12
game/event/frontlineattack.py
View File

@ -1,12 +0,0 @@
from .event import Event
class FrontlineAttackEvent(Event):
"""
An event centered on a FrontLine Conflict.
Currently the same as its parent, but here for legacy compatibility as well as to allow for
future unique Event handling
"""
def __str__(self) -> str:
return "Frontline attack"

52
game/game.py
View File

@ -6,9 +6,9 @@ import math
from collections import Iterator
from datetime import date, datetime, timedelta
from enum import Enum
from typing import Any, List, Type, Union, cast, TYPE_CHECKING
from typing import Any, List, TYPE_CHECKING, Type, Union, cast
from dcs.countries import Switzerland, UnitedNationsPeacekeepers, USAFAggressors
from dcs.countries import Switzerland, USAFAggressors, UnitedNationsPeacekeepers
from dcs.country import Country
from dcs.mapping import Point
from dcs.task import CAP, CAS, PinpointStrike
@ -19,19 +19,16 @@ from game.models.game_stats import GameStats
from game.plugins import LuaPluginManager
from gen import naming
from gen.flights.closestairfields import ObjectiveDistanceCache
from .ato.flighttype import FlightType
from gen.ground_forces.ai_ground_planner import GroundPlanner
from . import persistency
from .ato.flighttype import FlightType
from .campaignloader import CampaignAirWingConfig
from .coalition import Coalition
from .debriefing import Debriefing
from .event.event import Event
from .event.frontlineattack import FrontlineAttackEvent
from .factions.faction import Faction
from .infos.information import Information
from .profiling import logged_duration
from .settings import Settings
from .theater import ConflictTheater, ControlPoint
from .theater import ConflictTheater
from .theater.bullseye import Bullseye
from .theater.transitnetwork import TransitNetwork, TransitNetworkBuilder
from .weather import Conditions, TimeOfDay
@ -44,7 +41,6 @@ if TYPE_CHECKING:
from .navmesh import NavMesh
from .squadrons import AirWing
from .threatzones import ThreatZones
from .unitmap import UnitMap
COMMISION_UNIT_VARIETY = 4
COMMISION_LIMITS_SCALE = 1.5
@ -99,7 +95,6 @@ class Game:
enemy_budget: float,
) -> None:
self.settings = settings
self.events: List[Event] = []
self.theater = theater
self.turn = 0
# NB: This is the *start* date. It is never updated.
@ -183,20 +178,6 @@ class Game:
def bullseye_for(self, player: bool) -> Bullseye:
return self.coalition_for(player).bullseye
def _generate_player_event(
self, event_class: Type[Event], player_cp: ControlPoint, enemy_cp: ControlPoint
) -> None:
self.events.append(
event_class(
self,
player_cp,
enemy_cp,
enemy_cp.position,
self.blue.faction.name,
self.red.faction.name,
)
)
@property
def neutral_country(self) -> Type[Country]:
"""Return the best fitting country that can be used as neutral faction in the generated mission"""
@ -208,14 +189,6 @@ class Game:
else:
return USAFAggressors
def _generate_events(self) -> None:
for front_line in self.theater.conflicts():
self._generate_player_event(
FrontlineAttackEvent,
front_line.blue_cp,
front_line.red_cp,
)
def coalition_for(self, player: bool) -> Coalition:
if player:
return self.blue
@ -224,21 +197,6 @@ class Game:
def adjust_budget(self, amount: float, player: bool) -> None:
self.coalition_for(player).adjust_budget(amount)
@staticmethod
def initiate_event(event: Event) -> UnitMap:
# assert event in self.events
logging.info("Generating {} (regular)".format(event))
return event.generate()
def finish_event(self, event: Event, debriefing: Debriefing) -> None:
logging.info("Finishing event {}".format(event))
event.commit(debriefing)
if event in self.events:
self.events.remove(event)
else:
logging.info("finish_event: event not in the events!")
def on_load(self, game_still_initializing: bool = False) -> None:
if not hasattr(self, "name_generator"):
self.name_generator = naming.namegen
@ -378,8 +336,6 @@ class Game:
for_red: True if opfor should be re-initialized.
for_blue: True if the player coalition should be re-initialized.
"""
self.events = []
self._generate_events()
self.set_bullseye()
# Update statistics

1
game/sim/__init__.py Normal file
View File

@ -0,0 +1 @@
from .missionsimulation import MissionSimulation

80
game/event/event.py → game/sim/missionresultsprocessor.py
View File

@ -1,18 +1,12 @@
from __future__ import annotations
import logging
from typing import List, TYPE_CHECKING, Type
from typing import TYPE_CHECKING
from dcs.mapping import Point
from dcs.task import Task
from game import persistency
from game.debriefing import Debriefing
from game.theater import ControlPoint
from gen.ground_forces.combat_stance import CombatStance
from ..ato.airtaaskingorder import AirTaskingOrder
from ..missiongenerator import MissionGenerator
from ..unitmap import UnitMap
if TYPE_CHECKING:
from ..game import Game
@ -23,44 +17,24 @@ DEFEAT_INFLUENCE = 0.3
STRONG_DEFEAT_INFLUENCE = 0.5
class Event:
silent = False
informational = False
game = None # type: Game
location = None # type: Point
from_cp = None # type: ControlPoint
to_cp = None # type: ControlPoint
difficulty = 1 # type: int
def __init__(
self,
game: Game,
from_cp: ControlPoint,
target_cp: ControlPoint,
location: Point,
attacker_name: str,
defender_name: str,
) -> None:
class MissionResultsProcessor:
def __init__(self, game: Game) -> None:
self.game = game
self.from_cp = from_cp
self.to_cp = target_cp
self.location = location
self.attacker_name = attacker_name
self.defender_name = defender_name
@property
def is_player_attacking(self) -> bool:
return self.attacker_name == self.game.blue.faction.name
@property
def tasks(self) -> List[Type[Task]]:
return []
def generate(self) -> UnitMap:
return MissionGenerator(self.game).generate_miz(
persistency.mission_path_for("liberation_nextturn.miz")
)
def commit(self, debriefing: Debriefing) -> None:
logging.info("Committing mission results")
self.commit_air_losses(debriefing)
self.commit_pilot_experience()
self.commit_front_line_losses(debriefing)
self.commit_convoy_losses(debriefing)
self.commit_cargo_ship_losses(debriefing)
self.commit_airlift_losses(debriefing)
self.commit_ground_object_losses(debriefing)
self.commit_building_losses(debriefing)
self.commit_damaged_runways(debriefing)
self.commit_captures(debriefing)
self.commit_front_line_battle_impact(debriefing)
self.record_carcasses(debriefing)
def commit_air_losses(self, debriefing: Debriefing) -> None:
for loss in debriefing.air_losses.losses:
@ -200,27 +174,11 @@ class Event:
except Exception:
logging.exception(f"Could not process base capture {captured}")
def commit(self, debriefing: Debriefing) -> None:
logging.info("Committing mission results")
self.commit_air_losses(debriefing)
self.commit_pilot_experience()
self.commit_front_line_losses(debriefing)
self.commit_convoy_losses(debriefing)
self.commit_cargo_ship_losses(debriefing)
self.commit_airlift_losses(debriefing)
self.commit_ground_object_losses(debriefing)
self.commit_building_losses(debriefing)
self.commit_damaged_runways(debriefing)
self.commit_captures(debriefing)
# Destroyed units carcass
# -------------------------
def record_carcasses(self, debriefing: Debriefing) -> None:
for destroyed_unit in debriefing.state_data.destroyed_statics:
self.game.add_destroyed_units(destroyed_unit)
# -----------------------------------
# Compute damage to bases
def commit_front_line_battle_impact(self, debriefing: Debriefing) -> None:
for cp in self.game.theater.player_points():
enemy_cps = [e for e in cp.connected_points if not e.captured]
for enemy_cp in enemy_cps:

49
game/sim/missionsimulation.py Normal file
View File

@ -0,0 +1,49 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Optional, TYPE_CHECKING
from game.debriefing import Debriefing
from game.missiongenerator import MissionGenerator
from game.sim.missionresultsprocessor import MissionResultsProcessor
from game.unitmap import UnitMap
if TYPE_CHECKING:
from game import Game
class MissionSimulation:
def __init__(self, game: Game) -> None:
self.game = game
self.unit_map: Optional[UnitMap] = None
def generate_miz(self, output: Path) -> None:
self.unit_map = MissionGenerator(self.game).generate_miz(output)
def debrief_current_state(
self, state_path: Path, force_end: bool = False
) -> Debriefing:
if self.unit_map is None:
raise RuntimeError(
"Simulation has no unit map. Results processing began before a mission "
"was generated."
)
with state_path.open("r", encoding="utf-8") as state_file:
data = json.load(state_file)
if force_end:
data["mission_ended"] = True
return Debriefing(data, self.game, self.unit_map)
def process_results(self, debriefing: Debriefing) -> None:
if self.unit_map is None:
raise RuntimeError(
"Simulation has no unit map. Results processing began before a mission "
"was generated."
)
MissionResultsProcessor(self.game).commit(debriefing)
def finish(self) -> None:
self.unit_map = None

23
qt_ui/widgets/QTopPanel.py
View File

@ -10,11 +10,11 @@ from PySide2.QtWidgets import (
)
import qt_ui.uiconstants as CONST
from game import Game
from game.event.airwar import AirWarEvent
from game.profiling import logged_duration
from game.utils import meters
from game import Game, persistency
from game.ato.package import Package
from game.profiling import logged_duration
from game.sim import MissionSimulation
from game.utils import meters
from gen.flights.traveltime import TotEstimator
from qt_ui.models import GameModel
from qt_ui.widgets.QBudgetBox import QBudgetBox
@ -276,18 +276,11 @@ class QTopPanel(QFrame):
if negative_starts:
if not self.confirm_negative_start_time(negative_starts):
return
closest_cps = self.game.theater.closest_opposing_control_points()
game_event = AirWarEvent(
self.game,
closest_cps[0],
closest_cps[1],
self.game.theater.controlpoints[0].position,
self.game.blue.faction.name,
self.game.red.faction.name,
)
unit_map = self.game.initiate_event(game_event)
waiting = QWaitingForMissionResultWindow(game_event, self.game, unit_map, self)
sim = MissionSimulation(self.game)
sim.generate_miz(persistency.mission_path_for("liberation_nextturn.miz"))
waiting = QWaitingForMissionResultWindow(self.game, sim, self)
waiting.exec_()
def budget_update(self, game: Game):

2
qt_ui/windows/GameUpdateSignal.py
View File

@ -5,7 +5,7 @@ from typing import Optional
from PySide2.QtCore import QObject, Signal
from game import Game
from game.event import Debriefing
from game.debriefing import Debriefing
class GameUpdateSignal(QObject):

49
qt_ui/windows/QWaitingForMissionResultWindow.py
View File

@ -1,12 +1,13 @@
from __future__ import annotations
import json
import logging
import os
from typing import Sized, Optional
from pathlib import Path
from typing import Optional
from PySide2 import QtCore
from PySide2.QtCore import QObject, Qt, Signal
from PySide2.QtGui import QIcon, QMovie, QPixmap, QWindow
from PySide2.QtCore import QObject, Signal
from PySide2.QtGui import QIcon, QMovie, QPixmap
from PySide2.QtWidgets import (
QDialog,
QFileDialog,
@ -14,18 +15,17 @@ from PySide2.QtWidgets import (
QGroupBox,
QHBoxLayout,
QLabel,
QMessageBox,
QPushButton,
QTextBrowser,
QWidget,
)
from jinja2 import Environment, FileSystemLoader, select_autoescape
from game import Game
from game.debriefing import Debriefing, wait_for_debriefing
from game.game import Event, Game, logging
from game.persistency import base_path
from game.profiling import logged_duration
from game.unitmap import UnitMap
from game.sim import MissionSimulation
from qt_ui.windows.GameUpdateSignal import GameUpdateSignal
@ -52,16 +52,14 @@ DebriefingFileWrittenSignal()
class QWaitingForMissionResultWindow(QDialog):
def __init__(
self,
gameEvent: Event,
game: Game,
unit_map: UnitMap,
mission_simulation: MissionSimulation,
parent: Optional[QWidget] = None,
) -> None:
super(QWaitingForMissionResultWindow, self).__init__(parent=parent)
self.setWindowModality(QtCore.Qt.WindowModal)
self.gameEvent = gameEvent
self.game = game
self.unit_map = unit_map
self.mission_sim = mission_simulation
self.setWindowTitle("Waiting for mission completion.")
self.setWindowIcon(QIcon("./resources/icon.png"))
self.setMinimumHeight(570)
@ -71,9 +69,7 @@ class QWaitingForMissionResultWindow(QDialog):
self.updateLayout
)
self.wait_thread = wait_for_debriefing(
lambda debriefing: self.on_debriefing_update(debriefing),
self.game,
self.unit_map,
lambda debriefing: self.on_debriefing_update(debriefing), self.mission_sim
)
def initUi(self):
@ -209,12 +205,12 @@ class QWaitingForMissionResultWindow(QDialog):
except Exception:
logging.exception("Got an error while sending debriefing")
self.wait_thread = wait_for_debriefing(
lambda d: self.on_debriefing_update(d), self.game, self.unit_map
lambda d: self.on_debriefing_update(d), self.mission_sim
)
def process_debriefing(self):
with logged_duration("Turn processing"):
self.game.finish_event(event=self.gameEvent, debriefing=self.debriefing)
self.mission_sim.process_results(self.debriefing)
self.game.pass_turn()
GameUpdateSignal.get_instance().sendDebriefing(self.debriefing)
@ -233,20 +229,7 @@ class QWaitingForMissionResultWindow(QDialog):
file = QFileDialog.getOpenFileName(
self, "Select game file to open", filter="json(*.json)", dir="."
)
print(file)
try:
with open(file[0], "r", encoding="utf-8") as json_file:
json_data = json.load(json_file)
json_data["mission_ended"] = True
debriefing = Debriefing(json_data, self.game, self.unit_map)
self.on_debriefing_update(debriefing)
except Exception as e:
logging.error(e)
msg = QMessageBox()
msg.setIcon(QMessageBox.Information)
msg.setText("Invalid file : " + file[0])
msg.setWindowTitle("Invalid file.")
msg.setStandardButtons(QMessageBox.Ok)
msg.setWindowFlags(Qt.WindowStaysOnTopHint)
msg.exec_()
return
logging.debug("Processing manually submitted %s", file[0])
self.on_debriefing_update(
self.mission_sim.debrief_current_state(Path(file[0], force_end=True))
)