mirror of
https://github.com/dcs-retribution/dcs-retribution.git
synced 2025-11-10 15:41:24 +00:00
17751e52fd
Also changes when we clear the convoys. Because we plan when transfers are added (to plan UI orders immediately) we were planning convoys when delivering units, then clearing the convoys, then planning them again. Aside from the wasted effort, when we cleared the convoys we forgot to tell the transfer that they no longer had transport, so when replanning they did not get a new convoy.
496 lines
17 KiB
Python
496 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from functools import singledispatchmethod
|
|
from typing import Dict, Iterator, List, Optional, TYPE_CHECKING, Type
|
|
|
|
from dcs.mapping import Point
|
|
from dcs.unittype import FlyingType, VehicleType
|
|
|
|
from game.procurement import AircraftProcurementRequest
|
|
from game.utils import meters, nautical_miles
|
|
from gen.ato import Package
|
|
from gen.flights.ai_flight_planner_db import TRANSPORT_CAPABLE
|
|
from gen.flights.closestairfields import ObjectiveDistanceCache
|
|
from gen.flights.flightplan import FlightPlanBuilder
|
|
from game.theater import ControlPoint, MissionTarget
|
|
from game.theater.supplyroutes import SupplyRoute
|
|
from gen.naming import namegen
|
|
from gen.flights.flight import Flight, FlightType
|
|
|
|
if TYPE_CHECKING:
|
|
from game import Game
|
|
from game.inventory import ControlPointAircraftInventory
|
|
|
|
|
|
class Transport:
|
|
def find_escape_route(self) -> Optional[ControlPoint]:
|
|
raise NotImplementedError
|
|
|
|
def description(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
|
|
@dataclass
|
|
class TransferOrder:
|
|
"""The base type of all transfer orders.
|
|
|
|
A transfer order can transfer multiple units of multiple types.
|
|
"""
|
|
|
|
#: The location the units are transferring from.
|
|
origin: ControlPoint
|
|
|
|
#: The location the units are transferring to.
|
|
destination: ControlPoint
|
|
|
|
#: The current position of the group being transferred. Groups may make multiple
|
|
#: stops and can switch transport modes before reaching their destination.
|
|
position: ControlPoint = field(init=False)
|
|
|
|
#: True if the transfer order belongs to the player.
|
|
player: bool = field(init=False)
|
|
|
|
#: The units being transferred.
|
|
units: Dict[Type[VehicleType], int]
|
|
|
|
transport: Optional[Transport] = field(default=None)
|
|
|
|
def __post_init__(self) -> None:
|
|
self.position = self.origin
|
|
self.player = self.origin.is_friendly(to_player=True)
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
if self.transport is None:
|
|
return "No transports available"
|
|
return self.transport.description()
|
|
|
|
def kill_all(self) -> None:
|
|
self.units.clear()
|
|
|
|
def kill_unit(self, unit_type: Type[VehicleType]) -> None:
|
|
if unit_type in self.units:
|
|
self.units[unit_type] -= 1
|
|
return
|
|
raise KeyError
|
|
|
|
@property
|
|
def size(self) -> int:
|
|
return sum(c for c in self.units.values())
|
|
|
|
def iter_units(self) -> Iterator[Type[VehicleType]]:
|
|
for unit_type, count in self.units.items():
|
|
for _ in range(count):
|
|
yield unit_type
|
|
|
|
@property
|
|
def completed(self) -> bool:
|
|
return self.destination == self.position or not self.units
|
|
|
|
def disband_at(self, location: ControlPoint) -> None:
|
|
logging.info(f"Units halting at {location}.")
|
|
location.base.commision_units(self.units)
|
|
self.units.clear()
|
|
|
|
def proceed(self) -> None:
|
|
if self.transport is None:
|
|
return
|
|
|
|
if not self.destination.is_friendly(self.player):
|
|
logging.info(f"Transfer destination {self.destination} was captured.")
|
|
if self.position.is_friendly(self.player):
|
|
self.disband_at(self.position)
|
|
elif (escape_route := self.transport.find_escape_route()) is not None:
|
|
self.disband_at(escape_route)
|
|
else:
|
|
logging.info(
|
|
f"No escape route available. Units were surrounded and destroyed "
|
|
"during transfer."
|
|
)
|
|
self.kill_all()
|
|
return
|
|
|
|
self.position = self.destination
|
|
self.transport = None
|
|
|
|
if self.completed:
|
|
self.disband_at(self.position)
|
|
|
|
|
|
@dataclass
|
|
class Airlift(Transport):
|
|
"""A transfer order that moves units by cargo planes and helicopters."""
|
|
|
|
transfer: TransferOrder
|
|
|
|
flight: Flight
|
|
|
|
@property
|
|
def units(self) -> Dict[Type[VehicleType], int]:
|
|
return self.transfer.units
|
|
|
|
@property
|
|
def player_owned(self) -> bool:
|
|
return self.transfer.player
|
|
|
|
def find_escape_route(self) -> Optional[ControlPoint]:
|
|
# TODO: Move units to closest base.
|
|
return None
|
|
|
|
def description(self) -> str:
|
|
return f"Being airlifted by {self.flight}"
|
|
|
|
|
|
class AirliftPlanner:
|
|
#: Maximum range from for any link in the route of takeoff, pickup, dropoff, and RTB
|
|
#: for a helicopter to be considered for airlift. Total route length is not
|
|
#: considered because the helicopter can refuel at each stop. Cargo planes have no
|
|
#: maximum range.
|
|
HELO_MAX_RANGE = nautical_miles(100)
|
|
|
|
def __init__(self, game: Game, transfer: TransferOrder) -> None:
|
|
self.game = game
|
|
self.transfer = transfer
|
|
self.for_player = transfer.destination.captured
|
|
self.package = Package(target=transfer.destination, auto_asap=True)
|
|
|
|
def compatible_with_mission(
|
|
self, unit_type: Type[FlyingType], airfield: ControlPoint
|
|
) -> bool:
|
|
if not unit_type in TRANSPORT_CAPABLE:
|
|
return False
|
|
if not self.transfer.origin.can_operate(unit_type):
|
|
return False
|
|
if not self.transfer.destination.can_operate(unit_type):
|
|
return False
|
|
|
|
# Cargo planes have no maximum range.
|
|
if not unit_type.helicopter:
|
|
return True
|
|
|
|
# A helicopter that is transport capable and able to operate at both bases. Need
|
|
# to check that no leg of the journey exceeds the maximum range. This doesn't
|
|
# account for any routing around threats that might take place, but it's close
|
|
# enough.
|
|
|
|
home = airfield.position
|
|
pickup = self.transfer.position.position
|
|
drop_off = self.transfer.position.position
|
|
if meters(home.distance_to_point(pickup)) > self.HELO_MAX_RANGE:
|
|
return False
|
|
|
|
if meters(pickup.distance_to_point(drop_off)) > self.HELO_MAX_RANGE:
|
|
return False
|
|
|
|
if meters(drop_off.distance_to_point(home)) > self.HELO_MAX_RANGE:
|
|
return False
|
|
|
|
return True
|
|
|
|
def create_package_for_airlift(self) -> None:
|
|
distance_cache = ObjectiveDistanceCache.get_closest_airfields(
|
|
self.transfer.position
|
|
)
|
|
for cp in distance_cache.closest_airfields:
|
|
if cp.captured != self.for_player:
|
|
continue
|
|
|
|
inventory = self.game.aircraft_inventory.for_control_point(cp)
|
|
for unit_type, available in inventory.all_aircraft:
|
|
if self.compatible_with_mission(unit_type, cp):
|
|
while available and self.transfer.transport is None:
|
|
flight_size = self.create_airlift_flight(unit_type, inventory)
|
|
available -= flight_size
|
|
if self.package.flights:
|
|
self.game.ato_for(self.for_player).add_package(self.package)
|
|
|
|
def create_airlift_flight(
|
|
self, unit_type: Type[FlyingType], inventory: ControlPointAircraftInventory
|
|
) -> int:
|
|
available = inventory.available(unit_type)
|
|
# 4 is the max flight size in DCS.
|
|
flight_size = min(self.transfer.size, available, 4)
|
|
|
|
if flight_size < self.transfer.size:
|
|
transfer = self.game.transfers.split_transfer(self.transfer, flight_size)
|
|
else:
|
|
transfer = self.transfer
|
|
|
|
flight = Flight(
|
|
self.package,
|
|
self.game.player_country,
|
|
unit_type,
|
|
flight_size,
|
|
FlightType.TRANSPORT,
|
|
self.game.settings.default_start_type,
|
|
departure=inventory.control_point,
|
|
arrival=inventory.control_point,
|
|
divert=None,
|
|
cargo=transfer,
|
|
)
|
|
|
|
transport = Airlift(transfer, flight)
|
|
transfer.transport = transport
|
|
|
|
self.package.add_flight(flight)
|
|
planner = FlightPlanBuilder(self.game, self.package, self.for_player)
|
|
planner.populate_flight_plan(flight)
|
|
self.game.aircraft_inventory.claim_for_flight(flight)
|
|
return flight_size
|
|
|
|
|
|
class Convoy(MissionTarget, Transport):
|
|
def __init__(self, origin: ControlPoint, destination: ControlPoint) -> None:
|
|
super().__init__(namegen.next_convoy_name(), origin.position)
|
|
self.origin = origin
|
|
self.destination = destination
|
|
self.transfers: List[TransferOrder] = []
|
|
|
|
def mission_types(self, for_player: bool) -> Iterator[FlightType]:
|
|
if self.is_friendly(for_player):
|
|
return
|
|
|
|
yield FlightType.BAI
|
|
yield from super().mission_types(for_player)
|
|
|
|
def is_friendly(self, to_player: bool) -> bool:
|
|
return self.origin.captured
|
|
|
|
def add_units(self, transfer: TransferOrder) -> None:
|
|
self.transfers.append(transfer)
|
|
transfer.transport = self
|
|
|
|
def remove_units(self, transfer: TransferOrder) -> None:
|
|
transfer.transport = None
|
|
self.transfers.remove(transfer)
|
|
|
|
def kill_unit(self, unit_type: Type[VehicleType]) -> None:
|
|
for transfer in self.transfers:
|
|
try:
|
|
transfer.kill_unit(unit_type)
|
|
return
|
|
except KeyError:
|
|
pass
|
|
raise KeyError
|
|
|
|
def disband(self) -> None:
|
|
for transfer in list(self.transfers):
|
|
self.remove_units(transfer)
|
|
self.transfers.clear()
|
|
|
|
@property
|
|
def size(self) -> int:
|
|
return sum(sum(t.units.values()) for t in self.transfers)
|
|
|
|
@property
|
|
def units(self) -> Dict[Type[VehicleType], int]:
|
|
units: Dict[Type[VehicleType], int] = defaultdict(int)
|
|
for transfer in self.transfers:
|
|
for unit_type, count in transfer.units.items():
|
|
units[unit_type] += count
|
|
return units
|
|
|
|
@property
|
|
def player_owned(self) -> bool:
|
|
return self.origin.captured
|
|
|
|
def find_escape_route(self) -> Optional[ControlPoint]:
|
|
return None
|
|
|
|
def description(self) -> str:
|
|
return f"In a convoy to {self.destination}"
|
|
|
|
@property
|
|
def route_start(self) -> Point:
|
|
return self.origin.convoy_spawns[self.destination]
|
|
|
|
@property
|
|
def route_end(self) -> Point:
|
|
return self.destination.convoy_spawns[self.origin]
|
|
|
|
|
|
class ConvoyMap:
|
|
def __init__(self) -> None:
|
|
# Dict of origin -> destination -> convoy.
|
|
self.convoys: Dict[ControlPoint, Dict[ControlPoint, Convoy]] = defaultdict(dict)
|
|
|
|
def convoy_exists(self, origin: ControlPoint, destination: ControlPoint) -> bool:
|
|
return destination in self.convoys[origin]
|
|
|
|
def find_convoy(
|
|
self, origin: ControlPoint, destination: ControlPoint
|
|
) -> Optional[Convoy]:
|
|
return self.convoys[origin].get(destination)
|
|
|
|
def find_or_create_convoy(
|
|
self, origin: ControlPoint, destination: ControlPoint
|
|
) -> Convoy:
|
|
convoy = self.find_convoy(origin, destination)
|
|
if convoy is None:
|
|
convoy = Convoy(origin, destination)
|
|
self.convoys[origin][destination] = convoy
|
|
return convoy
|
|
|
|
def departing_from(self, origin: ControlPoint) -> Iterator[Convoy]:
|
|
yield from self.convoys[origin].values()
|
|
|
|
def travelling_to(self, destination: ControlPoint) -> Iterator[Convoy]:
|
|
for destination_dict in self.convoys.values():
|
|
if destination in destination_dict:
|
|
yield destination_dict[destination]
|
|
|
|
def disband_convoy(self, convoy: Convoy) -> None:
|
|
self.convoys[convoy.origin][convoy.destination].disband()
|
|
del self.convoys[convoy.origin][convoy.destination]
|
|
|
|
@staticmethod
|
|
def path_for(transfer: TransferOrder) -> List[ControlPoint]:
|
|
supply_route = SupplyRoute.for_control_point(transfer.position)
|
|
return supply_route.shortest_path_between(
|
|
transfer.position, transfer.destination
|
|
)
|
|
|
|
def next_stop_for(self, transfer: TransferOrder) -> ControlPoint:
|
|
return self.path_for(transfer)[0]
|
|
|
|
def add(self, transfer: TransferOrder) -> None:
|
|
next_stop = self.next_stop_for(transfer)
|
|
self.find_or_create_convoy(transfer.position, next_stop).add_units(transfer)
|
|
|
|
def remove(self, convoy: Convoy, transfer: TransferOrder) -> None:
|
|
convoy.remove_units(transfer)
|
|
if not convoy.transfers:
|
|
self.disband_convoy(convoy)
|
|
|
|
def disband_all(self) -> None:
|
|
for convoy in list(self):
|
|
self.disband_convoy(convoy)
|
|
|
|
def __iter__(self) -> Iterator[Convoy]:
|
|
for destination_dict in self.convoys.values():
|
|
yield from destination_dict.values()
|
|
|
|
|
|
class PendingTransfers:
|
|
def __init__(self, game: Game) -> None:
|
|
self.game = game
|
|
self.convoys = ConvoyMap()
|
|
self.pending_transfers: List[TransferOrder] = []
|
|
|
|
def __iter__(self) -> Iterator[TransferOrder]:
|
|
yield from self.pending_transfers
|
|
|
|
@property
|
|
def pending_transfer_count(self) -> int:
|
|
return len(self.pending_transfers)
|
|
|
|
def transfer_at_index(self, index: int) -> TransferOrder:
|
|
return self.pending_transfers[index]
|
|
|
|
def index_of_transfer(self, transfer: TransferOrder) -> int:
|
|
return self.pending_transfers.index(transfer)
|
|
|
|
def arrange_transport(self, transfer: TransferOrder) -> None:
|
|
supply_route = SupplyRoute.for_control_point(transfer.position)
|
|
if transfer.destination in supply_route:
|
|
self.convoys.add(transfer)
|
|
else:
|
|
AirliftPlanner(self.game, transfer).create_package_for_airlift()
|
|
|
|
def new_transfer(self, transfer: TransferOrder) -> None:
|
|
transfer.origin.base.commit_losses(transfer.units)
|
|
self.pending_transfers.append(transfer)
|
|
self.arrange_transport(transfer)
|
|
|
|
def split_transfer(self, transfer: TransferOrder, size: int) -> TransferOrder:
|
|
"""Creates a smaller transfer that is a subset of the original."""
|
|
if transfer.size <= size:
|
|
raise ValueError
|
|
|
|
units = {}
|
|
for unit_type, remaining in transfer.units.items():
|
|
take = min(remaining, size)
|
|
size -= take
|
|
transfer.units[unit_type] -= take
|
|
units[unit_type] = take
|
|
if not size:
|
|
break
|
|
new_transfer = TransferOrder(transfer.origin, transfer.destination, units)
|
|
self.pending_transfers.append(new_transfer)
|
|
return new_transfer
|
|
|
|
@singledispatchmethod
|
|
def cancel_transport(self, transfer: TransferOrder, transport) -> None:
|
|
pass
|
|
|
|
@cancel_transport.register
|
|
def _cancel_transport_air(
|
|
self, _transfer: TransferOrder, transport: Airlift
|
|
) -> None:
|
|
flight = transport.flight
|
|
flight.package.remove_flight(flight)
|
|
self.game.aircraft_inventory.return_from_flight(flight)
|
|
|
|
def _cancel_transport_convoy(
|
|
self, transfer: TransferOrder, transport: Convoy
|
|
) -> None:
|
|
self.convoys.remove(transport, transfer)
|
|
|
|
def cancel_transfer(self, transfer: TransferOrder) -> None:
|
|
if transfer.transport is not None:
|
|
self.cancel_transport(transfer, transfer.transport)
|
|
self.pending_transfers.remove(transfer)
|
|
transfer.origin.base.commision_units(transfer.units)
|
|
|
|
def perform_transfers(self) -> None:
|
|
incomplete = []
|
|
for transfer in self.pending_transfers:
|
|
transfer.proceed()
|
|
if not transfer.completed:
|
|
incomplete.append(transfer)
|
|
self.pending_transfers = incomplete
|
|
self.convoys.disband_all()
|
|
|
|
def plan_transports(self) -> None:
|
|
for transfer in self.pending_transfers:
|
|
if transfer.transport is None:
|
|
self.arrange_transport(transfer)
|
|
|
|
def order_airlift_assets(self) -> None:
|
|
for control_point in self.game.theater.controlpoints:
|
|
self.order_airlift_assets_at(control_point)
|
|
|
|
@staticmethod
|
|
def desired_airlift_capacity(control_point: ControlPoint) -> int:
|
|
return 4 if control_point.has_factory else 0
|
|
|
|
def current_airlift_capacity(self, control_point: ControlPoint) -> int:
|
|
inventory = self.game.aircraft_inventory.for_control_point(control_point)
|
|
return sum(
|
|
count
|
|
for unit_type, count in inventory.all_aircraft
|
|
if unit_type in TRANSPORT_CAPABLE
|
|
)
|
|
|
|
def order_airlift_assets_at(self, control_point: ControlPoint) -> None:
|
|
gap = self.desired_airlift_capacity(
|
|
control_point
|
|
) - self.current_airlift_capacity(control_point)
|
|
|
|
if gap <= 0:
|
|
return
|
|
|
|
if gap % 2:
|
|
# Always buy in pairs since we're not trying to fill odd squadrons. Purely
|
|
# aesthetic.
|
|
gap += 1
|
|
|
|
self.game.procurement_requests_for(player=control_point.captured).append(
|
|
AircraftProcurementRequest(
|
|
control_point, nautical_miles(200), FlightType.TRANSPORT, gap
|
|
)
|
|
)
|