mirror of
https://github.com/dcs-retribution/dcs-retribution.git
synced 2025-11-10 15:41:24 +00:00
Assign aircraft to squadrons rather than bases.
This is needed to support the upcoming squadron transfers, since squadrons need to bring their aircraft with them. https://github.com/dcs-liberation/dcs_liberation/issues/1145
This commit is contained in:
@@ -10,7 +10,6 @@ from game.campaignloader.defaultsquadronassigner import DefaultSquadronAssigner
|
||||
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.orderedset import OrderedSet
|
||||
from game.profiling import logged_duration, MultiEventTracer
|
||||
@@ -88,10 +87,6 @@ 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
|
||||
@@ -196,7 +191,9 @@ class Coalition:
|
||||
return
|
||||
|
||||
for cp in self.game.theater.control_points_for(self.player):
|
||||
cp.pending_unit_deliveries.refund_all(self)
|
||||
cp.ground_unit_orders.refund_all(self)
|
||||
for squadron in self.air_wing.iter_squadrons():
|
||||
squadron.refund_orders()
|
||||
|
||||
def plan_missions(self) -> None:
|
||||
color = "Blue" if self.player else "Red"
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from game.commander.missionproposals import ProposedFlight
|
||||
from game.inventory import GlobalAircraftInventory
|
||||
from game.squadrons.squadron import Squadron
|
||||
from game.squadrons.airwing import AirWing
|
||||
from game.squadrons.squadron import Squadron
|
||||
from game.theater import ControlPoint, MissionTarget
|
||||
from game.utils import meters
|
||||
from gen.flights.ai_flight_planner_db import aircraft_for_task
|
||||
@@ -15,15 +14,10 @@ class AircraftAllocator:
|
||||
"""Finds suitable aircraft for proposed missions."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
air_wing: AirWing,
|
||||
closest_airfields: ClosestAirfields,
|
||||
global_inventory: GlobalAircraftInventory,
|
||||
is_player: bool,
|
||||
self, air_wing: AirWing, closest_airfields: ClosestAirfields, 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(
|
||||
@@ -56,12 +50,9 @@ class AircraftAllocator:
|
||||
for airfield in self.closest_airfields.operational_airfields:
|
||||
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
|
||||
distance_to_target = meters(target.distance_to(airfield))
|
||||
if distance_to_target > aircraft.max_mission_range:
|
||||
continue
|
||||
@@ -71,9 +62,8 @@ class AircraftAllocator:
|
||||
aircraft, task, airfield
|
||||
)
|
||||
for squadron in squadrons:
|
||||
if squadron.operates_from(airfield) and squadron.can_provide_pilots(
|
||||
if squadron.operates_from(airfield) and squadron.can_fulfill_flight(
|
||||
flight.num_aircraft
|
||||
):
|
||||
inventory.remove_aircraft(aircraft, flight.num_aircraft)
|
||||
return airfield, squadron
|
||||
return None
|
||||
|
||||
@@ -157,7 +157,10 @@ class ObjectiveFinder:
|
||||
for control_point in self.enemy_control_points():
|
||||
if not isinstance(control_point, Airfield):
|
||||
continue
|
||||
if control_point.base.total_aircraft >= min_aircraft:
|
||||
if (
|
||||
control_point.allocated_aircraft(self.game).total_present
|
||||
>= min_aircraft
|
||||
):
|
||||
airfields.append(control_point)
|
||||
return self._targets_by_range(airfields)
|
||||
|
||||
|
||||
@@ -1,13 +1,12 @@
|
||||
from typing import Optional
|
||||
|
||||
from game.commander.aircraftallocator import AircraftAllocator
|
||||
from game.commander.missionproposals import ProposedFlight
|
||||
from game.dcs.aircrafttype import AircraftType
|
||||
from game.inventory import GlobalAircraftInventory
|
||||
from game.squadrons.airwing import AirWing
|
||||
from game.theater import MissionTarget, OffMapSpawn, ControlPoint
|
||||
from game.utils import nautical_miles
|
||||
from gen.ato import Package
|
||||
from game.commander.aircraftallocator import AircraftAllocator
|
||||
from gen.flights.closestairfields import ClosestAirfields
|
||||
from gen.flights.flight import Flight
|
||||
|
||||
@@ -19,7 +18,6 @@ class PackageBuilder:
|
||||
self,
|
||||
location: MissionTarget,
|
||||
closest_airfields: ClosestAirfields,
|
||||
global_inventory: GlobalAircraftInventory,
|
||||
air_wing: AirWing,
|
||||
is_player: bool,
|
||||
package_country: str,
|
||||
@@ -30,10 +28,7 @@ class PackageBuilder:
|
||||
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.allocator = AircraftAllocator(air_wing, closest_airfields, is_player)
|
||||
self.start_type = start_type
|
||||
|
||||
def plan_flight(self, plan: ProposedFlight) -> bool:
|
||||
@@ -93,6 +88,5 @@ class PackageBuilder:
|
||||
"""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()
|
||||
flight.return_pilots_and_aircraft()
|
||||
self.package.remove_flight(flight)
|
||||
|
||||
@@ -5,15 +5,14 @@ from collections import defaultdict
|
||||
from typing import Set, Iterable, Dict, TYPE_CHECKING, Optional
|
||||
|
||||
from game.commander.missionproposals import ProposedMission, ProposedFlight, EscortType
|
||||
from game.commander.packagebuilder import PackageBuilder
|
||||
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 game.commander.packagebuilder import PackageBuilder
|
||||
from gen.ato import AirTaskingOrder, Package
|
||||
from gen.flights.closestairfields import ObjectiveDistanceCache
|
||||
from gen.flights.flight import FlightType
|
||||
@@ -27,15 +26,10 @@ class PackageFulfiller:
|
||||
"""Responsible for package aircraft allocation and flight plan layout."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coalition: Coalition,
|
||||
theater: ConflictTheater,
|
||||
aircraft_inventory: GlobalAircraftInventory,
|
||||
settings: Settings,
|
||||
self, coalition: Coalition, theater: ConflictTheater, 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
|
||||
|
||||
@@ -137,7 +131,6 @@ class PackageFulfiller:
|
||||
builder = PackageBuilder(
|
||||
mission.location,
|
||||
ObjectiveDistanceCache.get_closest_airfields(mission.location),
|
||||
self.aircraft_inventory,
|
||||
self.air_wing,
|
||||
self.is_player,
|
||||
self.coalition.country_name,
|
||||
|
||||
@@ -53,8 +53,6 @@ class PackagePlanningTask(TheaterCommanderTask, Generic[MissionTargetT]):
|
||||
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
|
||||
@@ -99,7 +97,6 @@ class PackagePlanningTask(TheaterCommanderTask, Generic[MissionTargetT]):
|
||||
fulfiller = PackageFulfiller(
|
||||
state.context.coalition,
|
||||
state.context.theater,
|
||||
state.available_aircraft,
|
||||
state.context.settings,
|
||||
)
|
||||
self.package = fulfiller.plan_mission(
|
||||
|
||||
@@ -10,9 +10,9 @@ from typing import TYPE_CHECKING, Any, Union, Optional
|
||||
from game.commander.garrisons import Garrisons
|
||||
from game.commander.objectivefinder import ObjectiveFinder
|
||||
from game.htn import WorldState
|
||||
from game.inventory import GlobalAircraftInventory
|
||||
from game.profiling import MultiEventTracer
|
||||
from game.settings import Settings
|
||||
from game.squadrons import AirWing
|
||||
from game.theater import ControlPoint, FrontLine, MissionTarget, ConflictTheater
|
||||
from game.theater.theatergroundobject import (
|
||||
TheaterGroundObject,
|
||||
@@ -58,7 +58,6 @@ class TheaterState(WorldState["TheaterState"]):
|
||||
strike_targets: list[TheaterGroundObject[Any]]
|
||||
enemy_barcaps: list[ControlPoint]
|
||||
threat_zones: ThreatZones
|
||||
available_aircraft: GlobalAircraftInventory
|
||||
|
||||
def _rebuild_threat_zones(self) -> None:
|
||||
"""Recreates the theater's threat zones based on the current planned state."""
|
||||
@@ -122,7 +121,6 @@ class TheaterState(WorldState["TheaterState"]):
|
||||
strike_targets=list(self.strike_targets),
|
||||
enemy_barcaps=list(self.enemy_barcaps),
|
||||
threat_zones=self.threat_zones,
|
||||
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
|
||||
@@ -172,5 +170,4 @@ 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),
|
||||
available_aircraft=game.aircraft_inventory.clone(),
|
||||
)
|
||||
|
||||
@@ -7,13 +7,12 @@ from dcs.mapping import Point
|
||||
from dcs.task import Task
|
||||
|
||||
from game import persistency
|
||||
from game.debriefing import AirLosses, Debriefing
|
||||
from game.debriefing import Debriefing
|
||||
from game.infos.information import Information
|
||||
from game.operation.operation import Operation
|
||||
from game.theater import ControlPoint
|
||||
from gen.ato import AirTaskingOrder
|
||||
from gen.ground_forces.combat_stance import CombatStance
|
||||
from ..dcs.groundunittype import GroundUnitType
|
||||
from ..unitmap import UnitMap
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -67,59 +66,6 @@ class Event:
|
||||
)
|
||||
return unit_map
|
||||
|
||||
@staticmethod
|
||||
def _transfer_aircraft(
|
||||
ato: AirTaskingOrder, losses: AirLosses, for_player: bool
|
||||
) -> None:
|
||||
for package in ato.packages:
|
||||
for flight in package.flights:
|
||||
# No need to transfer to the same location.
|
||||
if flight.departure == flight.arrival:
|
||||
continue
|
||||
|
||||
# Don't transfer to bases that were captured. Note that if the
|
||||
# airfield was back-filling transfers it may overflow. We could
|
||||
# attempt to be smarter in the future by performing transfers in
|
||||
# order up a graph to prevent transfers to full airports and
|
||||
# send overflow off-map, but overflow is fine for now.
|
||||
if flight.arrival.captured != for_player:
|
||||
logging.info(
|
||||
f"Not transferring {flight} because {flight.arrival} "
|
||||
"was captured"
|
||||
)
|
||||
continue
|
||||
|
||||
transfer_count = losses.surviving_flight_members(flight)
|
||||
if transfer_count < 0:
|
||||
logging.error(
|
||||
f"{flight} had {flight.count} aircraft but "
|
||||
f"{transfer_count} losses were recorded."
|
||||
)
|
||||
continue
|
||||
|
||||
aircraft = flight.unit_type
|
||||
available = flight.departure.base.total_units_of_type(aircraft)
|
||||
if available < transfer_count:
|
||||
logging.error(
|
||||
f"Found killed {aircraft} from {flight.departure} but "
|
||||
f"that airbase has only {available} available."
|
||||
)
|
||||
continue
|
||||
|
||||
flight.departure.base.aircraft[aircraft] -= transfer_count
|
||||
if aircraft not in flight.arrival.base.aircraft:
|
||||
# TODO: Should use defaultdict.
|
||||
flight.arrival.base.aircraft[aircraft] = 0
|
||||
flight.arrival.base.aircraft[aircraft] += transfer_count
|
||||
|
||||
def complete_aircraft_transfers(self, debriefing: Debriefing) -> None:
|
||||
self._transfer_aircraft(
|
||||
self.game.blue.ato, debriefing.air_losses, for_player=True
|
||||
)
|
||||
self._transfer_aircraft(
|
||||
self.game.red.ato, debriefing.air_losses, for_player=False
|
||||
)
|
||||
|
||||
def commit_air_losses(self, debriefing: Debriefing) -> None:
|
||||
for loss in debriefing.air_losses.losses:
|
||||
if loss.pilot is not None and (
|
||||
@@ -127,18 +73,18 @@ class Event:
|
||||
or not self.game.settings.invulnerable_player_pilots
|
||||
):
|
||||
loss.pilot.kill()
|
||||
squadron = loss.flight.squadron
|
||||
aircraft = loss.flight.unit_type
|
||||
cp = loss.flight.departure
|
||||
available = cp.base.total_units_of_type(aircraft)
|
||||
available = squadron.owned_aircraft
|
||||
if available <= 0:
|
||||
logging.error(
|
||||
f"Found killed {aircraft} from {cp} but that airbase has "
|
||||
f"Found killed {aircraft} from {squadron} but that airbase has "
|
||||
"none available."
|
||||
)
|
||||
continue
|
||||
|
||||
logging.info(f"{aircraft} destroyed from {cp}")
|
||||
cp.base.aircraft[aircraft] -= 1
|
||||
logging.info(f"{aircraft} destroyed from {squadron}")
|
||||
squadron.owned_aircraft -= 1
|
||||
|
||||
@staticmethod
|
||||
def _commit_pilot_experience(ato: AirTaskingOrder) -> None:
|
||||
@@ -276,7 +222,6 @@ class Event:
|
||||
self.commit_building_losses(debriefing)
|
||||
self.commit_damaged_runways(debriefing)
|
||||
self.commit_captures(debriefing)
|
||||
self.complete_aircraft_transfers(debriefing)
|
||||
|
||||
# Destroyed units carcass
|
||||
# -------------------------
|
||||
@@ -458,15 +403,10 @@ class Event:
|
||||
source.base.commit_losses(moved_units)
|
||||
|
||||
# Also transfer pending deliveries.
|
||||
for unit_type, count in source.pending_unit_deliveries.units.items():
|
||||
if not isinstance(unit_type, GroundUnitType):
|
||||
continue
|
||||
if count <= 0:
|
||||
# Don't transfer *sales*...
|
||||
continue
|
||||
for unit_type, count in source.ground_unit_orders.units.items():
|
||||
move_count = int(count * move_factor)
|
||||
source.pending_unit_deliveries.sell({unit_type: move_count})
|
||||
destination.pending_unit_deliveries.order({unit_type: move_count})
|
||||
source.ground_unit_orders.sell({unit_type: move_count})
|
||||
destination.ground_unit_orders.order({unit_type: move_count})
|
||||
total_units_redeployed += move_count
|
||||
|
||||
if total_units_redeployed > 0:
|
||||
|
||||
13
game/game.py
13
game/game.py
@@ -13,7 +13,6 @@ from dcs.task import CAP, CAS, PinpointStrike
|
||||
from dcs.vehicles import AirDefence
|
||||
from faker import Faker
|
||||
|
||||
from game.inventory import GlobalAircraftInventory
|
||||
from game.models.game_stats import GameStats
|
||||
from game.plugins import LuaPluginManager
|
||||
from gen import naming
|
||||
@@ -127,8 +126,6 @@ class Game:
|
||||
self.blue.configure_default_air_wing(air_wing_config)
|
||||
self.red.configure_default_air_wing(air_wing_config)
|
||||
|
||||
self.aircraft_inventory = GlobalAircraftInventory(self.theater.controlpoints)
|
||||
|
||||
self.on_load(game_still_initializing=True)
|
||||
|
||||
def __setstate__(self, state: dict[str, Any]) -> None:
|
||||
@@ -392,9 +389,9 @@ class Game:
|
||||
|
||||
# Plan Coalition specific turn
|
||||
if for_blue:
|
||||
self.initialize_turn_for(player=True)
|
||||
self.blue.initialize_turn()
|
||||
if for_red:
|
||||
self.initialize_turn_for(player=False)
|
||||
self.red.initialize_turn()
|
||||
|
||||
# Plan GroundWar
|
||||
self.ground_planners = {}
|
||||
@@ -404,12 +401,6 @@ class Game:
|
||||
gplanner.plan_groundwar()
|
||||
self.ground_planners[cp.id] = gplanner
|
||||
|
||||
def initialize_turn_for(self, player: bool) -> None:
|
||||
self.aircraft_inventory.reset(player)
|
||||
for cp in self.theater.control_points_for(player):
|
||||
self.aircraft_inventory.set_from_control_point(cp)
|
||||
self.coalition_for(player).initialize_turn()
|
||||
|
||||
def message(self, text: str) -> None:
|
||||
self.informations.append(Information(text, turn=self.turn))
|
||||
|
||||
|
||||
@@ -2,13 +2,11 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, TYPE_CHECKING, Any
|
||||
from typing import Optional, TYPE_CHECKING
|
||||
|
||||
from game.theater import ControlPoint
|
||||
from .coalition import Coalition
|
||||
from .dcs.groundunittype import GroundUnitType
|
||||
from .dcs.unittype import UnitType
|
||||
from .theater.transitnetwork import (
|
||||
NoPathError,
|
||||
TransitNetwork,
|
||||
@@ -19,58 +17,41 @@ if TYPE_CHECKING:
|
||||
from .game import Game
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GroundUnitSource:
|
||||
control_point: ControlPoint
|
||||
|
||||
|
||||
class PendingUnitDeliveries:
|
||||
class GroundUnitOrders:
|
||||
def __init__(self, destination: ControlPoint) -> None:
|
||||
self.destination = destination
|
||||
|
||||
# Maps unit type to order quantity.
|
||||
self.units: dict[UnitType[Any], int] = defaultdict(int)
|
||||
self.units: dict[GroundUnitType, int] = defaultdict(int)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"Pending delivery to {self.destination}"
|
||||
return f"Pending ground unit delivery to {self.destination}"
|
||||
|
||||
def order(self, units: dict[UnitType[Any], int]) -> None:
|
||||
def order(self, units: dict[GroundUnitType, int]) -> None:
|
||||
for k, v in units.items():
|
||||
self.units[k] += v
|
||||
|
||||
def sell(self, units: dict[UnitType[Any], int]) -> None:
|
||||
def sell(self, units: dict[GroundUnitType, int]) -> None:
|
||||
for k, v in units.items():
|
||||
self.units[k] -= v
|
||||
if self.units[k] == 0:
|
||||
del self.units[k]
|
||||
|
||||
def refund_all(self, coalition: Coalition) -> None:
|
||||
self.refund(coalition, self.units)
|
||||
self._refund(coalition, self.units)
|
||||
self.units = defaultdict(int)
|
||||
|
||||
def refund_ground_units(self, coalition: Coalition) -> None:
|
||||
ground_units: dict[UnitType[Any], int] = {
|
||||
u: self.units[u] for u in self.units.keys() if isinstance(u, GroundUnitType)
|
||||
}
|
||||
self.refund(coalition, ground_units)
|
||||
for gu in ground_units.keys():
|
||||
del self.units[gu]
|
||||
|
||||
def refund(self, coalition: Coalition, units: dict[UnitType[Any], int]) -> None:
|
||||
def _refund(self, coalition: Coalition, units: dict[GroundUnitType, int]) -> None:
|
||||
for unit_type, count in units.items():
|
||||
logging.info(f"Refunding {count} {unit_type} at {self.destination.name}")
|
||||
coalition.adjust_budget(unit_type.price * count)
|
||||
|
||||
def pending_orders(self, unit_type: UnitType[Any]) -> int:
|
||||
def pending_orders(self, unit_type: GroundUnitType) -> int:
|
||||
pending_units = self.units.get(unit_type)
|
||||
if pending_units is None:
|
||||
pending_units = 0
|
||||
return pending_units
|
||||
|
||||
def available_next_turn(self, unit_type: UnitType[Any]) -> int:
|
||||
current_units = self.destination.base.total_units_of_type(unit_type)
|
||||
return self.pending_orders(unit_type) + current_units
|
||||
|
||||
def process(self, game: Game) -> None:
|
||||
coalition = game.coalition_for(self.destination.captured)
|
||||
ground_unit_source = self.find_ground_unit_source(game)
|
||||
@@ -79,36 +60,33 @@ class PendingUnitDeliveries:
|
||||
f"{self.destination.name} lost its source for ground unit "
|
||||
"reinforcements. Refunding purchase price."
|
||||
)
|
||||
self.refund_ground_units(coalition)
|
||||
self.refund_all(coalition)
|
||||
|
||||
bought_units: dict[UnitType[Any], int] = {}
|
||||
bought_units: dict[GroundUnitType, int] = {}
|
||||
units_needing_transfer: dict[GroundUnitType, int] = {}
|
||||
sold_units: dict[UnitType[Any], int] = {}
|
||||
for unit_type, count in self.units.items():
|
||||
allegiance = "Ally" if self.destination.captured else "Enemy"
|
||||
d: dict[Any, int]
|
||||
if (
|
||||
isinstance(unit_type, GroundUnitType)
|
||||
and self.destination != ground_unit_source
|
||||
):
|
||||
d: dict[GroundUnitType, int]
|
||||
if self.destination != ground_unit_source:
|
||||
source = ground_unit_source
|
||||
d = units_needing_transfer
|
||||
else:
|
||||
source = self.destination
|
||||
d = bought_units
|
||||
|
||||
if count >= 0:
|
||||
if count < 0:
|
||||
logging.error(
|
||||
f"Attempted sale of {unit_type} at {self.destination} but ground "
|
||||
"units cannot be sold"
|
||||
)
|
||||
elif count > 0:
|
||||
d[unit_type] = count
|
||||
game.message(
|
||||
f"{allegiance} reinforcements: {unit_type} x {count} at {source}"
|
||||
)
|
||||
else:
|
||||
sold_units[unit_type] = -count
|
||||
game.message(f"{allegiance} sold: {unit_type} x {-count} at {source}")
|
||||
|
||||
self.units = defaultdict(int)
|
||||
self.destination.base.commission_units(bought_units)
|
||||
self.destination.base.commit_losses(sold_units)
|
||||
|
||||
if units_needing_transfer:
|
||||
if ground_unit_source is None:
|
||||
@@ -1,142 +0,0 @@
|
||||
"""Inventory management APIs."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict, Iterator, Iterable
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from game.dcs.aircrafttype import AircraftType
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from game.theater import ControlPoint
|
||||
from gen.flights.flight import Flight
|
||||
|
||||
|
||||
class ControlPointAircraftInventory:
|
||||
"""Aircraft inventory for a single control point."""
|
||||
|
||||
def __init__(self, control_point: ControlPoint) -> None:
|
||||
self.control_point = control_point
|
||||
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.
|
||||
|
||||
Args:
|
||||
aircraft: The type of aircraft to add.
|
||||
count: The number of aircraft to add.
|
||||
"""
|
||||
self.inventory[aircraft] += count
|
||||
|
||||
def remove_aircraft(self, aircraft: AircraftType, count: int) -> None:
|
||||
"""Removes aircraft from the inventory.
|
||||
|
||||
Args:
|
||||
aircraft: The type of aircraft to remove.
|
||||
count: The number of aircraft to remove.
|
||||
|
||||
Raises:
|
||||
ValueError: The control point cannot fulfill the requested number of
|
||||
aircraft.
|
||||
"""
|
||||
available = self.inventory[aircraft]
|
||||
if available < count:
|
||||
raise ValueError(
|
||||
f"Cannot remove {count} {aircraft} from "
|
||||
f"{self.control_point.name}. Only have {available}."
|
||||
)
|
||||
self.inventory[aircraft] -= count
|
||||
|
||||
def available(self, aircraft: AircraftType) -> int:
|
||||
"""Returns the number of available aircraft of the given type.
|
||||
|
||||
Args:
|
||||
aircraft: The type of aircraft to query.
|
||||
"""
|
||||
try:
|
||||
return self.inventory[aircraft]
|
||||
except KeyError:
|
||||
return 0
|
||||
|
||||
@property
|
||||
def types_available(self) -> Iterator[AircraftType]:
|
||||
"""Iterates over all available aircraft types."""
|
||||
for aircraft, count in self.inventory.items():
|
||||
if count > 0:
|
||||
yield aircraft
|
||||
|
||||
@property
|
||||
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:
|
||||
yield aircraft, count
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Clears all aircraft from the inventory."""
|
||||
self.inventory.clear()
|
||||
|
||||
|
||||
class GlobalAircraftInventory:
|
||||
"""Game-wide aircraft inventory."""
|
||||
|
||||
def __init__(self, control_points: Iterable[ControlPoint]) -> None:
|
||||
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():
|
||||
if inventory.control_point.captured == for_player:
|
||||
inventory.clear()
|
||||
|
||||
def set_from_control_point(self, control_point: ControlPoint) -> None:
|
||||
"""Set the control point's aircraft inventory.
|
||||
|
||||
If the inventory for the given control point has already been set for
|
||||
the turn, it will be overwritten.
|
||||
"""
|
||||
inventory = self.inventories[control_point]
|
||||
for aircraft, count in control_point.base.aircraft.items():
|
||||
inventory.add_aircraft(aircraft, count)
|
||||
|
||||
def for_control_point(
|
||||
self, control_point: ControlPoint
|
||||
) -> ControlPointAircraftInventory:
|
||||
"""Returns the inventory specific to the given control point."""
|
||||
return self.inventories[control_point]
|
||||
|
||||
@property
|
||||
def available_types_for_player(self) -> Iterator[AircraftType]:
|
||||
"""Iterates over all aircraft types available to the player."""
|
||||
seen: set[AircraftType] = set()
|
||||
for control_point, inventory in self.inventories.items():
|
||||
if control_point.captured:
|
||||
for aircraft in inventory.types_available:
|
||||
if not control_point.can_operate(aircraft):
|
||||
continue
|
||||
if aircraft not in seen:
|
||||
seen.add(aircraft)
|
||||
yield aircraft
|
||||
|
||||
def claim_for_flight(self, flight: Flight) -> None:
|
||||
"""Removes aircraft from the inventory for the given flight."""
|
||||
inventory = self.for_control_point(flight.from_cp)
|
||||
inventory.remove_aircraft(flight.unit_type, flight.count)
|
||||
|
||||
def return_from_flight(self, flight: Flight) -> None:
|
||||
"""Returns a flight's aircraft to the inventory."""
|
||||
inventory = self.for_control_point(flight.from_cp)
|
||||
inventory.add_aircraft(flight.unit_type, flight.count)
|
||||
@@ -56,10 +56,12 @@ class GameStats:
|
||||
|
||||
for cp in game.theater.controlpoints:
|
||||
if cp.captured:
|
||||
turn_data.allied_units.aircraft_count += sum(cp.base.aircraft.values())
|
||||
for squadron in cp.squadrons:
|
||||
turn_data.allied_units.aircraft_count += squadron.owned_aircraft
|
||||
turn_data.allied_units.vehicles_count += sum(cp.base.armor.values())
|
||||
else:
|
||||
turn_data.enemy_units.aircraft_count += sum(cp.base.aircraft.values())
|
||||
for squadron in cp.squadrons:
|
||||
turn_data.enemy_units.aircraft_count += squadron.owned_aircraft
|
||||
turn_data.enemy_units.vehicles_count += sum(cp.base.armor.values())
|
||||
|
||||
self.data_per_turn.append(turn_data)
|
||||
|
||||
@@ -7,7 +7,6 @@ from typing import Iterator, List, Optional, TYPE_CHECKING, Tuple
|
||||
|
||||
from game import db
|
||||
from game.data.groundunitclass import GroundUnitClass
|
||||
from game.dcs.aircrafttype import AircraftType
|
||||
from game.dcs.groundunittype import GroundUnitType
|
||||
from game.factions.faction import Faction
|
||||
from game.squadrons import Squadron
|
||||
@@ -98,37 +97,10 @@ class ProcurementAi:
|
||||
budget -= armor_budget
|
||||
budget += self.reinforce_front_line(armor_budget)
|
||||
|
||||
# Don't sell overstock aircraft until after we've bought runways and
|
||||
# front lines. Any budget we free up should be earmarked for aircraft.
|
||||
if not self.is_player:
|
||||
budget += self.sell_incomplete_squadrons()
|
||||
if self.manage_aircraft:
|
||||
budget = self.purchase_aircraft(budget)
|
||||
return budget
|
||||
|
||||
def sell_incomplete_squadrons(self) -> float:
|
||||
# Selling incomplete squadrons gives us more money to spend on the next
|
||||
# turn. This serves as a short term fix for
|
||||
# https://github.com/dcs-liberation/dcs_liberation/issues/41.
|
||||
#
|
||||
# Only incomplete squadrons which are unlikely to get used will be sold
|
||||
# rather than all unused aircraft because the unused aircraft are what
|
||||
# make OCA strikes worthwhile.
|
||||
#
|
||||
# This option is only used by the AI since players cannot cancel sales
|
||||
# (https://github.com/dcs-liberation/dcs_liberation/issues/365).
|
||||
total = 0.0
|
||||
for cp in self.game.theater.control_points_for(self.is_player):
|
||||
inventory = self.game.aircraft_inventory.for_control_point(cp)
|
||||
for aircraft, available in inventory.all_aircraft:
|
||||
# We only ever plan even groups, so the odd aircraft is unlikely
|
||||
# to get used.
|
||||
if available % 2 == 0:
|
||||
continue
|
||||
inventory.remove_aircraft(aircraft, 1)
|
||||
total += aircraft.price
|
||||
return total
|
||||
|
||||
def repair_runways(self, budget: float) -> float:
|
||||
for control_point in self.owned_points:
|
||||
if budget < db.RUNWAY_REPAIR_COST:
|
||||
@@ -181,7 +153,7 @@ class ProcurementAi:
|
||||
break
|
||||
|
||||
budget -= unit.price
|
||||
cp.pending_unit_deliveries.order({unit: 1})
|
||||
cp.ground_unit_orders.order({unit: 1})
|
||||
|
||||
return budget
|
||||
|
||||
@@ -211,64 +183,28 @@ class ProcurementAi:
|
||||
return worst_balanced
|
||||
|
||||
@staticmethod
|
||||
def _compatible_squadron_at(
|
||||
aircraft: AircraftType, airbase: ControlPoint, task: FlightType, count: int
|
||||
) -> Optional[Squadron]:
|
||||
for squadron in airbase.squadrons:
|
||||
if squadron.aircraft != aircraft:
|
||||
continue
|
||||
if not squadron.can_auto_assign(task):
|
||||
continue
|
||||
if not squadron.can_provide_pilots(count):
|
||||
continue
|
||||
return squadron
|
||||
return None
|
||||
|
||||
def affordable_aircraft_for(
|
||||
self, request: AircraftProcurementRequest, airbase: ControlPoint, budget: float
|
||||
) -> Optional[AircraftType]:
|
||||
for unit in aircraft_for_task(request.task_capability):
|
||||
if unit.price * request.number > budget:
|
||||
continue
|
||||
|
||||
squadron = self._compatible_squadron_at(
|
||||
unit, airbase, request.task_capability, request.number
|
||||
)
|
||||
if squadron is None:
|
||||
continue
|
||||
|
||||
distance_to_target = meters(request.near.distance_to(airbase))
|
||||
if distance_to_target > unit.max_mission_range:
|
||||
continue
|
||||
|
||||
# Affordable, compatible, and we have a squadron capable of the task.
|
||||
return unit
|
||||
return None
|
||||
|
||||
def fulfill_aircraft_request(
|
||||
self, request: AircraftProcurementRequest, budget: float
|
||||
squadrons: list[Squadron], quantity: int, budget: float
|
||||
) -> Tuple[float, bool]:
|
||||
for airbase in self.best_airbases_for(request):
|
||||
unit = self.affordable_aircraft_for(request, airbase, budget)
|
||||
if unit is None:
|
||||
# Can't afford any aircraft capable of performing the
|
||||
# required mission that can operate from this airbase. We
|
||||
# might be able to afford aircraft at other airbases though,
|
||||
# in the case where the airbase we attempted to use is only
|
||||
# able to operate expensive aircraft.
|
||||
for squadron in squadrons:
|
||||
price = squadron.aircraft.price * quantity
|
||||
if price > budget:
|
||||
continue
|
||||
|
||||
budget -= unit.price * request.number
|
||||
airbase.pending_unit_deliveries.order({unit: request.number})
|
||||
squadron.pending_deliveries += quantity
|
||||
budget -= price
|
||||
return budget, True
|
||||
return budget, False
|
||||
|
||||
def purchase_aircraft(self, budget: float) -> float:
|
||||
for request in self.game.coalition_for(self.is_player).procurement_requests:
|
||||
if not list(self.best_airbases_for(request)):
|
||||
squadrons = list(self.best_squadrons_for(request))
|
||||
if not squadrons:
|
||||
# No airbases in range of this request. Skip it.
|
||||
continue
|
||||
budget, fulfilled = self.fulfill_aircraft_request(request, budget)
|
||||
budget, fulfilled = self.fulfill_aircraft_request(
|
||||
squadrons, request.number, budget
|
||||
)
|
||||
if not fulfilled:
|
||||
# The request was not fulfilled because we could not afford any suitable
|
||||
# aircraft. Rather than continuing, which could proceed to buy tons of
|
||||
@@ -285,9 +221,32 @@ class ProcurementAi:
|
||||
else:
|
||||
return self.game.theater.enemy_points()
|
||||
|
||||
def best_airbases_for(
|
||||
@staticmethod
|
||||
def squadron_rank_for_task(squadron: Squadron, task: FlightType) -> int:
|
||||
return aircraft_for_task(task).index(squadron.aircraft)
|
||||
|
||||
def compatible_squadrons_at_airbase(
|
||||
self, airbase: ControlPoint, request: AircraftProcurementRequest
|
||||
) -> Iterator[Squadron]:
|
||||
compatible: list[Squadron] = []
|
||||
for squadron in airbase.squadrons:
|
||||
if not squadron.can_auto_assign(request.task_capability):
|
||||
continue
|
||||
if not squadron.can_provide_pilots(request.number):
|
||||
continue
|
||||
|
||||
distance_to_target = meters(request.near.distance_to(airbase))
|
||||
if distance_to_target > squadron.aircraft.max_mission_range:
|
||||
continue
|
||||
compatible.append(squadron)
|
||||
yield from sorted(
|
||||
compatible,
|
||||
key=lambda s: self.squadron_rank_for_task(s, request.task_capability),
|
||||
)
|
||||
|
||||
def best_squadrons_for(
|
||||
self, request: AircraftProcurementRequest
|
||||
) -> Iterator[ControlPoint]:
|
||||
) -> Iterator[Squadron]:
|
||||
distance_cache = ObjectiveDistanceCache.get_closest_airfields(request.near)
|
||||
threatened = []
|
||||
for cp in distance_cache.operational_airfields:
|
||||
@@ -297,8 +256,10 @@ class ProcurementAi:
|
||||
continue
|
||||
if self.threat_zones.threatened(cp.position):
|
||||
threatened.append(cp)
|
||||
yield cp
|
||||
yield from threatened
|
||||
continue
|
||||
yield from self.compatible_squadrons_at_airbase(cp, request)
|
||||
for threatened_base in threatened:
|
||||
yield from self.compatible_squadrons_at_airbase(threatened_base, request)
|
||||
|
||||
def ground_reinforcement_candidate(self) -> Optional[ControlPoint]:
|
||||
worst_supply = math.inf
|
||||
|
||||
180
game/purchaseadapter.py
Normal file
180
game/purchaseadapter.py
Normal file
@@ -0,0 +1,180 @@
|
||||
from abc import abstractmethod
|
||||
from typing import TypeVar, Generic
|
||||
|
||||
from game import Game
|
||||
from game.coalition import Coalition
|
||||
from game.dcs.groundunittype import GroundUnitType
|
||||
from game.squadrons import Squadron
|
||||
from game.theater import ControlPoint
|
||||
|
||||
ItemType = TypeVar("ItemType")
|
||||
|
||||
|
||||
class TransactionError(RuntimeError):
|
||||
def __init__(self, message: str) -> None:
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class PurchaseAdapter(Generic[ItemType]):
|
||||
def __init__(self, coalition: Coalition) -> None:
|
||||
self.coalition = coalition
|
||||
|
||||
def buy(self, item: ItemType, quantity: int) -> None:
|
||||
for _ in range(quantity):
|
||||
if self.has_pending_sales(item):
|
||||
self.do_cancel_sale(item)
|
||||
elif self.can_buy(item):
|
||||
self.do_purchase(item)
|
||||
else:
|
||||
raise TransactionError(f"Cannot buy more {item}")
|
||||
self.coalition.adjust_budget(-self.price_of(item))
|
||||
|
||||
def sell(self, item: ItemType, quantity: int) -> None:
|
||||
for _ in range(quantity):
|
||||
if self.has_pending_orders(item):
|
||||
self.do_cancel_purchase(item)
|
||||
elif self.can_sell(item):
|
||||
self.do_sale(item)
|
||||
else:
|
||||
raise TransactionError(f"Cannot sell more {item}")
|
||||
self.coalition.adjust_budget(self.price_of(item))
|
||||
|
||||
def has_pending_orders(self, item: ItemType) -> bool:
|
||||
return self.pending_delivery_quantity(item) > 0
|
||||
|
||||
def has_pending_sales(self, item: ItemType) -> bool:
|
||||
return self.pending_delivery_quantity(item) < 0
|
||||
|
||||
@abstractmethod
|
||||
def current_quantity_of(self, item: ItemType) -> int:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def pending_delivery_quantity(self, item: ItemType) -> int:
|
||||
...
|
||||
|
||||
def expected_quantity_next_turn(self, item: ItemType) -> int:
|
||||
return self.current_quantity_of(item) + self.pending_delivery_quantity(item)
|
||||
|
||||
def can_buy(self, item: ItemType) -> bool:
|
||||
return self.coalition.budget >= self.price_of(item)
|
||||
|
||||
def can_sell_or_cancel(self, item: ItemType) -> bool:
|
||||
return self.can_sell(item) or self.has_pending_orders(item)
|
||||
|
||||
@abstractmethod
|
||||
def can_sell(self, item: ItemType) -> bool:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def do_purchase(self, item: ItemType) -> None:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def do_cancel_purchase(self, item: ItemType) -> None:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def do_sale(self, item: ItemType) -> None:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def do_cancel_sale(self, item: ItemType) -> None:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def price_of(self, item: ItemType) -> int:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def name_of(self, item: ItemType, multiline: bool = False) -> str:
|
||||
...
|
||||
|
||||
|
||||
class AircraftPurchaseAdapter(PurchaseAdapter[Squadron]):
|
||||
def __init__(
|
||||
self, control_point: ControlPoint, coalition: Coalition, game: Game
|
||||
) -> None:
|
||||
super().__init__(coalition)
|
||||
self.control_point = control_point
|
||||
self.game = game
|
||||
|
||||
def pending_delivery_quantity(self, item: Squadron) -> int:
|
||||
return item.pending_deliveries
|
||||
|
||||
def current_quantity_of(self, item: Squadron) -> int:
|
||||
return item.owned_aircraft
|
||||
|
||||
def can_buy(self, item: Squadron) -> bool:
|
||||
return (
|
||||
super().can_buy(item)
|
||||
and self.control_point.unclaimed_parking(self.game) > 0
|
||||
)
|
||||
|
||||
def can_sell(self, item: Squadron) -> bool:
|
||||
return item.untasked_aircraft > 0
|
||||
|
||||
def do_purchase(self, item: Squadron) -> None:
|
||||
item.pending_deliveries += 1
|
||||
|
||||
def do_cancel_purchase(self, item: Squadron) -> None:
|
||||
item.pending_deliveries -= 1
|
||||
|
||||
def do_sale(self, item: Squadron) -> None:
|
||||
item.untasked_aircraft -= 1
|
||||
item.pending_deliveries -= 1
|
||||
|
||||
def do_cancel_sale(self, item: Squadron) -> None:
|
||||
item.untasked_aircraft += 1
|
||||
item.pending_deliveries += 1
|
||||
|
||||
def price_of(self, item: Squadron) -> int:
|
||||
return item.aircraft.price
|
||||
|
||||
def name_of(self, item: Squadron, multiline: bool = False) -> str:
|
||||
if multiline:
|
||||
separator = "<br />"
|
||||
else:
|
||||
separator = " "
|
||||
return separator.join([item.aircraft.name, str(item)])
|
||||
|
||||
|
||||
class GroundUnitPurchaseAdapter(PurchaseAdapter[GroundUnitType]):
|
||||
def __init__(
|
||||
self, control_point: ControlPoint, coalition: Coalition, game: Game
|
||||
) -> None:
|
||||
super().__init__(coalition)
|
||||
self.control_point = control_point
|
||||
self.game = game
|
||||
|
||||
def pending_delivery_quantity(self, item: GroundUnitType) -> int:
|
||||
return self.control_point.ground_unit_orders.pending_orders(item)
|
||||
|
||||
def current_quantity_of(self, item: GroundUnitType) -> int:
|
||||
return self.control_point.base.total_units_of_type(item)
|
||||
|
||||
def can_buy(self, item: GroundUnitType) -> bool:
|
||||
return super().can_buy(item) and self.control_point.has_ground_unit_source(
|
||||
self.game
|
||||
)
|
||||
|
||||
def can_sell(self, item: GroundUnitType) -> bool:
|
||||
return False
|
||||
|
||||
def do_purchase(self, item: GroundUnitType) -> None:
|
||||
self.control_point.ground_unit_orders.order({item: 1})
|
||||
|
||||
def do_cancel_purchase(self, item: GroundUnitType) -> None:
|
||||
self.control_point.ground_unit_orders.sell({item: 1})
|
||||
|
||||
def do_sale(self, item: GroundUnitType) -> None:
|
||||
raise TransactionError("Sale of ground units not allowed")
|
||||
|
||||
def do_cancel_sale(self, item: GroundUnitType) -> None:
|
||||
raise TransactionError("Sale of ground units not allowed")
|
||||
|
||||
def price_of(self, item: GroundUnitType) -> int:
|
||||
return item.price
|
||||
|
||||
def name_of(self, item: GroundUnitType, multiline: bool = False) -> str:
|
||||
return f"{item}"
|
||||
@@ -5,12 +5,12 @@ from collections import defaultdict
|
||||
from typing import Sequence, Iterator, TYPE_CHECKING
|
||||
|
||||
from game.dcs.aircrafttype import AircraftType
|
||||
from gen.flights.flight import FlightType
|
||||
from .squadron import Squadron
|
||||
from ..theater import ControlPoint
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from game import Game
|
||||
from gen.flights.flight import FlightType
|
||||
|
||||
|
||||
class AirWing:
|
||||
@@ -32,11 +32,26 @@ class AirWing:
|
||||
except StopIteration:
|
||||
return False
|
||||
|
||||
@property
|
||||
def available_aircraft_types(self) -> Iterator[AircraftType]:
|
||||
for aircraft, squadrons in self.squadrons.items():
|
||||
for squadron in squadrons:
|
||||
if squadron.untasked_aircraft:
|
||||
yield aircraft
|
||||
break
|
||||
|
||||
def auto_assignable_for_task(self, task: FlightType) -> Iterator[Squadron]:
|
||||
for squadron in self.iter_squadrons():
|
||||
if squadron.can_auto_assign(task):
|
||||
yield squadron
|
||||
|
||||
def auto_assignable_for_task_at(
|
||||
self, task: FlightType, base: ControlPoint
|
||||
) -> Iterator[Squadron]:
|
||||
for squadron in self.iter_squadrons():
|
||||
if squadron.can_auto_assign(task) and squadron.location == base:
|
||||
yield squadron
|
||||
|
||||
def auto_assignable_for_task_with_type(
|
||||
self, aircraft: AircraftType, task: FlightType, base: ControlPoint
|
||||
) -> Iterator[Squadron]:
|
||||
@@ -67,7 +82,7 @@ class AirWing:
|
||||
|
||||
def reset(self) -> None:
|
||||
for squadron in self.iter_squadrons():
|
||||
squadron.return_all_pilots()
|
||||
squadron.return_all_pilots_and_aircraft()
|
||||
|
||||
@property
|
||||
def size(self) -> int:
|
||||
|
||||
@@ -54,6 +54,10 @@ class Squadron:
|
||||
|
||||
location: ControlPoint
|
||||
|
||||
owned_aircraft: int = field(init=False, hash=False, compare=False, default=0)
|
||||
untasked_aircraft: int = field(init=False, hash=False, compare=False, default=0)
|
||||
pending_deliveries: int = field(init=False, hash=False, compare=False, default=0)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.auto_assignable_mission_types = set(self.mission_types)
|
||||
|
||||
@@ -62,6 +66,17 @@ class Squadron:
|
||||
return self.name
|
||||
return f'{self.name} "{self.nickname}"'
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(
|
||||
(
|
||||
self.name,
|
||||
self.nickname,
|
||||
self.country,
|
||||
self.role,
|
||||
self.aircraft,
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def player(self) -> bool:
|
||||
return self.coalition.player
|
||||
@@ -165,8 +180,9 @@ class Squadron:
|
||||
if replenish_count > 0:
|
||||
self._recruit_pilots(replenish_count)
|
||||
|
||||
def return_all_pilots(self) -> None:
|
||||
def return_all_pilots_and_aircraft(self) -> None:
|
||||
self.available_pilots = list(self.active_pilots)
|
||||
self.untasked_aircraft = self.owned_aircraft
|
||||
|
||||
@staticmethod
|
||||
def send_on_leave(pilot: Pilot) -> None:
|
||||
@@ -238,6 +254,29 @@ class Squadron:
|
||||
def pilot_at_index(self, index: int) -> Pilot:
|
||||
return self.current_roster[index]
|
||||
|
||||
def claim_inventory(self, count: int) -> None:
|
||||
if self.untasked_aircraft < count:
|
||||
raise ValueError(
|
||||
f"Cannot remove {count} from {self.name}. Only have "
|
||||
f"{self.untasked_aircraft}."
|
||||
)
|
||||
self.untasked_aircraft -= count
|
||||
|
||||
def can_fulfill_flight(self, count: int) -> bool:
|
||||
return self.can_provide_pilots(count) and self.untasked_aircraft >= count
|
||||
|
||||
def refund_orders(self) -> None:
|
||||
self.coalition.adjust_budget(self.aircraft.price * self.pending_deliveries)
|
||||
self.pending_deliveries = 0
|
||||
|
||||
def deliver_orders(self) -> None:
|
||||
self.owned_aircraft += self.pending_deliveries
|
||||
self.pending_deliveries = 0
|
||||
|
||||
@property
|
||||
def max_fulfillable_aircraft(self) -> int:
|
||||
return max(self.number_of_available_pilots, self.untasked_aircraft)
|
||||
|
||||
@classmethod
|
||||
def create_from(
|
||||
cls,
|
||||
|
||||
@@ -1,10 +1,4 @@
|
||||
import itertools
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from game.dcs.aircrafttype import AircraftType
|
||||
from game.dcs.groundunittype import GroundUnitType
|
||||
from game.dcs.unittype import UnitType
|
||||
|
||||
BASE_MAX_STRENGTH = 1.0
|
||||
BASE_MIN_STRENGTH = 0.0
|
||||
@@ -12,14 +6,9 @@ BASE_MIN_STRENGTH = 0.0
|
||||
|
||||
class Base:
|
||||
def __init__(self) -> None:
|
||||
self.aircraft: dict[AircraftType, int] = {}
|
||||
self.armor: dict[GroundUnitType, int] = {}
|
||||
self.strength = 1.0
|
||||
|
||||
@property
|
||||
def total_aircraft(self) -> int:
|
||||
return sum(self.aircraft.values())
|
||||
|
||||
@property
|
||||
def total_armor(self) -> int:
|
||||
return sum(self.armor.values())
|
||||
@@ -31,49 +20,24 @@ class Base:
|
||||
total += unit_type.price * count
|
||||
return total
|
||||
|
||||
def total_units_of_type(self, unit_type: UnitType[Any]) -> int:
|
||||
return sum(
|
||||
[
|
||||
c
|
||||
for t, c in itertools.chain(self.aircraft.items(), self.armor.items())
|
||||
if t == unit_type
|
||||
]
|
||||
)
|
||||
def total_units_of_type(self, unit_type: GroundUnitType) -> int:
|
||||
return sum([c for t, c in self.armor.items() if t == unit_type])
|
||||
|
||||
def commission_units(self, units: dict[Any, int]) -> None:
|
||||
def commission_units(self, units: dict[GroundUnitType, int]) -> None:
|
||||
for unit_type, unit_count in units.items():
|
||||
if unit_count <= 0:
|
||||
continue
|
||||
self.armor[unit_type] = self.armor.get(unit_type, 0) + unit_count
|
||||
|
||||
target_dict: dict[Any, int]
|
||||
if isinstance(unit_type, AircraftType):
|
||||
target_dict = self.aircraft
|
||||
elif isinstance(unit_type, GroundUnitType):
|
||||
target_dict = self.armor
|
||||
else:
|
||||
logging.error(f"Unexpected unit type of {unit_type}")
|
||||
return
|
||||
|
||||
target_dict[unit_type] = target_dict.get(unit_type, 0) + unit_count
|
||||
|
||||
def commit_losses(self, units_lost: dict[Any, int]) -> None:
|
||||
def commit_losses(self, units_lost: dict[GroundUnitType, int]) -> None:
|
||||
for unit_type, count in units_lost.items():
|
||||
target_dict: dict[Any, int]
|
||||
if unit_type in self.aircraft:
|
||||
target_dict = self.aircraft
|
||||
elif unit_type in self.armor:
|
||||
target_dict = self.armor
|
||||
else:
|
||||
print("Base didn't find event type {}".format(unit_type))
|
||||
if unit_type not in self.armor:
|
||||
print("Base didn't find unit type {}".format(unit_type))
|
||||
continue
|
||||
|
||||
if unit_type not in target_dict:
|
||||
print("Base didn't find event type {}".format(unit_type))
|
||||
continue
|
||||
|
||||
target_dict[unit_type] = max(target_dict[unit_type] - count, 0)
|
||||
if target_dict[unit_type] == 0:
|
||||
del target_dict[unit_type]
|
||||
self.armor[unit_type] = max(self.armor[unit_type] - count, 0)
|
||||
if self.armor[unit_type] == 0:
|
||||
del self.armor[unit_type]
|
||||
|
||||
def affect_strength(self, amount: float) -> None:
|
||||
self.strength += amount
|
||||
|
||||
@@ -317,9 +317,9 @@ class ControlPoint(MissionTarget, ABC):
|
||||
self.cptype = cptype
|
||||
# TODO: Should be Airbase specific.
|
||||
self.stances: Dict[int, CombatStance] = {}
|
||||
from ..unitdelivery import PendingUnitDeliveries
|
||||
from ..groundunitorders import GroundUnitOrders
|
||||
|
||||
self.pending_unit_deliveries = PendingUnitDeliveries(self)
|
||||
self.ground_unit_orders = GroundUnitOrders(self)
|
||||
|
||||
self.target_position: Optional[Point] = None
|
||||
|
||||
@@ -578,25 +578,14 @@ class ControlPoint(MissionTarget, ABC):
|
||||
return airbase
|
||||
return None
|
||||
|
||||
def _retreat_air_units(
|
||||
self, game: Game, airframe: AircraftType, count: int
|
||||
) -> None:
|
||||
while count:
|
||||
logging.debug(f"Retreating {count} {airframe} from {self.name}")
|
||||
destination = self.aircraft_retreat_destination(game, airframe)
|
||||
if destination is None:
|
||||
self.capture_aircraft(game, airframe, count)
|
||||
return
|
||||
parking = destination.unclaimed_parking(game)
|
||||
transfer_amount = min([parking, count])
|
||||
destination.base.commission_units({airframe: transfer_amount})
|
||||
count -= transfer_amount
|
||||
@staticmethod
|
||||
def _retreat_squadron(squadron: Squadron) -> None:
|
||||
logging.error("Air unit retreat not currently implemented")
|
||||
|
||||
def retreat_air_units(self, game: Game) -> None:
|
||||
# TODO: Capture in order of price to retain maximum value?
|
||||
while self.base.aircraft:
|
||||
airframe, count = self.base.aircraft.popitem()
|
||||
self._retreat_air_units(game, airframe, count)
|
||||
for squadron in self.squadrons:
|
||||
self._retreat_squadron(squadron)
|
||||
|
||||
def depopulate_uncapturable_tgos(self) -> None:
|
||||
for tgo in self.connected_objectives:
|
||||
@@ -605,7 +594,10 @@ class ControlPoint(MissionTarget, ABC):
|
||||
|
||||
# TODO: Should be Airbase specific.
|
||||
def capture(self, game: Game, for_player: bool) -> None:
|
||||
self.pending_unit_deliveries.refund_all(game.coalition_for(for_player))
|
||||
coalition = game.coalition_for(for_player)
|
||||
self.ground_unit_orders.refund_all(coalition)
|
||||
for squadron in self.squadrons:
|
||||
squadron.refund_orders()
|
||||
self.retreat_ground_units(game)
|
||||
self.retreat_air_units(game)
|
||||
self.depopulate_uncapturable_tgos()
|
||||
@@ -621,19 +613,6 @@ class ControlPoint(MissionTarget, ABC):
|
||||
def can_operate(self, aircraft: AircraftType) -> bool:
|
||||
...
|
||||
|
||||
def aircraft_transferring(self, game: Game) -> dict[AircraftType, int]:
|
||||
ato = game.coalition_for(self.captured).ato
|
||||
transferring: defaultdict[AircraftType, int] = defaultdict(int)
|
||||
for package in ato.packages:
|
||||
for flight in package.flights:
|
||||
if flight.departure == flight.arrival:
|
||||
continue
|
||||
if flight.departure == self:
|
||||
transferring[flight.unit_type] -= flight.count
|
||||
elif flight.arrival == self:
|
||||
transferring[flight.unit_type] += flight.count
|
||||
return transferring
|
||||
|
||||
def unclaimed_parking(self, game: Game) -> int:
|
||||
return self.total_aircraft_parking - self.allocated_aircraft(game).total
|
||||
|
||||
@@ -663,7 +642,9 @@ class ControlPoint(MissionTarget, ABC):
|
||||
self.runway_status.begin_repair()
|
||||
|
||||
def process_turn(self, game: Game) -> None:
|
||||
self.pending_unit_deliveries.process(game)
|
||||
self.ground_unit_orders.process(game)
|
||||
for squadron in self.squadrons:
|
||||
squadron.deliver_orders()
|
||||
|
||||
runway_status = self.runway_status
|
||||
if runway_status is not None:
|
||||
@@ -685,21 +666,22 @@ class ControlPoint(MissionTarget, ABC):
|
||||
u.position.x = u.position.x + delta.x
|
||||
u.position.y = u.position.y + delta.y
|
||||
|
||||
def allocated_aircraft(self, game: Game) -> AircraftAllocations:
|
||||
on_order = {}
|
||||
for unit_bought, count in self.pending_unit_deliveries.units.items():
|
||||
if isinstance(unit_bought, AircraftType):
|
||||
on_order[unit_bought] = count
|
||||
def allocated_aircraft(self, _game: Game) -> AircraftAllocations:
|
||||
present: dict[AircraftType, int] = defaultdict(int)
|
||||
on_order: dict[AircraftType, int] = defaultdict(int)
|
||||
for squadron in self.squadrons:
|
||||
present[squadron.aircraft] += squadron.owned_aircraft
|
||||
# TODO: Only if this is the squadron destination, not location.
|
||||
on_order[squadron.aircraft] += squadron.pending_deliveries
|
||||
|
||||
return AircraftAllocations(
|
||||
self.base.aircraft, on_order, self.aircraft_transferring(game)
|
||||
)
|
||||
# TODO: Implement squadron transfers.
|
||||
return AircraftAllocations(present, on_order, transferring={})
|
||||
|
||||
def allocated_ground_units(
|
||||
self, transfers: PendingTransfers
|
||||
) -> GroundUnitAllocations:
|
||||
on_order = {}
|
||||
for unit_bought, count in self.pending_unit_deliveries.units.items():
|
||||
for unit_bought, count in self.ground_unit_orders.units.items():
|
||||
if isinstance(unit_bought, GroundUnitType):
|
||||
on_order[unit_bought] = count
|
||||
|
||||
|
||||
@@ -66,7 +66,6 @@ from gen.naming import namegen
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from game import Game
|
||||
from game.inventory import ControlPointAircraftInventory
|
||||
from game.squadrons import Squadron
|
||||
|
||||
|
||||
@@ -315,29 +314,20 @@ class AirliftPlanner:
|
||||
if cp.captured != self.for_player:
|
||||
continue
|
||||
|
||||
inventory = self.game.aircraft_inventory.for_control_point(cp)
|
||||
for unit_type, available in inventory.all_aircraft:
|
||||
squadrons = air_wing.auto_assignable_for_task_with_type(
|
||||
unit_type, FlightType.TRANSPORT, cp
|
||||
)
|
||||
for squadron in squadrons:
|
||||
if self.compatible_with_mission(unit_type, cp):
|
||||
while (
|
||||
available
|
||||
and squadron.has_available_pilots
|
||||
and self.transfer.transport is None
|
||||
):
|
||||
flight_size = self.create_airlift_flight(
|
||||
squadron, inventory
|
||||
)
|
||||
available -= flight_size
|
||||
squadrons = air_wing.auto_assignable_for_task_at(FlightType.TRANSPORT, cp)
|
||||
for squadron in squadrons:
|
||||
if self.compatible_with_mission(squadron.aircraft, cp):
|
||||
while (
|
||||
squadron.untasked_aircraft
|
||||
and squadron.has_available_pilots
|
||||
and self.transfer.transport is None
|
||||
):
|
||||
self.create_airlift_flight(squadron)
|
||||
if self.package.flights:
|
||||
self.game.ato_for(self.for_player).add_package(self.package)
|
||||
|
||||
def create_airlift_flight(
|
||||
self, squadron: Squadron, inventory: ControlPointAircraftInventory
|
||||
) -> int:
|
||||
available_aircraft = inventory.available(squadron.aircraft)
|
||||
def create_airlift_flight(self, squadron: Squadron) -> int:
|
||||
available_aircraft = squadron.untasked_aircraft
|
||||
capacity_each = 1 if squadron.aircraft.dcs_unit_type.helicopter else 2
|
||||
required = math.ceil(self.transfer.size / capacity_each)
|
||||
flight_size = min(
|
||||
@@ -348,8 +338,8 @@ class AirliftPlanner:
|
||||
# TODO: Use number_of_available_pilots directly once feature flag is gone.
|
||||
# The number of currently available pilots is not relevant when pilot limits
|
||||
# are disabled.
|
||||
if not squadron.can_provide_pilots(flight_size):
|
||||
flight_size = squadron.number_of_available_pilots
|
||||
if not squadron.can_fulfill_flight(flight_size):
|
||||
flight_size = squadron.max_fulfillable_aircraft
|
||||
capacity = flight_size * capacity_each
|
||||
|
||||
if capacity < self.transfer.size:
|
||||
@@ -359,16 +349,15 @@ class AirliftPlanner:
|
||||
else:
|
||||
transfer = self.transfer
|
||||
|
||||
player = inventory.control_point.captured
|
||||
flight = Flight(
|
||||
self.package,
|
||||
self.game.country_for(player),
|
||||
self.game.country_for(squadron.player),
|
||||
squadron,
|
||||
flight_size,
|
||||
FlightType.TRANSPORT,
|
||||
self.game.settings.default_start_type,
|
||||
departure=inventory.control_point,
|
||||
arrival=inventory.control_point,
|
||||
departure=squadron.location,
|
||||
arrival=squadron.location,
|
||||
divert=None,
|
||||
cargo=transfer,
|
||||
)
|
||||
@@ -381,7 +370,6 @@ class AirliftPlanner:
|
||||
self.package, self.game.coalition_for(self.for_player), self.game.theater
|
||||
)
|
||||
planner.populate_flight_plan(flight)
|
||||
self.game.aircraft_inventory.claim_for_flight(flight)
|
||||
return flight_size
|
||||
|
||||
|
||||
@@ -652,8 +640,7 @@ class PendingTransfers:
|
||||
flight.package.remove_flight(flight)
|
||||
if not flight.package.flights:
|
||||
self.game.ato_for(self.player).remove_package(flight.package)
|
||||
self.game.aircraft_inventory.return_from_flight(flight)
|
||||
flight.clear_roster()
|
||||
flight.return_pilots_and_aircraft()
|
||||
|
||||
@cancel_transport.register
|
||||
def _cancel_transport_convoy(
|
||||
@@ -756,16 +743,12 @@ class PendingTransfers:
|
||||
|
||||
return 0
|
||||
|
||||
def current_airlift_capacity(self, control_point: ControlPoint) -> int:
|
||||
inventory = self.game.aircraft_inventory.for_control_point(control_point)
|
||||
squadrons = self.game.air_wing_for(
|
||||
control_point.captured
|
||||
).auto_assignable_for_task(FlightType.TRANSPORT)
|
||||
unit_types = {s.aircraft for s in squadrons}
|
||||
@staticmethod
|
||||
def current_airlift_capacity(control_point: ControlPoint) -> int:
|
||||
return sum(
|
||||
count
|
||||
for unit_type, count in inventory.all_aircraft
|
||||
if unit_type in unit_types
|
||||
s.owned_aircraft
|
||||
for s in control_point.squadrons
|
||||
if s.can_auto_assign(FlightType.TRANSPORT)
|
||||
)
|
||||
|
||||
def order_airlift_assets_at(self, control_point: ControlPoint) -> None:
|
||||
|
||||
Reference in New Issue
Block a user