mirror of
https://github.com/dcs-retribution/dcs-retribution.git
synced 2025-11-10 15:41:24 +00:00
422 lines
16 KiB
Python
422 lines
16 KiB
Python
"""Generators for creating the groups for ground objectives.
|
|
|
|
The classes in this file are responsible for creating the vehicle groups, ship
|
|
groups, statics, missile sites, and AA sites for the mission. Each of these
|
|
objectives is defined in the Theater by a TheaterGroundObject. These classes
|
|
create the pydcs groups and statics for those areas and add them to the mission.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import random
|
|
from typing import Dict, Iterator, Optional, TYPE_CHECKING
|
|
|
|
from dcs import Mission
|
|
from dcs.country import Country
|
|
from dcs.statics import fortification_map, warehouse_map
|
|
from dcs.task import (
|
|
ActivateBeaconCommand,
|
|
ActivateICLSCommand,
|
|
EPLRS,
|
|
OptAlarmState,
|
|
)
|
|
from dcs.unit import Ship, Vehicle, Unit
|
|
from dcs.unitgroup import Group, ShipGroup, StaticGroup
|
|
from dcs.unittype import StaticType, UnitType
|
|
|
|
from game import db
|
|
from game.data.building_data import FORTIFICATION_UNITS, FORTIFICATION_UNITS_ID
|
|
from game.db import unit_type_from_name
|
|
from theater import ControlPoint, TheaterGroundObject
|
|
from theater.theatergroundobject import (
|
|
BuildingGroundObject, CarrierGroundObject,
|
|
GenericCarrierGroundObject,
|
|
LhaGroundObject, ShipGroundObject,
|
|
)
|
|
from .conflictgen import Conflict
|
|
from .radios import RadioFrequency, RadioRegistry
|
|
from .runways import RunwayData
|
|
from .tacan import TacanBand, TacanChannel, TacanRegistry
|
|
|
|
if TYPE_CHECKING:
|
|
from game import Game
|
|
|
|
|
|
FARP_FRONTLINE_DISTANCE = 10000
|
|
AA_CP_MIN_DISTANCE = 40000
|
|
|
|
|
|
class GenericGroundObjectGenerator:
|
|
"""An unspecialized ground object generator.
|
|
|
|
Currently used only for SAM and missile (V1/V2) sites.
|
|
"""
|
|
def __init__(self, ground_object: TheaterGroundObject, country: Country,
|
|
game: Game, mission: Mission) -> None:
|
|
self.ground_object = ground_object
|
|
self.country = country
|
|
self.game = game
|
|
self.m = mission
|
|
|
|
def generate(self) -> None:
|
|
if self.game.position_culled(self.ground_object.position):
|
|
return
|
|
|
|
for group in self.ground_object.groups:
|
|
if not group.units:
|
|
logging.warning(f"Found empty group in {self.ground_object}")
|
|
continue
|
|
|
|
unit_type = unit_type_from_name(group.units[0].type)
|
|
if unit_type is None:
|
|
raise RuntimeError(
|
|
f"Unrecognized unit type: {group.units[0].type}")
|
|
|
|
vg = self.m.vehicle_group(self.country, group.name, unit_type,
|
|
position=group.position,
|
|
heading=group.units[0].heading)
|
|
vg.units[0].name = self.m.string(group.units[0].name)
|
|
vg.units[0].player_can_drive = True
|
|
for i, u in enumerate(group.units):
|
|
if i > 0:
|
|
vehicle = Vehicle(self.m.next_unit_id(),
|
|
self.m.string(u.name), u.type)
|
|
vehicle.position.x = u.position.x
|
|
vehicle.position.y = u.position.y
|
|
vehicle.heading = u.heading
|
|
vehicle.player_can_drive = True
|
|
vg.add_unit(vehicle)
|
|
|
|
self.enable_eplrs(vg, unit_type)
|
|
self.set_alarm_state(vg)
|
|
|
|
@staticmethod
|
|
def enable_eplrs(group: Group, unit_type: UnitType) -> None:
|
|
if hasattr(unit_type, 'eplrs'):
|
|
if unit_type.eplrs:
|
|
group.points[0].tasks.append(EPLRS(group.id))
|
|
|
|
def set_alarm_state(self, group: Group) -> None:
|
|
if self.game.settings.perf_red_alert_state:
|
|
group.points[0].tasks.append(OptAlarmState(2))
|
|
else:
|
|
group.points[0].tasks.append(OptAlarmState(1))
|
|
|
|
|
|
class BuildingSiteGenerator(GenericGroundObjectGenerator):
|
|
"""Generator for building sites.
|
|
|
|
Building sites are the primary type of non-airbase objective locations that
|
|
appear on the map. They come in a handful of variants each with different
|
|
types of buildings and ground units.
|
|
"""
|
|
|
|
def generate(self) -> None:
|
|
if self.game.position_culled(self.ground_object.position):
|
|
return
|
|
|
|
if self.ground_object.dcs_identifier in warehouse_map:
|
|
static_type = warehouse_map[self.ground_object.dcs_identifier]
|
|
self.generate_static(static_type)
|
|
elif self.ground_object.dcs_identifier in fortification_map:
|
|
static_type = fortification_map[self.ground_object.dcs_identifier]
|
|
self.generate_static(static_type)
|
|
elif self.ground_object.dcs_identifier in FORTIFICATION_UNITS_ID:
|
|
for f in FORTIFICATION_UNITS:
|
|
if f.id == self.ground_object.dcs_identifier:
|
|
unit_type = f
|
|
self.generate_vehicle_group(unit_type)
|
|
break
|
|
else:
|
|
logging.error(
|
|
f"{self.ground_object.dcs_identifier} not found in static maps")
|
|
|
|
def generate_vehicle_group(self, unit_type: UnitType) -> None:
|
|
if not self.ground_object.is_dead:
|
|
self.m.vehicle_group(
|
|
country=self.country,
|
|
name=self.ground_object.group_name,
|
|
_type=unit_type,
|
|
position=self.ground_object.position,
|
|
heading=self.ground_object.heading,
|
|
)
|
|
|
|
def generate_static(self, static_type: StaticType) -> None:
|
|
self.m.static_group(
|
|
country=self.country,
|
|
name=self.ground_object.group_name,
|
|
_type=static_type,
|
|
position=self.ground_object.position,
|
|
heading=self.ground_object.heading,
|
|
dead=self.ground_object.is_dead,
|
|
)
|
|
|
|
|
|
class GenericCarrierGenerator(GenericGroundObjectGenerator):
|
|
"""Base type for carrier group generation.
|
|
|
|
Used by both CV(N) groups and LHA groups.
|
|
"""
|
|
def __init__(self, ground_object: GenericCarrierGroundObject,
|
|
control_point: ControlPoint, country: Country, game: Game,
|
|
mission: Mission, radio_registry: RadioRegistry,
|
|
tacan_registry: TacanRegistry, icls_alloc: Iterator[int],
|
|
runways: Dict[str, RunwayData]) -> None:
|
|
super().__init__(ground_object, country, game, mission)
|
|
self.ground_object = ground_object
|
|
self.control_point = control_point
|
|
self.radio_registry = radio_registry
|
|
self.tacan_registry = tacan_registry
|
|
self.icls_alloc = icls_alloc
|
|
self.runways = runways
|
|
|
|
def generate(self) -> None:
|
|
# TODO: Require single group?
|
|
for group in self.ground_object.groups:
|
|
if not group.units:
|
|
logging.warning(
|
|
f"Found empty carrier group in {self.control_point}")
|
|
continue
|
|
|
|
atc = self.radio_registry.alloc_uhf()
|
|
ship_group = self.configure_carrier(group, atc)
|
|
for unit in group.units[1:]:
|
|
ship_group.add_unit(self.create_ship(unit, atc))
|
|
|
|
tacan = self.tacan_registry.alloc_for_band(TacanBand.X)
|
|
tacan_callsign = self.tacan_callsign()
|
|
icls = next(self.icls_alloc)
|
|
|
|
brc = self.steam_into_wind(ship_group)
|
|
self.activate_beacons(ship_group, tacan, tacan_callsign, icls)
|
|
self.add_runway_data(brc or 0, atc, tacan, tacan_callsign, icls)
|
|
|
|
def get_carrier_type(self, group: Group) -> UnitType:
|
|
unit_type = unit_type_from_name(group.units[0].type)
|
|
if unit_type is None:
|
|
raise RuntimeError(
|
|
f"Unrecognized carrier name: {group.units[0].type}")
|
|
return unit_type
|
|
|
|
def configure_carrier(self, group: Group,
|
|
atc_channel: RadioFrequency) -> ShipGroup:
|
|
unit_type = self.get_carrier_type(group)
|
|
|
|
ship_group = self.m.ship_group(self.country, group.name, unit_type,
|
|
position=group.position,
|
|
heading=group.units[0].heading)
|
|
ship_group.set_frequency(atc_channel.hertz)
|
|
ship_group.units[0].name = self.m.string(group.units[0].name)
|
|
return ship_group
|
|
|
|
def create_ship(self, unit: Unit, atc_channel: RadioFrequency) -> Ship:
|
|
ship = Ship(self.m.next_unit_id(),
|
|
self.m.string(unit.name),
|
|
unit_type_from_name(unit.type))
|
|
ship.position.x = unit.position.x
|
|
ship.position.y = unit.position.y
|
|
ship.heading = unit.heading
|
|
# TODO: Verify.
|
|
ship.set_frequency(atc_channel.hertz)
|
|
return ship
|
|
|
|
def steam_into_wind(self, group: ShipGroup) -> Optional[int]:
|
|
brc = self.m.weather.wind_at_ground.direction + 180
|
|
for attempt in range(5):
|
|
point = group.points[0].position.point_from_heading(
|
|
brc, 100000 - attempt * 20000)
|
|
if self.game.theater.is_in_sea(point):
|
|
group.add_waypoint(point)
|
|
return brc
|
|
return None
|
|
|
|
def tacan_callsign(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
@staticmethod
|
|
def activate_beacons(group: ShipGroup, tacan: TacanChannel,
|
|
callsign: str, icls: int) -> None:
|
|
group.points[0].tasks.append(ActivateBeaconCommand(
|
|
channel=tacan.number,
|
|
modechannel=tacan.band.value,
|
|
callsign=callsign,
|
|
unit_id=group.units[0].id,
|
|
aa=False
|
|
))
|
|
group.points[0].tasks.append(ActivateICLSCommand(
|
|
icls, unit_id=group.units[0].id
|
|
))
|
|
|
|
def add_runway_data(self, brc: int, atc: RadioFrequency,
|
|
tacan: TacanChannel, callsign: str, icls: int) -> None:
|
|
# TODO: Make unit name usable.
|
|
# This relies on one control point mapping exactly
|
|
# to one LHA, carrier, or other usable "runway".
|
|
# This isn't wholly true, since the DD escorts of
|
|
# the carrier group are valid for helicopters, but
|
|
# they aren't exposed as such to the game. Should
|
|
# clean this up so that's possible. We can't use the
|
|
# unit name since it's an arbitrary ID.
|
|
self.runways[self.control_point.name] = RunwayData(
|
|
self.control_point.name,
|
|
brc,
|
|
"N/A",
|
|
atc=atc,
|
|
tacan=tacan,
|
|
tacan_callsign=callsign,
|
|
icls=icls,
|
|
)
|
|
|
|
|
|
class CarrierGenerator(GenericCarrierGenerator):
|
|
"""Generator for CV(N) groups."""
|
|
|
|
def get_carrier_type(self, group: Group) -> UnitType:
|
|
unit_type = super().get_carrier_type(group)
|
|
if self.game.settings.supercarrier:
|
|
unit_type = db.upgrade_to_supercarrier(unit_type,
|
|
self.control_point.name)
|
|
return unit_type
|
|
|
|
def tacan_callsign(self) -> str:
|
|
# TODO: Assign these properly.
|
|
return random.choice([
|
|
"STE",
|
|
"CVN",
|
|
"CVH",
|
|
"CCV",
|
|
"ACC",
|
|
"ARC",
|
|
"GER",
|
|
"ABR",
|
|
"LIN",
|
|
"TRU",
|
|
])
|
|
|
|
|
|
class LhaGenerator(GenericCarrierGenerator):
|
|
"""Generator for LHA groups."""
|
|
|
|
def tacan_callsign(self) -> str:
|
|
# TODO: Assign these properly.
|
|
return random.choice([
|
|
"LHD",
|
|
"LHA",
|
|
"LHB",
|
|
"LHC",
|
|
"LHD",
|
|
"LDS",
|
|
])
|
|
|
|
|
|
class ShipObjectGenerator(GenericGroundObjectGenerator):
|
|
"""Generator for non-carrier naval groups."""
|
|
|
|
def generate(self) -> None:
|
|
if self.game.position_culled(self.ground_object.position):
|
|
return
|
|
|
|
for group in self.ground_object.groups:
|
|
if not group.units:
|
|
logging.warning(f"Found empty group in {self.ground_object}")
|
|
continue
|
|
|
|
unit_type = unit_type_from_name(group.units[0].type)
|
|
if unit_type is None:
|
|
raise RuntimeError(
|
|
f"Unrecognized unit type: {group.units[0].type}")
|
|
|
|
self.generate_group(group, unit_type)
|
|
|
|
def generate_group(self, group_def: Group, unit_type: UnitType):
|
|
group = self.m.ship_group(self.country, group_def.name, unit_type,
|
|
position=group_def.position,
|
|
heading=group_def.units[0].heading)
|
|
group.units[0].name = self.m.string(group_def.units[0].name)
|
|
# TODO: Skipping the first unit looks like copy pasta from the carrier.
|
|
for unit in group_def.units[1:]:
|
|
unit_type = unit_type_from_name(unit.type)
|
|
ship = Ship(self.m.next_unit_id(),
|
|
self.m.string(unit.name), unit_type)
|
|
ship.position.x = unit.position.x
|
|
ship.position.y = unit.position.y
|
|
ship.heading = unit.heading
|
|
group.add_unit(ship)
|
|
self.set_alarm_state(group)
|
|
|
|
|
|
class GroundObjectsGenerator:
|
|
"""Creates DCS groups and statics for the theater during mission generation.
|
|
|
|
Most of the work of group/static generation is delegated to the other
|
|
generator classes. This class is responsible for finding each of the
|
|
locations for spawning ground objects, determining their types, and creating
|
|
the appropriate generators.
|
|
"""
|
|
FARP_CAPACITY = 4
|
|
|
|
def __init__(self, mission: Mission, conflict: Conflict, game,
|
|
radio_registry: RadioRegistry, tacan_registry: TacanRegistry):
|
|
self.m = mission
|
|
self.conflict = conflict
|
|
self.game = game
|
|
self.radio_registry = radio_registry
|
|
self.tacan_registry = tacan_registry
|
|
self.icls_alloc = iter(range(1, 21))
|
|
self.runways: Dict[str, RunwayData] = {}
|
|
|
|
def generate_farps(self, number_of_units=1) -> Iterator[StaticGroup]:
|
|
if self.conflict.is_vector:
|
|
center = self.conflict.center
|
|
heading = self.conflict.heading - 90
|
|
else:
|
|
center, heading = self.conflict.frontline_position(self.conflict.from_cp, self.conflict.to_cp)
|
|
heading -= 90
|
|
|
|
initial_position = center.point_from_heading(heading, FARP_FRONTLINE_DISTANCE)
|
|
position = self.conflict.find_ground_position(initial_position, heading)
|
|
if not position:
|
|
position = initial_position
|
|
|
|
for i, _ in enumerate(range(0, number_of_units, self.FARP_CAPACITY)):
|
|
position = position.point_from_heading(0, i * 275)
|
|
|
|
yield self.m.farp(
|
|
country=self.m.country(self.game.player_country),
|
|
name="FARP",
|
|
position=position,
|
|
)
|
|
|
|
def generate(self):
|
|
for cp in self.game.theater.controlpoints:
|
|
if cp.captured:
|
|
country_name = self.game.player_country
|
|
else:
|
|
country_name = self.game.enemy_country
|
|
country = self.m.country(country_name)
|
|
|
|
for ground_object in cp.ground_objects:
|
|
if isinstance(ground_object, BuildingGroundObject):
|
|
generator = BuildingSiteGenerator(ground_object, country,
|
|
self.game, self.m)
|
|
elif isinstance(ground_object, CarrierGroundObject):
|
|
generator = CarrierGenerator(ground_object, cp, country,
|
|
self.game, self.m,
|
|
self.radio_registry,
|
|
self.tacan_registry,
|
|
self.icls_alloc, self.runways)
|
|
elif isinstance(ground_object, LhaGroundObject):
|
|
generator = CarrierGenerator(ground_object, cp, country,
|
|
self.game, self.m,
|
|
self.radio_registry,
|
|
self.tacan_registry,
|
|
self.icls_alloc, self.runways)
|
|
elif isinstance(ground_object, ShipGroundObject):
|
|
generator = ShipObjectGenerator(ground_object, country,
|
|
self.game, self.m)
|
|
else:
|
|
generator = GenericGroundObjectGenerator(ground_object,
|
|
country, self.game,
|
|
self.m)
|
|
generator.generate()
|