import logging import os from pathlib import Path from typing import List, Optional, Set from dcs import Mission from dcs.action import DoScript, DoScriptFile from dcs.coalition import Coalition from dcs.countries import country_dict from dcs.lua.parse import loads from dcs.mapping import Point from dcs.terrain.terrain import Terrain from dcs.translation import String from dcs.triggers import TriggerStart from dcs.unittype import UnitType from gen import Conflict, VisualGenerator, FlightType from gen.aircraft import AIRCRAFT_DATA, AircraftConflictGenerator, FlightData from gen.airfields import AIRFIELD_DATA from gen.airsupportgen import AirSupport, AirSupportConflictGenerator from gen.armor import GroundConflictGenerator, JtacInfo from gen.beacons import load_beacons_for_terrain from gen.briefinggen import BriefingGenerator from gen.environmentgen import EnviromentGenerator from gen.forcedoptionsgen import ForcedOptionsGenerator from gen.groundobjectsgen import GroundObjectsGenerator from gen.kneeboard import KneeboardGenerator from gen.radios import RadioFrequency, RadioRegistry from gen.tacan import TacanRegistry from gen.triggergen import TRIGGER_RADIUS_MEDIUM, TriggersGenerator from theater import ControlPoint from .. import db from ..debriefing import Debriefing from plugin import BasePlugin, INSTALLED_PLUGINS class Operation: attackers_starting_position = None # type: db.StartingPosition defenders_starting_position = None # type: db.StartingPosition current_mission = None # type: Mission regular_mission = None # type: Mission quick_mission = None # type: Mission conflict = None # type: Conflict airgen = None # type: AircraftConflictGenerator triggersgen = None # type: TriggersGenerator airsupportgen = None # type: AirSupportConflictGenerator visualgen = None # type: VisualGenerator envgen = None # type: EnviromentGenerator groundobjectgen = None # type: GroundObjectsGenerator briefinggen = None # type: BriefingGenerator forcedoptionsgen = None # type: ForcedOptionsGenerator radio_registry: Optional[RadioRegistry] = None tacan_registry: Optional[TacanRegistry] = None environment_settings = None trigger_radius = TRIGGER_RADIUS_MEDIUM is_quick = None is_awacs_enabled = False ca_slots = 0 def __init__(self, game, attacker_name: str, defender_name: str, from_cp: ControlPoint, departure_cp: ControlPoint, to_cp: ControlPoint): self.game = game self.attacker_name = attacker_name self.attacker_country = db.FACTIONS[attacker_name]["country"] self.defender_name = defender_name self.defender_country = db.FACTIONS[defender_name]["country"] print(self.defender_country, self.attacker_country) self.from_cp = from_cp self.departure_cp = departure_cp self.to_cp = to_cp self.is_quick = False self.listOfPluginsScripts = [] def units_of(self, country_name: str) -> List[UnitType]: return [] def is_successfull(self, debriefing: Debriefing) -> bool: return True @property def is_player_attack(self) -> bool: return self.from_cp.captured def initialize(self, mission: Mission, conflict: Conflict): self.current_mission = mission self.conflict = conflict self.briefinggen = BriefingGenerator(self.current_mission, self.conflict, self.game) def prepare(self, terrain: Terrain, is_quick: bool): with open("resources/default_options.lua", "r") as f: options_dict = loads(f.read())["options"] self.current_mission = Mission(terrain) print(self.game.player_country) print(country_dict[db.country_id_from_name(self.game.player_country)]) print(country_dict[db.country_id_from_name(self.game.player_country)]()) # Setup coalition : self.current_mission.coalition["blue"] = Coalition("blue") self.current_mission.coalition["red"] = Coalition("red") p_country = self.game.player_country e_country = self.game.enemy_country self.current_mission.coalition["blue"].add_country(country_dict[db.country_id_from_name(p_country)]()) self.current_mission.coalition["red"].add_country(country_dict[db.country_id_from_name(e_country)]()) print([c for c in self.current_mission.coalition["blue"].countries.keys()]) print([c for c in self.current_mission.coalition["red"].countries.keys()]) if is_quick: self.quick_mission = self.current_mission else: self.regular_mission = self.current_mission self.current_mission.options.load_from_dict(options_dict) self.is_quick = is_quick if is_quick: self.attackers_starting_position = None self.defenders_starting_position = None else: self.attackers_starting_position = self.departure_cp.at # TODO: Is this possible? if self.to_cp is not None: self.defenders_starting_position = self.to_cp.at else: self.defenders_starting_position = None def injectLuaTrigger(self, luascript, comment = "LUA script"): trigger = TriggerStart(comment=comment) trigger.add_action(DoScript(String(luascript))) self.current_mission.triggerrules.triggers.append(trigger) def bypassPluginScript(self, pluginName, scriptFileMnemonic): self.listOfPluginsScripts.append(scriptFileMnemonic) def injectPluginScript(self, pluginName, scriptFile, scriptFileMnemonic): if not scriptFileMnemonic in self.listOfPluginsScripts: self.listOfPluginsScripts.append(scriptFileMnemonic) if pluginName == None: pluginName = "custom" plugin_path = Path("./resources/plugins",pluginName) logging.debug(f"plugin_path = {plugin_path}") if scriptFile != None: scriptFile_path = Path(plugin_path, scriptFile) if scriptFile_path.exists(): trigger = TriggerStart(comment="Load " + scriptFileMnemonic) filename = scriptFile_path.resolve() fileref = self.current_mission.map_resource.add_resource_file(filename) trigger.add_action(DoScriptFile(fileref)) self.current_mission.triggerrules.triggers.append(trigger) else: logging.error(f"Cannot find script file {scriptFile} for plugin {pluginName}") else: logging.debug(f"Skipping script file {scriptFile} for plugin {pluginName}") def generate(self): radio_registry = RadioRegistry() tacan_registry = TacanRegistry() # Dedup beacon/radio frequencies, since some maps have some frequencies # used multiple times. beacons = load_beacons_for_terrain(self.game.theater.terrain.name) unique_map_frequencies: Set[RadioFrequency] = set() for beacon in beacons: unique_map_frequencies.add(beacon.frequency) if beacon.is_tacan: if beacon.channel is None: logging.error( f"TACAN beacon has no channel: {beacon.callsign}") else: tacan_registry.reserve(beacon.tacan_channel) for airfield, data in AIRFIELD_DATA.items(): if data.theater == self.game.theater.terrain.name: unique_map_frequencies.add(data.atc.hf) unique_map_frequencies.add(data.atc.vhf_fm) unique_map_frequencies.add(data.atc.vhf_am) unique_map_frequencies.add(data.atc.uhf) # No need to reserve ILS or TACAN because those are in the # beacon list. for frequency in unique_map_frequencies: radio_registry.reserve(frequency) # Generate meteo envgen = EnviromentGenerator(self.current_mission, self.conflict, self.game) if self.environment_settings is None: self.environment_settings = envgen.generate() else: envgen.load(self.environment_settings) # Generate ground object first groundobjectgen = GroundObjectsGenerator( self.current_mission, self.conflict, self.game, radio_registry, tacan_registry ) groundobjectgen.generate() # Generate destroyed units for d in self.game.get_destroyed_units(): try: utype = db.unit_type_from_name(d["type"]) except KeyError: continue pos = Point(d["x"], d["z"]) if utype is not None and not self.game.position_culled(pos) and self.game.settings.perf_destroyed_units: self.current_mission.static_group( country=self.current_mission.country(self.game.player_country), name="", _type=utype, hidden=True, position=pos, heading=d["orientation"], dead=True, ) # Air Support (Tanker & Awacs) airsupportgen = AirSupportConflictGenerator( self.current_mission, self.conflict, self.game, radio_registry, tacan_registry) airsupportgen.generate(self.is_awacs_enabled) # Generate Activity on the map airgen = AircraftConflictGenerator( self.current_mission, self.conflict, self.game.settings, self.game, radio_registry) airgen.generate_flights( self.current_mission.country(self.game.player_country), self.game.blue_ato, groundobjectgen.runways ) airgen.generate_flights( self.current_mission.country(self.game.enemy_country), self.game.red_ato, groundobjectgen.runways ) # Generate ground units on frontline everywhere jtacs: List[JtacInfo] = [] for player_cp, enemy_cp in self.game.theater.conflicts(True): conflict = Conflict.frontline_cas_conflict(self.attacker_name, self.defender_name, self.current_mission.country(self.attacker_country), self.current_mission.country(self.defender_country), player_cp, enemy_cp, self.game.theater) # Generate frontline ops player_gp = self.game.ground_planners[player_cp.id].units_per_cp[enemy_cp.id] enemy_gp = self.game.ground_planners[enemy_cp.id].units_per_cp[player_cp.id] groundConflictGen = GroundConflictGenerator(self.current_mission, conflict, self.game, player_gp, enemy_gp, player_cp.stances[enemy_cp.id]) groundConflictGen.generate() jtacs.extend(groundConflictGen.jtacs) # Setup combined arms parameters self.current_mission.groundControl.pilot_can_control_vehicles = self.ca_slots > 0 if self.game.player_country in [country.name for country in self.current_mission.coalition["blue"].countries.values()]: self.current_mission.groundControl.blue_tactical_commander = self.ca_slots else: self.current_mission.groundControl.red_tactical_commander = self.ca_slots # Triggers triggersgen = TriggersGenerator(self.current_mission, self.conflict, self.game) triggersgen.generate() # Options forcedoptionsgen = ForcedOptionsGenerator(self.current_mission, self.conflict, self.game) forcedoptionsgen.generate() # Generate Visuals Smoke Effects visualgen = VisualGenerator(self.current_mission, self.conflict, self.game) if self.game.settings.perf_smoke_gen: visualgen.generate() luaData = {} luaData["AircraftCarriers"] = {} luaData["Tankers"] = {} luaData["AWACs"] = {} luaData["JTACs"] = {} luaData["TargetPoints"] = {} self.assign_channels_to_flights(airgen.flights, airsupportgen.air_support) kneeboard_generator = KneeboardGenerator(self.current_mission) for dynamic_runway in groundobjectgen.runways.values(): self.briefinggen.add_dynamic_runway(dynamic_runway) for tanker in airsupportgen.air_support.tankers: self.briefinggen.add_tanker(tanker) kneeboard_generator.add_tanker(tanker) luaData["Tankers"][tanker.callsign] = { "dcsGroupName": tanker.dcsGroupName, "callsign": tanker.callsign, "variant": tanker.variant, "radio": tanker.freq.mhz, "tacan": str(tanker.tacan.number) + tanker.tacan.band.name } if self.is_awacs_enabled: for awacs in airsupportgen.air_support.awacs: self.briefinggen.add_awacs(awacs) kneeboard_generator.add_awacs(awacs) luaData["AWACs"][awacs.callsign] = { "dcsGroupName": awacs.dcsGroupName, "callsign": awacs.callsign, "radio": awacs.freq.mhz } for jtac in jtacs: self.briefinggen.add_jtac(jtac) kneeboard_generator.add_jtac(jtac) luaData["JTACs"][jtac.callsign] = { "dcsGroupName": jtac.dcsGroupName, "callsign": jtac.callsign, "zone": jtac.region, "dcsUnit": jtac.unit_name, "laserCode": jtac.code } for flight in airgen.flights: self.briefinggen.add_flight(flight) kneeboard_generator.add_flight(flight) if flight.friendly and flight.flight_type in [FlightType.ANTISHIP, FlightType.DEAD, FlightType.SEAD, FlightType.STRIKE]: flightType = flight.flight_type.name flightTarget = flight.targetPoint if flightTarget: flightTargetName = None flightTargetType = None if hasattr(flightTarget, 'obj_name'): flightTargetName = flightTarget.obj_name flightTargetType = flightType + f" TGT ({flightTarget.category})" elif hasattr(flightTarget, 'name'): flightTargetName = flightTarget.name flightTargetType = flightType + " TGT (Airbase)" luaData["TargetPoints"][flightTargetName] = { "name": flightTargetName, "type": flightTargetType, "position": { "x": flightTarget.position.x, "y": flightTarget.position.y} } self.briefinggen.generate() kneeboard_generator.generate() # set a LUA table with data from Liberation that we want to set # at the moment it contains Liberation's install path, and an overridable definition for the JTACAutoLase function # later, we'll add data about the units and points having been generated, in order to facilitate the configuration of the plugin lua scripts state_location = "[[" + os.path.abspath("state.json") + "]]" lua = """ -- setting configuration table env.info("DCSLiberation|: setting configuration table") -- all data in this table is overridable. dcsLiberation = {} -- the base location for state.json; if non-existent, it'll be replaced with LIBERATION_EXPORT_DIR, TEMP, or DCS working directory dcsLiberation.installPath=""" + state_location + """ -- you can override dcsLiberation.JTACAutoLase to make it use your own function ; it will be called with these parameters : ({jtac.unit_name}, {jtac.code}, {smoke}, 'vehicle') for all JTACs if ctld then dcsLiberation.JTACAutoLase=ctld.JTACAutoLase elseif JTACAutoLase then dcsLiberation.JTACAutoLase=JTACAutoLase end """ # Process the tankers lua += """ -- list the tankers generated by Liberation dcsLiberation.Tankers = { """ for key in luaData["Tankers"]: data = luaData["Tankers"][key] dcsGroupName= data["dcsGroupName"] callsign = data["callsign"] variant = data["variant"] tacan = data["tacan"] radio = data["radio"] lua += f" {{dcsGroupName='{dcsGroupName}', callsign='{callsign}', variant='{variant}', tacan='{tacan}', radio='{radio}' }}, \n" #lua += f" {{name='{dcsGroupName}', description='{callsign} ({variant})', information='Tacan:{tacan} Radio:{radio}' }}, \n" lua += "}" # Process the AWACSes lua += """ -- list the AWACs generated by Liberation dcsLiberation.AWACs = { """ for key in luaData["AWACs"]: data = luaData["AWACs"][key] dcsGroupName= data["dcsGroupName"] callsign = data["callsign"] radio = data["radio"] lua += f" {{dcsGroupName='{dcsGroupName}', callsign='{callsign}', radio='{radio}' }}, \n" #lua += f" {{name='{dcsGroupName}', description='{callsign} (AWACS)', information='Radio:{radio}' }}, \n" lua += "}" # Process the JTACs lua += """ -- list the JTACs generated by Liberation dcsLiberation.JTACs = { """ for key in luaData["JTACs"]: data = luaData["JTACs"][key] dcsGroupName= data["dcsGroupName"] callsign = data["callsign"] zone = data["zone"] laserCode = data["laserCode"] dcsUnit = data["dcsUnit"] lua += f" {{dcsGroupName='{dcsGroupName}', callsign='{callsign}', zone='{zone}', laserCode='{laserCode}', dcsUnit='{dcsUnit}' }}, \n" #lua += f" {{name='{dcsGroupName}', description='JTAC {callsign} ', information='Laser:{laserCode}', jtac={laserCode} }}, \n" lua += "}" # Process the Target Points lua += """ -- list the target points generated by Liberation dcsLiberation.TargetPoints = { """ for key in luaData["TargetPoints"]: data = luaData["TargetPoints"][key] name = data["name"] pointType = data["type"] positionX = data["position"]["x"] positionY = data["position"]["y"] lua += f" {{name='{name}', pointType='{pointType}', positionX='{positionX}', positionY='{positionY}' }}, \n" #lua += f" {{name='{pointType} {name}', point{{x={positionX}, z={positionY} }} }}, \n" lua += "}" lua += """ -- list the airbases generated by Liberation -- dcsLiberation.Airbases = {} -- list the aircraft carriers generated by Liberation -- dcsLiberation.Carriers = {} -- later, we'll add more data to the table """ trigger = TriggerStart(comment="Set DCS Liberation data") trigger.add_action(DoScript(String(lua))) self.current_mission.triggerrules.triggers.append(trigger) # Inject Plugins Lua Scripts and data self.listOfPluginsScripts = [] for pluginName in INSTALLED_PLUGINS: plugin = INSTALLED_PLUGINS[pluginName] plugin.injectScripts(self) plugin.injectConfiguration(self) self.assign_channels_to_flights(airgen.flights, airsupportgen.air_support) kneeboard_generator = KneeboardGenerator(self.current_mission) for dynamic_runway in groundobjectgen.runways.values(): self.briefinggen.add_dynamic_runway(dynamic_runway) for tanker in airsupportgen.air_support.tankers: self.briefinggen.add_tanker(tanker) kneeboard_generator.add_tanker(tanker) if self.is_awacs_enabled: for awacs in airsupportgen.air_support.awacs: self.briefinggen.add_awacs(awacs) kneeboard_generator.add_awacs(awacs) for jtac in jtacs: self.briefinggen.add_jtac(jtac) kneeboard_generator.add_jtac(jtac) for flight in airgen.flights: self.briefinggen.add_flight(flight) kneeboard_generator.add_flight(flight) self.briefinggen.generate() kneeboard_generator.generate() def assign_channels_to_flights(self, flights: List[FlightData], air_support: AirSupport) -> None: """Assigns preset radio channels for client flights.""" for flight in flights: if not flight.client_units: continue self.assign_channels_to_flight(flight, air_support) def assign_channels_to_flight(self, flight: FlightData, air_support: AirSupport) -> None: """Assigns preset radio channels for a client flight.""" airframe = flight.aircraft_type try: aircraft_data = AIRCRAFT_DATA[airframe.id] except KeyError: logging.warning(f"No aircraft data for {airframe.id}") return if aircraft_data.channel_allocator is not None: aircraft_data.channel_allocator.assign_channels_for_flight( flight, air_support )