mirror of
https://github.com/dcs-retribution/dcs-retribution.git
synced 2025-11-10 15:41:24 +00:00
489 lines
19 KiB
Python
489 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import random
|
|
from datetime import datetime
|
|
from functools import cached_property
|
|
from typing import Any, Dict, List, TYPE_CHECKING, Tuple
|
|
from uuid import UUID
|
|
|
|
from dcs import Point
|
|
from dcs.country import Country
|
|
from dcs.mission import Mission
|
|
from dcs.unitgroup import FlyingGroup, StaticGroup
|
|
|
|
from game.ato.airtaaskingorder import AirTaskingOrder
|
|
from game.ato.flight import Flight
|
|
from game.ato.flightstate import Completed, WaitingForStart, Navigating
|
|
from game.ato.flighttype import FlightType
|
|
from game.ato.package import Package
|
|
from game.ato.starttype import StartType
|
|
from game.missiongenerator.aircraft.flightgroupconfigurator import (
|
|
FlightGroupConfigurator,
|
|
)
|
|
from game.missiongenerator.lasercoderegistry import LaserCodeRegistry
|
|
from game.missiongenerator.missiondata import MissionData
|
|
from game.radio.radios import RadioRegistry
|
|
from game.radio.tacan import TacanRegistry
|
|
from game.runways import RunwayData
|
|
from game.settings import Settings
|
|
from game.squadrons import AirWing
|
|
from game.theater.controlpoint import (
|
|
ControlPoint,
|
|
OffMapSpawn,
|
|
ParkingType,
|
|
)
|
|
from game.unitmap import UnitMap
|
|
from game.missiongenerator.aircraft.aircraftpainter import AircraftPainter
|
|
from game.missiongenerator.aircraft.flightdata import FlightData
|
|
from game.data.weapons import WeaponType
|
|
|
|
if TYPE_CHECKING:
|
|
from game import Game
|
|
|
|
|
|
PRETENSE_SQUADRON_DEF_RETRIES = 100
|
|
PRETENSE_SEAD_FLIGHTS_PER_CP = 1
|
|
PRETENSE_CAS_FLIGHTS_PER_CP = 1
|
|
PRETENSE_STRIKE_FLIGHTS_PER_CP = 1
|
|
PRETENSE_BARCAP_FLIGHTS_PER_CP = 1
|
|
PRETENSE_AI_AIRCRAFT_PER_FLIGHT = 2
|
|
PRETENSE_AI_AWACS_PER_FLIGHT = 1
|
|
PRETENSE_AI_TANKERS_PER_FLIGHT = 1
|
|
|
|
|
|
class PretenseAircraftGenerator:
|
|
def __init__(
|
|
self,
|
|
mission: Mission,
|
|
settings: Settings,
|
|
game: Game,
|
|
time: datetime,
|
|
radio_registry: RadioRegistry,
|
|
tacan_registry: TacanRegistry,
|
|
laser_code_registry: LaserCodeRegistry,
|
|
unit_map: UnitMap,
|
|
mission_data: MissionData,
|
|
helipads: dict[ControlPoint, list[StaticGroup]],
|
|
ground_spawns_roadbase: dict[ControlPoint, list[Tuple[StaticGroup, Point]]],
|
|
ground_spawns: dict[ControlPoint, list[Tuple[StaticGroup, Point]]],
|
|
) -> None:
|
|
self.mission = mission
|
|
self.settings = settings
|
|
self.game = game
|
|
self.time = time
|
|
self.radio_registry = radio_registry
|
|
self.tacan_registy = tacan_registry
|
|
self.laser_code_registry = laser_code_registry
|
|
self.unit_map = unit_map
|
|
self.flights: List[FlightData] = []
|
|
self.mission_data = mission_data
|
|
self.helipads = helipads
|
|
self.ground_spawns_roadbase = ground_spawns_roadbase
|
|
self.ground_spawns = ground_spawns
|
|
|
|
self.ewrj_package_dict: Dict[int, List[FlyingGroup[Any]]] = {}
|
|
self.ewrj = settings.plugins.get("ewrj")
|
|
self.need_ecm = settings.plugin_option("ewrj.ecm_required")
|
|
|
|
@cached_property
|
|
def use_client(self) -> bool:
|
|
"""True if Client should be used instead of Player."""
|
|
blue_clients = self.client_slots_in_ato(self.game.blue.ato)
|
|
red_clients = self.client_slots_in_ato(self.game.red.ato)
|
|
return blue_clients + red_clients > 1
|
|
|
|
@staticmethod
|
|
def client_slots_in_ato(ato: AirTaskingOrder) -> int:
|
|
total = 0
|
|
for package in ato.packages:
|
|
for flight in package.flights:
|
|
total += flight.client_count
|
|
return total
|
|
|
|
def clear_parking_slots(self) -> None:
|
|
for cp in self.game.theater.controlpoints:
|
|
for parking_slot in cp.parking_slots:
|
|
parking_slot.unit_id = None
|
|
|
|
def find_pretense_cargo_plane_cp(self, cp: ControlPoint) -> ControlPoint:
|
|
"""
|
|
Finds the location (ControlPoint) for Pretense off-map transport planes.
|
|
"""
|
|
distance_to_flot = 0.0
|
|
offmap_transport_cp_id = cp.id
|
|
parking_type = ParkingType(
|
|
fixed_wing=True, fixed_wing_stol=True, rotary_wing=False
|
|
)
|
|
for front_line_cp in self.game.theater.controlpoints:
|
|
if isinstance(front_line_cp, OffMapSpawn):
|
|
continue
|
|
for front_line in self.game.theater.conflicts():
|
|
if front_line_cp.captured == cp.captured:
|
|
if (
|
|
front_line_cp.total_aircraft_parking(parking_type) > 0
|
|
and front_line_cp.position.distance_to_point(
|
|
front_line.position
|
|
)
|
|
> distance_to_flot
|
|
):
|
|
distance_to_flot = front_line_cp.position.distance_to_point(
|
|
front_line.position
|
|
)
|
|
offmap_transport_cp_id = front_line_cp.id
|
|
return self.game.theater.find_control_point_by_id(offmap_transport_cp_id)
|
|
|
|
def should_generate_pretense_transports(
|
|
self, air_wing: AirWing
|
|
) -> tuple[bool, bool]:
|
|
"""
|
|
Returns a tuple of booleans, telling whether transport helicopter and aircraft
|
|
squadrons should be generated from the faction squadron definitions.
|
|
|
|
This helps to ensure that the faction has at least one transport helicopter and one cargo plane squadron.
|
|
|
|
(autogenerate_transport_helicopter_squadron, autogenerate_cargo_plane_squadron)
|
|
"""
|
|
autogenerate_transport_helicopter_squadron = True
|
|
autogenerate_cargo_plane_squadron = True
|
|
for aircraft_type in air_wing.squadrons:
|
|
for squadron in air_wing.squadrons[aircraft_type]:
|
|
if squadron.aircraft.helicopter and (
|
|
squadron.aircraft.capable_of(FlightType.TRANSPORT)
|
|
or squadron.aircraft.capable_of(FlightType.AIR_ASSAULT)
|
|
):
|
|
autogenerate_transport_helicopter_squadron = False
|
|
elif not squadron.aircraft.helicopter and (
|
|
squadron.aircraft.capable_of(FlightType.TRANSPORT)
|
|
or squadron.aircraft.capable_of(FlightType.AIR_ASSAULT)
|
|
):
|
|
autogenerate_cargo_plane_squadron = False
|
|
return (
|
|
autogenerate_transport_helicopter_squadron,
|
|
autogenerate_cargo_plane_squadron,
|
|
)
|
|
|
|
def generate_pretense_transport_squadron(
|
|
self,
|
|
cp: ControlPoint,
|
|
flight_type: FlightType,
|
|
fixed_wing: bool,
|
|
num_retries: int,
|
|
) -> None:
|
|
from game.squadrons import Squadron
|
|
|
|
"""
|
|
Generates a Pretense transport squadron from the faction squadron definitions. Use FlightType AIR_ASSAULT
|
|
for Pretense supply helicopters and TRANSPORT for off-map cargo plane squadrons.
|
|
|
|
Retribution does not differentiate between fixed wing and rotary wing transport squadron definitions, which
|
|
is why there is a retry mechanism in case the wrong type is returned. Use fixed_wing False
|
|
for Pretense supply helicopters and fixed_wing True for off-map cargo plane squadrons.
|
|
|
|
TODO: Find out if Pretense can handle rotary wing "cargo planes".
|
|
"""
|
|
|
|
squadron_def = cp.coalition.air_wing.squadron_def_generator.generate_for_task(
|
|
flight_type, cp
|
|
)
|
|
print(
|
|
f"Generating a squadron definition for fixed-wing {fixed_wing} squadron at {cp}"
|
|
)
|
|
for retries in range(num_retries):
|
|
if squadron_def is None or fixed_wing == squadron_def.aircraft.helicopter:
|
|
squadron_def = (
|
|
cp.coalition.air_wing.squadron_def_generator.generate_for_task(
|
|
flight_type, cp
|
|
)
|
|
)
|
|
|
|
# Failed, stop here
|
|
if squadron_def is None:
|
|
return
|
|
|
|
squadron = Squadron.create_from(
|
|
squadron_def,
|
|
flight_type,
|
|
2,
|
|
cp,
|
|
cp.coalition,
|
|
self.game,
|
|
)
|
|
if squadron.aircraft not in cp.coalition.air_wing.squadrons:
|
|
cp.coalition.air_wing.squadrons[squadron.aircraft] = list()
|
|
cp.coalition.air_wing.add_squadron(squadron)
|
|
return
|
|
|
|
def generate_pretense_aircraft(
|
|
self, cp: ControlPoint, ato: AirTaskingOrder
|
|
) -> None:
|
|
"""
|
|
Plans and generates AI aircraft groups/packages for Pretense.
|
|
|
|
Aircraft generation is done by walking the control points which will be made into
|
|
Pretense "zones" and spawning flights for different missions.
|
|
After the flight is generated the package is added to the ATO so the flights
|
|
can be configured.
|
|
|
|
Args:
|
|
cp: Control point to generate aircraft for.
|
|
ato: The ATO to generate aircraft for.
|
|
"""
|
|
num_of_sead = 0
|
|
num_of_cas = 0
|
|
num_of_strike = 0
|
|
num_of_cap = 0
|
|
|
|
for squadron in cp.squadrons:
|
|
# Intentionally don't spawn anything at OffMapSpawns in Pretense
|
|
if isinstance(squadron.location, OffMapSpawn):
|
|
continue
|
|
|
|
squadron.owned_aircraft += PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
squadron.untasked_aircraft += PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
package = Package(cp, squadron.flight_db, auto_asap=False)
|
|
mission_types = squadron.auto_assignable_mission_types
|
|
aircraft_per_flight = 1
|
|
if squadron.aircraft.helicopter and (
|
|
FlightType.TRANSPORT in mission_types
|
|
or FlightType.AIR_ASSAULT in mission_types
|
|
):
|
|
flight_type = FlightType.AIR_ASSAULT
|
|
elif not squadron.aircraft.helicopter and (
|
|
FlightType.TRANSPORT in mission_types
|
|
or FlightType.AIR_ASSAULT in mission_types
|
|
):
|
|
flight_type = FlightType.TRANSPORT
|
|
elif (
|
|
FlightType.SEAD in mission_types
|
|
or FlightType.SEAD_SWEEP in mission_types
|
|
or FlightType.SEAD_ESCORT in mission_types
|
|
) and num_of_sead < PRETENSE_SEAD_FLIGHTS_PER_CP:
|
|
flight_type = FlightType.SEAD
|
|
num_of_sead += 1
|
|
aircraft_per_flight = PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
elif (
|
|
FlightType.DEAD in mission_types
|
|
and num_of_sead < PRETENSE_SEAD_FLIGHTS_PER_CP
|
|
):
|
|
flight_type = FlightType.DEAD
|
|
num_of_sead += 1
|
|
aircraft_per_flight = PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
elif (
|
|
FlightType.CAS in mission_types or FlightType.BAI in mission_types
|
|
) and num_of_cas < PRETENSE_CAS_FLIGHTS_PER_CP:
|
|
flight_type = FlightType.CAS
|
|
num_of_cas += 1
|
|
aircraft_per_flight = PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
elif (
|
|
FlightType.STRIKE in mission_types
|
|
or FlightType.OCA_RUNWAY in mission_types
|
|
or FlightType.OCA_AIRCRAFT in mission_types
|
|
) and num_of_strike < PRETENSE_STRIKE_FLIGHTS_PER_CP:
|
|
flight_type = FlightType.STRIKE
|
|
num_of_strike += 1
|
|
aircraft_per_flight = PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
elif (
|
|
FlightType.BARCAP in mission_types
|
|
or FlightType.TARCAP in mission_types
|
|
or FlightType.ESCORT in mission_types
|
|
) and num_of_cap < PRETENSE_BARCAP_FLIGHTS_PER_CP:
|
|
flight_type = FlightType.BARCAP
|
|
num_of_cap += 1
|
|
aircraft_per_flight = PRETENSE_AI_AIRCRAFT_PER_FLIGHT
|
|
elif FlightType.AEWC in mission_types:
|
|
aircraft_per_flight = PRETENSE_AI_AWACS_PER_FLIGHT
|
|
elif FlightType.REFUELING in mission_types:
|
|
aircraft_per_flight = PRETENSE_AI_TANKERS_PER_FLIGHT
|
|
else:
|
|
flight_type = random.choice(list(mission_types))
|
|
|
|
print(
|
|
f"Generating flight of {aircraft_per_flight} for {flight_type}: {squadron.aircraft}"
|
|
)
|
|
if flight_type == FlightType.TRANSPORT:
|
|
flight = Flight(
|
|
package,
|
|
squadron,
|
|
aircraft_per_flight,
|
|
FlightType.PRETENSE_CARGO,
|
|
StartType.IN_FLIGHT,
|
|
divert=cp,
|
|
)
|
|
package.add_flight(flight)
|
|
flight.state = Navigating(flight, self.game.settings, waypoint_index=1)
|
|
else:
|
|
flight = Flight(
|
|
package,
|
|
squadron,
|
|
aircraft_per_flight,
|
|
flight_type,
|
|
StartType.COLD,
|
|
divert=cp,
|
|
)
|
|
package.add_flight(flight)
|
|
flight.state = WaitingForStart(
|
|
flight, self.game.settings, self.game.conditions.start_time
|
|
)
|
|
|
|
ato.add_package(package)
|
|
return
|
|
|
|
def initialize_pretense_data_structures(
|
|
self, cp: ControlPoint, flight: Flight
|
|
) -> None:
|
|
"""
|
|
Ensures that the data structures used to pass flight group information
|
|
to the Pretense init script lua are initialized for use in
|
|
PretenseFlightGroupSpawner and PretenseLuaGenerator.
|
|
|
|
Args:
|
|
cp: Control point to generate aircraft for.
|
|
flight: The current flight being generated.
|
|
"""
|
|
flight_type = flight.flight_type.name
|
|
cp_side = 2 if cp.captured else 1
|
|
cp_name_trimmed = "".join([i for i in cp.name.lower() if i.isalnum()])
|
|
|
|
if cp_name_trimmed not in flight.coalition.game.pretense_air[cp_side]:
|
|
flight.coalition.game.pretense_air[cp_side][cp_name_trimmed] = {}
|
|
if (
|
|
flight_type
|
|
not in flight.coalition.game.pretense_air[cp_side][cp_name_trimmed]
|
|
):
|
|
flight.coalition.game.pretense_air[cp_side][cp_name_trimmed][
|
|
flight_type
|
|
] = list()
|
|
return
|
|
|
|
def generate_flights(
|
|
self,
|
|
country: Country,
|
|
cp: ControlPoint,
|
|
ato: AirTaskingOrder,
|
|
dynamic_runways: Dict[str, RunwayData],
|
|
) -> None:
|
|
"""Adds aircraft to the mission for every flight in the ATO.
|
|
|
|
Args:
|
|
country: The country from the mission to use for this ATO.
|
|
cp: Control point to generate aircraft for.
|
|
ato: The ATO to generate aircraft for.
|
|
dynamic_runways: Runway data for carriers and FARPs.
|
|
"""
|
|
|
|
offmap_transport_cp = self.find_pretense_cargo_plane_cp(cp)
|
|
|
|
(
|
|
autogenerate_transport_helicopter_squadron,
|
|
autogenerate_cargo_plane_squadron,
|
|
) = self.should_generate_pretense_transports(cp.coalition.air_wing)
|
|
|
|
if autogenerate_transport_helicopter_squadron:
|
|
self.generate_pretense_transport_squadron(
|
|
offmap_transport_cp,
|
|
FlightType.AIR_ASSAULT,
|
|
False,
|
|
PRETENSE_SQUADRON_DEF_RETRIES,
|
|
)
|
|
if autogenerate_cargo_plane_squadron:
|
|
self.generate_pretense_transport_squadron(
|
|
offmap_transport_cp,
|
|
FlightType.TRANSPORT,
|
|
True,
|
|
PRETENSE_SQUADRON_DEF_RETRIES,
|
|
)
|
|
|
|
self.generate_pretense_aircraft(cp, ato)
|
|
|
|
self._reserve_frequencies_and_tacan(ato)
|
|
|
|
for package in reversed(sorted(ato.packages, key=lambda x: x.time_over_target)):
|
|
logging.info(f"Generating package for target: {package.target.name}")
|
|
if not package.flights:
|
|
continue
|
|
for flight in package.flights:
|
|
self.initialize_pretense_data_structures(cp, flight)
|
|
|
|
if flight.alive:
|
|
if not flight.squadron.location.runway_is_operational():
|
|
logging.warning(
|
|
f"Runway not operational, skipping flight: {flight.flight_type}"
|
|
)
|
|
flight.return_pilots_and_aircraft()
|
|
continue
|
|
logging.info(f"Generating flight: {flight.unit_type}")
|
|
group = self.create_and_configure_flight(
|
|
flight, country, dynamic_runways
|
|
)
|
|
self.unit_map.add_aircraft(group, flight)
|
|
|
|
def create_and_configure_flight(
|
|
self, flight: Flight, country: Country, dynamic_runways: Dict[str, RunwayData]
|
|
) -> FlyingGroup[Any]:
|
|
from game.pretense.pretenseflightgroupspawner import PretenseFlightGroupSpawner
|
|
|
|
"""Creates and configures the flight group in the mission."""
|
|
group = PretenseFlightGroupSpawner(
|
|
flight,
|
|
country,
|
|
self.mission,
|
|
self.helipads,
|
|
self.ground_spawns_roadbase,
|
|
self.ground_spawns,
|
|
self.mission_data,
|
|
).create_flight_group()
|
|
if flight.flight_type in [
|
|
FlightType.REFUELING,
|
|
FlightType.AEWC,
|
|
]:
|
|
self.flights.append(
|
|
FlightGroupConfigurator(
|
|
flight,
|
|
group,
|
|
self.game,
|
|
self.mission,
|
|
self.time,
|
|
self.radio_registry,
|
|
self.tacan_registy,
|
|
self.laser_code_registry,
|
|
self.mission_data,
|
|
dynamic_runways,
|
|
self.use_client,
|
|
).configure()
|
|
)
|
|
|
|
if self.ewrj:
|
|
self._track_ewrj_flight(flight, group)
|
|
|
|
return group
|
|
|
|
def _track_ewrj_flight(self, flight: Flight, group: FlyingGroup[Any]) -> None:
|
|
if not self.ewrj_package_dict.get(id(flight.package)):
|
|
self.ewrj_package_dict[id(flight.package)] = []
|
|
if (
|
|
flight.package.primary_flight
|
|
and flight is flight.package.primary_flight
|
|
or flight.client_count
|
|
and (
|
|
not self.need_ecm
|
|
or flight.loadout.has_weapon_of_type(WeaponType.JAMMER)
|
|
)
|
|
):
|
|
self.ewrj_package_dict[id(flight.package)].append(group)
|
|
|
|
def _reserve_frequencies_and_tacan(self, ato: AirTaskingOrder) -> None:
|
|
for package in ato.packages:
|
|
if package.frequency is None:
|
|
continue
|
|
if package.frequency not in self.radio_registry.allocated_channels:
|
|
self.radio_registry.reserve(package.frequency)
|
|
for f in package.flights:
|
|
if (
|
|
f.frequency
|
|
and f.frequency not in self.radio_registry.allocated_channels
|
|
):
|
|
self.radio_registry.reserve(f.frequency)
|
|
if f.tacan and f.tacan not in self.tacan_registy.allocated_channels:
|
|
self.tacan_registy.mark_unavailable(f.tacan)
|