Improve DEAD mission prioritization.

This alters the DEAD task planning to be the *least* preferred task, but
prevents other tasks from being planned unless they are excepted to be
clear of air defenses first. Even so, missions are a guaranteed success
so those other missions will still get SEAD escorts if there's potential
for a SAM in the area.

This means that air defenses that are not protecting a more useful
target (like a convoy, armor column, building, etc) will no longer be
considered by the mission planner. This isn't *quite* right since we
currently only check the target area for air defenses rather than the
entire flight plan, so there's a chance that we ignore IADS that have
threatened ingress points (though that's mostly solved by the flight
plan layout).

This also is still slightly limited because it's not checking for
aircraft availability at this stage yet, so we may aggressively plan
missions that we should be skipping unless we can guarantee that the
DEAD mission was planned. However, that's not new behavior.

changelog.md

@@ -4,6 +4,8 @@ Saves from 3.x are not compatible with 5.0.
 
 ## Features/Improvements
 
+* **[Campaign AI]** Overhauled campaign AI target prioritization. This currently only affects the ordering of DEAD missions.
+
 ## Fixes
 
 # 4.1.0

game/commander/objectivefinder.py

@@ -43,53 +43,15 @@ class ObjectiveFinder:
         self.game = game
         self.is_player = is_player
 
-    def enemy_air_defenses(self) -> Iterator[tuple[IadsGroundObject, Distance]]:
+    def enemy_air_defenses(self) -> Iterator[IadsGroundObject]:
         """Iterates over all enemy SAM sites."""
-        doctrine = self.game.faction_for(self.is_player).doctrine
-        threat_zones = self.game.threat_zone_for(not self.is_player)
         for cp in self.enemy_control_points():
             for ground_object in cp.ground_objects:
                 if ground_object.is_dead:
                     continue
 
-                if isinstance(ground_object, EwrGroundObject):
-                    if threat_zones.threatened_by_air_defense(ground_object):
-                        # This is a very weak heuristic for determining whether the EWR
-                        # is close enough to be worth targeting before a SAM that is
-                        # covering it. Ingress distance corresponds to the beginning of
-                        # the attack range and is sufficient for most standoff weapons,
-                        # so treating the ingress distance as the threat distance sorts
-                        # these EWRs such that they will be attacked before SAMs that do
-                        # not threaten the ingress point, but after those that do.
-                        target_range = doctrine.ingress_egress_distance
-                    else:
-                        # But if the EWR isn't covered then we should only be worrying
-                        # about its detection range.
-                        target_range = ground_object.max_detection_range()
-                elif isinstance(ground_object, SamGroundObject):
-                    target_range = ground_object.max_threat_range()
-                else:
-                    continue
-
-                yield ground_object, target_range
-
-    def threatening_air_defenses(self) -> Iterator[IadsGroundObject]:
-        """Iterates over enemy SAMs in threat range of friendly control points.
-
-        SAM sites are sorted by their closest proximity to any friendly control
-        point (airfield or fleet).
-        """
-
-        target_ranges: list[tuple[IadsGroundObject, Distance]] = []
-        for target, threat_range in self.enemy_air_defenses():
-            ranges: list[Distance] = []
-            for cp in self.friendly_control_points():
-                ranges.append(meters(target.distance_to(cp)) - threat_range)
-            target_ranges.append((target, min(ranges)))
-
-        target_ranges = sorted(target_ranges, key=operator.itemgetter(1))
-        for target, _range in target_ranges:
-            yield target
+                if isinstance(ground_object, IadsGroundObject):
+                    yield ground_object
 
     def enemy_vehicle_groups(self) -> Iterator[VehicleGroupGroundObject]:
         """Iterates over all enemy vehicle groups."""

game/commander/tasks/compound/degradeiads.py

@@ -1,11 +1,16 @@
 from collections import Iterator
 
+from game.commander.tasks.primitive.antiship import PlanAntiShip
 from game.commander.tasks.primitive.dead import PlanDead
 from game.commander.theaterstate import TheaterState
 from game.htn import CompoundTask, Method
+from game.theater.theatergroundobject import IadsGroundObject
 
 
 class DegradeIads(CompoundTask[TheaterState]):
     def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]:
         for air_defense in state.threatening_air_defenses:
-            yield [PlanDead(air_defense)]
+            if isinstance(air_defense, IadsGroundObject):
+                yield [PlanDead(air_defense)]
+            else:
+                yield [PlanAntiShip(air_defense)]

game/commander/tasks/compound/destroyships.py

@@ -1,11 +0,0 @@
-from collections import Iterator
-
-from game.commander.tasks.primitive.antiship import PlanAntiShip
-from game.commander.theaterstate import TheaterState
-from game.htn import CompoundTask, Method
-
-
-class DestroyShips(CompoundTask[TheaterState]):
-    def each_valid_method(self, state: TheaterState) -> Iterator[Method[TheaterState]]:
-        for ship in state.threatening_ships:
-            yield [PlanAntiShip(ship)]

game/commander/tasks/compound/nextaction.py

@@ -8,7 +8,6 @@ from game.commander.tasks.compound.attackairinfrastructure import (
 from game.commander.tasks.compound.attackbuildings import AttackBuildings
 from game.commander.tasks.compound.attackgarrisons import AttackGarrisons
 from game.commander.tasks.compound.degradeiads import DegradeIads
-from game.commander.tasks.compound.destroyships import DestroyShips
 from game.commander.tasks.compound.frontlinedefense import FrontLineDefense
 from game.commander.tasks.compound.interdictreinforcements import (
     InterdictReinforcements,
@@ -28,9 +27,8 @@ class PlanNextAction(CompoundTask[TheaterState]):
         yield [PlanRefuelingSupport()]
         yield [ProtectAirSpace()]
         yield [FrontLineDefense()]
-        yield [DegradeIads()]
         yield [InterdictReinforcements()]
-        yield [DestroyShips()]
         yield [AttackGarrisons()]
         yield [AttackAirInfrastructure(self.aircraft_cold_start)]
         yield [AttackBuildings()]
+        yield [DegradeIads()]

game/commander/tasks/packageplanningtask.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
+import itertools
+import operator
 from abc import abstractmethod
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Optional, Generic, TypeVar
+from typing import TYPE_CHECKING, Optional, Generic, TypeVar, Iterator, Union
 
 from game.commander.missionproposals import ProposedFlight, EscortType, ProposedMission
+from game.commander.tasks.theatercommandertask import TheaterCommanderTask
 from game.commander.theaterstate import TheaterState
 from game.data.doctrine import Doctrine
-from game.htn import PrimitiveTask
 from game.profiling import MultiEventTracer
 from game.theater import MissionTarget
-from game.utils import Distance
+from game.theater.theatergroundobject import IadsGroundObject, NavalGroundObject
+from game.utils import Distance, meters
 from gen.flights.flight import FlightType
 
 if TYPE_CHECKING:
@@ -23,7 +26,7 @@ MissionTargetT = TypeVar("MissionTargetT", bound=MissionTarget)
 # TODO: Refactor so that we don't need to call up to the mission planner.
 # Bypass type checker due to https://github.com/python/mypy/issues/5374
 @dataclass  # type: ignore
-class PackagePlanningTask(PrimitiveTask[TheaterState], Generic[MissionTargetT]):
+class PackagePlanningTask(TheaterCommanderTask, Generic[MissionTargetT]):
     target: MissionTargetT
     flights: list[ProposedFlight] = field(init=False)
 
@@ -71,3 +74,44 @@ class PackagePlanningTask(PrimitiveTask[TheaterState], Generic[MissionTargetT]):
             doctrine.mission_ranges.offensive,
             EscortType.AirToAir,
         )
+
+    def iter_iads_threats(
+        self, state: TheaterState
+    ) -> Iterator[Union[IadsGroundObject, NavalGroundObject]]:
+        target_ranges: list[
+            tuple[Union[IadsGroundObject, NavalGroundObject], Distance]
+        ] = []
+        all_iads: Iterator[
+            Union[IadsGroundObject, NavalGroundObject]
+        ] = itertools.chain(state.enemy_air_defenses, state.enemy_ships)
+        for target in all_iads:
+            distance = meters(target.distance_to(self.target))
+            threat_range = target.max_threat_range()
+            if not threat_range:
+                continue
+            # IADS out of range of our target area will have a positive
+            # distance_to_threat and should be pruned. The rest have a decreasing
+            # distance_to_threat as overlap increases. The most negative distance has
+            # the greatest coverage of the target and should be treated as the highest
+            # priority threat.
+            distance_to_threat = distance - threat_range
+            if distance_to_threat > meters(0):
+                continue
+            target_ranges.append((target, distance_to_threat))
+
+        # TODO: Prioritize IADS by vulnerability?
+        target_ranges = sorted(target_ranges, key=operator.itemgetter(1))
+        for target, _range in target_ranges:
+            yield target
+
+    def target_area_preconditions_met(
+        self, state: TheaterState, ignore_iads: bool = False
+    ) -> bool:
+        """Checks if the target area has been cleared of threats."""
+        threatened = False
+        if not ignore_iads:
+            for iads_threat in self.iter_iads_threats(state):
+                threatened = True
+                if iads_threat not in state.threatening_air_defenses:
+                    state.threatening_air_defenses.append(iads_threat)
+        return not threatened

game/commander/tasks/primitive/antiship.py

@@ -13,10 +13,12 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanAntiShip(PackagePlanningTask[NavalGroundObject]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.threatening_ships
+        if self.target not in state.threatening_air_defenses:
+            return False
+        return self.target_area_preconditions_met(state, ignore_iads=True)
 
     def apply_effects(self, state: TheaterState) -> None:
-        state.threatening_ships.remove(self.target)
+        state.eliminate_ship(self.target)
 
     def propose_flights(self, doctrine: Doctrine) -> None:
         self.propose_flight(FlightType.ANTISHIP, 2, doctrine.mission_ranges.offensive)

game/commander/tasks/primitive/antishipping.py

@@ -12,7 +12,9 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanAntiShipping(PackagePlanningTask[CargoShip]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.enemy_shipping
+        if self.target not in state.enemy_shipping:
+            return False
+        return self.target_area_preconditions_met(state)
 
     def apply_effects(self, state: TheaterState) -> None:
         state.enemy_shipping.remove(self.target)

game/commander/tasks/primitive/bai.py

@@ -12,7 +12,9 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanBai(PackagePlanningTask[VehicleGroupGroundObject]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.enemy_garrisons
+        if self.target not in state.enemy_garrisons:
+            return False
+        return self.target_area_preconditions_met(state)
 
     def apply_effects(self, state: TheaterState) -> None:
         state.enemy_garrisons.remove(self.target)

game/commander/tasks/primitive/convoyinterdiction.py

@@ -12,7 +12,9 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanConvoyInterdiction(PackagePlanningTask[Convoy]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.enemy_convoys
+        if self.target not in state.enemy_convoys:
+            return False
+        return self.target_area_preconditions_met(state)
 
     def apply_effects(self, state: TheaterState) -> None:
         state.enemy_convoys.remove(self.target)

game/commander/tasks/primitive/dead.py

@@ -13,10 +13,12 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanDead(PackagePlanningTask[IadsGroundObject]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.threatening_air_defenses
+        if self.target not in state.threatening_air_defenses:
+            return False
+        return self.target_area_preconditions_met(state, ignore_iads=True)
 
     def apply_effects(self, state: TheaterState) -> None:
-        state.threatening_air_defenses.remove(self.target)
+        state.eliminate_air_defense(self.target)
 
     def propose_flights(self, doctrine: Doctrine) -> None:
         self.propose_flight(FlightType.DEAD, 2, doctrine.mission_ranges.offensive)

game/commander/tasks/primitive/oca.py

@@ -14,7 +14,9 @@ class PlanOcaStrike(PackagePlanningTask[ControlPoint]):
     aircraft_cold_start: bool
 
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.oca_targets
+        if self.target not in state.oca_targets:
+            return False
+        return self.target_area_preconditions_met(state)
 
     def apply_effects(self, state: TheaterState) -> None:
         state.oca_targets.remove(self.target)

game/commander/tasks/primitive/strike.py

@@ -13,7 +13,9 @@ from gen.flights.flight import FlightType
 @dataclass
 class PlanStrike(PackagePlanningTask[TheaterGroundObject[Any]]):
     def preconditions_met(self, state: TheaterState) -> bool:
-        return self.target in state.strike_targets
+        if self.target not in state.strike_targets:
+            return False
+        return self.target_area_preconditions_met(state)
 
     def apply_effects(self, state: TheaterState) -> None:
         state.strike_targets.remove(self.target)

game/commander/theaterstate.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import itertools
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Union
 
 from game.commander.objectivefinder import ObjectiveFinder
+from game.data.doctrine import Doctrine
 from game.htn import WorldState
 from game.theater import ControlPoint, FrontLine, MissionTarget
 from game.theater.theatergroundobject import (
@@ -12,6 +14,7 @@ from game.theater.theatergroundobject import (
     NavalGroundObject,
     IadsGroundObject,
 )
+from game.threatzones import ThreatZones
 from game.transfers import Convoy, CargoShip
 
 if TYPE_CHECKING:
@@ -24,13 +27,35 @@ class TheaterState(WorldState["TheaterState"]):
     vulnerable_front_lines: list[FrontLine]
     aewc_targets: list[MissionTarget]
     refueling_targets: list[MissionTarget]
-    threatening_air_defenses: list[IadsGroundObject]
+    enemy_air_defenses: list[IadsGroundObject]
+    threatening_air_defenses: list[Union[IadsGroundObject, NavalGroundObject]]
     enemy_convoys: list[Convoy]
     enemy_shipping: list[CargoShip]
-    threatening_ships: list[NavalGroundObject]
+    enemy_ships: list[NavalGroundObject]
     enemy_garrisons: list[VehicleGroupGroundObject]
     oca_targets: list[ControlPoint]
     strike_targets: list[TheaterGroundObject[Any]]
+    enemy_barcaps: list[ControlPoint]
+    threat_zones: ThreatZones
+    opposing_doctrine: Doctrine
+
+    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,
+            barcap_locations=self.enemy_barcaps,
+            air_defenses=itertools.chain(self.enemy_air_defenses, self.enemy_ships),
+        )
+
+    def eliminate_air_defense(self, target: IadsGroundObject) -> None:
+        self.threatening_air_defenses.remove(target)
+        self.enemy_air_defenses.remove(target)
+        self._rebuild_threat_zones()
+
+    def eliminate_ship(self, target: NavalGroundObject) -> None:
+        self.threatening_air_defenses.remove(target)
+        self.enemy_ships.remove(target)
+        self._rebuild_threat_zones()
 
     def clone(self) -> TheaterState:
         # Do not use copy.deepcopy. Copying every TGO, control point, etc is absurdly
@@ -40,13 +65,23 @@ class TheaterState(WorldState["TheaterState"]):
             vulnerable_front_lines=list(self.vulnerable_front_lines),
             aewc_targets=list(self.aewc_targets),
             refueling_targets=list(self.refueling_targets),
-            threatening_air_defenses=list(self.threatening_air_defenses),
+            enemy_air_defenses=list(self.enemy_air_defenses),
             enemy_convoys=list(self.enemy_convoys),
             enemy_shipping=list(self.enemy_shipping),
-            threatening_ships=list(self.threatening_ships),
+            enemy_ships=list(self.enemy_ships),
             enemy_garrisons=list(self.enemy_garrisons),
             oca_targets=list(self.oca_targets),
             strike_targets=list(self.strike_targets),
+            enemy_barcaps=list(self.enemy_barcaps),
+            threat_zones=self.threat_zones,
+            opposing_doctrine=self.opposing_doctrine,
+            # 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
+            # protection. In that case, the preconditions of PlanBai would fail, but
+            # would add the IADS that prevented it from being planned to the list of
+            # IADS threats so that DegradeIads will consider it a threat later.
+            threatening_air_defenses=self.threatening_air_defenses,
         )
 
     @classmethod
@@ -57,11 +92,15 @@ class TheaterState(WorldState["TheaterState"]):
             vulnerable_front_lines=list(finder.front_lines()),
             aewc_targets=[finder.farthest_friendly_control_point()],
             refueling_targets=[finder.closest_friendly_control_point()],
-            threatening_air_defenses=list(finder.threatening_air_defenses()),
+            enemy_air_defenses=list(finder.enemy_air_defenses()),
+            threatening_air_defenses=[],
             enemy_convoys=list(finder.convoys()),
             enemy_shipping=list(finder.cargo_ships()),
-            threatening_ships=list(finder.threatening_ships()),
+            enemy_ships=list(finder.enemy_ships()),
             enemy_garrisons=list(finder.threatening_vehicle_groups()),
             oca_targets=list(finder.oca_targets(min_aircraft=20)),
             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,
         )

game/threatzones.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from functools import singledispatchmethod
-from typing import Optional, TYPE_CHECKING, Union, Iterable
+from typing import Optional, TYPE_CHECKING, Union, Iterable, Any
 
 from dcs.mapping import Point as DcsPoint
 from shapely.geometry import (
@@ -13,7 +13,8 @@ from shapely.geometry import (
 from shapely.geometry.base import BaseGeometry
 from shapely.ops import nearest_points, unary_union
 
-from game.theater import ControlPoint, MissionTarget
+from game.data.doctrine import Doctrine
+from game.theater import ControlPoint, MissionTarget, TheaterGroundObject
 from game.utils import Distance, meters, nautical_miles
 from gen.flights.closestairfields import ObjectiveDistanceCache
 from gen.flights.flight import Flight, FlightWaypoint
@@ -82,6 +83,10 @@ class ThreatZones:
             LineString((self.dcs_to_shapely_point(p.position) for p in flight.points))
         )
 
+    @threatened_by_aircraft.register
+    def _threatened_by_aircraft_mission_target(self, target: MissionTarget) -> bool:
+        return self.threatened_by_aircraft(self.dcs_to_shapely_point(target.position))
+
     def waypoints_threatened_by_aircraft(
         self, waypoints: Iterable[FlightWaypoint]
     ) -> bool:
@@ -145,8 +150,9 @@ class ThreatZones:
         return None
 
     @classmethod
-    def barcap_threat_range(cls, game: Game, control_point: ControlPoint) -> Distance:
-        doctrine = game.faction_for(control_point.captured).doctrine
+    def barcap_threat_range(
+        cls, doctrine: Doctrine, control_point: ControlPoint
+    ) -> Distance:
         cap_threat_range = (
             doctrine.cap_max_distance_from_cp + doctrine.cap_engagement_range
         )
@@ -185,33 +191,59 @@ class ThreatZones:
         """
         air_threats = []
         air_defenses = []
-        radar_sam_threats = []
-        for control_point in game.theater.controlpoints:
-            if control_point.captured != player:
-                continue
-            if control_point.runway_is_operational():
-                point = ShapelyPoint(control_point.position.x, control_point.position.y)
-                cap_threat_range = cls.barcap_threat_range(game, control_point)
-                air_threats.append(point.buffer(cap_threat_range.meters))
+        for control_point in game.theater.control_points_for(player):
+            air_threats.append(control_point)
+            air_defenses.extend(control_point.ground_objects)
 
-            for tgo in control_point.ground_objects:
-                for group in tgo.groups:
-                    threat_range = tgo.threat_range(group)
-                    # Any system with a shorter range than this is not worth
-                    # even avoiding.
-                    if threat_range > nautical_miles(3):
-                        point = ShapelyPoint(tgo.position.x, tgo.position.y)
-                        threat_zone = point.buffer(threat_range.meters)
-                        air_defenses.append(threat_zone)
-                    radar_threat_range = tgo.threat_range(group, radar_only=True)
-                    if radar_threat_range > nautical_miles(3):
-                        point = ShapelyPoint(tgo.position.x, tgo.position.y)
-                        threat_zone = point.buffer(threat_range.meters)
-                        radar_sam_threats.append(threat_zone)
+        return cls.for_threats(
+            game.faction_for(player).doctrine, air_threats, air_defenses
+        )
+
+    @classmethod
+    def for_threats(
+        cls,
+        doctrine: Doctrine,
+        barcap_locations: Iterable[ControlPoint],
+        air_defenses: Iterable[TheaterGroundObject[Any]],
+    ) -> ThreatZones:
+        """Generates the threat zones projected by the given locations.
+
+        Args:
+            doctrine: The doctrine of the owning coalition.
+            barcap_locations: The locations that will be considered for BARCAP planning.
+            air_defenses: TGOs that may have air defenses.
+
+        Returns:
+            The threat zones projected by the given locations. If the threat zone
+            belongs to the player, it is the zone that will be avoided by the enemy and
+            vice versa.
+        """
+        air_threats = []
+        air_defense_threats = []
+        radar_sam_threats = []
+        for barcap in barcap_locations:
+            point = ShapelyPoint(barcap.position.x, barcap.position.y)
+            cap_threat_range = cls.barcap_threat_range(doctrine, barcap)
+            air_threats.append(point.buffer(cap_threat_range.meters))
+
+        for tgo in air_defenses:
+            for group in tgo.groups:
+                threat_range = tgo.threat_range(group)
+                # Any system with a shorter range than this is not worth
+                # even avoiding.
+                if threat_range > nautical_miles(3):
+                    point = ShapelyPoint(tgo.position.x, tgo.position.y)
+                    threat_zone = point.buffer(threat_range.meters)
+                    air_defense_threats.append(threat_zone)
+                radar_threat_range = tgo.threat_range(group, radar_only=True)
+                if radar_threat_range > nautical_miles(3):
+                    point = ShapelyPoint(tgo.position.x, tgo.position.y)
+                    threat_zone = point.buffer(threat_range.meters)
+                    radar_sam_threats.append(threat_zone)
 
         return cls(
             airbases=unary_union(air_threats),
-            air_defenses=unary_union(air_defenses),
+            air_defenses=unary_union(air_defense_threats),
             radar_sam_threats=unary_union(radar_sam_threats),
         )