diff --git a/game/pretense/pretenseluagenerator.py b/game/pretense/pretenseluagenerator.py new file mode 100644 index 00000000..f8698ec7 --- /dev/null +++ b/game/pretense/pretenseluagenerator.py @@ -0,0 +1,392 @@ +from __future__ import annotations + +import logging +import os +from abc import ABC, abstractmethod +from pathlib import Path +from typing import TYPE_CHECKING, Optional + +from dcs import Mission +from dcs.action import DoScript, DoScriptFile +from dcs.translation import String +from dcs.triggers import TriggerStart + +from game.ato import FlightType +from game.dcs.aircrafttype import AircraftType +from game.plugins import LuaPluginManager +from game.theater import TheaterGroundObject +from game.theater.iadsnetwork.iadsrole import IadsRole +from game.utils import escape_string_for_lua +from .missiondata import MissionData + +if TYPE_CHECKING: + from game import Game + + +class LuaGenerator: + def __init__( + self, + game: Game, + mission: Mission, + mission_data: MissionData, + ) -> None: + self.game = game + self.mission = mission + self.mission_data = mission_data + self.plugin_scripts: list[str] = [] + + def generate(self) -> None: + ewrj_triggers = [ + x for x in self.mission.triggerrules.triggers if isinstance(x, TriggerStart) + ] + self.generate_plugin_data() + self.inject_plugins() + for t in ewrj_triggers: + self.mission.triggerrules.triggers.remove(t) + self.mission.triggerrules.triggers.append(t) + + def generate_plugin_data(self) -> None: + lua_data = LuaData("dcsRetribution") + + install_path = lua_data.add_item("installPath") + install_path.set_value(os.path.abspath(".")) + + lua_data.add_item("Airbases") + carriers_object = lua_data.add_item("Carriers") + + for carrier in self.mission_data.carriers: + carrier_item = carriers_object.add_item() + carrier_item.add_key_value("dcsGroupName", carrier.group_name) + carrier_item.add_key_value("unit_name", carrier.unit_name) + carrier_item.add_key_value("callsign", carrier.callsign) + carrier_item.add_key_value("radio", str(carrier.freq.mhz)) + carrier_item.add_key_value( + "tacan", str(carrier.tacan.number) + carrier.tacan.band.name + ) + + tankers_object = lua_data.add_item("Tankers") + for tanker in self.mission_data.tankers: + tanker_item = tankers_object.add_item() + tanker_item.add_key_value("dcsGroupName", tanker.group_name) + tanker_item.add_key_value("callsign", tanker.callsign) + tanker_item.add_key_value("variant", tanker.variant) + tanker_item.add_key_value("radio", str(tanker.freq.mhz)) + tanker_item.add_key_value( + "tacan", str(tanker.tacan.number) + tanker.tacan.band.name + ) + + awacs_object = lua_data.add_item("AWACs") + for awacs in self.mission_data.awacs: + awacs_item = awacs_object.add_item() + awacs_item.add_key_value("dcsGroupName", awacs.group_name) + awacs_item.add_key_value("callsign", awacs.callsign) + awacs_item.add_key_value("radio", str(awacs.freq.mhz)) + + jtacs_object = lua_data.add_item("JTACs") + for jtac in self.mission_data.jtacs: + jtac_item = jtacs_object.add_item() + jtac_item.add_key_value("dcsGroupName", jtac.group_name) + jtac_item.add_key_value("callsign", jtac.callsign) + jtac_item.add_key_value("zone", jtac.region) + jtac_item.add_key_value("dcsUnit", jtac.unit_name) + jtac_item.add_key_value("laserCode", jtac.code) + jtac_item.add_key_value("radio", str(jtac.freq.mhz)) + jtac_item.add_key_value("modulation", jtac.freq.modulation.name) + + logistics_object = lua_data.add_item("Logistics") + logistics_flights = logistics_object.add_item("flights") + crates_object = logistics_object.add_item("crates") + spawnable_crates: dict[str, str] = {} + transports: list[AircraftType] = [] + for logistic_info in self.mission_data.logistics: + if logistic_info.transport not in transports: + transports.append(logistic_info.transport) + coalition_color = "blue" if logistic_info.blue else "red" + logistics_item = logistics_flights.add_item() + logistics_item.add_data_array("pilot_names", logistic_info.pilot_names) + logistics_item.add_key_value("pickup_zone", logistic_info.pickup_zone) + logistics_item.add_key_value("drop_off_zone", logistic_info.drop_off_zone) + logistics_item.add_key_value("target_zone", logistic_info.target_zone) + logistics_item.add_key_value("side", str(2 if logistic_info.blue else 1)) + logistics_item.add_key_value("logistic_unit", logistic_info.logistic_unit) + logistics_item.add_key_value( + "aircraft_type", logistic_info.transport.dcs_id + ) + logistics_item.add_key_value( + "preload", "true" if logistic_info.preload else "false" + ) + for cargo in logistic_info.cargo: + if cargo.unit_type not in spawnable_crates: + spawnable_crates[cargo.unit_type] = str(200 + len(spawnable_crates)) + crate_weight = spawnable_crates[cargo.unit_type] + for i in range(cargo.amount): + cargo_item = crates_object.add_item() + cargo_item.add_key_value("weight", crate_weight) + cargo_item.add_key_value("coalition", coalition_color) + cargo_item.add_key_value("zone", cargo.spawn_zone) + transport_object = logistics_object.add_item("transports") + for transport in transports: + transport_item = transport_object.add_item() + transport_item.add_key_value("aircraft_type", transport.dcs_id) + transport_item.add_key_value("cabin_size", str(transport.cabin_size)) + transport_item.add_key_value( + "troops", "true" if transport.cabin_size > 0 else "false" + ) + transport_item.add_key_value( + "crates", "true" if transport.can_carry_crates else "false" + ) + spawnable_crates_object = logistics_object.add_item("spawnable_crates") + for unit, weight in spawnable_crates.items(): + crate_item = spawnable_crates_object.add_item() + crate_item.add_key_value("unit", unit) + crate_item.add_key_value("weight", weight) + + target_points = lua_data.add_item("TargetPoints") + for flight in self.mission_data.flights: + if flight.friendly and flight.flight_type in [ + FlightType.ANTISHIP, + FlightType.DEAD, + FlightType.SEAD, + FlightType.STRIKE, + ]: + flight_type = str(flight.flight_type) + flight_target = flight.package.target + if flight_target: + flight_target_name = None + flight_target_type = None + if isinstance(flight_target, TheaterGroundObject): + flight_target_name = flight_target.obj_name + flight_target_type = ( + flight_type + f" TGT ({flight_target.category})" + ) + elif hasattr(flight_target, "name"): + flight_target_name = flight_target.name + flight_target_type = flight_type + " TGT (Airbase)" + target_item = target_points.add_item() + if flight_target_name: + target_item.add_key_value("name", flight_target_name) + if flight_target_type: + target_item.add_key_value("type", flight_target_type) + target_item.add_key_value( + "positionX", str(flight_target.position.x) + ) + target_item.add_key_value( + "positionY", str(flight_target.position.y) + ) + + for cp in self.game.theater.controlpoints: + coalition_object = ( + lua_data.get_or_create_item("BlueAA") + if cp.captured + else lua_data.get_or_create_item("RedAA") + ) + for ground_object in cp.ground_objects: + for g in ground_object.groups: + threat_range = g.max_threat_range() + + if not threat_range: + continue + + aa_item = coalition_object.add_item() + aa_item.add_key_value("name", ground_object.name) + aa_item.add_key_value("range", str(threat_range.meters)) + aa_item.add_key_value("positionX", str(ground_object.position.x)) + aa_item.add_key_value("positionY", str(ground_object.position.y)) + + # Generate IADS Lua Item + iads_object = lua_data.add_item("IADS") + for node in self.game.theater.iads_network.skynet_nodes(self.game): + coalition = iads_object.get_or_create_item("BLUE" if node.player else "RED") + iads_type = coalition.get_or_create_item(node.iads_role.value) + iads_element = iads_type.add_item() + iads_element.add_key_value("dcsGroupName", node.dcs_name) + if node.iads_role in [IadsRole.SAM, IadsRole.SAM_AS_EWR]: + # add additional SkynetProperties to SAM Sites + for property, value in node.properties.items(): + iads_element.add_key_value(property, value) + for role, connections in node.connections.items(): + iads_element.add_data_array(role, connections) + + trigger = TriggerStart(comment="Set DCS Retribution data") + trigger.add_action(DoScript(String(lua_data.create_operations_lua()))) + self.mission.triggerrules.triggers.append(trigger) + + def inject_lua_trigger(self, contents: str, comment: str) -> None: + trigger = TriggerStart(comment=comment) + trigger.add_action(DoScript(String(contents))) + self.mission.triggerrules.triggers.append(trigger) + + def bypass_plugin_script(self, mnemonic: str) -> None: + self.plugin_scripts.append(mnemonic) + + def inject_plugin_script( + self, plugin_mnemonic: str, script: str, script_mnemonic: str + ) -> None: + if script_mnemonic in self.plugin_scripts: + logging.debug(f"Skipping already loaded {script} for {plugin_mnemonic}") + return + + self.plugin_scripts.append(script_mnemonic) + + plugin_path = Path("./resources/plugins", plugin_mnemonic) + + script_path = Path(plugin_path, script) + if not script_path.exists(): + logging.error(f"Cannot find {script_path} for plugin {plugin_mnemonic}") + return + + trigger = TriggerStart(comment=f"Load {script_mnemonic}") + filename = script_path.resolve() + fileref = self.mission.map_resource.add_resource_file(filename) + trigger.add_action(DoScriptFile(fileref)) + self.mission.triggerrules.triggers.append(trigger) + + def inject_plugins(self) -> None: + for plugin in LuaPluginManager.plugins(): + if plugin.enabled: + plugin.inject_scripts(self) + plugin.inject_configuration(self) + + +class LuaValue: + key: Optional[str] + value: str | list[str] + + def __init__(self, key: Optional[str], value: str | list[str]): + self.key = key + self.value = value + + def serialize(self) -> str: + serialized_value = self.key + " = " if self.key else "" + if isinstance(self.value, str): + serialized_value += f'"{escape_string_for_lua(self.value)}"' + else: + escaped_values = [f'"{escape_string_for_lua(v)}"' for v in self.value] + serialized_value += "{" + ", ".join(escaped_values) + "}" + return serialized_value + + +class LuaItem(ABC): + value: LuaValue | list[LuaValue] + name: Optional[str] + + def __init__(self, name: Optional[str]): + self.value = [] + self.name = name + + def set_value(self, value: str) -> None: + self.value = LuaValue(None, value) + + def set_data_array(self, values: list[str]) -> None: + self.value = LuaValue(None, values) + + def add_data_array(self, key: str, values: list[str]) -> None: + self._add_value(LuaValue(key, values)) + + def add_key_value(self, key: str, value: str) -> None: + self._add_value(LuaValue(key, value)) + + def _add_value(self, value: LuaValue) -> None: + if isinstance(self.value, list): + self.value.append(value) + else: + self.value = value + + @abstractmethod + def add_item(self, item_name: Optional[str] = None) -> LuaItem: + """adds a new item to the LuaArray without checking the existence""" + raise NotImplementedError + + @abstractmethod + def get_item(self, item_name: str) -> Optional[LuaItem]: + """gets item from LuaArray. Returns None if it does not exist""" + raise NotImplementedError + + @abstractmethod + def get_or_create_item(self, item_name: Optional[str] = None) -> LuaItem: + """gets item from the LuaArray or creates one if it does not exist already""" + raise NotImplementedError + + @abstractmethod + def serialize(self) -> str: + if isinstance(self.value, LuaValue): + return self.value.serialize() + else: + serialized_data = [d.serialize() for d in self.value] + return "{" + ", ".join(serialized_data) + "}" + + +class LuaData(LuaItem): + objects: list[LuaData] + base_name: Optional[str] + + def __init__(self, name: Optional[str], is_base_name: bool = True): + self.objects = [] + self.base_name = name if is_base_name else None + super().__init__(name) + + def add_item(self, item_name: Optional[str] = None) -> LuaItem: + item = LuaData(item_name, False) + self.objects.append(item) + return item + + def get_item(self, item_name: str) -> Optional[LuaItem]: + for lua_object in self.objects: + if lua_object.name == item_name: + return lua_object + return None + + def get_or_create_item(self, item_name: Optional[str] = None) -> LuaItem: + if item_name: + item = self.get_item(item_name) + if item: + return item + return self.add_item(item_name) + + def serialize(self, level: int = 0) -> str: + """serialize the LuaData to a string""" + serialized_data: list[str] = [] + serialized_name = "" + linebreak = "\n" + tab = "\t" + tab_end = "" + for _ in range(level): + tab += "\t" + tab_end += "\t" + if self.base_name: + # Only used for initialization of the object in lua + serialized_name += self.base_name + " = " + if self.objects: + # nested objects + serialized_objects = [o.serialize(level + 1) for o in self.objects] + if self.name: + if self.name is not self.base_name: + serialized_name += self.name + " = " + serialized_data.append( + serialized_name + + "{" + + linebreak + + tab + + ("," + linebreak + tab).join(serialized_objects) + + linebreak + + tab_end + + "}" + ) + else: + # key with value + if self.name: + serialized_data.append(self.name + " = " + super().serialize()) + # only value + else: + serialized_data.append(super().serialize()) + + return "\n".join(serialized_data) + + def create_operations_lua(self) -> str: + """crates the liberation lua script for the dcs mission""" + lua_prefix = """ +-- setting configuration table +env.info("DCSRetribution|: setting configuration table") +""" + + return lua_prefix + self.serialize()