mirror of
https://github.com/dcs-liberation/dcs_liberation.git
synced 2025-11-10 14:22:26 +00:00
edfaaacd04
This way the results of committing the debriefing can't alter the view of the debriefing. It looks like it was probably that case that debriefing information displays (but not the committed results) would be incorrect after a base capture because the results might be shown after the results were committed. Maybe fixes https://github.com/Khopa/dcs_liberation/issues/513
273 lines
9.7 KiB
Python
273 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
import itertools
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from typing import (
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
Iterator,
|
|
List,
|
|
Type,
|
|
TYPE_CHECKING,
|
|
)
|
|
|
|
from dcs.unittype import FlyingType, UnitType
|
|
|
|
from game import db
|
|
from game.theater import Airfield, ControlPoint
|
|
from game.unitmap import Building, FrontLineUnit, GroundObjectUnit, UnitMap
|
|
from gen.flights.flight import Flight
|
|
|
|
if TYPE_CHECKING:
|
|
from game import Game
|
|
|
|
DEBRIEFING_LOG_EXTENSION = "log"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AirLosses:
|
|
player: List[Flight]
|
|
enemy: List[Flight]
|
|
|
|
@property
|
|
def losses(self) -> Iterator[Flight]:
|
|
return itertools.chain(self.player, self.enemy)
|
|
|
|
def by_type(self, player: bool) -> Dict[Type[FlyingType], int]:
|
|
losses_by_type: Dict[Type[FlyingType], int] = defaultdict(int)
|
|
losses = self.player if player else self.enemy
|
|
for loss in losses:
|
|
losses_by_type[loss.unit_type] += 1
|
|
return losses_by_type
|
|
|
|
def surviving_flight_members(self, flight: Flight) -> int:
|
|
losses = 0
|
|
for loss in self.losses:
|
|
if loss == flight:
|
|
losses += 1
|
|
return flight.count - losses
|
|
|
|
|
|
@dataclass
|
|
class GroundLosses:
|
|
player_front_line: List[FrontLineUnit] = field(default_factory=list)
|
|
enemy_front_line: List[FrontLineUnit] = field(default_factory=list)
|
|
|
|
player_ground_objects: List[GroundObjectUnit] = field(default_factory=list)
|
|
enemy_ground_objects: List[GroundObjectUnit] = field(default_factory=list)
|
|
|
|
player_buildings: List[Building] = field(default_factory=list)
|
|
enemy_buildings: List[Building] = field(default_factory=list)
|
|
|
|
player_airfields: List[Airfield] = field(default_factory=list)
|
|
enemy_airfields: List[Airfield] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StateData:
|
|
#: True if the mission ended. If False, the mission exited abnormally.
|
|
mission_ended: bool
|
|
|
|
#: Names of aircraft units that were killed during the mission.
|
|
killed_aircraft: List[str]
|
|
|
|
#: Names of vehicle (and ship) units that were killed during the mission.
|
|
killed_ground_units: List[str]
|
|
|
|
#: Names of static units that were destroyed during the mission.
|
|
destroyed_statics: List[str]
|
|
|
|
#: Mangled names of bases that were captured during the mission.
|
|
base_capture_events: List[str]
|
|
|
|
@classmethod
|
|
def from_json(cls, data: Dict[str, Any]) -> StateData:
|
|
return cls(
|
|
mission_ended=data["mission_ended"],
|
|
killed_aircraft=data["killed_aircrafts"],
|
|
# Airfields emit a new "dead" event every time a bomb is dropped on
|
|
# them when they've already dead. Dedup.
|
|
killed_ground_units=list(set(data["killed_ground_units"])),
|
|
destroyed_statics=data["destroyed_objects_positions"],
|
|
base_capture_events=data["base_capture_events"]
|
|
)
|
|
|
|
|
|
class Debriefing:
|
|
def __init__(self, state_data: Dict[str, Any], game: Game,
|
|
unit_map: UnitMap) -> None:
|
|
self.state_data = StateData.from_json(state_data)
|
|
self.unit_map = unit_map
|
|
|
|
self.player_country = game.player_country
|
|
self.enemy_country = game.enemy_country
|
|
self.player_country_id = db.country_id_from_name(game.player_country)
|
|
self.enemy_country_id = db.country_id_from_name(game.enemy_country)
|
|
|
|
self.air_losses = self.dead_aircraft()
|
|
self.ground_losses = self.dead_ground_units()
|
|
|
|
@property
|
|
def front_line_losses(self) -> Iterator[FrontLineUnit]:
|
|
yield from self.ground_losses.player_front_line
|
|
yield from self.ground_losses.enemy_front_line
|
|
|
|
@property
|
|
def ground_object_losses(self) -> Iterator[GroundObjectUnit]:
|
|
yield from self.ground_losses.player_ground_objects
|
|
yield from self.ground_losses.enemy_ground_objects
|
|
|
|
@property
|
|
def building_losses(self) -> Iterator[Building]:
|
|
yield from self.ground_losses.player_buildings
|
|
yield from self.ground_losses.enemy_buildings
|
|
|
|
@property
|
|
def damaged_runways(self) -> Iterator[Airfield]:
|
|
yield from self.ground_losses.player_airfields
|
|
yield from self.ground_losses.enemy_airfields
|
|
|
|
def casualty_count(self, control_point: ControlPoint) -> int:
|
|
return len(
|
|
[x for x in self.front_line_losses if x.origin == control_point]
|
|
)
|
|
|
|
def front_line_losses_by_type(
|
|
self, player: bool) -> Dict[Type[UnitType], int]:
|
|
losses_by_type: Dict[Type[UnitType], int] = defaultdict(int)
|
|
if player:
|
|
losses = self.ground_losses.player_front_line
|
|
else:
|
|
losses = self.ground_losses.enemy_front_line
|
|
for loss in losses:
|
|
losses_by_type[loss.unit_type] += 1
|
|
return losses_by_type
|
|
|
|
def building_losses_by_type(self, player: bool) -> Dict[str, int]:
|
|
losses_by_type: Dict[str, int] = defaultdict(int)
|
|
if player:
|
|
losses = self.ground_losses.player_buildings
|
|
else:
|
|
losses = self.ground_losses.enemy_buildings
|
|
for loss in losses:
|
|
if loss.ground_object.control_point.captured != player:
|
|
continue
|
|
|
|
losses_by_type[loss.ground_object.dcs_identifier] += 1
|
|
return losses_by_type
|
|
|
|
def dead_aircraft(self) -> AirLosses:
|
|
player_losses = []
|
|
enemy_losses = []
|
|
for unit_name in self.state_data.killed_aircraft:
|
|
flight = self.unit_map.flight(unit_name)
|
|
if flight is None:
|
|
logging.error(f"Could not find Flight matching {unit_name}")
|
|
continue
|
|
if flight.departure.captured:
|
|
player_losses.append(flight)
|
|
else:
|
|
enemy_losses.append(flight)
|
|
return AirLosses(player_losses, enemy_losses)
|
|
|
|
def dead_ground_units(self) -> GroundLosses:
|
|
losses = GroundLosses()
|
|
for unit_name in self.state_data.killed_ground_units:
|
|
front_line_unit = self.unit_map.front_line_unit(unit_name)
|
|
if front_line_unit is not None:
|
|
if front_line_unit.origin.captured:
|
|
losses.player_front_line.append(front_line_unit)
|
|
else:
|
|
losses.enemy_front_line.append(front_line_unit)
|
|
continue
|
|
|
|
ground_object_unit = self.unit_map.ground_object_unit(unit_name)
|
|
if ground_object_unit is not None:
|
|
if ground_object_unit.ground_object.control_point.captured:
|
|
losses.player_ground_objects.append(ground_object_unit)
|
|
else:
|
|
losses.enemy_ground_objects.append(ground_object_unit)
|
|
continue
|
|
|
|
building = self.unit_map.building_or_fortification(unit_name)
|
|
if building is not None:
|
|
if building.ground_object.control_point.captured:
|
|
losses.player_buildings.append(building)
|
|
else:
|
|
losses.enemy_buildings.append(building)
|
|
continue
|
|
|
|
airfield = self.unit_map.airfield(unit_name)
|
|
if airfield is not None:
|
|
if airfield.captured:
|
|
losses.player_airfields.append(airfield)
|
|
else:
|
|
losses.enemy_airfields.append(airfield)
|
|
continue
|
|
|
|
# Only logging as debug because we don't currently track infantry
|
|
# deaths, so we expect to see quite a few unclaimed dead ground
|
|
# units. We should start tracking those and covert this to a
|
|
# warning.
|
|
logging.debug(f"Death of untracked ground unit {unit_name} will "
|
|
"have no effect. This may be normal behavior.")
|
|
|
|
return losses
|
|
|
|
@property
|
|
def base_capture_events(self):
|
|
"""Keeps only the last instance of a base capture event for each base ID."""
|
|
reversed_captures = list(reversed(self.state_data.base_capture_events))
|
|
last_base_cap_indexes = []
|
|
for idx, base in enumerate(i.split("||")[0] for i in reversed_captures):
|
|
if base not in [x[1] for x in last_base_cap_indexes]:
|
|
last_base_cap_indexes.append((idx, base))
|
|
return [reversed_captures[idx[0]] for idx in last_base_cap_indexes]
|
|
|
|
|
|
class PollDebriefingFileThread(threading.Thread):
|
|
"""Thread class with a stop() method. The thread itself has to check
|
|
regularly for the stopped() condition."""
|
|
|
|
def __init__(self, callback: Callable[[Debriefing], None],
|
|
game: Game, unit_map: UnitMap) -> None:
|
|
super().__init__()
|
|
self._stop_event = threading.Event()
|
|
self.callback = callback
|
|
self.game = game
|
|
self.unit_map = unit_map
|
|
|
|
def stop(self):
|
|
self._stop_event.set()
|
|
|
|
def stopped(self):
|
|
return self._stop_event.is_set()
|
|
|
|
def run(self):
|
|
if os.path.isfile("state.json"):
|
|
last_modified = os.path.getmtime("state.json")
|
|
else:
|
|
last_modified = 0
|
|
while not self.stopped():
|
|
if os.path.isfile("state.json") and os.path.getmtime("state.json") > last_modified:
|
|
with open("state.json", "r") as json_file:
|
|
json_data = json.load(json_file)
|
|
debriefing = Debriefing(json_data, self.game, self.unit_map)
|
|
self.callback(debriefing)
|
|
break
|
|
time.sleep(5)
|
|
|
|
|
|
def wait_for_debriefing(callback: Callable[[Debriefing], None],
|
|
game: Game, unit_map) -> PollDebriefingFileThread:
|
|
thread = PollDebriefingFileThread(callback, game, unit_map)
|
|
thread.start()
|
|
return thread
|