From ccf6b6ef5f1369a967120500779cda9c656828dc Mon Sep 17 00:00:00 2001 From: Dan Albert Date: Tue, 13 Jul 2021 14:25:30 -0700 Subject: [PATCH] Check for available aircraft as task precondition. This makes it so that the mission planning effects are applied only if the package can be fulfilled. For example, breakthrough will be used only if all the BAI missions were fulfilled, not if they will *attempt* to be fulfilled. --- game/coalition.py | 16 +- game/commander/aircraftallocator.py | 76 +++ game/commander/missionscheduler.py | 76 +++ game/commander/packagebuilder.py | 98 ++++ game/commander/packagefulfiller.py | 213 ++++++++ game/commander/tasks/compound/capturebase.py | 2 +- game/commander/tasks/compound/defendbase.py | 4 +- .../tasks/compound/destroyenemygroundunits.py | 4 +- .../tasks/compound/protectairspace.py | 5 +- game/commander/tasks/frontlinestancetask.py | 12 +- game/commander/tasks/packageplanningtask.py | 40 +- game/commander/tasks/primitive/barcap.py | 39 +- game/commander/tasks/theatercommandertask.py | 8 +- game/commander/theatercommander.py | 9 +- game/commander/theaterstate.py | 60 ++- game/inventory.py | 24 +- gen/flights/ai_flight_planner.py | 489 ------------------ 17 files changed, 578 insertions(+), 597 deletions(-) create mode 100644 game/commander/aircraftallocator.py create mode 100644 game/commander/missionscheduler.py create mode 100644 game/commander/packagebuilder.py create mode 100644 game/commander/packagefulfiller.py delete mode 100644 gen/flights/ai_flight_planner.py diff --git a/game/coalition.py b/game/coalition.py index 06432c97..8b0ff812 100644 --- a/game/coalition.py +++ b/game/coalition.py @@ -6,12 +6,13 @@ from dcs import Point from faker import Faker from game.commander import TheaterCommander +from game.commander.missionscheduler import MissionScheduler from game.income import Income +from game.inventory import GlobalAircraftInventory from game.navmesh import NavMesh from game.profiling import logged_duration, MultiEventTracer from game.threatzones import ThreatZones from game.transfers import PendingTransfers -from gen.flights.ai_flight_planner import CoalitionMissionPlanner, MissionScheduler if TYPE_CHECKING: from game import Game @@ -84,6 +85,10 @@ class Coalition: assert self._navmesh is not None return self._navmesh + @property + def aircraft_inventory(self) -> GlobalAircraftInventory: + return self.game.aircraft_inventory + def __getstate__(self) -> dict[str, Any]: state = self.__dict__.copy() # Avoid persisting any volatile types that can be deterministically @@ -181,16 +186,9 @@ class Coalition: def plan_missions(self) -> None: color = "Blue" if self.player else "Red" with MultiEventTracer() as tracer: - mission_planner = CoalitionMissionPlanner( - self, - self.game.theater, - self.game.aircraft_inventory, - self.game.settings, - ) with tracer.trace(f"{color} mission planning"): with tracer.trace(f"{color} mission identification"): - commander = TheaterCommander(self.game, self.player) - commander.plan_missions(mission_planner, tracer) + TheaterCommander(self.game, self.player).plan_missions(tracer) with tracer.trace(f"{color} mission scheduling"): MissionScheduler( self, self.game.settings.desired_player_mission_duration diff --git a/game/commander/aircraftallocator.py b/game/commander/aircraftallocator.py new file mode 100644 index 00000000..16ea678a --- /dev/null +++ b/game/commander/aircraftallocator.py @@ -0,0 +1,76 @@ +from typing import Optional, Tuple + +from game.commander.missionproposals import ProposedFlight +from game.inventory import GlobalAircraftInventory +from game.squadrons import AirWing, Squadron +from game.theater import ControlPoint +from gen.flights.ai_flight_planner_db import aircraft_for_task +from gen.flights.closestairfields import ClosestAirfields +from gen.flights.flight import FlightType + + +class AircraftAllocator: + """Finds suitable aircraft for proposed missions.""" + + def __init__( + self, + air_wing: AirWing, + closest_airfields: ClosestAirfields, + global_inventory: GlobalAircraftInventory, + is_player: bool, + ) -> None: + self.air_wing = air_wing + self.closest_airfields = closest_airfields + self.global_inventory = global_inventory + self.is_player = is_player + + def find_squadron_for_flight( + self, flight: ProposedFlight + ) -> Optional[Tuple[ControlPoint, Squadron]]: + """Finds aircraft suitable for the given mission. + + Searches for aircraft capable of performing the given mission within the + maximum allowed range. If insufficient aircraft are available for the + mission, None is returned. + + Airfields are searched ordered nearest to farthest from the target and + searched twice. The first search looks for aircraft which prefer the + mission type, and the second search looks for any aircraft which are + capable of the mission type. For example, an F-14 from a nearby carrier + will be preferred for the CAP of an airfield that has only F-16s, but if + the carrier has only F/A-18s the F-16s will be used for CAP instead. + + Note that aircraft *will* be removed from the global inventory on + success. This is to ensure that the same aircraft are not matched twice + on subsequent calls. If the found aircraft are not used, the caller is + responsible for returning them to the inventory. + """ + return self.find_aircraft_for_task(flight, flight.task) + + def find_aircraft_for_task( + self, flight: ProposedFlight, task: FlightType + ) -> Optional[Tuple[ControlPoint, Squadron]]: + types = aircraft_for_task(task) + airfields_in_range = self.closest_airfields.operational_airfields_within( + flight.max_distance + ) + + for airfield in airfields_in_range: + if not airfield.is_friendly(self.is_player): + continue + inventory = self.global_inventory.for_control_point(airfield) + for aircraft in types: + if not airfield.can_operate(aircraft): + continue + if inventory.available(aircraft) < flight.num_aircraft: + continue + # Valid location with enough aircraft available. Find a squadron to fit + # the role. + squadrons = self.air_wing.auto_assignable_for_task_with_type( + aircraft, task + ) + for squadron in squadrons: + if squadron.can_provide_pilots(flight.num_aircraft): + inventory.remove_aircraft(aircraft, flight.num_aircraft) + return airfield, squadron + return None diff --git a/game/commander/missionscheduler.py b/game/commander/missionscheduler.py new file mode 100644 index 00000000..26889a97 --- /dev/null +++ b/game/commander/missionscheduler.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import logging +import random +from collections import defaultdict +from datetime import timedelta +from typing import Iterator, Dict, TYPE_CHECKING + +from game.theater import MissionTarget +from gen.flights.flight import FlightType +from gen.flights.traveltime import TotEstimator + +if TYPE_CHECKING: + from game.coalition import Coalition + + +class MissionScheduler: + def __init__(self, coalition: Coalition, desired_mission_length: timedelta) -> None: + self.coalition = coalition + self.desired_mission_length = desired_mission_length + + def schedule_missions(self) -> None: + """Identifies and plans mission for the turn.""" + + def start_time_generator( + count: int, earliest: int, latest: int, margin: int + ) -> Iterator[timedelta]: + interval = (latest - earliest) // count + for time in range(earliest, latest, interval): + error = random.randint(-margin, margin) + yield timedelta(seconds=max(0, time + error)) + + dca_types = { + FlightType.BARCAP, + FlightType.TARCAP, + } + + previous_cap_end_time: Dict[MissionTarget, timedelta] = defaultdict(timedelta) + non_dca_packages = [ + p for p in self.coalition.ato.packages if p.primary_task not in dca_types + ] + + start_time = start_time_generator( + count=len(non_dca_packages), + earliest=5 * 60, + latest=int(self.desired_mission_length.total_seconds()), + margin=5 * 60, + ) + for package in self.coalition.ato.packages: + tot = TotEstimator(package).earliest_tot() + if package.primary_task in dca_types: + previous_end_time = previous_cap_end_time[package.target] + if tot > previous_end_time: + # Can't get there exactly on time, so get there ASAP. This + # will typically only happen for the first CAP at each + # target. + package.time_over_target = tot + else: + package.time_over_target = previous_end_time + + departure_time = package.mission_departure_time + # Should be impossible for CAPs + if departure_time is None: + logging.error(f"Could not determine mission end time for {package}") + continue + previous_cap_end_time[package.target] = departure_time + elif package.auto_asap: + package.set_tot_asap() + else: + # But other packages should be spread out a bit. Note that take + # times are delayed, but all aircraft will become active at + # mission start. This makes it more worthwhile to attack enemy + # airfields to hit grounded aircraft, since they're more likely + # to be present. Runway and air started aircraft will be + # delayed until their takeoff time by AirConflictGenerator. + package.time_over_target = next(start_time) + tot diff --git a/game/commander/packagebuilder.py b/game/commander/packagebuilder.py new file mode 100644 index 00000000..490e0286 --- /dev/null +++ b/game/commander/packagebuilder.py @@ -0,0 +1,98 @@ +from typing import Optional + +from game.commander.missionproposals import ProposedFlight +from game.dcs.aircrafttype import AircraftType +from game.inventory import GlobalAircraftInventory +from game.squadrons import AirWing +from game.theater import MissionTarget, OffMapSpawn, ControlPoint +from game.utils import nautical_miles +from gen import Package +from game.commander.aircraftallocator import AircraftAllocator +from gen.flights.closestairfields import ClosestAirfields +from gen.flights.flight import Flight + + +class PackageBuilder: + """Builds a Package for the flights it receives.""" + + def __init__( + self, + location: MissionTarget, + closest_airfields: ClosestAirfields, + global_inventory: GlobalAircraftInventory, + air_wing: AirWing, + is_player: bool, + package_country: str, + start_type: str, + asap: bool, + ) -> None: + self.closest_airfields = closest_airfields + self.is_player = is_player + self.package_country = package_country + self.package = Package(location, auto_asap=asap) + self.allocator = AircraftAllocator( + air_wing, closest_airfields, global_inventory, is_player + ) + self.global_inventory = global_inventory + self.start_type = start_type + + def plan_flight(self, plan: ProposedFlight) -> bool: + """Allocates aircraft for the given flight and adds them to the package. + + If no suitable aircraft are available, False is returned. If the failed + flight was critical and the rest of the mission will be scrubbed, the + caller should return any previously planned flights to the inventory + using release_planned_aircraft. + """ + assignment = self.allocator.find_squadron_for_flight(plan) + if assignment is None: + return False + airfield, squadron = assignment + if isinstance(airfield, OffMapSpawn): + start_type = "In Flight" + else: + start_type = self.start_type + + flight = Flight( + self.package, + self.package_country, + squadron, + plan.num_aircraft, + plan.task, + start_type, + departure=airfield, + arrival=airfield, + divert=self.find_divert_field(squadron.aircraft, airfield), + ) + self.package.add_flight(flight) + return True + + def find_divert_field( + self, aircraft: AircraftType, arrival: ControlPoint + ) -> Optional[ControlPoint]: + divert_limit = nautical_miles(150) + for airfield in self.closest_airfields.operational_airfields_within( + divert_limit + ): + if airfield.captured != self.is_player: + continue + if airfield == arrival: + continue + if not airfield.can_operate(aircraft): + continue + if isinstance(airfield, OffMapSpawn): + continue + return airfield + return None + + def build(self) -> Package: + """Returns the built package.""" + return self.package + + def release_planned_aircraft(self) -> None: + """Returns any planned flights to the inventory.""" + flights = list(self.package.flights) + for flight in flights: + self.global_inventory.return_from_flight(flight) + flight.clear_roster() + self.package.remove_flight(flight) diff --git a/game/commander/packagefulfiller.py b/game/commander/packagefulfiller.py new file mode 100644 index 00000000..be903548 --- /dev/null +++ b/game/commander/packagefulfiller.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +import logging +from collections import defaultdict +from typing import Set, Iterable, Dict, TYPE_CHECKING, Optional + +from game.commander.missionproposals import ProposedMission, ProposedFlight, EscortType +from game.data.doctrine import Doctrine +from game.inventory import GlobalAircraftInventory +from game.procurement import AircraftProcurementRequest +from game.profiling import MultiEventTracer +from game.settings import Settings +from game.squadrons import AirWing +from game.theater import ConflictTheater +from game.threatzones import ThreatZones +from gen import AirTaskingOrder, Package +from game.commander.packagebuilder import PackageBuilder +from gen.flights.closestairfields import ObjectiveDistanceCache +from gen.flights.flight import FlightType +from gen.flights.flightplan import FlightPlanBuilder + +if TYPE_CHECKING: + from game.coalition import Coalition + + +class PackageFulfiller: + """Responsible for package aircraft allocation and flight plan layout.""" + + def __init__( + self, + coalition: Coalition, + theater: ConflictTheater, + aircraft_inventory: GlobalAircraftInventory, + settings: Settings, + ) -> None: + self.coalition = coalition + self.theater = theater + self.aircraft_inventory = aircraft_inventory + self.player_missions_asap = settings.auto_ato_player_missions_asap + self.default_start_type = settings.default_start_type + + @property + def is_player(self) -> bool: + return self.coalition.player + + @property + def ato(self) -> AirTaskingOrder: + return self.coalition.ato + + @property + def air_wing(self) -> AirWing: + return self.coalition.air_wing + + @property + def doctrine(self) -> Doctrine: + return self.coalition.doctrine + + @property + def threat_zones(self) -> ThreatZones: + return self.coalition.opponent.threat_zone + + def add_procurement_request(self, request: AircraftProcurementRequest) -> None: + self.coalition.procurement_requests.append(request) + + def air_wing_can_plan(self, mission_type: FlightType) -> bool: + """Returns True if it is possible for the air wing to plan this mission type. + + Not all mission types can be fulfilled by all air wings. Many factions do not + have AEW&C aircraft, so they will never be able to plan those missions. It's + also possible for the player to exclude mission types from their squadron + designs. + """ + return self.air_wing.can_auto_plan(mission_type) + + def plan_flight( + self, + mission: ProposedMission, + flight: ProposedFlight, + builder: PackageBuilder, + missing_types: Set[FlightType], + ) -> None: + if not builder.plan_flight(flight): + missing_types.add(flight.task) + purchase_order = AircraftProcurementRequest( + near=mission.location, + range=flight.max_distance, + task_capability=flight.task, + number=flight.num_aircraft, + ) + # Reserves are planned for critical missions, so prioritize those orders + # over aircraft needed for non-critical missions. + self.add_procurement_request(purchase_order) + + def scrub_mission_missing_aircraft( + self, + mission: ProposedMission, + builder: PackageBuilder, + missing_types: Set[FlightType], + not_attempted: Iterable[ProposedFlight], + ) -> None: + # Try to plan the rest of the mission just so we can count the missing + # types to buy. + for flight in not_attempted: + self.plan_flight(mission, flight, builder, missing_types) + + missing_types_str = ", ".join(sorted([t.name for t in missing_types])) + builder.release_planned_aircraft() + logging.debug( + f"Not enough aircraft in range for {mission.location.name} " + f"capable of: {missing_types_str}" + ) + + def check_needed_escorts(self, builder: PackageBuilder) -> Dict[EscortType, bool]: + threats = defaultdict(bool) + for flight in builder.package.flights: + if self.threat_zones.waypoints_threatened_by_aircraft( + flight.flight_plan.escorted_waypoints() + ): + threats[EscortType.AirToAir] = True + if self.threat_zones.waypoints_threatened_by_radar_sam( + list(flight.flight_plan.escorted_waypoints()) + ): + threats[EscortType.Sead] = True + return threats + + def plan_mission( + self, mission: ProposedMission, tracer: MultiEventTracer + ) -> Optional[Package]: + """Allocates aircraft for a proposed mission and adds it to the ATO.""" + builder = PackageBuilder( + mission.location, + ObjectiveDistanceCache.get_closest_airfields(mission.location), + self.aircraft_inventory, + self.air_wing, + self.is_player, + self.coalition.country_name, + self.default_start_type, + mission.asap, + ) + + # Attempt to plan all the main elements of the mission first. Escorts + # will be planned separately so we can prune escorts for packages that + # are not expected to encounter that type of threat. + missing_types: Set[FlightType] = set() + escorts = [] + for proposed_flight in mission.flights: + if not self.air_wing_can_plan(proposed_flight.task): + # This air wing can never plan this mission type because they do not + # have compatible aircraft or squadrons. Skip fulfillment so that we + # don't place the purchase request. + continue + if proposed_flight.escort_type is not None: + # Escorts are planned after the primary elements of the package. + # If the package does not need escorts they may be pruned. + escorts.append(proposed_flight) + continue + with tracer.trace("Flight planning"): + self.plan_flight(mission, proposed_flight, builder, missing_types) + + if missing_types: + self.scrub_mission_missing_aircraft( + mission, builder, missing_types, escorts + ) + return None + + if not builder.package.flights: + # The non-escort part of this mission is unplannable by this faction. Scrub + # the mission and do not attempt planning escorts because there's no reason + # to buy them because this mission will never be planned. + return None + + # Create flight plans for the main flights of the package so we can + # determine threats. This is done *after* creating all of the flights + # rather than as each flight is added because the flight plan for + # flights that will rendezvous with their package will be affected by + # the other flights in the package. Escorts will not be able to + # contribute to this. + flight_plan_builder = FlightPlanBuilder( + builder.package, self.coalition, self.theater + ) + for flight in builder.package.flights: + with tracer.trace("Flight plan population"): + flight_plan_builder.populate_flight_plan(flight) + + needed_escorts = self.check_needed_escorts(builder) + for escort in escorts: + # This list was generated from the not None set, so this should be + # impossible. + assert escort.escort_type is not None + if needed_escorts[escort.escort_type]: + with tracer.trace("Flight planning"): + self.plan_flight(mission, escort, builder, missing_types) + + # Check again for unavailable aircraft. If the escort was required and + # none were found, scrub the mission. + if missing_types: + self.scrub_mission_missing_aircraft( + mission, builder, missing_types, escorts + ) + return None + + package = builder.build() + # Add flight plans for escorts. + for flight in package.flights: + if not flight.flight_plan.waypoints: + with tracer.trace("Flight plan population"): + flight_plan_builder.populate_flight_plan(flight) + + if package.has_players and self.player_missions_asap: + package.auto_asap = True + package.set_tot_asap() + + return package diff --git a/game/commander/tasks/compound/capturebase.py b/game/commander/tasks/compound/capturebase.py index 747b7599..86d2b86e 100644 --- a/game/commander/tasks/compound/capturebase.py +++ b/game/commander/tasks/compound/capturebase.py @@ -15,5 +15,5 @@ class CaptureBase(CompoundTask[TheaterState]): front_line: FrontLine def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]: - yield [BreakthroughAttack(self.front_line, state.player)] + yield [BreakthroughAttack(self.front_line, state.context.coalition.player)] yield [DestroyEnemyGroundUnits(self.front_line)] diff --git a/game/commander/tasks/compound/defendbase.py b/game/commander/tasks/compound/defendbase.py index 69a008e5..e7071489 100644 --- a/game/commander/tasks/compound/defendbase.py +++ b/game/commander/tasks/compound/defendbase.py @@ -14,6 +14,6 @@ class DefendBase(CompoundTask[TheaterState]): front_line: FrontLine def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]: - yield [DefensiveStance(self.front_line, state.player)] - yield [RetreatStance(self.front_line, state.player)] + yield [DefensiveStance(self.front_line, state.context.coalition.player)] + yield [RetreatStance(self.front_line, state.context.coalition.player)] yield [PlanCas(self.front_line)] diff --git a/game/commander/tasks/compound/destroyenemygroundunits.py b/game/commander/tasks/compound/destroyenemygroundunits.py index cf83213f..327acecd 100644 --- a/game/commander/tasks/compound/destroyenemygroundunits.py +++ b/game/commander/tasks/compound/destroyenemygroundunits.py @@ -14,6 +14,6 @@ class DestroyEnemyGroundUnits(CompoundTask[TheaterState]): front_line: FrontLine def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]: - yield [EliminationAttack(self.front_line, state.player)] - yield [AggressiveAttack(self.front_line, state.player)] + yield [EliminationAttack(self.front_line, state.context.coalition.player)] + yield [AggressiveAttack(self.front_line, state.context.coalition.player)] yield [PlanCas(self.front_line)] diff --git a/game/commander/tasks/compound/protectairspace.py b/game/commander/tasks/compound/protectairspace.py index 5a13e486..67407010 100644 --- a/game/commander/tasks/compound/protectairspace.py +++ b/game/commander/tasks/compound/protectairspace.py @@ -7,5 +7,6 @@ from game.htn import CompoundTask, Method class ProtectAirSpace(CompoundTask[TheaterState]): def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]: - for cp in state.vulnerable_control_points: - yield [PlanBarcap(cp, state.barcap_rounds)] + for cp, needed in state.barcaps_needed.items(): + if needed > 0: + yield [PlanBarcap(cp)] diff --git a/game/commander/tasks/frontlinestancetask.py b/game/commander/tasks/frontlinestancetask.py index ae27f1a6..f8c3b8d1 100644 --- a/game/commander/tasks/frontlinestancetask.py +++ b/game/commander/tasks/frontlinestancetask.py @@ -6,12 +6,11 @@ from typing import TYPE_CHECKING from game.commander.tasks.theatercommandertask import TheaterCommanderTask from game.commander.theaterstate import TheaterState -from game.profiling import MultiEventTracer from game.theater import FrontLine from gen.ground_forces.combat_stance import CombatStance if TYPE_CHECKING: - from gen.flights.ai_flight_planner import CoalitionMissionPlanner + from game.coalition import Coalition class FrontLineStanceTask(TheaterCommanderTask, ABC): @@ -27,7 +26,10 @@ class FrontLineStanceTask(TheaterCommanderTask, ABC): @staticmethod def management_allowed(state: TheaterState) -> bool: - return not state.player or state.stance_automation_enabled + return ( + not state.context.coalition.player + or state.context.settings.automate_front_line_stance + ) def better_stance_already_set(self, state: TheaterState) -> bool: current_stance = state.front_line_stances[self.front_line] @@ -69,7 +71,5 @@ class FrontLineStanceTask(TheaterCommanderTask, ABC): def apply_effects(self, state: TheaterState) -> None: state.front_line_stances[self.front_line] = self.stance - def execute( - self, mission_planner: CoalitionMissionPlanner, tracer: MultiEventTracer - ) -> None: + def execute(self, coalition: Coalition) -> None: self.friendly_cp.stances[self.enemy_cp.id] = self.stance diff --git a/game/commander/tasks/packageplanningtask.py b/game/commander/tasks/packageplanningtask.py index 7196a0b9..fb50af23 100644 --- a/game/commander/tasks/packageplanningtask.py +++ b/game/commander/tasks/packageplanningtask.py @@ -8,18 +8,19 @@ from enum import unique, IntEnum, auto from typing import TYPE_CHECKING, Optional, Generic, TypeVar, Iterator, Union from game.commander.missionproposals import ProposedFlight, EscortType, ProposedMission +from game.commander.packagefulfiller import PackageFulfiller from game.commander.tasks.theatercommandertask import TheaterCommanderTask from game.commander.theaterstate import TheaterState from game.data.doctrine import Doctrine -from game.profiling import MultiEventTracer +from game.settings import AutoAtoBehavior from game.theater import MissionTarget from game.theater.theatergroundobject import IadsGroundObject, NavalGroundObject from game.utils import Distance, meters +from gen import Package from gen.flights.flight import FlightType if TYPE_CHECKING: - from gen.flights.ai_flight_planner import CoalitionMissionPlanner - + from game.coalition import Coalition MissionTargetT = TypeVar("MissionTargetT", bound=MissionTarget) @@ -36,18 +37,26 @@ class RangeType(IntEnum): class PackagePlanningTask(TheaterCommanderTask, Generic[MissionTargetT]): target: MissionTargetT flights: list[ProposedFlight] = field(init=False) + package: Optional[Package] = field(init=False, default=None) def __post_init__(self) -> None: self.flights = [] + self.package = Package(self.target) def preconditions_met(self, state: TheaterState) -> bool: - return not state.player or state.ato_automation_enabled + if ( + state.context.coalition.player + and state.context.settings.auto_ato_behavior is AutoAtoBehavior.Disabled + ): + return False + return self.fulfill_mission(state) - def execute( - self, mission_planner: CoalitionMissionPlanner, tracer: MultiEventTracer - ) -> None: - self.propose_flights(mission_planner.doctrine) - mission_planner.plan_mission(ProposedMission(self.target, self.flights), tracer) + def execute(self, coalition: Coalition) -> None: + if self.package is None: + raise RuntimeError("Attempted to execute failed package planning task") + for flight in self.package.flights: + coalition.aircraft_inventory.claim_for_flight(flight) + coalition.ato.add_package(self.package) @abstractmethod def propose_flights(self, doctrine: Doctrine) -> None: @@ -70,6 +79,19 @@ class PackagePlanningTask(TheaterCommanderTask, Generic[MissionTargetT]): def asap(self) -> bool: return False + def fulfill_mission(self, state: TheaterState) -> bool: + self.propose_flights(state.context.coalition.doctrine) + fulfiller = PackageFulfiller( + state.context.coalition, + state.context.theater, + state.available_aircraft, + state.context.settings, + ) + self.package = fulfiller.plan_mission( + ProposedMission(self.target, self.flights), state.context.tracer + ) + return self.package is not None + def propose_common_escorts(self, doctrine: Doctrine) -> None: self.propose_flight( FlightType.SEAD_ESCORT, diff --git a/game/commander/tasks/primitive/barcap.py b/game/commander/tasks/primitive/barcap.py index 8fd86fab..8d3cf456 100644 --- a/game/commander/tasks/primitive/barcap.py +++ b/game/commander/tasks/primitive/barcap.py @@ -1,46 +1,23 @@ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING -from game.commander.missionproposals import ProposedMission, ProposedFlight -from game.commander.tasks.theatercommandertask import TheaterCommanderTask +from game.commander.tasks.packageplanningtask import PackagePlanningTask from game.commander.theaterstate import TheaterState -from game.profiling import MultiEventTracer +from game.data.doctrine import Doctrine from game.theater import ControlPoint from gen.flights.flight import FlightType -if TYPE_CHECKING: - from gen.flights.ai_flight_planner import CoalitionMissionPlanner - @dataclass -class PlanBarcap(TheaterCommanderTask): - target: ControlPoint - rounds: int - +class PlanBarcap(PackagePlanningTask[ControlPoint]): def preconditions_met(self, state: TheaterState) -> bool: - if state.player and not state.ato_automation_enabled: + if not super().preconditions_met(state): return False - return self.target in state.vulnerable_control_points + return state.barcaps_needed[self.target] > 0 def apply_effects(self, state: TheaterState) -> None: - state.vulnerable_control_points.remove(self.target) + state.barcaps_needed[self.target] -= 1 - def execute( - self, mission_planner: CoalitionMissionPlanner, tracer: MultiEventTracer - ) -> None: - for _ in range(self.rounds): - mission_planner.plan_mission( - ProposedMission( - self.target, - [ - ProposedFlight( - FlightType.BARCAP, - 2, - mission_planner.doctrine.mission_ranges.cap, - ), - ], - ), - tracer, - ) + def propose_flights(self, doctrine: Doctrine) -> None: + self.propose_flight(FlightType.BARCAP, 2, doctrine.mission_ranges.cap) diff --git a/game/commander/tasks/theatercommandertask.py b/game/commander/tasks/theatercommandertask.py index aefaeea5..5daa6b6c 100644 --- a/game/commander/tasks/theatercommandertask.py +++ b/game/commander/tasks/theatercommandertask.py @@ -5,16 +5,12 @@ from typing import TYPE_CHECKING from game.commander.theaterstate import TheaterState from game.htn import PrimitiveTask -from game.profiling import MultiEventTracer if TYPE_CHECKING: - from gen.flights.ai_flight_planner import CoalitionMissionPlanner + from game.coalition import Coalition -# TODO: Refactor so that we don't need to call up to the mission planner. class TheaterCommanderTask(PrimitiveTask[TheaterState]): @abstractmethod - def execute( - self, mission_planner: CoalitionMissionPlanner, tracer: MultiEventTracer - ) -> None: + def execute(self, coalition: Coalition) -> None: ... diff --git a/game/commander/theatercommander.py b/game/commander/theatercommander.py index 1e085410..3066ff54 100644 --- a/game/commander/theatercommander.py +++ b/game/commander/theatercommander.py @@ -64,7 +64,6 @@ from game.profiling import MultiEventTracer if TYPE_CHECKING: from game import Game - from gen.flights.ai_flight_planner import CoalitionMissionPlanner class TheaterCommander(Planner[TheaterState, TheaterCommanderTask]): @@ -77,15 +76,13 @@ class TheaterCommander(Planner[TheaterState, TheaterCommanderTask]): self.game = game self.player = player - def plan_missions( - self, mission_planner: CoalitionMissionPlanner, tracer: MultiEventTracer - ) -> None: - state = TheaterState.from_game(self.game, self.player) + def plan_missions(self, tracer: MultiEventTracer) -> None: + state = TheaterState.from_game(self.game, self.player, tracer) while True: result = self.plan(state) if result is None: # Planned all viable tasks this turn. return for task in result.tasks: - task.execute(mission_planner, tracer) + task.execute(self.game.coalition_for(self.player)) state = result.end_state diff --git a/game/commander/theaterstate.py b/game/commander/theaterstate.py index 6da5f9f2..cf41f9a5 100644 --- a/game/commander/theaterstate.py +++ b/game/commander/theaterstate.py @@ -8,10 +8,11 @@ from typing import TYPE_CHECKING, Any, Union, Optional from game.commander.garrisons import Garrisons from game.commander.objectivefinder import ObjectiveFinder -from game.data.doctrine import Doctrine from game.htn import WorldState -from game.settings import AutoAtoBehavior -from game.theater import ControlPoint, FrontLine, MissionTarget +from game.inventory import GlobalAircraftInventory +from game.profiling import MultiEventTracer +from game.settings import Settings +from game.theater import ControlPoint, FrontLine, MissionTarget, ConflictTheater from game.theater.theatergroundobject import ( TheaterGroundObject, NavalGroundObject, @@ -23,16 +24,22 @@ from gen.ground_forces.combat_stance import CombatStance if TYPE_CHECKING: from game import Game + from game.coalition import Coalition from game.transfers import Convoy, CargoShip +@dataclass(frozen=True) +class PersistentContext: + coalition: Coalition + theater: ConflictTheater + settings: Settings + tracer: MultiEventTracer + + @dataclass class TheaterState(WorldState["TheaterState"]): - player: bool - stance_automation_enabled: bool - ato_automation_enabled: bool - barcap_rounds: int - vulnerable_control_points: list[ControlPoint] + context: PersistentContext + barcaps_needed: dict[ControlPoint, int] active_front_lines: list[FrontLine] front_line_stances: dict[FrontLine, Optional[CombatStance]] vulnerable_front_lines: list[FrontLine] @@ -49,12 +56,12 @@ class TheaterState(WorldState["TheaterState"]): strike_targets: list[TheaterGroundObject[Any]] enemy_barcaps: list[ControlPoint] threat_zones: ThreatZones - opposing_doctrine: Doctrine + available_aircraft: GlobalAircraftInventory def _rebuild_threat_zones(self) -> None: """Recreates the theater's threat zones based on the current planned state.""" self.threat_zones = ThreatZones.for_threats( - self.opposing_doctrine, + self.context.coalition.opponent.doctrine, barcap_locations=self.enemy_barcaps, air_defenses=itertools.chain(self.enemy_air_defenses, self.enemy_ships), ) @@ -85,11 +92,8 @@ class TheaterState(WorldState["TheaterState"]): # Do not use copy.deepcopy. Copying every TGO, control point, etc is absurdly # expensive. return TheaterState( - player=self.player, - stance_automation_enabled=self.stance_automation_enabled, - ato_automation_enabled=self.ato_automation_enabled, - barcap_rounds=self.barcap_rounds, - vulnerable_control_points=list(self.vulnerable_control_points), + context=self.context, + barcaps_needed=dict(self.barcaps_needed), active_front_lines=list(self.active_front_lines), front_line_stances=dict(self.front_line_stances), vulnerable_front_lines=list(self.vulnerable_front_lines), @@ -106,7 +110,7 @@ class TheaterState(WorldState["TheaterState"]): strike_targets=list(self.strike_targets), enemy_barcaps=list(self.enemy_barcaps), threat_zones=self.threat_zones, - opposing_doctrine=self.opposing_doctrine, + available_aircraft=self.available_aircraft.clone(), # Persistent properties are not copied. These are a way for failed subtasks # to communicate requirements to other tasks. For example, the task to # attack enemy garrisons might fail because the target area has IADS @@ -118,26 +122,26 @@ class TheaterState(WorldState["TheaterState"]): ) @classmethod - def from_game(cls, game: Game, player: bool) -> TheaterState: + def from_game( + cls, game: Game, player: bool, tracer: MultiEventTracer + ) -> TheaterState: + coalition = game.coalition_for(player) finder = ObjectiveFinder(game, player) - auto_stance = game.settings.automate_front_line_stance - auto_ato = game.settings.auto_ato_behavior is not AutoAtoBehavior.Disabled ordered_capturable_points = finder.prioritized_unisolated_points() + context = PersistentContext(coalition, game.theater, game.settings, tracer) + # Plan enough rounds of CAP that the target has coverage over the expected # mission duration. mission_duration = game.settings.desired_player_mission_duration.total_seconds() - barcap_duration = game.coalition_for( - player - ).doctrine.cap_duration.total_seconds() + barcap_duration = coalition.doctrine.cap_duration.total_seconds() barcap_rounds = math.ceil(mission_duration / barcap_duration) return TheaterState( - player=player, - stance_automation_enabled=auto_stance, - ato_automation_enabled=auto_ato, - barcap_rounds=barcap_rounds, - vulnerable_control_points=list(finder.vulnerable_control_points()), + context=context, + barcaps_needed={ + cp: barcap_rounds for cp in finder.vulnerable_control_points() + }, active_front_lines=list(finder.front_lines()), front_line_stances={f: None for f in finder.front_lines()}, vulnerable_front_lines=list(finder.front_lines()), @@ -156,5 +160,5 @@ class TheaterState(WorldState["TheaterState"]): strike_targets=list(finder.strike_targets()), enemy_barcaps=list(game.theater.control_points_for(not player)), threat_zones=game.threat_zone_for(not player), - opposing_doctrine=game.faction_for(not player).doctrine, + available_aircraft=game.aircraft_inventory.clone(), ) diff --git a/game/inventory.py b/game/inventory.py index 77587cb2..f7f0dbe1 100644 --- a/game/inventory.py +++ b/game/inventory.py @@ -1,8 +1,8 @@ """Inventory management APIs.""" from __future__ import annotations -from collections import defaultdict -from typing import Dict, Iterable, Iterator, Set, Tuple, TYPE_CHECKING +from collections import defaultdict, Iterator, Iterable +from typing import TYPE_CHECKING from game.dcs.aircrafttype import AircraftType from gen.flights.flight import Flight @@ -16,7 +16,12 @@ class ControlPointAircraftInventory: def __init__(self, control_point: ControlPoint) -> None: self.control_point = control_point - self.inventory: Dict[AircraftType, int] = defaultdict(int) + self.inventory: dict[AircraftType, int] = defaultdict(int) + + def clone(self) -> ControlPointAircraftInventory: + new = ControlPointAircraftInventory(self.control_point) + new.inventory = self.inventory.copy() + return new def add_aircraft(self, aircraft: AircraftType, count: int) -> None: """Adds aircraft to the inventory. @@ -65,7 +70,7 @@ class ControlPointAircraftInventory: yield aircraft @property - def all_aircraft(self) -> Iterator[Tuple[AircraftType, int]]: + def all_aircraft(self) -> Iterator[tuple[AircraftType, int]]: """Iterates over all available aircraft types, including amounts.""" for aircraft, count in self.inventory.items(): if count > 0: @@ -80,10 +85,17 @@ class GlobalAircraftInventory: """Game-wide aircraft inventory.""" def __init__(self, control_points: Iterable[ControlPoint]) -> None: - self.inventories: Dict[ControlPoint, ControlPointAircraftInventory] = { + self.inventories: dict[ControlPoint, ControlPointAircraftInventory] = { cp: ControlPointAircraftInventory(cp) for cp in control_points } + def clone(self) -> GlobalAircraftInventory: + new = GlobalAircraftInventory([]) + new.inventories = { + cp: inventory.clone() for cp, inventory in self.inventories.items() + } + return new + def reset(self, for_player: bool) -> None: """Clears the inventory of every control point owned by the given coalition.""" for inventory in self.inventories.values(): @@ -109,7 +121,7 @@ class GlobalAircraftInventory: @property def available_types_for_player(self) -> Iterator[AircraftType]: """Iterates over all aircraft types available to the player.""" - seen: Set[AircraftType] = set() + seen: set[AircraftType] = set() for control_point, inventory in self.inventories.items(): if control_point.captured: for aircraft in inventory.types_available: diff --git a/gen/flights/ai_flight_planner.py b/gen/flights/ai_flight_planner.py deleted file mode 100644 index e74ce0af..00000000 --- a/gen/flights/ai_flight_planner.py +++ /dev/null @@ -1,489 +0,0 @@ -from __future__ import annotations - -import logging -import random -from collections import defaultdict -from datetime import timedelta -from typing import ( - Dict, - Iterable, - Iterator, - Optional, - Set, - TYPE_CHECKING, - Tuple, -) - -from game.commander.missionproposals import ProposedFlight, ProposedMission, EscortType -from game.data.doctrine import Doctrine -from game.dcs.aircrafttype import AircraftType -from game.procurement import AircraftProcurementRequest -from game.profiling import MultiEventTracer -from game.settings import Settings -from game.squadrons import AirWing, Squadron -from game.theater import ( - ControlPoint, - MissionTarget, - OffMapSpawn, - ConflictTheater, -) -from game.threatzones import ThreatZones -from game.utils import nautical_miles -from gen.ato import Package, AirTaskingOrder -from gen.flights.ai_flight_planner_db import aircraft_for_task -from gen.flights.closestairfields import ( - ClosestAirfields, - ObjectiveDistanceCache, -) -from gen.flights.flight import ( - Flight, - FlightType, -) -from gen.flights.flightplan import FlightPlanBuilder -from gen.flights.traveltime import TotEstimator - -# Avoid importing some types that cause circular imports unless type checking. -if TYPE_CHECKING: - from game.coalition import Coalition - from game.inventory import GlobalAircraftInventory - - -class AircraftAllocator: - """Finds suitable aircraft for proposed missions.""" - - def __init__( - self, - air_wing: AirWing, - closest_airfields: ClosestAirfields, - global_inventory: GlobalAircraftInventory, - is_player: bool, - ) -> None: - self.air_wing = air_wing - self.closest_airfields = closest_airfields - self.global_inventory = global_inventory - self.is_player = is_player - - def find_squadron_for_flight( - self, flight: ProposedFlight - ) -> Optional[Tuple[ControlPoint, Squadron]]: - """Finds aircraft suitable for the given mission. - - Searches for aircraft capable of performing the given mission within the - maximum allowed range. If insufficient aircraft are available for the - mission, None is returned. - - Airfields are searched ordered nearest to farthest from the target and - searched twice. The first search looks for aircraft which prefer the - mission type, and the second search looks for any aircraft which are - capable of the mission type. For example, an F-14 from a nearby carrier - will be preferred for the CAP of an airfield that has only F-16s, but if - the carrier has only F/A-18s the F-16s will be used for CAP instead. - - Note that aircraft *will* be removed from the global inventory on - success. This is to ensure that the same aircraft are not matched twice - on subsequent calls. If the found aircraft are not used, the caller is - responsible for returning them to the inventory. - """ - return self.find_aircraft_for_task(flight, flight.task) - - def find_aircraft_for_task( - self, flight: ProposedFlight, task: FlightType - ) -> Optional[Tuple[ControlPoint, Squadron]]: - types = aircraft_for_task(task) - airfields_in_range = self.closest_airfields.operational_airfields_within( - flight.max_distance - ) - - for airfield in airfields_in_range: - if not airfield.is_friendly(self.is_player): - continue - inventory = self.global_inventory.for_control_point(airfield) - for aircraft in types: - if not airfield.can_operate(aircraft): - continue - if inventory.available(aircraft) < flight.num_aircraft: - continue - # Valid location with enough aircraft available. Find a squadron to fit - # the role. - squadrons = self.air_wing.auto_assignable_for_task_with_type( - aircraft, task - ) - for squadron in squadrons: - if squadron.can_provide_pilots(flight.num_aircraft): - inventory.remove_aircraft(aircraft, flight.num_aircraft) - return airfield, squadron - return None - - -class PackageBuilder: - """Builds a Package for the flights it receives.""" - - def __init__( - self, - location: MissionTarget, - closest_airfields: ClosestAirfields, - global_inventory: GlobalAircraftInventory, - air_wing: AirWing, - is_player: bool, - package_country: str, - start_type: str, - asap: bool, - ) -> None: - self.closest_airfields = closest_airfields - self.is_player = is_player - self.package_country = package_country - self.package = Package(location, auto_asap=asap) - self.allocator = AircraftAllocator( - air_wing, closest_airfields, global_inventory, is_player - ) - self.global_inventory = global_inventory - self.start_type = start_type - - def plan_flight(self, plan: ProposedFlight) -> bool: - """Allocates aircraft for the given flight and adds them to the package. - - If no suitable aircraft are available, False is returned. If the failed - flight was critical and the rest of the mission will be scrubbed, the - caller should return any previously planned flights to the inventory - using release_planned_aircraft. - """ - assignment = self.allocator.find_squadron_for_flight(plan) - if assignment is None: - return False - airfield, squadron = assignment - if isinstance(airfield, OffMapSpawn): - start_type = "In Flight" - else: - start_type = self.start_type - - flight = Flight( - self.package, - self.package_country, - squadron, - plan.num_aircraft, - plan.task, - start_type, - departure=airfield, - arrival=airfield, - divert=self.find_divert_field(squadron.aircraft, airfield), - ) - self.package.add_flight(flight) - return True - - def find_divert_field( - self, aircraft: AircraftType, arrival: ControlPoint - ) -> Optional[ControlPoint]: - divert_limit = nautical_miles(150) - for airfield in self.closest_airfields.operational_airfields_within( - divert_limit - ): - if airfield.captured != self.is_player: - continue - if airfield == arrival: - continue - if not airfield.can_operate(aircraft): - continue - if isinstance(airfield, OffMapSpawn): - continue - return airfield - return None - - def build(self) -> Package: - """Returns the built package.""" - return self.package - - def release_planned_aircraft(self) -> None: - """Returns any planned flights to the inventory.""" - flights = list(self.package.flights) - for flight in flights: - self.global_inventory.return_from_flight(flight) - flight.clear_roster() - self.package.remove_flight(flight) - - -class MissionScheduler: - def __init__(self, coalition: Coalition, desired_mission_length: timedelta) -> None: - self.coalition = coalition - self.desired_mission_length = desired_mission_length - - def schedule_missions(self) -> None: - """Identifies and plans mission for the turn.""" - - def start_time_generator( - count: int, earliest: int, latest: int, margin: int - ) -> Iterator[timedelta]: - interval = (latest - earliest) // count - for time in range(earliest, latest, interval): - error = random.randint(-margin, margin) - yield timedelta(seconds=max(0, time + error)) - - dca_types = { - FlightType.BARCAP, - FlightType.TARCAP, - } - - previous_cap_end_time: Dict[MissionTarget, timedelta] = defaultdict(timedelta) - non_dca_packages = [ - p for p in self.coalition.ato.packages if p.primary_task not in dca_types - ] - - start_time = start_time_generator( - count=len(non_dca_packages), - earliest=5 * 60, - latest=int(self.desired_mission_length.total_seconds()), - margin=5 * 60, - ) - for package in self.coalition.ato.packages: - tot = TotEstimator(package).earliest_tot() - if package.primary_task in dca_types: - previous_end_time = previous_cap_end_time[package.target] - if tot > previous_end_time: - # Can't get there exactly on time, so get there ASAP. This - # will typically only happen for the first CAP at each - # target. - package.time_over_target = tot - else: - package.time_over_target = previous_end_time - - departure_time = package.mission_departure_time - # Should be impossible for CAPs - if departure_time is None: - logging.error(f"Could not determine mission end time for {package}") - continue - previous_cap_end_time[package.target] = departure_time - elif package.auto_asap: - package.set_tot_asap() - else: - # But other packages should be spread out a bit. Note that take - # times are delayed, but all aircraft will become active at - # mission start. This makes it more worthwhile to attack enemy - # airfields to hit grounded aircraft, since they're more likely - # to be present. Runway and air started aircraft will be - # delayed until their takeoff time by AirConflictGenerator. - package.time_over_target = next(start_time) + tot - - -class CoalitionMissionPlanner: - """Coalition flight planning AI. - - This class is responsible for automatically planning missions for the - coalition at the start of the turn. - - The primary goal of the mission planner is to protect existing friendly - assets. Missions will be planned with the following priorities: - - 1. CAP for airfields/fleets in close proximity to the enemy to prevent heavy - losses of friendly aircraft. - 2. CAP for front line areas to protect ground and CAS units. - 3. DEAD to reduce necessity of SEAD for future missions. - 4. CAS to protect friendly ground units. - 5. Strike missions to reduce the enemy's resources. - - TODO: Anti-ship and airfield strikes to reduce enemy sortie rates. - TODO: BAI to prevent enemy forces from reaching the front line. - TODO: Should fleets always have a CAP? - - TODO: Stance and doctrine-specific planning behavior. - """ - - def __init__( - self, - coalition: Coalition, - theater: ConflictTheater, - aircraft_inventory: GlobalAircraftInventory, - settings: Settings, - ) -> None: - self.coalition = coalition - self.theater = theater - self.aircraft_inventory = aircraft_inventory - self.player_missions_asap = settings.auto_ato_player_missions_asap - self.default_start_type = settings.default_start_type - - @property - def is_player(self) -> bool: - return self.coalition.player - - @property - def ato(self) -> AirTaskingOrder: - return self.coalition.ato - - @property - def air_wing(self) -> AirWing: - return self.coalition.air_wing - - @property - def doctrine(self) -> Doctrine: - return self.coalition.doctrine - - @property - def threat_zones(self) -> ThreatZones: - return self.coalition.opponent.threat_zone - - def add_procurement_request( - self, request: AircraftProcurementRequest, priority: bool - ) -> None: - if priority: - self.coalition.procurement_requests.insert(0, request) - else: - self.coalition.procurement_requests.append(request) - - def air_wing_can_plan(self, mission_type: FlightType) -> bool: - """Returns True if it is possible for the air wing to plan this mission type. - - Not all mission types can be fulfilled by all air wings. Many factions do not - have AEW&C aircraft, so they will never be able to plan those missions. It's - also possible for the player to exclude mission types from their squadron - designs. - """ - return self.air_wing.can_auto_plan(mission_type) - - def plan_flight( - self, - mission: ProposedMission, - flight: ProposedFlight, - builder: PackageBuilder, - missing_types: Set[FlightType], - for_reserves: bool, - ) -> None: - if not builder.plan_flight(flight): - missing_types.add(flight.task) - purchase_order = AircraftProcurementRequest( - near=mission.location, - range=flight.max_distance, - task_capability=flight.task, - number=flight.num_aircraft, - ) - # Reserves are planned for critical missions, so prioritize those orders - # over aircraft needed for non-critical missions. - self.add_procurement_request(purchase_order, priority=for_reserves) - - def scrub_mission_missing_aircraft( - self, - mission: ProposedMission, - builder: PackageBuilder, - missing_types: Set[FlightType], - not_attempted: Iterable[ProposedFlight], - reserves: bool, - ) -> None: - # Try to plan the rest of the mission just so we can count the missing - # types to buy. - for flight in not_attempted: - self.plan_flight(mission, flight, builder, missing_types, reserves) - - missing_types_str = ", ".join(sorted([t.name for t in missing_types])) - builder.release_planned_aircraft() - desc = "reserve aircraft" if reserves else "aircraft" - logging.debug( - f"Not enough {desc} in range for {mission.location.name} " - f"capable of: {missing_types_str}" - ) - - def check_needed_escorts(self, builder: PackageBuilder) -> Dict[EscortType, bool]: - threats = defaultdict(bool) - for flight in builder.package.flights: - if self.threat_zones.waypoints_threatened_by_aircraft( - flight.flight_plan.escorted_waypoints() - ): - threats[EscortType.AirToAir] = True - if self.threat_zones.waypoints_threatened_by_radar_sam( - list(flight.flight_plan.escorted_waypoints()) - ): - threats[EscortType.Sead] = True - return threats - - def plan_mission( - self, mission: ProposedMission, tracer: MultiEventTracer, reserves: bool = False - ) -> None: - """Allocates aircraft for a proposed mission and adds it to the ATO.""" - builder = PackageBuilder( - mission.location, - ObjectiveDistanceCache.get_closest_airfields(mission.location), - self.aircraft_inventory, - self.air_wing, - self.is_player, - self.coalition.country_name, - self.default_start_type, - mission.asap, - ) - - # Attempt to plan all the main elements of the mission first. Escorts - # will be planned separately so we can prune escorts for packages that - # are not expected to encounter that type of threat. - missing_types: Set[FlightType] = set() - escorts = [] - for proposed_flight in mission.flights: - if not self.air_wing_can_plan(proposed_flight.task): - # This air wing can never plan this mission type because they do not - # have compatible aircraft or squadrons. Skip fulfillment so that we - # don't place the purchase request. - continue - if proposed_flight.escort_type is not None: - # Escorts are planned after the primary elements of the package. - # If the package does not need escorts they may be pruned. - escorts.append(proposed_flight) - continue - with tracer.trace("Flight planning"): - self.plan_flight( - mission, proposed_flight, builder, missing_types, reserves - ) - - if missing_types: - self.scrub_mission_missing_aircraft( - mission, builder, missing_types, escorts, reserves - ) - return - - if not builder.package.flights: - # The non-escort part of this mission is unplannable by this faction. Scrub - # the mission and do not attempt planning escorts because there's no reason - # to buy them because this mission will never be planned. - return - - # Create flight plans for the main flights of the package so we can - # determine threats. This is done *after* creating all of the flights - # rather than as each flight is added because the flight plan for - # flights that will rendezvous with their package will be affected by - # the other flights in the package. Escorts will not be able to - # contribute to this. - flight_plan_builder = FlightPlanBuilder( - builder.package, self.coalition, self.theater - ) - for flight in builder.package.flights: - with tracer.trace("Flight plan population"): - flight_plan_builder.populate_flight_plan(flight) - - needed_escorts = self.check_needed_escorts(builder) - for escort in escorts: - # This list was generated from the not None set, so this should be - # impossible. - assert escort.escort_type is not None - if needed_escorts[escort.escort_type]: - with tracer.trace("Flight planning"): - self.plan_flight(mission, escort, builder, missing_types, reserves) - - # Check again for unavailable aircraft. If the escort was required and - # none were found, scrub the mission. - if missing_types: - self.scrub_mission_missing_aircraft( - mission, builder, missing_types, escorts, reserves - ) - return - - if reserves: - # Mission is planned reserves which will not be used this turn. - # Return reserves to the inventory. - builder.release_planned_aircraft() - return - - package = builder.build() - # Add flight plans for escorts. - for flight in package.flights: - if not flight.flight_plan.waypoints: - with tracer.trace("Flight plan population"): - flight_plan_builder.populate_flight_plan(flight) - - if package.has_players and self.player_missions_asap: - package.auto_asap = True - package.set_tot_asap() - - self.ato.add_package(package)