weyne/dcs-retribution
1
0
Fork 0
mirror of https://github.com/dcs-retribution/dcs-retribution.git synced 2025-11-10 15:41:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move mission generation code into game.

Browse Source 
Operation has been renamed MissionGenerator and is no longer a static
class.
This commit is contained in:
Dan Albert
2021-10-22 13:31:43 -07:00
parent b0787d9a3f
commit 74291271e3
43 changed files with 805 additions and 774 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1922
gen/aircraft.py
View File

File diff suppressed because it is too large Load Diff

4
gen/airfields.py
View File

@@ -8,8 +8,8 @@ from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple
from .radios import MHz, RadioFrequency
from .tacan import TacanBand, TacanChannel
from game.radio.radios import MHz, RadioFrequency
from game.radio.tacan import TacanBand, TacanChannel
@dataclass

56
gen/airsupport.py
View File

@@ -1,56 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Optional, TYPE_CHECKING
if TYPE_CHECKING:
from gen.radios import RadioFrequency
from gen.tacan import TacanChannel
@dataclass
class AwacsInfo:
"""AWACS information for the kneeboard."""
group_name: str
callsign: str
freq: RadioFrequency
depature_location: Optional[str]
start_time: Optional[timedelta]
end_time: Optional[timedelta]
blue: bool
@dataclass
class TankerInfo:
"""Tanker information for the kneeboard."""
group_name: str
callsign: str
variant: str
freq: RadioFrequency
tacan: TacanChannel
start_time: Optional[timedelta]
end_time: Optional[timedelta]
blue: bool
@dataclass(frozen=True)
class JtacInfo:
"""JTAC information."""
group_name: str
unit_name: str
callsign: str
region: str
code: str
blue: bool
freq: RadioFrequency
@dataclass
class AirSupport:
awacs: list[AwacsInfo] = field(default_factory=list)
tankers: list[TankerInfo] = field(default_factory=list)
jtacs: list[JtacInfo] = field(default_factory=list)

210
gen/airsupportgen.py
View File

@@ -1,210 +0,0 @@
from __future__ import annotations
import logging
from typing import List, Type, Tuple, TYPE_CHECKING
from dcs.mission import Mission, StartType
from dcs.planes import IL_78M, KC130, KC135MPRS, KC_135, PlaneType
from dcs.task import (
AWACS,
ActivateBeaconCommand,
MainTask,
Refueling,
SetImmortalCommand,
SetInvisibleCommand,
)
from dcs.unittype import UnitType
from game.utils import Heading
from .airsupport import AirSupport, TankerInfo, AwacsInfo
from .callsigns import callsign_for_support_unit
from .conflictgen import Conflict
from .flights.ai_flight_planner_db import AEWC_CAPABLE
from .naming import namegen
from .radios import RadioRegistry
from .tacan import TacanBand, TacanRegistry, TacanUsage
if TYPE_CHECKING:
from game import Game
TANKER_DISTANCE = 15000
TANKER_ALT = 4572
TANKER_HEADING_OFFSET = 45
AWACS_DISTANCE = 150000
AWACS_ALT = 13000
class AirSupportConflictGenerator:
def __init__(
self,
mission: Mission,
conflict: Conflict,
game: Game,
radio_registry: RadioRegistry,
tacan_registry: TacanRegistry,
air_support: AirSupport,
) -> None:
self.mission = mission
self.conflict = conflict
self.game = game
self.radio_registry = radio_registry
self.tacan_registry = tacan_registry
self.air_support = air_support
@classmethod
def support_tasks(cls) -> List[Type[MainTask]]:
return [Refueling, AWACS]
@staticmethod
def _get_tanker_params(unit_type: Type[UnitType]) -> Tuple[int, int]:
if unit_type is KC130:
return TANKER_ALT - 500, 596
elif unit_type is KC_135:
return TANKER_ALT, 770
elif unit_type is KC135MPRS:
return TANKER_ALT + 500, 596
return TANKER_ALT, 574
def generate(self) -> None:
player_cp = (
self.conflict.blue_cp
if self.conflict.blue_cp.captured
else self.conflict.red_cp
)
country = self.mission.country(self.game.blue.country_name)
if not self.game.settings.disable_legacy_tanker:
fallback_tanker_number = 0
for i, tanker_unit_type in enumerate(
self.game.faction_for(player=True).tankers
):
unit_type = tanker_unit_type.dcs_unit_type
if not issubclass(unit_type, PlaneType):
logging.warning(f"Refueling aircraft {unit_type} must be a plane")
continue
# TODO: Make loiter altitude a property of the unit type.
alt, airspeed = self._get_tanker_params(tanker_unit_type.dcs_unit_type)
freq = self.radio_registry.alloc_uhf()
tacan = self.tacan_registry.alloc_for_band(
TacanBand.Y, TacanUsage.AirToAir
)
tanker_heading = Heading.from_degrees(
self.conflict.red_cp.position.heading_between_point(
self.conflict.blue_cp.position
)
+ TANKER_HEADING_OFFSET * i
)
tanker_position = player_cp.position.point_from_heading(
tanker_heading.degrees, TANKER_DISTANCE
)
tanker_group = self.mission.refuel_flight(
country=country,
name=namegen.next_tanker_name(country, tanker_unit_type),
airport=None,
plane_type=unit_type,
position=tanker_position,
altitude=alt,
race_distance=58000,
frequency=freq.mhz,
start_type=StartType.Warm,
speed=airspeed,
tacanchannel=str(tacan),
)
tanker_group.set_frequency(freq.mhz)
callsign = callsign_for_support_unit(tanker_group)
tacan_callsign = {
"Texaco": "TEX",
"Arco": "ARC",
"Shell": "SHL",
}.get(callsign)
if tacan_callsign is None:
# The dict above is all the callsigns currently in the game, but
# non-Western countries don't use the callsigns and instead just
# use numbers. It's possible that none of those nations have
# TACAN compatible refueling aircraft, but fallback just in
# case.
tacan_callsign = f"TK{fallback_tanker_number}"
fallback_tanker_number += 1
if tanker_unit_type != IL_78M:
# Override PyDCS tacan channel.
tanker_group.points[0].tasks.pop()
tanker_group.points[0].tasks.append(
ActivateBeaconCommand(
tacan.number,
tacan.band.value,
tacan_callsign,
True,
tanker_group.units[0].id,
True,
)
)
tanker_group.points[0].tasks.append(SetInvisibleCommand(True))
tanker_group.points[0].tasks.append(SetImmortalCommand(True))
self.air_support.tankers.append(
TankerInfo(
str(tanker_group.name),
callsign,
tanker_unit_type.name,
freq,
tacan,
start_time=None,
end_time=None,
blue=True,
)
)
if not self.game.settings.disable_legacy_aewc:
possible_awacs = [
a
for a in self.game.faction_for(player=True).aircrafts
if a in AEWC_CAPABLE
]
if not possible_awacs:
logging.warning("No AWACS for faction")
return
awacs_unit = possible_awacs[0]
freq = self.radio_registry.alloc_uhf()
unit_type = awacs_unit.dcs_unit_type
if not issubclass(unit_type, PlaneType):
logging.warning(f"AWACS aircraft {unit_type} must be a plane")
return
awacs_flight = self.mission.awacs_flight(
country=country,
name=namegen.next_awacs_name(country),
plane_type=unit_type,
altitude=AWACS_ALT,
airport=None,
position=self.conflict.position.random_point_within(
AWACS_DISTANCE, AWACS_DISTANCE
),
frequency=freq.mhz,
start_type=StartType.Warm,
)
awacs_flight.set_frequency(freq.mhz)
awacs_flight.points[0].tasks.append(SetInvisibleCommand(True))
awacs_flight.points[0].tasks.append(SetImmortalCommand(True))
self.air_support.awacs.append(
AwacsInfo(
group_name=str(awacs_flight.name),
callsign=callsign_for_support_unit(awacs_flight),
freq=freq,
depature_location=None,
start_time=None,
end_time=None,
blue=True,
)
)

802
gen/armor.py
View File

@@ -1,802 +0,0 @@
from __future__ import annotations
import logging
import math
import random
from dataclasses import dataclass
from typing import TYPE_CHECKING, List, Optional, Tuple
from dcs import Mission
from dcs.action import AITaskPush
from dcs.condition import GroupLifeLess, Or, TimeAfter, UnitDamaged
from dcs.country import Country
from dcs.mapping import Point
from dcs.point import PointAction
from dcs.task import (
AFAC,
EPLRS,
AttackGroup,
ControlledTask,
FAC,
FireAtPoint,
GoToWaypoint,
Hold,
OrbitAction,
SetImmortalCommand,
SetInvisibleCommand,
)
from dcs.triggers import Event, TriggerOnce
from dcs.unit import Vehicle, Skill
from dcs.unitgroup import VehicleGroup
from game.data.groundunitclass import GroundUnitClass
from game.dcs.aircrafttype import AircraftType
from game.dcs.groundunittype import GroundUnitType
from game.theater.controlpoint import ControlPoint
from game.unitmap import UnitMap
from game.utils import Heading
from gen.ground_forces.ai_ground_planner import (
DISTANCE_FROM_FRONTLINE,
CombatGroup,
CombatGroupRole,
)
from .airsupport import AirSupport, JtacInfo
from .callsigns import callsign_for_support_unit
from .conflictgen import Conflict
from .ground_forces.combat_stance import CombatStance
from .lasercoderegistry import LaserCodeRegistry
from .naming import namegen
from .radios import MHz, RadioFrequency, RadioRegistry
if TYPE_CHECKING:
from game import Game
SPREAD_DISTANCE_FACTOR = 0.1, 0.3
SPREAD_DISTANCE_SIZE_FACTOR = 0.1
FRONTLINE_CAS_FIGHTS_COUNT = 16, 24
FRONTLINE_CAS_GROUP_MIN = 1, 2
FRONTLINE_CAS_PADDING = 12000
RETREAT_DISTANCE = 20000
BREAKTHROUGH_OFFENSIVE_DISTANCE = 35000
AGGRESIVE_MOVE_DISTANCE = 16000
FIGHT_DISTANCE = 3500
RANDOM_OFFSET_ATTACK = 250
INFANTRY_GROUP_SIZE = 5
class GroundConflictGenerator:
def __init__(
self,
mission: Mission,
conflict: Conflict,
game: Game,
player_planned_combat_groups: List[CombatGroup],
enemy_planned_combat_groups: List[CombatGroup],
player_stance: CombatStance,
enemy_stance: CombatStance,
unit_map: UnitMap,
radio_registry: RadioRegistry,
air_support: AirSupport,
laser_code_registry: LaserCodeRegistry,
) -> None:
self.mission = mission
self.conflict = conflict
self.enemy_planned_combat_groups = enemy_planned_combat_groups
self.player_planned_combat_groups = player_planned_combat_groups
self.player_stance = player_stance
self.enemy_stance = enemy_stance
self.game = game
self.unit_map = unit_map
self.radio_registry = radio_registry
self.air_support = air_support
self.laser_code_registry = laser_code_registry
def generate(self) -> None:
position = Conflict.frontline_position(
self.conflict.front_line, self.game.theater
)
frontline_vector = Conflict.frontline_vector(
self.conflict.front_line, self.game.theater
)
# Create player groups at random position
player_groups = self._generate_groups(
self.player_planned_combat_groups, frontline_vector, True
)
# Create enemy groups at random position
enemy_groups = self._generate_groups(
self.enemy_planned_combat_groups, frontline_vector, False
)
# TODO: Differentiate AirConflict and GroundConflict classes.
if self.conflict.heading is None:
raise RuntimeError(
"Cannot generate ground units for non-ground conflict. Ground unit "
"conflicts cannot have the heading `None`."
)
# Plan combat actions for groups
self.plan_action_for_groups(
self.player_stance,
player_groups,
enemy_groups,
self.conflict.heading.right,
self.conflict.blue_cp,
self.conflict.red_cp,
)
self.plan_action_for_groups(
self.enemy_stance,
enemy_groups,
player_groups,
self.conflict.heading.left,
self.conflict.red_cp,
self.conflict.blue_cp,
)
# Add JTAC
if self.game.blue.faction.has_jtac:
n = "JTAC" + str(self.conflict.blue_cp.id) + str(self.conflict.red_cp.id)
code: int
freq = self.radio_registry.alloc_uhf()
# If the option fc3LaserCode is enabled, force all JTAC
# laser codes to 1113 to allow lasing for Su-25 Frogfoots and A-10A Warthogs.
# Otherwise use 1688 for the first JTAC, 1687 for the second etc.
if self.game.settings.plugins["plugins.jtacautolase.fc3LaserCode"]:
code = 1113
else:
code = self.laser_code_registry.get_next_laser_code()
utype = self.game.blue.faction.jtac_unit
if utype is None:
utype = AircraftType.named("MQ-9 Reaper")
jtac = self.mission.flight_group(
country=self.mission.country(self.game.blue.country_name),
name=n,
aircraft_type=utype.dcs_unit_type,
position=position[0],
airport=None,
altitude=5000,
maintask=AFAC,
)
jtac.points[0].tasks.append(
FAC(callsign=len(self.air_support.jtacs) + 1, frequency=int(freq.mhz))
)
jtac.points[0].tasks.append(SetInvisibleCommand(True))
jtac.points[0].tasks.append(SetImmortalCommand(True))
jtac.points[0].tasks.append(
OrbitAction(5000, 300, OrbitAction.OrbitPattern.Circle)
)
frontline = (
f"Frontline {self.conflict.blue_cp.name}/{self.conflict.red_cp.name}"
)
# Note: Will need to change if we ever add ground based JTAC.
callsign = callsign_for_support_unit(jtac)
self.air_support.jtacs.append(
JtacInfo(
str(jtac.name),
n,
callsign,
frontline,
str(code),
blue=True,
freq=freq,
)
)
def gen_infantry_group_for_group(
self,
group: VehicleGroup,
is_player: bool,
side: Country,
forward_heading: Heading,
) -> None:
infantry_position = self.conflict.find_ground_position(
group.points[0].position.random_point_within(250, 50),
500,
forward_heading,
self.conflict.theater,
)
if not infantry_position:
logging.warning("Could not find infantry position")
return
if side == self.conflict.attackers_country:
cp = self.conflict.blue_cp
else:
cp = self.conflict.red_cp
faction = self.game.faction_for(is_player)
# Disable infantry unit gen if disabled
if not self.game.settings.perf_infantry:
if self.game.settings.manpads:
# 50% of armored units protected by manpad
if random.choice([True, False]):
manpads = list(faction.infantry_with_class(GroundUnitClass.Manpads))
if manpads:
u = random.choices(
manpads, weights=[m.spawn_weight for m in manpads]
)[0]
self.mission.vehicle_group(
side,
namegen.next_infantry_name(side, cp.id, u),
u.dcs_unit_type,
position=infantry_position,
group_size=1,
heading=forward_heading.degrees,
move_formation=PointAction.OffRoad,
)
return
possible_infantry_units = set(
faction.infantry_with_class(GroundUnitClass.Infantry)
)
if self.game.settings.manpads:
possible_infantry_units |= set(
faction.infantry_with_class(GroundUnitClass.Manpads)
)
if not possible_infantry_units:
return
infantry_choices = list(possible_infantry_units)
units = random.choices(
infantry_choices,
weights=[u.spawn_weight for u in infantry_choices],
k=INFANTRY_GROUP_SIZE,
)
self.mission.vehicle_group(
side,
namegen.next_infantry_name(side, cp.id, units[0]),
units[0].dcs_unit_type,
position=infantry_position,
group_size=1,
heading=forward_heading.degrees,
move_formation=PointAction.OffRoad,
)
for unit in units[1:]:
position = infantry_position.random_point_within(55, 5)
self.mission.vehicle_group(
side,
namegen.next_infantry_name(side, cp.id, unit),
unit.dcs_unit_type,
position=position,
group_size=1,
heading=forward_heading.degrees,
move_formation=PointAction.OffRoad,
)
def _set_reform_waypoint(
self, dcs_group: VehicleGroup, forward_heading: Heading
) -> None:
"""Setting a waypoint close to the spawn position allows the group to reform gracefully
rather than spin
"""
reform_point = dcs_group.position.point_from_heading(
forward_heading.degrees, 50
)
dcs_group.add_waypoint(reform_point)
def _plan_artillery_action(
self,
stance: CombatStance,
gen_group: CombatGroup,
dcs_group: VehicleGroup,
forward_heading: Heading,
target: Point,
) -> bool:
"""
Handles adding the DCS tasks for artillery groups for all combat stances.
Returns True if tasking was added, returns False if the stance was not a combat stance.
"""
self._set_reform_waypoint(dcs_group, forward_heading)
if stance != CombatStance.RETREAT:
hold_task = Hold()
hold_task.number = 1
dcs_group.add_trigger_action(hold_task)
# Artillery strike random start
artillery_trigger = TriggerOnce(
Event.NoEvent, "ArtilleryFireTask #" + str(dcs_group.id)
)
artillery_trigger.add_condition(TimeAfter(seconds=random.randint(1, 45) * 60))
# TODO: Update to fire at group instead of point
fire_task = FireAtPoint(target, gen_group.size * 10, 100)
fire_task.number = 2 if stance != CombatStance.RETREAT else 1
dcs_group.add_trigger_action(fire_task)
artillery_trigger.add_action(AITaskPush(dcs_group.id, len(dcs_group.tasks)))
self.mission.triggerrules.triggers.append(artillery_trigger)
# Artillery will fall back when under attack
if stance != CombatStance.RETREAT:
# Hold position
dcs_group.points[1].tasks.append(Hold())
retreat = self.find_retreat_point(
dcs_group, forward_heading, (int)(RETREAT_DISTANCE / 3)
)
dcs_group.add_waypoint(
dcs_group.position.point_from_heading(forward_heading.degrees, 1),
PointAction.OffRoad,
)
dcs_group.points[2].tasks.append(Hold())
dcs_group.add_waypoint(retreat, PointAction.OffRoad)
artillery_fallback = TriggerOnce(
Event.NoEvent, "ArtilleryRetreat #" + str(dcs_group.id)
)
for i, u in enumerate(dcs_group.units):
artillery_fallback.add_condition(UnitDamaged(u.id))
if i < len(dcs_group.units) - 1:
artillery_fallback.add_condition(Or())
hold_2 = Hold()
hold_2.number = 3
dcs_group.add_trigger_action(hold_2)
retreat_task = GoToWaypoint(to_index=3)
retreat_task.number = 4
dcs_group.add_trigger_action(retreat_task)
artillery_fallback.add_action(
AITaskPush(dcs_group.id, len(dcs_group.tasks))
)
self.mission.triggerrules.triggers.append(artillery_fallback)
for u in dcs_group.units:
u.heading = (forward_heading + Heading.random(-5, 5)).degrees
return True
return False
def _plan_tank_ifv_action(
self,
stance: CombatStance,
enemy_groups: List[Tuple[VehicleGroup, CombatGroup]],
dcs_group: VehicleGroup,
forward_heading: Heading,
to_cp: ControlPoint,
) -> bool:
"""
Handles adding the DCS tasks for tank and IFV groups for all combat stances.
Returns True if tasking was added, returns False if the stance was not a combat stance.
"""
self._set_reform_waypoint(dcs_group, forward_heading)
if stance == CombatStance.AGGRESSIVE:
# Attack nearest enemy if any
# Then move forward OR Attack enemy base if it is not too far away
target = self.find_nearest_enemy_group(dcs_group, enemy_groups)
if target is not None:
rand_offset = Point(
random.randint(-RANDOM_OFFSET_ATTACK, RANDOM_OFFSET_ATTACK),
random.randint(-RANDOM_OFFSET_ATTACK, RANDOM_OFFSET_ATTACK),
)
target_point = self.conflict.theater.nearest_land_pos(
target.points[0].position + rand_offset
)
dcs_group.add_waypoint(target_point)
dcs_group.points[2].tasks.append(AttackGroup(target.id))
if (
to_cp.position.distance_to_point(dcs_group.points[0].position)
<= AGGRESIVE_MOVE_DISTANCE
):
attack_point = self.conflict.theater.nearest_land_pos(
to_cp.position.random_point_within(500, 0)
)
else:
# We use an offset heading here because DCS doesn't always
# force vehicles to move if there's no heading change.
offset_heading = forward_heading - Heading.from_degrees(2)
attack_point = self.find_offensive_point(
dcs_group, offset_heading, AGGRESIVE_MOVE_DISTANCE
)
dcs_group.add_waypoint(attack_point, PointAction.OffRoad)
elif stance == CombatStance.BREAKTHROUGH:
# In breakthrough mode, the units will move forward
# If the enemy base is close enough, the units will attack the base
if (
to_cp.position.distance_to_point(dcs_group.points[0].position)
<= BREAKTHROUGH_OFFENSIVE_DISTANCE
):
attack_point = self.conflict.theater.nearest_land_pos(
to_cp.position.random_point_within(500, 0)
)
else:
# We use an offset heading here because DCS doesn't always
# force vehicles to move if there's no heading change.
offset_heading = forward_heading - Heading.from_degrees(1)
attack_point = self.find_offensive_point(
dcs_group, offset_heading, BREAKTHROUGH_OFFENSIVE_DISTANCE
)
dcs_group.add_waypoint(attack_point, PointAction.OffRoad)
elif stance == CombatStance.ELIMINATION:
# In elimination mode, the units focus on destroying as much enemy groups as possible
targets = self.find_n_nearest_enemy_groups(dcs_group, enemy_groups, 3)
for i, target in enumerate(targets, start=1):
rand_offset = Point(
random.randint(-RANDOM_OFFSET_ATTACK, RANDOM_OFFSET_ATTACK),
random.randint(-RANDOM_OFFSET_ATTACK, RANDOM_OFFSET_ATTACK),
)
target_point = self.conflict.theater.nearest_land_pos(
target.points[0].position + rand_offset
)
dcs_group.add_waypoint(target_point, PointAction.OffRoad)
dcs_group.points[i + 1].tasks.append(AttackGroup(target.id))
if (
to_cp.position.distance_to_point(dcs_group.points[0].position)
<= AGGRESIVE_MOVE_DISTANCE
):
attack_point = self.conflict.theater.nearest_land_pos(
to_cp.position.random_point_within(500, 0)
)
dcs_group.add_waypoint(attack_point)
if stance != CombatStance.RETREAT:
self.add_morale_trigger(dcs_group, forward_heading)
return True
return False
def _plan_apc_atgm_action(
self,
stance: CombatStance,
dcs_group: VehicleGroup,
forward_heading: Heading,
to_cp: ControlPoint,
) -> bool:
"""
Handles adding the DCS tasks for APC and ATGM groups for all combat stances.
Returns True if tasking was added, returns False if the stance was not a combat stance.
"""
self._set_reform_waypoint(dcs_group, forward_heading)
if stance in [
CombatStance.AGGRESSIVE,
CombatStance.BREAKTHROUGH,
CombatStance.ELIMINATION,
]:
# APC & ATGM will never move too much forward, but will follow along any offensive
if (
to_cp.position.distance_to_point(dcs_group.points[0].position)
<= AGGRESIVE_MOVE_DISTANCE
):
attack_point = self.conflict.theater.nearest_land_pos(
to_cp.position.random_point_within(500, 0)
)
else:
attack_point = self.find_offensive_point(
dcs_group, forward_heading, AGGRESIVE_MOVE_DISTANCE
)
dcs_group.add_waypoint(attack_point, PointAction.OffRoad)
if stance != CombatStance.RETREAT:
self.add_morale_trigger(dcs_group, forward_heading)
return True
return False
def plan_action_for_groups(
self,
stance: CombatStance,
ally_groups: List[Tuple[VehicleGroup, CombatGroup]],
enemy_groups: List[Tuple[VehicleGroup, CombatGroup]],
forward_heading: Heading,
from_cp: ControlPoint,
to_cp: ControlPoint,
) -> None:
if not self.game.settings.perf_moving_units:
return
for dcs_group, group in ally_groups:
if group.unit_type.eplrs_capable:
dcs_group.points[0].tasks.append(EPLRS(dcs_group.id))
if group.role == CombatGroupRole.ARTILLERY:
if self.game.settings.perf_artillery:
target = self.get_artillery_target_in_range(
dcs_group, group, enemy_groups
)
if target is not None:
self._plan_artillery_action(
stance, group, dcs_group, forward_heading, target
)
elif group.role in [CombatGroupRole.TANK, CombatGroupRole.IFV]:
self._plan_tank_ifv_action(
stance, enemy_groups, dcs_group, forward_heading, to_cp
)
elif group.role in [CombatGroupRole.APC, CombatGroupRole.ATGM]:
self._plan_apc_atgm_action(stance, dcs_group, forward_heading, to_cp)
if stance == CombatStance.RETREAT:
# In retreat mode, the units will fall back
# If the ally base is close enough, the units will even regroup there
if (
from_cp.position.distance_to_point(dcs_group.points[0].position)
<= RETREAT_DISTANCE
):
retreat_point = from_cp.position.random_point_within(500, 250)
else:
retreat_point = self.find_retreat_point(dcs_group, forward_heading)
reposition_point = retreat_point.point_from_heading(
forward_heading.degrees, 10
) # Another point to make the unit face the enemy
dcs_group.add_waypoint(retreat_point, PointAction.OffRoad)
dcs_group.add_waypoint(reposition_point, PointAction.OffRoad)
def add_morale_trigger(
self, dcs_group: VehicleGroup, forward_heading: Heading
) -> None:
"""
This add a trigger to manage units fleeing whenever their group is hit hard, or being engaged by CAS
"""
if len(dcs_group.units) == 1:
return
# Units should hold position on last waypoint
dcs_group.points[len(dcs_group.points) - 1].tasks.append(Hold())
# Force unit heading
for unit in dcs_group.units:
unit.heading = forward_heading.degrees
dcs_group.manualHeading = True
# We add a new retreat waypoint
dcs_group.add_waypoint(
self.find_retreat_point(
dcs_group, forward_heading, (int)(RETREAT_DISTANCE / 8)
),
PointAction.OffRoad,
)
# Fallback task
task = ControlledTask(GoToWaypoint(to_index=len(dcs_group.points)))
task.enabled = False
dcs_group.add_trigger_action(Hold())
dcs_group.add_trigger_action(task)
# Create trigger
fallback = TriggerOnce(Event.NoEvent, "Morale manager #" + str(dcs_group.id))
# Usually more than 50% casualties = RETREAT
fallback.add_condition(GroupLifeLess(dcs_group.id, random.randint(51, 76)))
# Do retreat to the configured retreat waypoint
fallback.add_action(AITaskPush(dcs_group.id, len(dcs_group.tasks)))
self.mission.triggerrules.triggers.append(fallback)
def find_retreat_point(
self,
dcs_group: VehicleGroup,
frontline_heading: Heading,
distance: int = RETREAT_DISTANCE,
) -> Point:
"""
Find a point to retreat to
:param dcs_group: DCS mission group we are searching a retreat point for
:param frontline_heading: Heading of the frontline
:return: dcs.mapping.Point object with the desired position
"""
desired_point = dcs_group.points[0].position.point_from_heading(
frontline_heading.opposite.degrees, distance
)
if self.conflict.theater.is_on_land(desired_point):
return desired_point
return self.conflict.theater.nearest_land_pos(desired_point)
def find_offensive_point(
self, dcs_group: VehicleGroup, frontline_heading: Heading, distance: int
) -> Point:
"""
Find a point to attack
:param dcs_group: DCS mission group we are searching an attack point for
:param frontline_heading: Heading of the frontline
:param distance: Distance of the offensive (how far unit should move)
:return: dcs.mapping.Point object with the desired position
"""
desired_point = dcs_group.points[0].position.point_from_heading(
frontline_heading.degrees, distance
)
if self.conflict.theater.is_on_land(desired_point):
return desired_point
return self.conflict.theater.nearest_land_pos(desired_point)
@staticmethod
def find_n_nearest_enemy_groups(
player_group: VehicleGroup,
enemy_groups: List[Tuple[VehicleGroup, CombatGroup]],
n: int,
) -> List[VehicleGroup]:
"""
Return the nearest enemy group for the player group
@param group Group for which we should find the nearest ennemies
@param enemy_groups Potential enemy groups
@param n number of nearby groups to take
"""
targets = [] # type: List[VehicleGroup]
sorted_list = sorted(
enemy_groups,
key=lambda group: player_group.points[0].position.distance_to_point(
group[0].points[0].position
),
)
for i in range(n):
# TODO: Is this supposed to return no groups if enemy_groups is less than n?
if len(sorted_list) <= i:
break
else:
targets.append(sorted_list[i][0])
return targets
@staticmethod
def find_nearest_enemy_group(
player_group: VehicleGroup, enemy_groups: List[Tuple[VehicleGroup, CombatGroup]]
) -> Optional[VehicleGroup]:
"""
Search the enemy groups for a potential target suitable to armored assault
@param group Group for which we should find the nearest ennemy
@param enemy_groups Potential enemy groups
"""
min_distance = math.inf
target = None
for dcs_group, _ in enemy_groups:
dist = player_group.points[0].position.distance_to_point(
dcs_group.points[0].position
)
if dist < min_distance:
min_distance = dist
target = dcs_group
return target
@staticmethod
def get_artillery_target_in_range(
dcs_group: VehicleGroup,
group: CombatGroup,
enemy_groups: List[Tuple[VehicleGroup, CombatGroup]],
) -> Optional[Point]:
"""
Search the enemy groups for a potential target suitable to an artillery unit
"""
# TODO: Update to return a list of groups instead of a single point
rng = getattr(group.unit_type.dcs_unit_type, "threat_range", 0)
if not enemy_groups:
return None
for _ in range(10):
potential_target = random.choice(enemy_groups)[0]
distance_to_target = dcs_group.points[0].position.distance_to_point(
potential_target.points[0].position
)
if distance_to_target < rng:
return potential_target.points[0].position
return None
@staticmethod
def get_artilery_group_distance_from_frontline(group: CombatGroup) -> int:
"""
For artilery group, decide the distance from frontline with the range of the unit
"""
rg = group.unit_type.dcs_unit_type.threat_range - 7500
if rg > DISTANCE_FROM_FRONTLINE[CombatGroupRole.ARTILLERY][1]:
rg = random.randint(
DISTANCE_FROM_FRONTLINE[CombatGroupRole.ARTILLERY][0],
DISTANCE_FROM_FRONTLINE[CombatGroupRole.ARTILLERY][1],
)
elif rg < DISTANCE_FROM_FRONTLINE[CombatGroupRole.ARTILLERY][1]:
rg = random.randint(
DISTANCE_FROM_FRONTLINE[CombatGroupRole.TANK][0],
DISTANCE_FROM_FRONTLINE[CombatGroupRole.TANK][1],
)
return rg
def get_valid_position_for_group(
self,
conflict_position: Point,
combat_width: int,
distance_from_frontline: int,
heading: Heading,
spawn_heading: Heading,
) -> Optional[Point]:
shifted = conflict_position.point_from_heading(
heading.degrees, random.randint(0, combat_width)
)
desired_point = shifted.point_from_heading(
spawn_heading.degrees, distance_from_frontline
)
return Conflict.find_ground_position(
desired_point, combat_width, heading, self.conflict.theater
)
def _generate_groups(
self,
groups: list[CombatGroup],
frontline_vector: Tuple[Point, Heading, int],
is_player: bool,
) -> List[Tuple[VehicleGroup, CombatGroup]]:
"""Finds valid positions for planned groups and generates a pydcs group for them"""
positioned_groups = []
position, heading, combat_width = frontline_vector
spawn_heading = heading.left if is_player else heading.right
country = self.game.coalition_for(is_player).country_name
for group in groups:
if group.role == CombatGroupRole.ARTILLERY:
distance_from_frontline = (
self.get_artilery_group_distance_from_frontline(group)
)
else:
distance_from_frontline = random.randint(
DISTANCE_FROM_FRONTLINE[group.role][0],
DISTANCE_FROM_FRONTLINE[group.role][1],
)
final_position = self.get_valid_position_for_group(
position, combat_width, distance_from_frontline, heading, spawn_heading
)
if final_position is not None:
g = self._generate_group(
self.mission.country(country),
group.unit_type,
group.size,
final_position,
heading=spawn_heading.opposite,
)
if is_player:
g.set_skill(Skill(self.game.settings.player_skill))
else:
g.set_skill(Skill(self.game.settings.enemy_vehicle_skill))
positioned_groups.append((g, group))
if group.role in [CombatGroupRole.APC, CombatGroupRole.IFV]:
self.gen_infantry_group_for_group(
g,
is_player,
self.mission.country(country),
spawn_heading.opposite,
)
else:
logging.warning(f"Unable to get valid position for {group}")
return positioned_groups
def _generate_group(
self,
side: Country,
unit_type: GroundUnitType,
count: int,
at: Point,
move_formation: PointAction = PointAction.OffRoad,
heading: Heading = Heading.from_degrees(0),
) -> VehicleGroup:
if side == self.conflict.attackers_country:
cp = self.conflict.blue_cp
else:
cp = self.conflict.red_cp
group = self.mission.vehicle_group(
side,
namegen.next_unit_name(side, cp.id, unit_type),
unit_type.dcs_unit_type,
position=at,
group_size=count,
heading=heading.degrees,
move_formation=move_formation,
)
self.unit_map.add_front_line_units(group, cp, unit_type)
for c in range(count):
vehicle: Vehicle = group.units[c]
vehicle.player_can_drive = True
return group

74
gen/beacons.py
View File

@@ -1,74 +0,0 @@
from dataclasses import dataclass
from enum import auto, IntEnum
import json
from pathlib import Path
from typing import Iterable, Optional
from gen.radios import RadioFrequency
from gen.tacan import TacanBand, TacanChannel
BEACONS_RESOURCE_PATH = Path("resources/dcs/beacons")
class BeaconType(IntEnum):
BEACON_TYPE_NULL = auto()
BEACON_TYPE_VOR = auto()
BEACON_TYPE_DME = auto()
BEACON_TYPE_VOR_DME = auto()
BEACON_TYPE_TACAN = auto()
BEACON_TYPE_VORTAC = auto()
BEACON_TYPE_RSBN = auto()
BEACON_TYPE_BROADCAST_STATION = auto()
BEACON_TYPE_HOMER = auto()
BEACON_TYPE_AIRPORT_HOMER = auto()
BEACON_TYPE_AIRPORT_HOMER_WITH_MARKER = auto()
BEACON_TYPE_ILS_FAR_HOMER = auto()
BEACON_TYPE_ILS_NEAR_HOMER = auto()
BEACON_TYPE_ILS_LOCALIZER = auto()
BEACON_TYPE_ILS_GLIDESLOPE = auto()
BEACON_TYPE_PRMG_LOCALIZER = auto()
BEACON_TYPE_PRMG_GLIDESLOPE = auto()
BEACON_TYPE_ICLS_LOCALIZER = auto()
BEACON_TYPE_ICLS_GLIDESLOPE = auto()
BEACON_TYPE_NAUTICAL_HOMER = auto()
@dataclass(frozen=True)
class Beacon:
name: str
callsign: str
beacon_type: BeaconType
hertz: int
channel: Optional[int]
@property
def frequency(self) -> RadioFrequency:
return RadioFrequency(self.hertz)
@property
def is_tacan(self) -> bool:
return self.beacon_type in (
BeaconType.BEACON_TYPE_VORTAC,
BeaconType.BEACON_TYPE_TACAN,
)
@property
def tacan_channel(self) -> TacanChannel:
assert self.is_tacan
assert self.channel is not None
return TacanChannel(self.channel, TacanBand.X)
def load_beacons_for_terrain(name: str) -> Iterable[Beacon]:
beacons_file = BEACONS_RESOURCE_PATH / f"{name.lower()}.json"
if not beacons_file.exists():
raise RuntimeError(f"Beacon file {beacons_file.resolve()} is missing")
for beacon in json.loads(beacons_file.read_text()):
yield Beacon(**beacon)

192
gen/briefinggen.py
View File

@@ -1,192 +0,0 @@
"""
Briefing generation logic
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from datetime import timedelta
from typing import Dict, List, TYPE_CHECKING
from dcs.mission import Mission
from jinja2 import Environment, FileSystemLoader, select_autoescape
from game.theater import ControlPoint, FrontLine
from .aircraft import FlightData
from .airsupportgen import AwacsInfo, TankerInfo
from .armor import JtacInfo
from game.ato.flightwaypoint import FlightWaypoint
from .ground_forces.combat_stance import CombatStance
from .radios import RadioFrequency
from .runways import RunwayData
if TYPE_CHECKING:
from game import Game
@dataclass
class CommInfo:
"""Communications information for the kneeboard."""
name: str
freq: RadioFrequency
class FrontLineInfo:
def __init__(self, front_line: FrontLine):
self.front_line: FrontLine = front_line
self.player_base: ControlPoint = front_line.blue_cp
self.enemy_base: ControlPoint = front_line.red_cp
self.player_zero: bool = self.player_base.base.total_armor == 0
self.enemy_zero: bool = self.enemy_base.base.total_armor == 0
self.advantage: bool = (
self.player_base.base.total_armor > self.enemy_base.base.total_armor
)
self.stance: CombatStance = self.player_base.stances[self.enemy_base.id]
self.combat_stances = CombatStance
class MissionInfoGenerator:
"""Base type for generators of mission information for the player.
Examples of subtypes include briefing generators, kneeboard generators, etc.
"""
def __init__(self, mission: Mission, game: Game) -> None:
self.mission = mission
self.game = game
self.awacs: List[AwacsInfo] = []
self.comms: List[CommInfo] = []
self.flights: List[FlightData] = []
self.jtacs: List[JtacInfo] = []
self.tankers: List[TankerInfo] = []
self.frontlines: List[FrontLineInfo] = []
self.dynamic_runways: List[RunwayData] = []
def add_awacs(self, awacs: AwacsInfo) -> None:
"""Adds an AWACS/GCI to the mission.
Args:
awacs: AWACS information.
"""
self.awacs.append(awacs)
def add_comm(self, name: str, freq: RadioFrequency) -> None:
"""Adds communications info to the mission.
Args:
name: Name of the radio channel.
freq: Frequency of the radio channel.
"""
self.comms.append(CommInfo(name, freq))
def add_flight(self, flight: FlightData) -> None:
"""Adds flight info to the mission.
Args:
flight: Flight information.
"""
self.flights.append(flight)
def add_jtac(self, jtac: JtacInfo) -> None:
"""Adds a JTAC to the mission.
Args:
jtac: JTAC information.
"""
self.jtacs.append(jtac)
def add_tanker(self, tanker: TankerInfo) -> None:
"""Adds a tanker to the mission.
Args:
tanker: Tanker information.
"""
self.tankers.append(tanker)
def add_frontline(self, frontline: FrontLineInfo) -> None:
"""Adds a frontline to the briefing
Arguments:
frontline: Frontline conflict information
"""
self.frontlines.append(frontline)
def add_dynamic_runway(self, runway: RunwayData) -> None:
"""Adds a dynamically generated runway to the briefing.
Dynamic runways are any valid landing point that is a unit rather than a
map feature. These include carriers, ships with a helipad, and FARPs.
"""
self.dynamic_runways.append(runway)
def generate(self) -> None:
"""Generates the mission information."""
raise NotImplementedError
def format_waypoint_time(waypoint: FlightWaypoint, depart_prefix: str) -> str:
if waypoint.tot is not None:
time = timedelta(seconds=int(waypoint.tot.total_seconds()))
return f"T+{time} "
elif waypoint.departure_time is not None:
time = timedelta(seconds=int(waypoint.departure_time.total_seconds()))
return f"{depart_prefix} T+{time} "
return ""
def format_intra_flight_channel(flight: FlightData) -> str:
frequency = flight.intra_flight_channel
channel = flight.channel_for(frequency)
if channel is None:
return str(frequency)
channel_name = flight.aircraft_type.channel_name(channel.radio_id, channel.channel)
return f"{channel_name} ({frequency})"
class BriefingGenerator(MissionInfoGenerator):
def __init__(self, mission: Mission, game: Game):
super().__init__(mission, game)
self.allied_flights_by_departure: Dict[str, List[FlightData]] = {}
env = Environment(
loader=FileSystemLoader("resources/briefing/templates"),
autoescape=select_autoescape(
disabled_extensions=("",),
default_for_string=True,
default=True,
),
trim_blocks=True,
lstrip_blocks=True,
)
env.filters["waypoint_timing"] = format_waypoint_time
env.filters["intra_flight_channel"] = format_intra_flight_channel
self.template = env.get_template("briefingtemplate_EN.j2")
def generate(self) -> None:
"""Generate the mission briefing"""
self._generate_frontline_info()
self.generate_allied_flights_by_departure()
self.mission.set_description_text(self.template.render(vars(self)))
self.mission.add_picture_blue(
os.path.abspath("./resources/ui/splash_screen.png")
)
def _generate_frontline_info(self) -> None:
"""Build FrontLineInfo objects from FrontLine type and append to briefing."""
for front_line in self.game.theater.conflicts():
self.add_frontline(FrontLineInfo(front_line))
# TODO: This should determine if runway is friendly through a method more robust than the existing string match
def generate_allied_flights_by_departure(self) -> None:
"""Create iterable to display allied flights grouped by departure airfield."""
for flight in self.flights:
if not flight.client_units and flight.friendly:
name = flight.departure.airfield_name
if (
name in self.allied_flights_by_departure
): # where else can we get this?
self.allied_flights_by_departure[name].append(flight)
else:
self.allied_flights_by_departure[name] = [flight]

48
gen/cargoshipgen.py
View File

@@ -1,48 +0,0 @@
from __future__ import annotations
import itertools
from typing import TYPE_CHECKING
from dcs import Mission
from dcs.ships import HandyWind
from dcs.unitgroup import ShipGroup
from game.transfers import CargoShip
from game.unitmap import UnitMap
from game.utils import knots
if TYPE_CHECKING:
from game import Game
class CargoShipGenerator:
def __init__(self, mission: Mission, game: Game, unit_map: UnitMap) -> None:
self.mission = mission
self.game = game
self.unit_map = unit_map
self.count = itertools.count()
def generate(self) -> None:
# Reset the count to make generation deterministic.
for coalition in self.game.coalitions:
for ship in coalition.transfers.cargo_ships:
self.generate_cargo_ship(ship)
def generate_cargo_ship(self, ship: CargoShip) -> ShipGroup:
country = self.mission.country(
self.game.coalition_for(ship.player_owned).country_name
)
waypoints = ship.route
group = self.mission.ship_group(
country,
ship.name,
HandyWind,
position=waypoints[0],
group_size=1,
)
for waypoint in waypoints[1:]:
# 12 knots is very slow but it's also nearly the max allowed by DCS for this
# type of ship.
group.add_waypoint(waypoint, speed=knots(12).kph)
self.unit_map.add_cargo_ship(group, ship)
return group

170
gen/conflictgen.py
View File

@@ -1,170 +0,0 @@
from __future__ import annotations
import logging
from typing import Tuple, Optional
from dcs.country import Country
from dcs.mapping import Point
from shapely.geometry import LineString, Point as ShapelyPoint
from game.theater.conflicttheater import ConflictTheater, FrontLine
from game.theater.controlpoint import ControlPoint
from game.utils import Heading
FRONTLINE_LENGTH = 80000
class Conflict:
def __init__(
self,
theater: ConflictTheater,
front_line: FrontLine,
attackers_side: str,
defenders_side: str,
attackers_country: Country,
defenders_country: Country,
position: Point,
heading: Optional[Heading] = None,
size: Optional[int] = None,
):
self.attackers_side = attackers_side
self.defenders_side = defenders_side
self.attackers_country = attackers_country
self.defenders_country = defenders_country
self.front_line = front_line
self.theater = theater
self.position = position
self.heading = heading
self.size = size
@property
def blue_cp(self) -> ControlPoint:
return self.front_line.blue_cp
@property
def red_cp(self) -> ControlPoint:
return self.front_line.red_cp
@classmethod
def has_frontline_between(cls, from_cp: ControlPoint, to_cp: ControlPoint) -> bool:
return from_cp.has_frontline and to_cp.has_frontline
@classmethod
def frontline_position(
cls, frontline: FrontLine, theater: ConflictTheater
) -> Tuple[Point, Heading]:
attack_heading = frontline.attack_heading
position = cls.find_ground_position(
frontline.position,
FRONTLINE_LENGTH,
attack_heading.right,
theater,
)
if position is None:
raise RuntimeError("Could not find front line position")
return position, attack_heading.opposite
@classmethod
def frontline_vector(
cls, front_line: FrontLine, theater: ConflictTheater
) -> Tuple[Point, Heading, int]:
"""
Returns a vector for a valid frontline location avoiding exclusion zones.
"""
center_position, heading = cls.frontline_position(front_line, theater)
left_heading = heading.left
right_heading = heading.right
left_position = cls.extend_ground_position(
center_position, int(FRONTLINE_LENGTH / 2), left_heading, theater
)
right_position = cls.extend_ground_position(
center_position, int(FRONTLINE_LENGTH / 2), right_heading, theater
)
distance = int(left_position.distance_to_point(right_position))
return left_position, right_heading, distance
@classmethod
def frontline_cas_conflict(
cls,
attacker_name: str,
defender_name: str,
attacker: Country,
defender: Country,
front_line: FrontLine,
theater: ConflictTheater,
) -> Conflict:
assert cls.has_frontline_between(front_line.blue_cp, front_line.red_cp)
position, heading, distance = cls.frontline_vector(front_line, theater)
conflict = cls(
position=position,
heading=heading,
theater=theater,
front_line=front_line,
attackers_side=attacker_name,
defenders_side=defender_name,
attackers_country=attacker,
defenders_country=defender,
size=distance,
)
return conflict
@classmethod
def extend_ground_position(
cls,
initial: Point,
max_distance: int,
heading: Heading,
theater: ConflictTheater,
) -> Point:
"""Finds the first intersection with an exclusion zone in one heading from an initial point up to max_distance"""
extended = initial.point_from_heading(heading.degrees, max_distance)
if theater.landmap is None:
# TODO: Why is this possible?
return extended
p0 = ShapelyPoint(initial.x, initial.y)
p1 = ShapelyPoint(extended.x, extended.y)
line = LineString([p0, p1])
intersection = line.intersection(theater.landmap.inclusion_zone_only.boundary)
if intersection.is_empty:
# Max extent does not intersect with the boundary of the inclusion
# zone, so the full front line is usable. This does assume that the
# front line was centered on a valid location.
return extended
# Otherwise extend the front line only up to the intersection.
return initial.point_from_heading(heading.degrees, p0.distance(intersection))
@classmethod
def find_ground_position(
cls,
initial: Point,
max_distance: int,
heading: Heading,
theater: ConflictTheater,
coerce: bool = True,
) -> Optional[Point]:
"""
Finds the nearest valid ground position along a provided heading and it's inverse up to max_distance.
`coerce=True` will return the closest land position to `initial` regardless of heading or distance
`coerce=False` will return None if a point isn't found
"""
pos = initial
if theater.is_on_land(pos):
return pos
for distance in range(0, int(max_distance), 100):
pos = initial.point_from_heading(heading.degrees, distance)
if theater.is_on_land(pos):
return pos
pos = initial.point_from_heading(heading.opposite.degrees, distance)
if theater.is_on_land(pos):
return pos
if coerce:
pos = theater.nearest_land_pos(initial)
return pos
logging.error("Didn't find ground position ({})!".format(initial))
return None

91
gen/convoygen.py
View File

@@ -1,91 +0,0 @@
from __future__ import annotations
import itertools
from typing import TYPE_CHECKING
from dcs import Mission
from dcs.mapping import Point
from dcs.point import PointAction
from dcs.unit import Vehicle
from dcs.unitgroup import VehicleGroup
from game.dcs.groundunittype import GroundUnitType
from game.transfers import Convoy
from game.unitmap import UnitMap
from game.utils import kph
if TYPE_CHECKING:
from game import Game
class ConvoyGenerator:
def __init__(self, mission: Mission, game: Game, unit_map: UnitMap) -> None:
self.mission = mission
self.game = game
self.unit_map = unit_map
self.count = itertools.count()
def generate(self) -> None:
# Reset the count to make generation deterministic.
for coalition in self.game.coalitions:
for convoy in coalition.transfers.convoys:
self.generate_convoy(convoy)
def generate_convoy(self, convoy: Convoy) -> VehicleGroup:
group = self._create_mixed_unit_group(
convoy.name,
convoy.route_start,
convoy.units,
convoy.player_owned,
)
group.add_waypoint(
convoy.route_end,
speed=kph(40).kph,
move_formation=PointAction.OnRoad,
)
self.make_drivable(group)
self.unit_map.add_convoy_units(group, convoy)
return group
def _create_mixed_unit_group(
self,
name: str,
position: Point,
units: dict[GroundUnitType, int],
for_player: bool,
) -> VehicleGroup:
country = self.mission.country(self.game.coalition_for(for_player).country_name)
unit_types = list(units.items())
main_unit_type, main_unit_count = unit_types[0]
group = self.mission.vehicle_group(
country,
name,
main_unit_type.dcs_unit_type,
position=position,
group_size=main_unit_count,
move_formation=PointAction.OnRoad,
)
unit_name_counter = itertools.count(main_unit_count + 1)
# pydcs spreads units out by 20 in the Y axis by default. Pick up where it left
# off.
y = itertools.count(position.y + main_unit_count * 20, 20)
for unit_type, count in unit_types[1:]:
for i in range(count):
v = self.mission.vehicle(
f"{name} Unit #{next(unit_name_counter)}", unit_type.dcs_unit_type
)
v.position.x = position.x
v.position.y = next(y)
v.heading = 0
group.add_unit(v)
return group
@staticmethod
def make_drivable(group: VehicleGroup) -> None:
for v in group.units:
if isinstance(v, Vehicle):
v.player_can_drive = True

42
gen/environmentgen.py
View File

@@ -1,42 +0,0 @@
from typing import Optional
from dcs.mission import Mission
from game.weather import Clouds, Fog, Conditions, WindConditions, AtmosphericConditions
class EnvironmentGenerator:
def __init__(self, mission: Mission, conditions: Conditions) -> None:
self.mission = mission
self.conditions = conditions
def set_atmospheric(self, atmospheric: AtmosphericConditions) -> None:
self.mission.weather.qnh = atmospheric.qnh.mm_hg
self.mission.weather.season_temperature = atmospheric.temperature_celsius
def set_clouds(self, clouds: Optional[Clouds]) -> None:
if clouds is None:
return
self.mission.weather.clouds_base = clouds.base
self.mission.weather.clouds_thickness = clouds.thickness
self.mission.weather.clouds_density = clouds.density
self.mission.weather.clouds_iprecptns = clouds.precipitation
self.mission.weather.clouds_preset = clouds.preset
def set_fog(self, fog: Optional[Fog]) -> None:
if fog is None:
return
self.mission.weather.fog_visibility = int(fog.visibility.meters)
self.mission.weather.fog_thickness = fog.thickness
def set_wind(self, wind: WindConditions) -> None:
self.mission.weather.wind_at_ground = wind.at_0m
self.mission.weather.wind_at_2000 = wind.at_2000m
self.mission.weather.wind_at_8000 = wind.at_8000m
def generate(self) -> None:
self.mission.start_time = self.conditions.start_time
self.set_atmospheric(self.conditions.weather.atmospheric)
self.set_clouds(self.conditions.weather.clouds)
self.set_fog(self.conditions.weather.fog)
self.set_wind(self.conditions.weather.wind)

11
gen/flights/flightplan.py
View File

@@ -46,7 +46,6 @@ from game.ato.flightwaypoint import FlightWaypoint
from game.ato.flight import Flight
from .traveltime import GroundSpeed, TravelTime
from .waypointbuilder import StrikeTarget, WaypointBuilder
from ..conflictgen import Conflict, FRONTLINE_LENGTH
if TYPE_CHECKING:
from game.ato.package import Package
@@ -1605,7 +1604,13 @@ class FlightPlanBuilder:
if not isinstance(location, FrontLine):
raise InvalidObjectiveLocation(flight.flight_type, location)
ingress, heading, distance = Conflict.frontline_vector(location, self.theater)
from game.missiongenerator.frontlineconflictdescription import (
FrontLineConflictDescription,
)
ingress, heading, distance = FrontLineConflictDescription.frontline_vector(
location, self.theater
)
center = ingress.point_from_heading(heading.degrees, distance / 2)
egress = ingress.point_from_heading(heading.degrees, distance)
@@ -1626,6 +1631,8 @@ class FlightPlanBuilder:
patrol_speed = flight.unit_type.preferred_patrol_speed(ingress_egress_altitude)
use_agl_ingress_egress = is_helo
from game.missiongenerator.frontlineconflictdescription import FRONTLINE_LENGTH
return CasFlightPlan(
package=self.package,
flight=flight,

56
gen/forcedoptionsgen.py
View File

@@ -1,56 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from dcs.forcedoptions import ForcedOptions
from dcs.mission import Mission
if TYPE_CHECKING:
from game.game import Game
class ForcedOptionsGenerator:
def __init__(self, mission: Mission, game: Game) -> None:
self.mission = mission
self.game = game
def _set_options_view(self) -> None:
self.mission.forced_options.options_view = (
self.game.settings.map_coalition_visibility
)
def _set_external_views(self) -> None:
if not self.game.settings.external_views_allowed:
self.mission.forced_options.external_views = (
self.game.settings.external_views_allowed
)
def _set_labels(self) -> None:
# TODO: Fix settings to use the real type.
# TODO: Allow forcing "full" and have default do nothing.
if self.game.settings.labels == "Abbreviated":
self.mission.forced_options.labels = ForcedOptions.Labels.Abbreviate
elif self.game.settings.labels == "Dot Only":
self.mission.forced_options.labels = ForcedOptions.Labels.DotOnly
elif self.game.settings.labels == "Neutral Dot":
self.mission.forced_options.labels = ForcedOptions.Labels.NeutralDot
elif self.game.settings.labels == "Off":
self.mission.forced_options.labels = ForcedOptions.Labels.None_
def _set_unrestricted_satnav(self) -> None:
blue = self.game.blue.faction
red = self.game.red.faction
if blue.unrestricted_satnav or red.unrestricted_satnav:
self.mission.forced_options.unrestricted_satnav = True
def _set_battle_damage_assessment(self) -> None:
self.mission.forced_options.battle_damage_assessment = (
self.game.settings.battle_damage_assessment
)
def generate(self) -> None:
self._set_options_view()
self._set_external_views()
self._set_labels()
self._set_unrestricted_satnav()
self._set_battle_damage_assessment()

723
gen/groundobjectsgen.py
View File

@@ -1,723 +0,0 @@
"""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 collections import defaultdict
from typing import (
Dict,
Iterator,
Optional,
TYPE_CHECKING,
Type,
List,
TypeVar,
Any,
Generic,
Union,
)
from dcs import Mission, Point, unitgroup
from dcs.action import SceneryDestructionZone
from dcs.country import Country
from dcs.point import StaticPoint
from dcs.statics import Fortification, fortification_map, warehouse_map
from dcs.task import (
ActivateBeaconCommand,
ActivateICLSCommand,
EPLRS,
OptAlarmState,
FireAtPoint,
)
from dcs.triggers import TriggerStart, TriggerZone
from dcs.unit import Ship, Unit, Vehicle, InvisibleFARP
from dcs.unitgroup import ShipGroup, StaticGroup, VehicleGroup
from dcs.unittype import StaticType, ShipType, VehicleType
from dcs.vehicles import vehicle_map
from game import db
from game.data.building_data import FORTIFICATION_UNITS, FORTIFICATION_UNITS_ID
from game.db import unit_type_from_name, ship_type_from_name, vehicle_type_from_name
from game.theater import ControlPoint, TheaterGroundObject
from game.theater.theatergroundobject import (
BuildingGroundObject,
CarrierGroundObject,
FactoryGroundObject,
GenericCarrierGroundObject,
LhaGroundObject,
ShipGroundObject,
MissileSiteGroundObject,
SceneryGroundObject,
)
from game.unitmap import UnitMap
from game.utils import Heading, feet, knots, mps
from .radios import RadioFrequency, RadioRegistry
from .runways import RunwayData
from .tacan import TacanBand, TacanChannel, TacanRegistry, TacanUsage
if TYPE_CHECKING:
from game import Game
FARP_FRONTLINE_DISTANCE = 10000
AA_CP_MIN_DISTANCE = 40000
TgoT = TypeVar("TgoT", bound=TheaterGroundObject[Any])
class GenericGroundObjectGenerator(Generic[TgoT]):
"""An unspecialized ground object generator.
Currently used only for SAM
"""
def __init__(
self,
ground_object: TgoT,
country: Country,
game: Game,
mission: Mission,
unit_map: UnitMap,
) -> None:
self.ground_object = ground_object
self.country = country
self.game = game
self.m = mission
self.unit_map = unit_map
@property
def culled(self) -> bool:
return self.game.position_culled(self.ground_object.position)
def generate(self) -> None:
if self.culled:
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 = vehicle_type_from_name(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 = 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(), 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)
self._register_unit_group(group, vg)
@staticmethod
def enable_eplrs(group: VehicleGroup, unit_type: Type[VehicleType]) -> None:
if unit_type.eplrs:
group.points[0].tasks.append(EPLRS(group.id))
def set_alarm_state(self, group: Union[ShipGroup, VehicleGroup]) -> None:
if self.game.settings.perf_red_alert_state:
group.points[0].tasks.append(OptAlarmState(2))
else:
group.points[0].tasks.append(OptAlarmState(1))
def _register_unit_group(
self,
persistence_group: Union[ShipGroup, VehicleGroup],
miz_group: Union[ShipGroup, VehicleGroup],
) -> None:
self.unit_map.add_ground_object_units(
self.ground_object, persistence_group, miz_group
)
class MissileSiteGenerator(GenericGroundObjectGenerator[MissileSiteGroundObject]):
@property
def culled(self) -> bool:
# Don't cull missile sites - their range is long enough to make them easily
# culled despite being a threat.
return False
def generate(self) -> None:
super(MissileSiteGenerator, self).generate()
# Note : Only the SCUD missiles group can fire (V1 site cannot fire in game right now)
# TODO : Should be pre-planned ?
# TODO : Add delay to task to spread fire task over mission duration ?
for group in self.ground_object.groups:
vg = self.m.find_group(group.name)
if vg is not None:
targets = self.possible_missile_targets()
if targets:
target = random.choice(targets)
real_target = target.point_from_heading(
Heading.random().degrees, random.randint(0, 2500)
)
vg.points[0].add_task(FireAtPoint(real_target))
logging.info("Set up fire task for missile group.")
else:
logging.info(
"Couldn't setup missile site to fire, no valid target in range."
)
else:
logging.info(
"Couldn't setup missile site to fire, group was not generated."
)
def possible_missile_targets(self) -> List[Point]:
"""
Find enemy control points in range
:param vg: Vehicle group we are searching a target for (There is always only oe group right now)
:return: List of possible missile targets
"""
targets: List[Point] = []
for cp in self.game.theater.controlpoints:
if cp.captured != self.ground_object.control_point.captured:
distance = cp.position.distance_to_point(self.ground_object.position)
if distance < self.missile_site_range:
targets.append(cp.position)
return targets
@property
def missile_site_range(self) -> int:
"""
Get the missile site range
:return: Missile site range
"""
site_range = 0
for group in self.ground_object.groups:
vg = self.m.find_group(group.name)
if vg is not None:
for u in vg.units:
if u.type in vehicle_map:
if vehicle_map[u.type].threat_range > site_range:
site_range = vehicle_map[u.type].threat_range
return site_range
class BuildingSiteGenerator(GenericGroundObjectGenerator[BuildingGroundObject]):
"""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: Type[VehicleType]) -> None:
if not self.ground_object.is_dead:
group = 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.degrees,
)
self._register_fortification(group)
def generate_static(self, static_type: Type[StaticType]) -> None:
group = 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.degrees,
dead=self.ground_object.is_dead,
)
self._register_building(group)
def _register_fortification(self, fortification: VehicleGroup) -> None:
assert isinstance(self.ground_object, BuildingGroundObject)
self.unit_map.add_fortification(self.ground_object, fortification)
def _register_building(self, building: StaticGroup) -> None:
assert isinstance(self.ground_object, BuildingGroundObject)
self.unit_map.add_building(self.ground_object, building)
class FactoryGenerator(BuildingSiteGenerator):
"""Generator for factory sites.
Factory sites are the buildings that allow the recruitment of ground units.
Destroying these prevents the owner from recruiting ground units at the connected
control point.
"""
def generate(self) -> None:
if self.game.position_culled(self.ground_object.position):
return
# TODO: Faction specific?
self.generate_static(Fortification.Workshop_A)
class SceneryGenerator(BuildingSiteGenerator):
def generate(self) -> None:
assert isinstance(self.ground_object, SceneryGroundObject)
trigger_zone = self.generate_trigger_zone(self.ground_object)
# DCS only visually shows a scenery object is dead when
# this trigger rule is applied. Otherwise you can kill a
# structure twice.
if self.ground_object.is_dead:
self.generate_dead_trigger_rule(trigger_zone)
# Tell Liberation to manage this groundobjectsgen as part of the campaign.
self.register_scenery()
def generate_trigger_zone(self, scenery: SceneryGroundObject) -> TriggerZone:
zone = scenery.zone
# Align the trigger zones to the faction color on the DCS briefing/F10 map.
if scenery.is_friendly(to_player=True):
color = {1: 0.2, 2: 0.7, 3: 1, 4: 0.15}
else:
color = {1: 1, 2: 0.2, 3: 0.2, 4: 0.15}
# Create the smallest valid size trigger zone (16 feet) so that risk of overlap is minimized.
# As long as the triggerzone is over the scenery object, we're ok.
smallest_valid_radius = feet(16).meters
return self.m.triggers.add_triggerzone(
zone.position,
smallest_valid_radius,
zone.hidden,
zone.name,
color,
zone.properties,
)
def generate_dead_trigger_rule(self, trigger_zone: TriggerZone) -> None:
# Add destruction zone trigger
t = TriggerStart(comment="Destruction")
t.actions.append(
SceneryDestructionZone(destruction_level=100, zone=trigger_zone.id)
)
self.m.triggerrules.triggers.append(t)
def register_scenery(self) -> None:
scenery = self.ground_object
assert isinstance(scenery, SceneryGroundObject)
self.unit_map.add_scenery(scenery)
class GenericCarrierGenerator(GenericGroundObjectGenerator[GenericCarrierGroundObject]):
"""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],
unit_map: UnitMap,
) -> None:
super().__init__(ground_object, country, game, mission, unit_map)
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, TacanUsage.TransmitReceive
)
tacan_callsign = self.tacan_callsign()
icls = next(self.icls_alloc)
# Always steam into the wind, even if the carrier is being moved.
# There are multiple unsimulated hours between turns, so we can
# count those as the time the carrier uses to move and the mission
# time as the recovery window.
brc = self.steam_into_wind(ship_group)
self.activate_beacons(ship_group, tacan, tacan_callsign, icls)
self.add_runway_data(
brc or Heading.from_degrees(0), atc, tacan, tacan_callsign, icls
)
self._register_unit_group(group, ship_group)
def get_carrier_type(self, group: ShipGroup) -> Type[ShipType]:
return ship_type_from_name(group.units[0].type)
def configure_carrier(
self, group: ShipGroup, 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 = group.units[0].name
return ship_group
def create_ship(self, unit: Unit, atc_channel: RadioFrequency) -> Ship:
ship = Ship(
self.m.next_unit_id(),
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[Heading]:
wind = self.game.conditions.weather.wind.at_0m
brc = Heading.from_degrees(wind.direction).opposite
# Aim for 25kts over the deck.
carrier_speed = knots(25) - mps(wind.speed)
for attempt in range(5):
point = group.points[0].position.point_from_heading(
brc.degrees, 100000 - attempt * 20000
)
if self.game.theater.is_in_sea(point):
group.points[0].speed = carrier_speed.meters_per_second
group.add_waypoint(point, carrier_speed.kph)
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: Heading,
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: ShipGroup) -> Type[ShipType]:
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.
if self.control_point.name == "Carrier Strike Group 8":
return "TRU"
else:
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[ShipGroundObject]):
"""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
self.generate_group(group, ship_type_from_name(group.units[0].type))
def generate_group(
self, group_def: ShipGroup, first_unit_type: Type[ShipType]
) -> None:
group = self.m.ship_group(
self.country,
group_def.name,
first_unit_type,
position=group_def.position,
heading=group_def.units[0].heading,
)
group.units[0].name = 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(), 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)
self._register_unit_group(group_def, group)
class HelipadGenerator:
"""
Generates helipads for given control point
"""
def __init__(
self,
mission: Mission,
cp: ControlPoint,
game: Game,
radio_registry: RadioRegistry,
tacan_registry: TacanRegistry,
):
self.m = mission
self.cp = cp
self.game = game
self.radio_registry = radio_registry
self.tacan_registry = tacan_registry
self.helipads: list[StaticGroup] = []
def generate(self) -> None:
# Note : Helipad are generated as neutral object in order not to interfer with capture triggers
neutral_country = self.m.country(self.game.neutral_country.name)
country = self.m.country(self.game.coalition_for(self.cp.captured).country_name)
for i, helipad in enumerate(self.cp.helipads):
name = self.cp.name + "_helipad_" + str(i)
logging.info("Generating helipad static : " + name)
pad = InvisibleFARP(name=name)
pad.position = Point(helipad.x, helipad.y)
pad.heading = helipad.heading.degrees
sg = unitgroup.StaticGroup(self.m.next_group_id(), name)
sg.add_unit(pad)
sp = StaticPoint()
sp.position = pad.position
sg.add_point(sp)
neutral_country.add_static_group(sg)
self.helipads.append(sg)
# Generate a FARP Ammo and Fuel stack for each pad
self.m.static_group(
country=country,
name=(name + "_fuel"),
_type=Fortification.FARP_Fuel_Depot,
position=pad.position.point_from_heading(helipad.heading.degrees, 35),
heading=pad.heading,
)
self.m.static_group(
country=country,
name=(name + "_ammo"),
_type=Fortification.FARP_Ammo_Dump_Coating,
position=pad.position.point_from_heading(
helipad.heading.degrees, 35
).point_from_heading(helipad.heading.degrees + 90, 10),
heading=pad.heading,
)
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.
"""
def __init__(
self,
mission: Mission,
game: Game,
radio_registry: RadioRegistry,
tacan_registry: TacanRegistry,
unit_map: UnitMap,
) -> None:
self.m = mission
self.game = game
self.radio_registry = radio_registry
self.tacan_registry = tacan_registry
self.unit_map = unit_map
self.icls_alloc = iter(range(1, 21))
self.runways: Dict[str, RunwayData] = {}
self.helipads: dict[ControlPoint, list[StaticGroup]] = defaultdict(list)
def generate(self) -> None:
for cp in self.game.theater.controlpoints:
country = self.m.country(self.game.coalition_for(cp.captured).country_name)
# Generate helipads
helipad_gen = HelipadGenerator(
self.m, cp, self.game, self.radio_registry, self.tacan_registry
)
helipad_gen.generate()
self.helipads[cp] = helipad_gen.helipads
for ground_object in cp.ground_objects:
generator: GenericGroundObjectGenerator[Any]
if isinstance(ground_object, FactoryGroundObject):
generator = FactoryGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
elif isinstance(ground_object, SceneryGroundObject):
generator = SceneryGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
elif isinstance(ground_object, BuildingGroundObject):
generator = BuildingSiteGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
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,
self.unit_map,
)
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,
self.unit_map,
)
elif isinstance(ground_object, ShipGroundObject):
generator = ShipObjectGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
elif isinstance(ground_object, MissileSiteGroundObject):
generator = MissileSiteGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
else:
generator = GenericGroundObjectGenerator(
ground_object, country, self.game, self.m, self.unit_map
)
generator.generate()

746
gen/kneeboard.py
View File

@@ -1,746 +0,0 @@
"""Generates kneeboard pages relevant to the player's mission.
The player kneeboard includes the following information:
* Airfield (departure, arrival, divert) info.
* Flight plan (waypoint numbers, names, altitudes).
* Comm channels.
* AWACS info.
* Tanker info.
* JTAC info.
Things we should add:
* Flight plan ToT and fuel ladder (current have neither available).
* Support for planning an arrival/divert airfield separate from departure.
* Mission package infrastructure to include information about the larger
mission, i.e. information about the escort flight for a strike package.
* Target information. Steerpoints, preplanned objectives, ToT, etc.
For multiplayer missions, a kneeboard will be generated per flight.
https://forums.eagle.ru/showthread.php?t=206360 claims that kneeboard pages can
only be added per airframe, so PvP missions where each side have the same
aircraft will be able to see the enemy's kneeboard for the same airframe.
"""
import datetime
import math
import textwrap
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional, TYPE_CHECKING, Tuple, Iterator
from PIL import Image, ImageDraw, ImageFont
from dcs.mission import Mission
from dcs.unit import Unit
from tabulate import tabulate
from game.data.alic import AlicCodes
from game.db import unit_type_from_name
from game.dcs.aircrafttype import AircraftType
from game.theater import ConflictTheater, TheaterGroundObject, LatLon
from game.theater.bullseye import Bullseye
from game.utils import meters
from game.weather import Weather
from .aircraft import FlightData
from .airsupportgen import AwacsInfo, TankerInfo
from .briefinggen import CommInfo, JtacInfo, MissionInfoGenerator
from game.ato.flighttype import FlightType
from game.ato.flightwaypointtype import FlightWaypointType
from game.ato.flightwaypoint import FlightWaypoint
from .radios import RadioFrequency
from .runways import RunwayData
if TYPE_CHECKING:
from game import Game
class KneeboardPageWriter:
"""Creates kneeboard images."""
def __init__(
self, page_margin: int = 24, line_spacing: int = 12, dark_theme: bool = False
) -> None:
if dark_theme:
self.foreground_fill = (215, 200, 200)
self.background_fill = (10, 5, 5)
else:
self.foreground_fill = (15, 15, 15)
self.background_fill = (255, 252, 252)
self.image_size = (768, 1024)
self.image = Image.new("RGB", self.image_size, self.background_fill)
# These font sizes create a relatively full page for current sorties. If
# we start generating more complicated flight plans, or start including
# more information in the comm ladder (the latter of which we should
# probably do), we'll need to split some of this information off into a
# second page.
self.title_font = ImageFont.truetype(
"arial.ttf", 32, layout_engine=ImageFont.LAYOUT_BASIC
)
self.heading_font = ImageFont.truetype(
"arial.ttf", 24, layout_engine=ImageFont.LAYOUT_BASIC
)
self.content_font = ImageFont.truetype(
"arial.ttf", 16, layout_engine=ImageFont.LAYOUT_BASIC
)
self.table_font = ImageFont.truetype(
"resources/fonts/Inconsolata.otf", 20, layout_engine=ImageFont.LAYOUT_BASIC
)
self.draw = ImageDraw.Draw(self.image)
self.page_margin = page_margin
self.x = page_margin
self.y = page_margin
self.line_spacing = line_spacing
@property
def position(self) -> Tuple[int, int]:
return self.x, self.y
def text(
self,
text: str,
font: Optional[ImageFont.FreeTypeFont] = None,
fill: Optional[Tuple[int, int, int]] = None,
wrap: bool = False,
) -> None:
if font is None:
font = self.content_font
if fill is None:
fill = self.foreground_fill
if wrap:
text = "\n".join(
self.wrap_line_with_font(
line, self.image_size[0] - self.page_margin - self.x, font
)
for line in text.splitlines()
)
self.draw.text(self.position, text, font=font, fill=fill)
width, height = self.draw.textsize(text, font=font)
self.y += height + self.line_spacing
def title(self, title: str) -> None:
self.text(title, font=self.title_font, fill=self.foreground_fill)
def heading(self, text: str) -> None:
self.text(text, font=self.heading_font, fill=self.foreground_fill)
def table(
self,
cells: List[List[str]],
headers: Optional[List[str]] = None,
font: Optional[ImageFont.FreeTypeFont] = None,
) -> None:
if headers is None:
headers = []
if font is None:
font = self.table_font
table = tabulate(cells, headers=headers, numalign="right")
self.text(table, font, fill=self.foreground_fill)
def write(self, path: Path) -> None:
self.image.save(path)
@staticmethod
def wrap_line(inputstr: str, max_length: int) -> str:
if len(inputstr) <= max_length:
return inputstr
tokens = inputstr.split(" ")
output = ""
segments = []
for token in tokens:
combo = output + " " + token
if len(combo) > max_length:
combo = output + "\n" + token
segments.append(combo)
output = ""
else:
output = combo
return "".join(segments + [output]).strip()
@staticmethod
def wrap_line_with_font(
inputstr: str, max_width: int, font: ImageFont.FreeTypeFont
) -> str:
if font.getsize(inputstr)[0] <= max_width:
return inputstr
tokens = inputstr.split(" ")
output = ""
segments = []
for token in tokens:
combo = output + " " + token
if font.getsize(combo)[0] > max_width:
segments.append(output + "\n")
output = token
else:
output = combo
return "".join(segments + [output]).strip()
class KneeboardPage:
"""Base class for all kneeboard pages."""
def write(self, path: Path) -> None:
"""Writes the kneeboard page to the given path."""
raise NotImplementedError
@staticmethod
def format_ll(ll: LatLon) -> str:
ns = "N" if ll.latitude >= 0 else "S"
ew = "E" if ll.longitude >= 0 else "W"
return f"{ll.latitude:.4}°{ns} {ll.longitude:.4}°{ew}"
@dataclass(frozen=True)
class NumberedWaypoint:
number: int
waypoint: FlightWaypoint
class FlightPlanBuilder:
WAYPOINT_DESC_MAX_LEN = 25
def __init__(self, start_time: datetime.datetime) -> None:
self.start_time = start_time
self.rows: List[List[str]] = []
self.target_points: List[NumberedWaypoint] = []
self.last_waypoint: Optional[FlightWaypoint] = None
def add_waypoint(self, waypoint_num: int, waypoint: FlightWaypoint) -> None:
if waypoint.waypoint_type == FlightWaypointType.TARGET_POINT:
self.target_points.append(NumberedWaypoint(waypoint_num, waypoint))
return
if self.target_points:
self.coalesce_target_points()
self.target_points = []
self.add_waypoint_row(NumberedWaypoint(waypoint_num, waypoint))
self.last_waypoint = waypoint
def coalesce_target_points(self) -> None:
if len(self.target_points) <= 4:
for steerpoint in self.target_points:
self.add_waypoint_row(steerpoint)
return
first_waypoint_num = self.target_points[0].number
last_waypoint_num = self.target_points[-1].number
self.rows.append(
[
f"{first_waypoint_num}-{last_waypoint_num}",
"Target points",
"0",
self._waypoint_distance(self.target_points[0].waypoint),
self._ground_speed(self.target_points[0].waypoint),
self._format_time(self.target_points[0].waypoint.tot),
self._format_time(self.target_points[0].waypoint.departure_time),
self._format_min_fuel(self.target_points[0].waypoint.min_fuel),
]
)
self.last_waypoint = self.target_points[-1].waypoint
def add_waypoint_row(self, waypoint: NumberedWaypoint) -> None:
self.rows.append(
[
str(waypoint.number),
KneeboardPageWriter.wrap_line(
waypoint.waypoint.pretty_name,
FlightPlanBuilder.WAYPOINT_DESC_MAX_LEN,
),
str(int(waypoint.waypoint.alt.feet)),
self._waypoint_distance(waypoint.waypoint),
self._ground_speed(waypoint.waypoint),
self._format_time(waypoint.waypoint.tot),
self._format_time(waypoint.waypoint.departure_time),
self._format_min_fuel(waypoint.waypoint.min_fuel),
]
)
def _format_time(self, time: Optional[datetime.timedelta]) -> str:
if time is None:
return ""
local_time = self.start_time + time
return local_time.strftime(f"%H:%M:%S")
def _waypoint_distance(self, waypoint: FlightWaypoint) -> str:
if self.last_waypoint is None:
return "-"
distance = meters(
self.last_waypoint.position.distance_to_point(waypoint.position)
)
return f"{distance.nautical_miles:.1f} NM"
def _ground_speed(self, waypoint: FlightWaypoint) -> str:
if self.last_waypoint is None:
return "-"
if waypoint.tot is None:
return "-"
if self.last_waypoint.departure_time is not None:
last_time = self.last_waypoint.departure_time
elif self.last_waypoint.tot is not None:
last_time = self.last_waypoint.tot
else:
return "-"
distance = meters(
self.last_waypoint.position.distance_to_point(waypoint.position)
)
duration = (waypoint.tot - last_time).total_seconds() / 3600
return f"{int(distance.nautical_miles / duration)} kt"
@staticmethod
def _format_min_fuel(min_fuel: Optional[float]) -> str:
if min_fuel is None:
return ""
return str(math.ceil(min_fuel / 100) * 100)
def build(self) -> List[List[str]]:
return self.rows
class BriefingPage(KneeboardPage):
"""A kneeboard page containing briefing information."""
def __init__(
self,
flight: FlightData,
bullseye: Bullseye,
theater: ConflictTheater,
weather: Weather,
start_time: datetime.datetime,
dark_kneeboard: bool,
) -> None:
self.flight = flight
self.bullseye = bullseye
self.theater = theater
self.weather = weather
self.start_time = start_time
self.dark_kneeboard = dark_kneeboard
self.flight_plan_font = ImageFont.truetype(
"resources/fonts/Inconsolata.otf",
16,
layout_engine=ImageFont.LAYOUT_BASIC,
)
def write(self, path: Path) -> None:
writer = KneeboardPageWriter(dark_theme=self.dark_kneeboard)
if self.flight.custom_name is not None:
custom_name_title = ' ("{}")'.format(self.flight.custom_name)
else:
custom_name_title = ""
writer.title(f"{self.flight.callsign} Mission Info{custom_name_title}")
# TODO: Handle carriers.
writer.heading("Airfield Info")
writer.table(
[
self.airfield_info_row("Departure", self.flight.departure),
self.airfield_info_row("Arrival", self.flight.arrival),
self.airfield_info_row("Divert", self.flight.divert),
],
headers=["", "Airbase", "ATC", "TCN", "I(C)LS", "RWY"],
)
writer.heading("Flight Plan")
flight_plan_builder = FlightPlanBuilder(self.start_time)
for num, waypoint in enumerate(self.flight.waypoints):
flight_plan_builder.add_waypoint(num, waypoint)
writer.table(
flight_plan_builder.build(),
headers=[
"#",
"Action",
"Alt",
"Dist",
"GSPD",
"Time",
"Departure",
"Min fuel",
],
font=self.flight_plan_font,
)
writer.text(f"Bullseye: {self.bullseye.to_lat_lon(self.theater).format_dms()}")
qnh_in_hg = f"{self.weather.atmospheric.qnh.inches_hg:.2f}"
qnh_mm_hg = f"{self.weather.atmospheric.qnh.mm_hg:.1f}"
qnh_hpa = f"{self.weather.atmospheric.qnh.hecto_pascals:.1f}"
writer.text(
f"Temperature: {round(self.weather.atmospheric.temperature_celsius)} °C at sea level"
)
writer.text(f"QNH: {qnh_in_hg} inHg / {qnh_mm_hg} mmHg / {qnh_hpa} hPa")
writer.table(
[
[
"{}lbs".format(self.flight.bingo_fuel),
"{}lbs".format(self.flight.joker_fuel),
]
],
["Bingo", "Joker"],
)
if any(self.flight.laser_codes):
codes: list[list[str]] = []
for idx, code in enumerate(self.flight.laser_codes, start=1):
codes.append([str(idx), "" if code is None else str(code)])
writer.table(codes, ["#", "Laser Code"])
writer.write(path)
def airfield_info_row(
self, row_title: str, runway: Optional[RunwayData]
) -> List[str]:
"""Creates a table row for a given airfield.
Args:
row_title: Purpose of the airfield. e.g. "Departure", "Arrival" or
"Divert".
runway: The runway described by this row.
Returns:
A list of strings to be used as a row of the airfield table.
"""
if runway is None:
return [row_title, "", "", "", "", ""]
atc = ""
if runway.atc is not None:
atc = self.format_frequency(runway.atc)
if runway.tacan is None:
tacan = ""
else:
tacan = str(runway.tacan)
if runway.ils is not None:
ils = str(runway.ils)
elif runway.icls is not None:
ils = str(runway.icls)
else:
ils = ""
return [
row_title,
"\n".join(textwrap.wrap(runway.airfield_name, width=24)),
atc,
tacan,
ils,
runway.runway_name,
]
def format_frequency(self, frequency: RadioFrequency) -> str:
channel = self.flight.channel_for(frequency)
if channel is None:
return str(frequency)
channel_name = self.flight.aircraft_type.channel_name(
channel.radio_id, channel.channel
)
return f"{channel_name}\n{frequency}"
class SupportPage(KneeboardPage):
"""A kneeboard page containing information about support units."""
JTAC_REGION_MAX_LEN = 25
def __init__(
self,
flight: FlightData,
comms: List[CommInfo],
awacs: List[AwacsInfo],
tankers: List[TankerInfo],
jtacs: List[JtacInfo],
start_time: datetime.datetime,
dark_kneeboard: bool,
) -> None:
self.flight = flight
self.comms = list(comms)
self.awacs = awacs
self.tankers = tankers
self.jtacs = jtacs
self.start_time = start_time
self.dark_kneeboard = dark_kneeboard
self.comms.append(CommInfo("Flight", self.flight.intra_flight_channel))
def write(self, path: Path) -> None:
writer = KneeboardPageWriter(dark_theme=self.dark_kneeboard)
if self.flight.custom_name is not None:
custom_name_title = ' ("{}")'.format(self.flight.custom_name)
else:
custom_name_title = ""
writer.title(f"{self.flight.callsign} Support Info{custom_name_title}")
# AEW&C
writer.heading("AEW&C")
aewc_ladder = []
for single_aewc in self.awacs:
if single_aewc.depature_location is None:
dep = "-"
arr = "-"
else:
dep = self._format_time(single_aewc.start_time)
arr = self._format_time(single_aewc.end_time)
aewc_ladder.append(
[
str(single_aewc.callsign),
self.format_frequency(single_aewc.freq),
str(single_aewc.depature_location),
str(dep),
str(arr),
]
)
writer.table(
aewc_ladder,
headers=["Callsign", "FREQ", "Depature", "ETD", "ETA"],
)
# Package Section
writer.heading("Comm ladder")
comm_ladder = []
for comm in self.comms:
comm_ladder.append(
[comm.name, "", "", "", self.format_frequency(comm.freq)]
)
for tanker in self.tankers:
comm_ladder.append(
[
tanker.callsign,
"Tanker",
tanker.variant,
str(tanker.tacan),
self.format_frequency(tanker.freq),
]
)
writer.table(comm_ladder, headers=["Callsign", "Task", "Type", "TACAN", "FREQ"])
writer.heading("JTAC")
jtacs = []
for jtac in self.jtacs:
jtacs.append(
[
jtac.callsign,
KneeboardPageWriter.wrap_line(
jtac.region,
self.JTAC_REGION_MAX_LEN,
),
jtac.code,
self.format_frequency(jtac.freq),
]
)
writer.table(jtacs, headers=["Callsign", "Region", "Laser Code", "FREQ"])
writer.write(path)
def format_frequency(self, frequency: RadioFrequency) -> str:
channel = self.flight.channel_for(frequency)
if channel is None:
return str(frequency)
channel_name = self.flight.aircraft_type.channel_name(
channel.radio_id, channel.channel
)
return f"{channel_name}\n{frequency}"
def _format_time(self, time: Optional[datetime.timedelta]) -> str:
if time is None:
return ""
local_time = self.start_time + time
return local_time.strftime(f"%H:%M:%S")
class SeadTaskPage(KneeboardPage):
"""A kneeboard page containing SEAD/DEAD target information."""
def __init__(
self, flight: FlightData, dark_kneeboard: bool, theater: ConflictTheater
) -> None:
self.flight = flight
self.dark_kneeboard = dark_kneeboard
self.theater = theater
@property
def target_units(self) -> Iterator[Unit]:
if isinstance(self.flight.package.target, TheaterGroundObject):
yield from self.flight.package.target.units
@staticmethod
def alic_for(unit: Unit) -> str:
try:
return str(AlicCodes.code_for(unit))
except KeyError:
return ""
def write(self, path: Path) -> None:
writer = KneeboardPageWriter(dark_theme=self.dark_kneeboard)
if self.flight.custom_name is not None:
custom_name_title = ' ("{}")'.format(self.flight.custom_name)
else:
custom_name_title = ""
task = "DEAD" if self.flight.flight_type == FlightType.DEAD else "SEAD"
writer.title(f"{self.flight.callsign} {task} Target Info{custom_name_title}")
writer.table(
[self.target_info_row(t) for t in self.target_units],
headers=["Description", "ALIC", "Location"],
)
writer.write(path)
def target_info_row(self, unit: Unit) -> List[str]:
ll = self.theater.point_to_ll(unit.position)
unit_type = unit_type_from_name(unit.type)
name = unit.name if unit_type is None else unit_type.name
return [name, self.alic_for(unit), ll.format_dms(include_decimal_seconds=True)]
class StrikeTaskPage(KneeboardPage):
"""A kneeboard page containing strike target information."""
def __init__(
self,
flight: FlightData,
dark_kneeboard: bool,
theater: ConflictTheater,
) -> None:
self.flight = flight
self.dark_kneeboard = dark_kneeboard
self.theater = theater
@property
def targets(self) -> Iterator[NumberedWaypoint]:
for idx, waypoint in enumerate(self.flight.waypoints):
if waypoint.waypoint_type == FlightWaypointType.TARGET_POINT:
yield NumberedWaypoint(idx, waypoint)
def write(self, path: Path) -> None:
writer = KneeboardPageWriter(dark_theme=self.dark_kneeboard)
if self.flight.custom_name is not None:
custom_name_title = ' ("{}")'.format(self.flight.custom_name)
else:
custom_name_title = ""
writer.title(f"{self.flight.callsign} Strike Task Info{custom_name_title}")
writer.table(
[self.target_info_row(t) for t in self.targets],
headers=["Steerpoint", "Description", "Location"],
)
writer.write(path)
def target_info_row(self, target: NumberedWaypoint) -> List[str]:
ll = self.theater.point_to_ll(target.waypoint.position)
return [
str(target.number),
target.waypoint.pretty_name,
ll.format_dms(include_decimal_seconds=True),
]
class NotesPage(KneeboardPage):
"""A kneeboard page containing the campaign owner's notes."""
def __init__(
self,
notes: str,
dark_kneeboard: bool,
) -> None:
self.notes = notes
self.dark_kneeboard = dark_kneeboard
def write(self, path: Path) -> None:
writer = KneeboardPageWriter(dark_theme=self.dark_kneeboard)
writer.title(f"Notes")
writer.text(self.notes, wrap=True)
writer.write(path)
class KneeboardGenerator(MissionInfoGenerator):
"""Creates kneeboard pages for each client flight in the mission."""
def __init__(self, mission: Mission, game: "Game") -> None:
super().__init__(mission, game)
self.dark_kneeboard = self.game.settings.generate_dark_kneeboard and (
self.mission.start_time.hour > 19 or self.mission.start_time.hour < 7
)
def generate(self) -> None:
"""Generates a kneeboard per client flight."""
temp_dir = Path("kneeboards")
temp_dir.mkdir(exist_ok=True)
for aircraft, pages in self.pages_by_airframe().items():
aircraft_dir = temp_dir / aircraft.dcs_unit_type.id
aircraft_dir.mkdir(exist_ok=True)
for idx, page in enumerate(pages):
page_path = aircraft_dir / f"page{idx:02}.png"
page.write(page_path)
self.mission.add_aircraft_kneeboard(aircraft.dcs_unit_type, page_path)
def pages_by_airframe(self) -> Dict[AircraftType, List[KneeboardPage]]:
"""Returns a list of kneeboard pages per airframe in the mission.
Only client flights will be included, but because DCS does not support
group-specific kneeboard pages, flights (possibly from opposing sides)
will be able to see the kneeboards of all aircraft of the same type.
Returns:
A dict mapping aircraft types to the list of kneeboard pages for
that aircraft.
"""
all_flights: Dict[AircraftType, List[KneeboardPage]] = defaultdict(list)
for flight in self.flights:
if not flight.client_units:
continue
all_flights[flight.aircraft_type].extend(
self.generate_flight_kneeboard(flight)
)
return all_flights
def generate_task_page(self, flight: FlightData) -> Optional[KneeboardPage]:
if flight.flight_type in (FlightType.DEAD, FlightType.SEAD):
return SeadTaskPage(flight, self.dark_kneeboard, self.game.theater)
elif flight.flight_type is FlightType.STRIKE:
return StrikeTaskPage(flight, self.dark_kneeboard, self.game.theater)
return None
def generate_flight_kneeboard(self, flight: FlightData) -> List[KneeboardPage]:
"""Returns a list of kneeboard pages for the given flight."""
pages: List[KneeboardPage] = [
BriefingPage(
flight,
self.game.bullseye_for(flight.friendly),
self.game.theater,
self.game.conditions.weather,
self.mission.start_time,
self.dark_kneeboard,
),
SupportPage(
flight,
self.comms,
self.awacs,
self.tankers,
self.jtacs,
self.mission.start_time,
self.dark_kneeboard,
),
]
# Only create the notes page if there are notes to show.
if notes := self.game.notes:
pages.append(NotesPage(notes, self.dark_kneeboard))
if (target_page := self.generate_task_page(flight)) is not None:
pages.append(target_page)
return pages

37
gen/lasercoderegistry.py
View File

@@ -1,37 +0,0 @@
from collections import deque
from typing import Iterator
class OutOfLaserCodesError(RuntimeError):
def __init__(self) -> None:
super().__init__(
f"All JTAC laser codes have been allocated. No available codes."
)
class LaserCodeRegistry:
def __init__(self) -> None:
self.allocated_codes: set[int] = set()
self.allocator: Iterator[int] = LaserCodeRegistry.__laser_code_generator()
def get_next_laser_code(self) -> int:
try:
while (code := next(self.allocator)) in self.allocated_codes:
pass
self.allocated_codes.add(code)
return code
except StopIteration:
raise OutOfLaserCodesError
@staticmethod
def __laser_code_generator() -> Iterator[int]:
# Valid laser codes are as follows
# First digit is always 1
# Second digit is 5-7
# Third and fourth digits are 1 - 8
# We iterate backward (reversed()) so that 1687 follows 1688
q = deque(int(oct(code)[2:]) + 11 for code in reversed(range(0o1500, 0o2000)))
# We start with the default of 1688 and wrap around when we reach the end
q.rotate(-q.index(1688))
return iter(q)

273
gen/radios.py
View File

@@ -1,273 +0,0 @@
"""Radio frequency types and allocators."""
import itertools
import logging
from dataclasses import dataclass
from typing import Dict, FrozenSet, Iterator, List, Reversible, Set, Tuple
@dataclass(frozen=True)
class RadioFrequency:
"""A radio frequency.
Not currently concerned with tracking modulation, just the frequency.
"""
#: The frequency in kilohertz.
hertz: int
def __str__(self) -> str:
if self.hertz >= 1000000:
return self.format("MHz", 1000000)
return self.format("kHz", 1000)
def format(self, units: str, divisor: int) -> str:
converted = self.hertz / divisor
if converted.is_integer():
return f"{int(converted)} {units}"
return f"{converted:0.3f} {units}"
@property
def mhz(self) -> float:
"""Returns the frequency in megahertz.
Returns:
The frequency in megahertz.
"""
return self.hertz / 1000000
def MHz(num: int, khz: int = 0) -> RadioFrequency:
return RadioFrequency(num * 1000000 + khz * 1000)
def kHz(num: int) -> RadioFrequency:
return RadioFrequency(num * 1000)
@dataclass(frozen=True)
class RadioRange:
"""Defines the minimum (inclusive) and maximum (exclusive) range of the radio."""
#: The minimum (inclusive) frequency tunable by this radio.
minimum: RadioFrequency
#: The maximum (exclusive) frequency tunable by this radio.
maximum: RadioFrequency
#: The spacing between adjacent frequencies.
step: RadioFrequency
#: Specific frequencies to exclude. (e.g. Guard channels)
excludes: FrozenSet[RadioFrequency] = frozenset()
def range(self) -> Iterator[RadioFrequency]:
"""Returns an iterator over the usable frequencies of this radio."""
return (
RadioFrequency(x)
for x in range(self.minimum.hertz, self.maximum.hertz, self.step.hertz)
if RadioFrequency(x) not in self.excludes
)
@property
def last_channel(self) -> RadioFrequency:
return next(
RadioFrequency(x)
for x in reversed(
range(self.minimum.hertz, self.maximum.hertz, self.step.hertz)
)
if RadioFrequency(x) not in self.excludes
)
@dataclass(frozen=True)
class Radio:
"""A radio.
Defines ranges of usable frequencies of the radio.
"""
#: The name of the radio.
name: str
#: List of usable frequency range of this radio.
ranges: Tuple[RadioRange, ...]
def __str__(self) -> str:
return self.name
def range(self) -> Iterator[RadioFrequency]:
"""Returns an iterator over the usable frequencies of this radio."""
return itertools.chain.from_iterable(rng.range() for rng in self.ranges)
@property
def last_channel(self) -> RadioFrequency:
return self.ranges[-1].last_channel
class ChannelInUseError(RuntimeError):
"""Raised when attempting to reserve an in-use frequency."""
def __init__(self, frequency: RadioFrequency) -> None:
super().__init__(f"{frequency} is already in use")
# TODO: Figure out appropriate steps for each radio. These are just guesses.
#: List of all known radios used by aircraft in the game.
RADIOS: List[Radio] = [
Radio("AN/ARC-164", (RadioRange(MHz(225), MHz(400), step=MHz(1)),)),
Radio("AN/ARC-186(V) AM", (RadioRange(MHz(116), MHz(152), step=MHz(1)),)),
Radio("AN/ARC-186(V) FM", (RadioRange(MHz(30), MHz(76), step=MHz(1)),)),
Radio(
"AN/ARC-210",
(
RadioRange(MHz(225), MHz(400), MHz(1), frozenset((MHz(243),))),
RadioRange(MHz(136), MHz(155), MHz(1)),
RadioRange(MHz(156), MHz(174), MHz(1)),
RadioRange(MHz(118), MHz(136), MHz(1)),
RadioRange(MHz(30), MHz(88), MHz(1)),
),
),
Radio("AN/ARC-222", (RadioRange(MHz(116), MHz(174), step=MHz(1)),)),
Radio("SCR-522", (RadioRange(MHz(100), MHz(156), step=MHz(1)),)),
Radio("A.R.I. 1063", (RadioRange(MHz(100), MHz(156), step=MHz(1)),)),
Radio("BC-1206", (RadioRange(kHz(200), kHz(400), step=kHz(10)),)),
# Note: The M2000C V/UHF can operate in both ranges, but has a gap between
# 150 MHz and 225 MHz. We can't allocate in that gap, and the current
# system doesn't model gaps, so just pretend it ends at 150 MHz for now. We
# can model gaps later if needed.
Radio("TRT ERA 7000 V/UHF", (RadioRange(MHz(118), MHz(150), step=MHz(1)),)),
Radio("TRT ERA 7200 UHF", (RadioRange(MHz(225), MHz(400), step=MHz(1)),)),
# Tomcat radios
# # https://www.heatblur.se/F-14Manual/general.html#an-arc-159-uhf-1-radio
Radio("AN/ARC-159", (RadioRange(MHz(225), MHz(400), step=MHz(1)),)),
# AN/ARC-182 can also operate from 30 MHz to 88 MHz, as well as from 225 MHz
# to 400 MHz range, but we can't model gaps with the current implementation.
# https://www.heatblur.se/F-14Manual/general.html#an-arc-182-v-uhf-2-radio
Radio("AN/ARC-182", (RadioRange(MHz(108), MHz(174), step=MHz(1)),)),
# Also capable of [103, 156) at 25 kHz intervals, but we can't do gaps.
Radio("FR 22", (RadioRange(MHz(225), MHz(400), step=kHz(50)),)),
# P-51 / P-47 Radio
# 4 preset channels (A/B/C/D)
Radio("SCR522", (RadioRange(MHz(100), MHz(156), step=kHz(25)),)),
Radio("R&S M3AR VHF", (RadioRange(MHz(120), MHz(174), step=MHz(1)),)),
Radio("R&S M3AR UHF", (RadioRange(MHz(225), MHz(400), step=MHz(1)),)),
# MiG-15bis
Radio("RSI-6K HF", (RadioRange(MHz(3, 750), MHz(5), step=kHz(25)),)),
# MiG-19P
Radio("RSIU-4V", (RadioRange(MHz(100), MHz(150), step=MHz(1)),)),
# MiG-21bis
Radio("RSIU-5V", (RadioRange(MHz(118), MHz(140), step=MHz(1)),)),
# Ka-50
# Note: Also capable of 100MHz-150MHz, but we can't model gaps.
Radio("R-800L1", (RadioRange(MHz(220), MHz(400), step=kHz(25)),)),
Radio("R-828", (RadioRange(MHz(20), MHz(60), step=kHz(25)),)),
# UH-1H
Radio("AN/ARC-51BX", (RadioRange(MHz(225), MHz(400), step=kHz(50)),)),
Radio("AN/ARC-131", (RadioRange(MHz(30), MHz(76), step=kHz(50)),)),
Radio("AN/ARC-134", (RadioRange(MHz(116), MHz(150), step=kHz(25)),)),
Radio("R&S Series 6000", (RadioRange(MHz(100), MHz(156), step=kHz(25)),)),
]
def get_radio(name: str) -> Radio:
"""Returns the radio with the given name.
Args:
name: Name of the radio to return.
Returns:
The radio matching name.
Raises:
KeyError: No matching radio was found.
"""
for radio in RADIOS:
if radio.name == name:
return radio
raise KeyError(f"Unknown radio: {name}")
class RadioRegistry:
"""Manages allocation of radio channels.
There's some room for improvement here. We could prefer to allocate
frequencies that are available to the fewest number of radios first, so
radios with wide bands like the AN/ARC-210 don't exhaust all the channels
available to narrower radios like the AN/ARC-186(V). In practice there are
probably plenty of channels, so we can deal with that later if we need to.
We could also allocate using a larger increment, returning to smaller
increments each time the range is exhausted. This would help with the
previous problem, as the AN/ARC-186(V) would still have plenty of 25 kHz
increment channels left after the AN/ARC-210 moved on to the higher
frequencies. This would also look a little nicer than having every flight
allocated in the 30 MHz range.
"""
# Not a real radio, but useful for allocating a channel usable for
# inter-flight communications.
BLUFOR_UHF = Radio("BLUFOR UHF", (RadioRange(MHz(225), MHz(400), step=MHz(1)),))
def __init__(self) -> None:
self.allocated_channels: Set[RadioFrequency] = set()
self.radio_allocators: Dict[Radio, Iterator[RadioFrequency]] = {}
radios = itertools.chain(RADIOS, [self.BLUFOR_UHF])
for radio in radios:
self.radio_allocators[radio] = radio.range()
def alloc_for_radio(self, radio: Radio) -> RadioFrequency:
"""Allocates a radio channel tunable by the given radio.
Args:
radio: The radio to allocate a channel for.
Returns:
A radio channel compatible with the given radio.
Raises:
OutOfChannelsError: All channels compatible with the given radio are
already allocated.
"""
allocator = self.radio_allocators[radio]
try:
while (channel := next(allocator)) in self.allocated_channels:
pass
self.reserve(channel)
return channel
except StopIteration:
# In the event of too many channel users, fail gracefully by reusing
# the last channel.
# https://github.com/dcs-liberation/dcs_liberation/issues/598
channel = radio.last_channel
logging.warning(
f"No more free channels for {radio.name}. Reusing {channel}."
)
return channel
def alloc_uhf(self) -> RadioFrequency:
"""Allocates a UHF radio channel suitable for inter-flight comms.
Returns:
A UHF radio channel suitable for inter-flight comms.
Raises:
OutOfChannelsError: All channels compatible with the given radio are
already allocated.
"""
return self.alloc_for_radio(self.BLUFOR_UHF)
def reserve(self, frequency: RadioFrequency) -> None:
"""Reserves the given channel.
Reserving a channel ensures that it will not be allocated in the future.
Args:
frequency: The channel to reserve.
Raises:
ChannelInUseError: The given frequency is already in use.
"""
if frequency in self.allocated_channels:
raise ChannelInUseError(frequency)
self.allocated_channels.add(frequency)

4
gen/runways.py
View File

@@ -10,8 +10,8 @@ from dcs.terrain.terrain import Airport
from game.weather import Conditions
from game.utils import Heading
from .airfields import AIRFIELD_DATA
from .radios import RadioFrequency
from .tacan import TacanChannel
from game.radio.radios import RadioFrequency
from game.radio.tacan import TacanChannel
@dataclass(frozen=True)

112
gen/tacan.py
View File

@@ -1,112 +0,0 @@
"""TACAN channel handling."""
from dataclasses import dataclass
from enum import Enum
from typing import Dict, Iterator, Set
class TacanUsage(Enum):
TransmitReceive = "transmit receive"
AirToAir = "air to air"
class TacanBand(Enum):
X = "X"
Y = "Y"
def range(self) -> Iterator["TacanChannel"]:
"""Returns an iterator over the channels in this band."""
return (TacanChannel(x, self) for x in range(1, 126 + 1))
def valid_channels(self, usage: TacanUsage) -> Iterator["TacanChannel"]:
for x in self.range():
if x.number not in UNAVAILABLE[usage][self]:
yield x
# Avoid certain TACAN channels for various reasons
# https://forums.eagle.ru/topic/276390-datalink-issue/
UNAVAILABLE = {
TacanUsage.TransmitReceive: {
TacanBand.X: set(range(2, 30 + 1)) | set(range(47, 63 + 1)),
TacanBand.Y: set(range(2, 30 + 1)) | set(range(64, 92 + 1)),
},
TacanUsage.AirToAir: {
TacanBand.X: set(range(1, 36 + 1)) | set(range(64, 99 + 1)),
TacanBand.Y: set(range(1, 36 + 1)) | set(range(64, 99 + 1)),
},
}
@dataclass(frozen=True)
class TacanChannel:
number: int
band: TacanBand
def __str__(self) -> str:
return f"{self.number}{self.band.value}"
class OutOfTacanChannelsError(RuntimeError):
"""Raised when all channels in this band have been allocated."""
def __init__(self, band: TacanBand) -> None:
super().__init__(f"No available channels in TACAN {band.value} band")
class TacanChannelInUseError(RuntimeError):
"""Raised when attempting to reserve an in-use channel."""
def __init__(self, channel: TacanChannel) -> None:
super().__init__(f"{channel} is already in use")
class TacanRegistry:
"""Manages allocation of TACAN channels."""
def __init__(self) -> None:
self.allocated_channels: Set[TacanChannel] = set()
self.allocators: Dict[TacanBand, Dict[TacanUsage, Iterator[TacanChannel]]] = {}
for band in TacanBand:
self.allocators[band] = {}
for usage in TacanUsage:
self.allocators[band][usage] = band.valid_channels(usage)
def alloc_for_band(
self, band: TacanBand, intended_usage: TacanUsage
) -> TacanChannel:
"""Allocates a TACAN channel in the given band.
Args:
band: The TACAN band to allocate a channel for.
intended_usage: What the caller intends to use the tacan channel for.
Returns:
A TACAN channel in the given band.
Raises:
OutOfTacanChannelsError: All channels compatible with the given radio are
already allocated.
"""
allocator = self.allocators[band][intended_usage]
try:
while (channel := next(allocator)) in self.allocated_channels:
pass
return channel
except StopIteration:
raise OutOfTacanChannelsError(band)
def mark_unavailable(self, channel: TacanChannel) -> None:
"""Reserves the given channel.
Reserving a channel ensures that it will not be allocated in the future.
Args:
channel: The channel to reserve.
Raises:
TacanChannelInUseError: The given channel is already in use.
"""
if channel in self.allocated_channels:
raise TacanChannelInUseError(channel)
self.allocated_channels.add(channel)

209
gen/triggergen.py
View File

@@ -1,209 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from dcs.action import MarkToAll, SetFlag, DoScript, ClearFlag
from dcs.condition import (
TimeAfter,
AllOfCoalitionOutsideZone,
PartOfCoalitionInZone,
FlagIsFalse,
FlagIsTrue,
)
from dcs.mission import Mission
from dcs.task import Option
from dcs.translation import String
from dcs.triggers import (
Event,
TriggerOnce,
TriggerCondition,
)
from dcs.unit import Skill
from game.theater import Airfield
from game.theater.controlpoint import Fob
if TYPE_CHECKING:
from game.game import Game
PUSH_TRIGGER_SIZE = 3000
PUSH_TRIGGER_ACTIVATION_AGL = 25
REGROUP_ZONE_DISTANCE = 12000
REGROUP_ALT = 5000
TRIGGER_WAYPOINT_OFFSET = 2
TRIGGER_MIN_DISTANCE_FROM_START = 10000
# modified since we now have advanced SAM units
TRIGGER_RADIUS_MINIMUM = 3000000
TRIGGER_RADIUS_SMALL = 50000
TRIGGER_RADIUS_MEDIUM = 100000
TRIGGER_RADIUS_LARGE = 150000
TRIGGER_RADIUS_ALL_MAP = 3000000
class Silence(Option):
Key = 7
class TriggersGenerator:
capture_zone_types = (Fob,)
capture_zone_flag = 600
def __init__(self, mission: Mission, game: Game) -> None:
self.mission = mission
self.game = game
def _set_allegiances(self, player_coalition: str, enemy_coalition: str) -> None:
"""
Set airbase initial coalition
"""
# Empty neutrals airports
cp_ids = [cp.id for cp in self.game.theater.controlpoints]
for airport in self.mission.terrain.airport_list():
if airport.id not in cp_ids:
airport.unlimited_fuel = False
airport.unlimited_munitions = False
airport.unlimited_aircrafts = False
airport.gasoline_init = 0
airport.methanol_mixture_init = 0
airport.diesel_init = 0
airport.jet_init = 0
airport.operating_level_air = 0
airport.operating_level_equipment = 0
airport.operating_level_fuel = 0
for airport in self.mission.terrain.airport_list():
if airport.id not in cp_ids:
airport.unlimited_fuel = True
airport.unlimited_munitions = True
airport.unlimited_aircrafts = True
for cp in self.game.theater.controlpoints:
if isinstance(cp, Airfield):
cp_airport = self.mission.terrain.airport_by_id(cp.airport.id)
if cp_airport is None:
raise RuntimeError(
f"Could not find {cp.airport.name} in the mission"
)
cp_airport.set_coalition(
cp.captured and player_coalition or enemy_coalition
)
def _set_skill(self, player_coalition: str, enemy_coalition: str) -> None:
"""
Set skill level for all aircraft in the mission
"""
for coalition_name, coalition in self.mission.coalition.items():
if coalition_name == player_coalition:
skill_level = Skill(self.game.settings.player_skill)
elif coalition_name == enemy_coalition:
skill_level = Skill(self.game.settings.enemy_vehicle_skill)
else:
continue
for country in coalition.countries.values():
for vehicle_group in country.vehicle_group:
vehicle_group.set_skill(skill_level)
def _gen_markers(self) -> None:
"""
Generate markers on F10 map for each existing objective
"""
if self.game.settings.generate_marks:
mark_trigger = TriggerOnce(Event.NoEvent, "Marks generator")
mark_trigger.add_condition(TimeAfter(1))
v = 10
for cp in self.game.theater.controlpoints:
seen = set()
for ground_object in cp.ground_objects:
if ground_object.obj_name in seen:
continue
seen.add(ground_object.obj_name)
for location in ground_object.mark_locations:
zone = self.mission.triggers.add_triggerzone(
location, radius=10, hidden=True, name="MARK"
)
if cp.captured:
name = ground_object.obj_name + " [ALLY]"
else:
name = ground_object.obj_name + " [ENEMY]"
mark_trigger.add_action(MarkToAll(v, zone.id, String(name)))
v += 1
self.mission.triggerrules.triggers.append(mark_trigger)
def _generate_capture_triggers(
self, player_coalition: str, enemy_coalition: str
) -> None:
"""Creates a pair of triggers for each control point of `cls.capture_zone_types`.
One for the initial capture of a control point, and one if it is recaptured.
Directly appends to the global `base_capture_events` var declared by `dcs_libaration.lua`
"""
for cp in self.game.theater.controlpoints:
if isinstance(cp, self.capture_zone_types):
if cp.captured:
attacking_coalition = enemy_coalition
attack_coalition_int = 1 # 1 is the Event int for Red
defending_coalition = player_coalition
defend_coalition_int = 2 # 2 is the Event int for Blue
else:
attacking_coalition = player_coalition
attack_coalition_int = 2
defending_coalition = enemy_coalition
defend_coalition_int = 1
trigger_zone = self.mission.triggers.add_triggerzone(
cp.position, radius=3000, hidden=False, name="CAPTURE"
)
flag = self.get_capture_zone_flag()
capture_trigger = TriggerCondition(Event.NoEvent, "Capture Trigger")
capture_trigger.add_condition(
AllOfCoalitionOutsideZone(defending_coalition, trigger_zone.id)
)
capture_trigger.add_condition(
PartOfCoalitionInZone(
attacking_coalition, trigger_zone.id, unit_type="GROUND"
)
)
capture_trigger.add_condition(FlagIsFalse(flag=flag))
script_string = String(
f'base_capture_events[#base_capture_events + 1] = "{cp.id}||{attack_coalition_int}||{cp.full_name}"'
)
capture_trigger.add_action(DoScript(script_string))
capture_trigger.add_action(SetFlag(flag=flag))
self.mission.triggerrules.triggers.append(capture_trigger)
recapture_trigger = TriggerCondition(Event.NoEvent, "Capture Trigger")
recapture_trigger.add_condition(
AllOfCoalitionOutsideZone(attacking_coalition, trigger_zone.id)
)
recapture_trigger.add_condition(
PartOfCoalitionInZone(
defending_coalition, trigger_zone.id, unit_type="GROUND"
)
)
recapture_trigger.add_condition(FlagIsTrue(flag=flag))
script_string = String(
f'base_capture_events[#base_capture_events + 1] = "{cp.id}||{defend_coalition_int}||{cp.full_name}"'
)
recapture_trigger.add_action(DoScript(script_string))
recapture_trigger.add_action(ClearFlag(flag=flag))
self.mission.triggerrules.triggers.append(recapture_trigger)
def generate(self) -> None:
player_coalition = "blue"
enemy_coalition = "red"
self._set_skill(player_coalition, enemy_coalition)
self._set_allegiances(player_coalition, enemy_coalition)
self._gen_markers()
self._generate_capture_triggers(player_coalition, enemy_coalition)
@classmethod
def get_capture_zone_flag(cls) -> int:
flag = cls.capture_zone_flag
cls.capture_zone_flag += 1
return flag

108
gen/visualgen.py
View File

@@ -1,108 +0,0 @@
from __future__ import annotations
import random
from typing import TYPE_CHECKING, Any
from dcs.mission import Mission
from dcs.unit import Static
from dcs.unittype import StaticType
if TYPE_CHECKING:
from game import Game
from .conflictgen import Conflict
class MarkerSmoke(StaticType):
id = "big_smoke"
category = "Effects"
name = "big_smoke"
shape_name = 5 # type: ignore
rate = 0.1 # type: ignore
class Smoke(StaticType):
id = "big_smoke"
category = "Effects"
name = "big_smoke"
shape_name = 2 # type: ignore
rate = 1
class BigSmoke(StaticType):
id = "big_smoke"
category = "Effects"
name = "big_smoke"
shape_name = 3 # type: ignore
rate = 1
class MassiveSmoke(StaticType):
id = "big_smoke"
category = "Effects"
name = "big_smoke"
shape_name = 4 # type: ignore
rate = 1
def __monkey_static_dict(self: Static) -> dict[str, Any]:
global __original_static_dict
d = __original_static_dict(self)
if self.type == "big_smoke":
d["effectPreset"] = self.shape_name
d["effectTransparency"] = self.rate
return d
__original_static_dict = Static.dict
Static.dict = __monkey_static_dict # type: ignore
FRONT_SMOKE_RANDOM_SPREAD = 4000
FRONT_SMOKE_TYPE_CHANCES = {
2: MassiveSmoke,
15: BigSmoke,
30: Smoke,
100: Smoke,
}
class VisualGenerator:
def __init__(self, mission: Mission, game: Game) -> None:
self.mission = mission
self.game = game
def _generate_frontline_smokes(self) -> None:
for front_line in self.game.theater.conflicts():
from_cp = front_line.blue_cp
to_cp = front_line.red_cp
if from_cp.is_global or to_cp.is_global:
continue
plane_start, heading, distance = Conflict.frontline_vector(
front_line, self.game.theater
)
if not plane_start:
continue
for offset in range(0, distance, self.game.settings.perf_smoke_spacing):
position = plane_start.point_from_heading(heading.degrees, offset)
for k, v in FRONT_SMOKE_TYPE_CHANCES.items():
if random.randint(0, 100) <= k:
pos = position.random_point_within(
FRONT_SMOKE_RANDOM_SPREAD, FRONT_SMOKE_RANDOM_SPREAD
)
if not self.game.theater.is_on_land(pos):
break
self.mission.static_group(
self.mission.country(self.game.red.country_name),
"",
_type=v,
position=pos,
)
break
def generate(self) -> None:
self._generate_frontline_smokes()