From 8f30e60e1b0d7cb45a7279f2d24cbdc16bfdaf02 Mon Sep 17 00:00:00 2001 From: Dan Albert Date: Sun, 6 Dec 2020 00:59:04 -0800 Subject: [PATCH] Use unplanned missions to guide aircraft purchase. AI aircraft purchase decisions are now driven by the missions that the flight planner was unable to fulfill. This way we're buying the aircraft we actually need instead of buying them at random, in the locations we need them, in the order they're needed. There's a bit more that could be improved here: * Unused aircraft could be ferried to where they would be more useful. * Partial squadrons could be completed rather than buying a whole flight at a time. * Aircraft could be ranked by their usefulness so we're not buying so many Hueys when a Hornet would do better. * Purchase a buffer of CAP capable aircraft in case too many are shot down and they are not available next turn. https://github.com/Khopa/dcs_liberation/issues/361 --- game/game.py | 33 ++++++-- game/procurement.py | 113 ++++++++++++++++------------ gen/flights/ai_flight_planner.py | 63 +++------------- gen/flights/ai_flight_planner_db.py | 63 +++++++++++++++- 4 files changed, 161 insertions(+), 111 deletions(-) diff --git a/game/game.py b/game/game.py index bc8396db..9f1579ea 100644 --- a/game/game.py +++ b/game/game.py @@ -106,10 +106,20 @@ class Game: cp.pending_unit_deliveries = self.units_delivery_event(cp) self.sanitize_sides() - # Turn 0 procurement. - self.plan_procurement() + self.on_load() + # Turn 0 procurement. We don't actually have any missions to plan, but + # the planner will tell us what it would like to plan so we can use that + # to drive purchase decisions. + blue_planner = CoalitionMissionPlanner(self, is_player=True) + blue_planner.plan_missions() + + red_planner = CoalitionMissionPlanner(self, is_player=False) + red_planner.plan_missions() + + self.plan_procurement(blue_planner, red_planner) + def generate_conditions(self) -> Conditions: return Conditions.generate(self.theater, self.date, self.current_turn_time_of_day, self.settings) @@ -164,6 +174,7 @@ class Game: self.budget += self.budget_reward_amount def process_enemy_income(self): + # TODO: Clean up save compat. if not hasattr(self, "enemy_budget"): self.enemy_budget = 0 @@ -262,17 +273,23 @@ class Game: self.ground_planners = {} self.blue_ato.clear() self.red_ato.clear() - CoalitionMissionPlanner(self, is_player=True).plan_missions() - CoalitionMissionPlanner(self, is_player=False).plan_missions() + + blue_planner = CoalitionMissionPlanner(self, is_player=True) + blue_planner.plan_missions() + + red_planner = CoalitionMissionPlanner(self, is_player=False) + red_planner.plan_missions() + for cp in self.theater.controlpoints: if cp.has_frontline: gplanner = GroundPlanner(cp, self) gplanner.plan_groundwar() self.ground_planners[cp.id] = gplanner - self.plan_procurement() + self.plan_procurement(blue_planner, red_planner) - def plan_procurement(self) -> None: + def plan_procurement(self, blue_planner: CoalitionMissionPlanner, + red_planner: CoalitionMissionPlanner) -> None: self.budget = ProcurementAi( self, for_player=True, @@ -280,7 +297,7 @@ class Game: manage_runways=self.settings.automate_runway_repair, manage_front_line=self.settings.automate_front_line_reinforcements, manage_aircraft=self.settings.automate_aircraft_reinforcements - ).spend_budget(self.budget) + ).spend_budget(self.budget, blue_planner.procurement_requests) self.enemy_budget = ProcurementAi( self, @@ -289,7 +306,7 @@ class Game: manage_runways=True, manage_front_line=True, manage_aircraft=True - ).spend_budget(self.enemy_budget) + ).spend_budget(self.enemy_budget, red_planner.procurement_requests) def message(self, text: str) -> None: self.informations.append(Information(text, turn=self.turn)) diff --git a/game/procurement.py b/game/procurement.py index 814a63e1..6852c534 100644 --- a/game/procurement.py +++ b/game/procurement.py @@ -1,23 +1,33 @@ from __future__ import annotations +from dataclasses import dataclass import math import random -from typing import List, Optional, TYPE_CHECKING, Type +from typing import Iterator, List, Optional, TYPE_CHECKING, Type from dcs.task import CAP, CAS from dcs.unittype import FlyingType, UnitType, VehicleType from game import db from game.factions.faction import Faction -from game.theater import ControlPoint +from game.theater import ControlPoint, MissionTarget +from gen.flights.ai_flight_planner_db import ( + capable_aircraft_for_task, + preferred_aircraft_for_task, +) +from gen.flights.closestairfields import ObjectiveDistanceCache +from gen.flights.flight import FlightType if TYPE_CHECKING: from game import Game -class AircraftProcurer: - def __init__(self, faction: Faction) -> None: - self.faction = faction +@dataclass(frozen=True) +class AircraftProcurementRequest: + near: MissionTarget + range: int + task_capability: FlightType + number: int class ProcurementAi: @@ -31,7 +41,9 @@ class ProcurementAi: self.manage_front_line = manage_front_line self.manage_aircraft = manage_aircraft - def spend_budget(self, budget: int) -> int: + def spend_budget( + self, budget: int, + aircraft_requests: List[AircraftProcurementRequest]) -> int: if self.manage_runways: budget = self.repair_runways(budget) if self.manage_front_line: @@ -39,7 +51,7 @@ class ProcurementAi: budget -= armor_budget budget += self.reinforce_front_line(armor_budget) if self.manage_aircraft: - budget = self.purchase_aircraft(budget) + budget = self.purchase_aircraft(budget, aircraft_requests) return budget def repair_runways(self, budget: int) -> int: @@ -90,38 +102,52 @@ class ProcurementAi: return budget - def random_affordable_aircraft_group( - self, budget: int, size: int) -> Optional[Type[FlyingType]]: - unit_pool = [u for u in self.faction.aircrafts - if u in db.UNIT_BY_TASK[CAS] or u in db.UNIT_BY_TASK[CAP]] - - affordable_units = [u for u in unit_pool - if db.PRICES[u] * size <= budget] + def _affordable_aircraft_of_types( + self, types: List[Type[FlyingType]], airbase: ControlPoint, + number: int, max_price: int) -> Optional[Type[FlyingType]]: + unit_pool = [u for u in self.faction.aircrafts if u in types] + affordable_units = [ + u for u in unit_pool + if db.PRICES[u] * number <= max_price and airbase.can_operate(u) + ] if not affordable_units: return None return random.choice(affordable_units) - def purchase_aircraft(self, budget: int) -> int: + def affordable_aircraft_for( + self, request: AircraftProcurementRequest, + airbase: ControlPoint, budget: int) -> Optional[Type[FlyingType]]: + aircraft = self._affordable_aircraft_of_types( + preferred_aircraft_for_task(request.task_capability), + airbase, request.number, budget) + if aircraft is not None: + return aircraft + return self._affordable_aircraft_of_types( + capable_aircraft_for_task(request.task_capability), + airbase, request.number, budget) + + def purchase_aircraft( + self, budget: int, + aircraft_requests: List[AircraftProcurementRequest]) -> int: unit_pool = [u for u in self.faction.aircrafts if u in db.UNIT_BY_TASK[CAS] or u in db.UNIT_BY_TASK[CAP]] if not unit_pool: return budget - while budget > 0: - group_size = 2 - candidates = self.airbase_candidates(group_size) - if not candidates: - break + for request in aircraft_requests: + for airbase in self.best_airbases_for(request): + unit = self.affordable_aircraft_for(request, airbase, budget) + if unit is None: + # Can't afford any aircraft capable of performing the + # required mission that can operate from this airbase. We + # might be able to afford aircraft at other airbases though, + # in the case where the airbase we attempted to use is only + # able to operate expensive aircraft. + continue - cp = random.choice(candidates) - unit = self.random_affordable_aircraft_group(budget, group_size) - if unit is None: - # Can't afford any more aircraft. - break - - budget -= db.PRICES[unit] * group_size - assert cp.pending_unit_deliveries is not None - cp.pending_unit_deliveries.deliver({unit: group_size}) + budget -= db.PRICES[unit] * request.number + assert airbase.pending_unit_deliveries is not None + airbase.pending_unit_deliveries.deliver({unit: request.number}) return budget @@ -132,27 +158,20 @@ class ProcurementAi: else: return self.game.theater.enemy_points() - def airbase_candidates(self, group_size: int) -> List[ControlPoint]: - all_usable = [] - preferred = [] - for cp in self.owned_points: + def best_airbases_for( + self, + request: AircraftProcurementRequest) -> Iterator[ControlPoint]: + distance_cache = ObjectiveDistanceCache.get_closest_airfields( + request.near + ) + for cp in distance_cache.airfields_within(request.range): + if not cp.is_friendly(self.is_player): + continue if not cp.runway_is_operational(): continue - if cp.unclaimed_parking(self.game) < group_size: + if cp.unclaimed_parking(self.game) < request.number: continue - - all_usable.append(cp) - for connected in cp.connected_points: - # Prefer to buy aircraft at active front lines. - # TODO: Buy aircraft where they are needed, not at front lines. - if not connected.is_friendly(to_player=self.is_player): - preferred.append(cp) - - if preferred: - return preferred - - # Otherwise buy them anywhere valid. - return all_usable + yield cp def front_line_candidates(self) -> List[ControlPoint]: candidates = [] diff --git a/gen/flights/ai_flight_planner.py b/gen/flights/ai_flight_planner.py index 4dc0944e..5d6ff05f 100644 --- a/gen/flights/ai_flight_planner.py +++ b/gen/flights/ai_flight_planner.py @@ -21,6 +21,7 @@ from dcs.unittype import FlyingType from game import db from game.data.radar_db import UNITS_WITH_RADAR from game.infos.information import Information +from game.procurement import AircraftProcurementRequest from game.theater import ( Airfield, ControlPoint, @@ -50,7 +51,7 @@ from gen.flights.ai_flight_planner_db import ( SEAD_CAPABLE, SEAD_PREFERRED, STRIKE_CAPABLE, - STRIKE_PREFERRED, + STRIKE_PREFERRED, capable_aircraft_for_task, preferred_aircraft_for_task, ) from gen.flights.closestairfields import ( ClosestAirfields, @@ -142,63 +143,14 @@ class AircraftAllocator: responsible for returning them to the inventory. """ result = self.find_aircraft_of_type( - flight, self.preferred_aircraft_for_task(flight.task) + flight, preferred_aircraft_for_task(flight.task) ) if result is not None: return result return self.find_aircraft_of_type( - flight, self.capable_aircraft_for_task(flight.task) + flight, capable_aircraft_for_task(flight.task) ) - @staticmethod - def preferred_aircraft_for_task(task: FlightType) -> List[Type[FlyingType]]: - cap_missions = (FlightType.BARCAP, FlightType.TARCAP) - if task in cap_missions: - return CAP_PREFERRED - elif task == FlightType.ANTISHIP: - return ANTISHIP_PREFERRED - elif task == FlightType.BAI: - return CAS_CAPABLE - elif task == FlightType.CAS: - return CAS_PREFERRED - elif task in (FlightType.DEAD, FlightType.SEAD): - return SEAD_PREFERRED - elif task == FlightType.OCA_AIRCRAFT: - return CAS_PREFERRED - elif task == FlightType.OCA_RUNWAY: - return RUNWAY_ATTACK_PREFERRED - elif task == FlightType.STRIKE: - return STRIKE_PREFERRED - elif task == FlightType.ESCORT: - return CAP_PREFERRED - else: - return [] - - @staticmethod - def capable_aircraft_for_task(task: FlightType) -> List[Type[FlyingType]]: - cap_missions = (FlightType.BARCAP, FlightType.TARCAP) - if task in cap_missions: - return CAP_CAPABLE - elif task == FlightType.ANTISHIP: - return ANTISHIP_CAPABLE - elif task == FlightType.BAI: - return CAS_CAPABLE - elif task == FlightType.CAS: - return CAS_CAPABLE - elif task in (FlightType.DEAD, FlightType.SEAD): - return SEAD_CAPABLE - elif task == FlightType.OCA_AIRCRAFT: - return CAS_CAPABLE - elif task == FlightType.OCA_RUNWAY: - return RUNWAY_ATTACK_CAPABLE - elif task == FlightType.STRIKE: - return STRIKE_CAPABLE - elif task == FlightType.ESCORT: - return CAP_CAPABLE - else: - logging.error(f"Unplannable flight type: {task}") - return [] - def find_aircraft_of_type( self, flight: ProposedFlight, types: List[Type[FlyingType]], ) -> Optional[Tuple[ControlPoint, Type[FlyingType]]]: @@ -528,6 +480,7 @@ class CoalitionMissionPlanner: self.is_player = is_player self.objective_finder = ObjectiveFinder(self.game, self.is_player) self.ato = self.game.blue_ato if is_player else self.game.red_ato + self.procurement_requests: List[AircraftProcurementRequest] = [] def propose_missions(self) -> Iterator[ProposedMission]: """Identifies and iterates over potential mission in priority order.""" @@ -620,6 +573,12 @@ class CoalitionMissionPlanner: for proposed_flight in mission.flights: if not builder.plan_flight(proposed_flight): missing_types.add(proposed_flight.task) + self.procurement_requests.append(AircraftProcurementRequest( + near=mission.location, + range=proposed_flight.max_distance, + task_capability=proposed_flight.task, + number=proposed_flight.num_aircraft + )) if missing_types: missing_types_str = ", ".join( diff --git a/gen/flights/ai_flight_planner_db.py b/gen/flights/ai_flight_planner_db.py index ea3dcb6e..2b00f2fe 100644 --- a/gen/flights/ai_flight_planner_db.py +++ b/gen/flights/ai_flight_planner_db.py @@ -1,3 +1,6 @@ +import logging +from typing import List, Type + from dcs.helicopters import ( AH_1W, AH_64A, @@ -80,19 +83,22 @@ from dcs.planes import ( WingLoong_I, I_16 ) +from dcs.unittype import FlyingType + +from gen.flights.flight import FlightType -# Interceptor are the aircraft prioritized for interception tasks -# If none is available, the AI will use regular CAP-capable aircraft instead from pydcs_extensions.a4ec.a4ec import A_4E_C from pydcs_extensions.f22a.f22a import F_22A from pydcs_extensions.mb339.mb339 import MB_339PAN from pydcs_extensions.rafale.rafale import Rafale_A_S, Rafale_M, Rafale_B +from pydcs_extensions.su57.su57 import Su_57 # TODO: These lists really ought to be era (faction) dependent. # Factions which have F-5s, F-86s, and A-4s will should prefer F-5s for CAP, but # factions that also have F-4s should not. -from pydcs_extensions.su57.su57 import Su_57 +# Interceptor are the aircraft prioritized for interception tasks +# If none is available, the AI will use regular CAP-capable aircraft instead INTERCEPT_CAPABLE = [ MiG_21Bis, MiG_25PD, @@ -528,4 +534,53 @@ DRONES = [ MQ_9_Reaper, RQ_1A_Predator, WingLoong_I -] \ No newline at end of file +] + + +def preferred_aircraft_for_task(task: FlightType) -> List[Type[FlyingType]]: + cap_missions = (FlightType.BARCAP, FlightType.TARCAP) + if task in cap_missions: + return CAP_PREFERRED + elif task == FlightType.ANTISHIP: + return ANTISHIP_PREFERRED + elif task == FlightType.BAI: + return CAS_CAPABLE + elif task == FlightType.CAS: + return CAS_PREFERRED + elif task in (FlightType.DEAD, FlightType.SEAD): + return SEAD_PREFERRED + elif task == FlightType.OCA_AIRCRAFT: + return CAS_PREFERRED + elif task == FlightType.OCA_RUNWAY: + return RUNWAY_ATTACK_PREFERRED + elif task == FlightType.STRIKE: + return STRIKE_PREFERRED + elif task == FlightType.ESCORT: + return CAP_PREFERRED + else: + return [] + + +def capable_aircraft_for_task(task: FlightType) -> List[Type[FlyingType]]: + cap_missions = (FlightType.BARCAP, FlightType.TARCAP) + if task in cap_missions: + return CAP_CAPABLE + elif task == FlightType.ANTISHIP: + return ANTISHIP_CAPABLE + elif task == FlightType.BAI: + return CAS_CAPABLE + elif task == FlightType.CAS: + return CAS_CAPABLE + elif task in (FlightType.DEAD, FlightType.SEAD): + return SEAD_CAPABLE + elif task == FlightType.OCA_AIRCRAFT: + return CAS_CAPABLE + elif task == FlightType.OCA_RUNWAY: + return RUNWAY_ATTACK_CAPABLE + elif task == FlightType.STRIKE: + return STRIKE_CAPABLE + elif task == FlightType.ESCORT: + return CAP_CAPABLE + else: + logging.error(f"Unplannable flight type: {task}") + return []