diff --git a/game/operation/operation.py b/game/operation/operation.py index 141f5b5b..09b5eb0a 100644 --- a/game/operation/operation.py +++ b/game/operation/operation.py @@ -255,7 +255,7 @@ class Operation: load_dcs_libe.add_action(DoScript(String(script))) self.current_mission.triggerrules.triggers.append(load_dcs_libe) - kneeboard_generator = KneeboardGenerator(self.current_mission, self.game) + kneeboard_generator = KneeboardGenerator(self.current_mission) # Briefing Generation for tanker in self.airsupportgen.air_support.tankers: @@ -269,9 +269,82 @@ class Operation: self.briefinggen.append_frequency(awacs.callsign, awacs.freq) kneeboard_generator.add_awacs(awacs) + self.assign_channels_to_flights() + # Generate the briefing self.briefinggen.generate() for region, code, name in self.game.jtacs: kneeboard_generator.add_jtac(name, region, code) - kneeboard_generator.generate() + kneeboard_generator.generate(self.airgen.flights) + + def assign_channels_to_flights(self) -> None: + """Assigns preset radio channels for client flights.""" + for flight in self.airgen.flights: + if not flight.client_units: + continue + self.assign_channels_to_flight(flight) + + def assign_channels_to_flight(self, flight: FlightData) -> 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 + + # Intra-flight channel is set up when the flight is created, however we + # do need to make sure we don't overwrite it. For cases where the + # inter-flight and intra-flight radios share presets (the AV-8B only has + # one set of channels, even though it can use two channels + # simultaneously), start assigning channels at 2. + radio_id = aircraft_data.inter_flight_radio_index + if aircraft_data.intra_flight_radio_index == radio_id: + first_channel = 2 + else: + first_channel = 1 + + last_channel = flight.num_radio_channels(radio_id) + channel_alloc = iter(range(first_channel, last_channel + 1)) + + # TODO: Fix departure/arrival to support carriers. + if flight.departure is not None: + try: + departure = AIRFIELD_DATA[flight.departure.name] + flight.assign_channel( + radio_id, next(channel_alloc), departure.atc.uhf) + except KeyError: + pass + + # TODO: If there ever are multiple AWACS, limit to mission relevant. + for awacs in self.airsupportgen.air_support.awacs: + flight.assign_channel(radio_id, next(channel_alloc), awacs.freq) + + # TODO: Fix departure/arrival to support carriers. + if flight.arrival is not None and flight.arrival != flight.departure: + try: + arrival = AIRFIELD_DATA[flight.arrival.name] + flight.assign_channel( + radio_id, next(channel_alloc), arrival.atc.uhf) + except KeyError: + pass + + try: + # TODO: Skip incompatible tankers. + for tanker in self.airsupportgen.air_support.tankers: + flight.assign_channel( + radio_id, next(channel_alloc), tanker.freq) + + if flight.divert is not None: + try: + divert = AIRFIELD_DATA[flight.divert.name] + flight.assign_channel( + radio_id, next(channel_alloc), divert.atc.uhf) + except KeyError: + pass + except StopIteration: + # Any remaining channels are nice-to-haves, but not necessary for + # the few aircraft with a small number of channels available. + pass diff --git a/gen/aircraft.py b/gen/aircraft.py index aa5a62e4..e0e8460e 100644 --- a/gen/aircraft.py +++ b/gen/aircraft.py @@ -1,15 +1,21 @@ from dataclasses import dataclass -from typing import Dict +from typing import Dict, List, Optional, Tuple from game.data.cap_capabilities_db import GUNFIGHTERS from game.settings import Settings from game.utils import nm_to_meter from gen.flights.ai_flight_planner import FlightPlanner -from gen.flights.flight import Flight, FlightType, FlightWaypointType +from gen.flights.flight import ( + Flight, + FlightType, + FlightWaypoint, + FlightWaypointType, +) from gen.radios import get_radio, MHz, Radio, RadioFrequency, RadioRegistry from pydcs.dcs import helicopters from pydcs.dcs.action import ActivateGroup, AITaskPush, MessageToAll from pydcs.dcs.condition import TimeAfter, CoalitionHasAirdrome, PartOfCoalitionInZone +from pydcs.dcs.flyingunit import FlyingUnit from pydcs.dcs.helicopters import helicopter_map, UH_1H from pydcs.dcs.mission import Mission, StartType from pydcs.dcs.planes import ( @@ -24,9 +30,9 @@ from pydcs.dcs.planes import ( SpitfireLFMkIX, SpitfireLFMkIXCW, ) -from pydcs.dcs.terrain.terrain import NoParkingSlotError +from pydcs.dcs.terrain.terrain import Airport, NoParkingSlotError from pydcs.dcs.triggers import TriggerOnce, Event -from pydcs.dcs.unittype import UnitType +from pydcs.dcs.unittype import FlyingType, UnitType from .conflictgen import * from .naming import * @@ -59,12 +65,40 @@ class AircraftData: #: The type of radio used for intra-flight communications. intra_flight_radio: Radio + #: Index of the radio used for intra-flight communications. Matches the + #: index of the panel_radio field of the pydcs.dcs.planes object. + inter_flight_radio_index: Optional[int] + + #: Index of the radio used for intra-flight communications. Matches the + #: index of the panel_radio field of the pydcs.dcs.planes object. + intra_flight_radio_index: Optional[int] + # Indexed by the id field of the pydcs PlaneType. AIRCRAFT_DATA: Dict[str, AircraftData] = { - "A-10C": AircraftData(get_radio("AN/ARC-186(V) AM")), - "F-16C_50": AircraftData(get_radio("AN/ARC-222")), - "F/A-18C": AircraftData(get_radio("AN/ARC-210")), + "A-10C": AircraftData( + get_radio("AN/ARC-186(V) AM"), + # The A-10's radio works differently than most aircraft. Doesn't seem to + # be a way to set these from the mission editor, let alone pydcs. + inter_flight_radio_index=None, + intra_flight_radio_index=None + ), + "F-16C_50": AircraftData( + get_radio("AN/ARC-222"), + # COM2 is the AN/ARC-222, which is the VHF radio we want to use for + # intra-flight communication to leave COM1 open for UHF inter-flight. + inter_flight_radio_index=1, + intra_flight_radio_index=2 + ), + "FA-18C_hornet": AircraftData( + get_radio("AN/ARC-210"), + # DCS will clobber channel 1 of the first radio compatible with the + # flight's assigned frequency. Since the F/A-18's two radios are both + # AN/ARC-210s, radio 1 will be compatible regardless of which frequency + # is assigned, so we must use radio 1 for the intra-flight radio. + inter_flight_radio_index=2, + intra_flight_radio_index=1 + ), } @@ -98,6 +132,100 @@ def get_fallback_channel(unit_type: UnitType) -> RadioFrequency: return UHF_FALLBACK_CHANNEL +@dataclass(frozen=True) +class ChannelAssignment: + radio_id: int + channel: int + + @property + def radio_name(self) -> str: + """Returns the name of the radio, i.e. COM1.""" + return f"COM{self.radio_id}" + + +@dataclass +class FlightData: + """Details of a planned flight.""" + + #: List of playable units in the flight. + client_units: List[FlyingUnit] + + # TODO: Arrival and departure should not be optional, but carriers don't count. + #: Arrival airport. + arrival: Optional[Airport] + + #: Departure airport. + departure: Optional[Airport] + + #: Diver airport. + divert: Optional[Airport] + + #: Waypoints of the flight plan. + waypoints: List[FlightWaypoint] + + #: Radio frequency for intra-flight communications. + intra_flight_channel: RadioFrequency + + #: Map of radio frequencies to their assigned radio and channel, if any. + frequency_to_channel_map: Dict[RadioFrequency, ChannelAssignment] + + def __init__(self, client_units: List[FlyingUnit], arrival: Airport, + departure: Airport, divert: Optional[Airport], + waypoints: List[FlightWaypoint], + intra_flight_channel: RadioFrequency) -> None: + self.client_units = client_units + self.arrival = arrival + self.departure = departure + self.divert = divert + self.waypoints = waypoints + self.intra_flight_channel = intra_flight_channel + self.frequency_to_channel_map = {} + + self.assign_intra_flight_channel() + + def assign_intra_flight_channel(self) -> None: + """Assigns a channel to the intra-flight frequency.""" + if not self.client_units: + return + + # pydcs will actually set up the channel for us, but we want to make + # sure that it ends up in frequency_to_channel_map. + try: + data = AIRCRAFT_DATA[self.aircraft_type.id] + self.assign_channel( + data.intra_flight_radio_index, 1, self.intra_flight_channel) + except KeyError: + logging.warning(f"No aircraft data for {self.aircraft_type.id}") + + @property + def aircraft_type(self) -> FlyingType: + """Returns the type of aircraft in this flight.""" + return self.client_units[0].unit_type + + def num_radio_channels(self, radio_id: int) -> int: + """Returns the number of preset channels for the given radio.""" + return self.client_units[0].num_radio_channels(radio_id) + + def channel_for( + self, frequency: RadioFrequency) -> Optional[ChannelAssignment]: + """Returns the radio and channel number for the given frequency.""" + return self.frequency_to_channel_map.get(frequency, None) + + def assign_channel(self, radio_id: int, channel_id: int, + frequency: RadioFrequency) -> None: + """Assigns a preset radio channel to the given frequency.""" + for unit in self.client_units: + unit.set_radio_channel_preset(radio_id, channel_id, frequency.mhz) + + # One frequency could be bound to multiple channels. Prefer the first, + # since with the current implementation it will be the lowest numbered + # channel. + if frequency not in self.frequency_to_channel_map: + self.frequency_to_channel_map[frequency] = ChannelAssignment( + radio_id, channel_id + ) + + class AircraftConflictGenerator: escort_targets = [] # type: typing.List[typing.Tuple[FlyingGroup, int]] @@ -109,14 +237,26 @@ class AircraftConflictGenerator: self.conflict = conflict self.radio_registry = radio_registry self.escort_targets = [] + self.flights: List[FlightData] = [] - def get_intra_flight_channel(self, airframe: UnitType) -> RadioFrequency: + def get_intra_flight_channel( + self, airframe: UnitType) -> Tuple[int, RadioFrequency]: + """Allocates an intra-flight channel to a group. + + Args: + airframe: The type of aircraft a channel should be allocated for. + + Returns: + A tuple of the radio index (for aircraft with multiple radios) and + the frequency of the intra-flight channel. + """ try: aircraft_data = AIRCRAFT_DATA[airframe.id] - return self.radio_registry.alloc_for_radio( + channel = self.radio_registry.alloc_for_radio( aircraft_data.intra_flight_radio) + return aircraft_data.intra_flight_radio_index, channel except KeyError: - return get_fallback_channel(airframe) + return 1, get_fallback_channel(airframe) def _start_type(self) -> StartType: return self.settings.cold_start and StartType.Cold or StartType.Warm @@ -156,12 +296,15 @@ class AircraftConflictGenerator: for unit_instance in group.units: unit_instance.livery_id = db.PLANE_LIVERY_OVERRIDES[unit_type] + clients: List[FlyingUnit] = [] single_client = flight.client_count == 1 for idx in range(0, min(len(group.units), flight.client_count)): + unit = group.units[idx] + clients.append(unit) if single_client: - group.units[idx].set_player() + unit.set_player() else: - group.units[idx].set_client() + unit.set_client() # Do not generate player group with late activation. if group.late_activation: @@ -169,14 +312,21 @@ class AircraftConflictGenerator: # Set up F-14 Client to have pre-stored alignement if unit_type is F_14B: - group.units[idx].set_property(F_14B.Properties.INSAlignmentStored.id, True) + unit.set_property(F_14B.Properties.INSAlignmentStored.id, True) group.points[0].tasks.append(OptReactOnThreat(OptReactOnThreat.Values.EvadeFire)) - channel = self.get_intra_flight_channel(unit_type) - group.set_frequency(channel.mhz) - flight.intra_flight_channel = channel + radio_id, channel = self.get_intra_flight_channel(unit_type) + group.set_frequency(channel.mhz, radio_id) + self.flights.append(FlightData( + client_units=clients, + departure=flight.from_cp.airport, + arrival=flight.from_cp.airport, + divert=None, + waypoints=flight.points, + intra_flight_channel=channel + )) # Special case so Su 33 carrier take off if unit_type is Su_33: diff --git a/gen/flights/flight.py b/gen/flights/flight.py index 912946e4..a9b5e9f8 100644 --- a/gen/flights/flight.py +++ b/gen/flights/flight.py @@ -1,10 +1,8 @@ from enum import Enum -from typing import List, Optional - -from pydcs.dcs.unittype import UnitType +from typing import List from game import db -from gen.radios import RadioFrequency +from pydcs.dcs.unittype import UnitType class FlightType(Enum): @@ -96,13 +94,6 @@ class Flight: # How long before this flight should take off scheduled_in = 0 - # Populated during mission generation time by AircraftConflictGenerator. - # TODO: Decouple radio planning from the Flight. - # Make AircraftConflictGenerator generate a FlightData object that is - # returned to the Operation rather than relying on the Flight object, which - # represents a game UI flight rather than a fully planned flight. - intra_flight_channel: Optional[RadioFrequency] - def __init__(self, unit_type: UnitType, count: int, from_cp, flight_type: FlightType): self.unit_type = unit_type self.count = count @@ -112,7 +103,6 @@ class Flight: self.targets = [] self.loadout = {} self.start_type = "Runway" - self.intra_flight_channel = None def __repr__(self): return self.flight_type.name + " | " + str(self.count) + "x" + db.unit_type_name(self.unit_type) \ diff --git a/gen/kneeboard.py b/gen/kneeboard.py index 1241078e..4c652757 100644 --- a/gen/kneeboard.py +++ b/gen/kneeboard.py @@ -34,9 +34,9 @@ from pydcs.dcs.mission import Mission from pydcs.dcs.terrain.terrain import Airport from pydcs.dcs.unittype import FlyingType from . import units +from .aircraft import FlightData from .airfields import AIRFIELD_DATA from .airsupportgen import AwacsInfo, TankerInfo -from .flights.flight import Flight from .radios import RadioFrequency @@ -96,24 +96,6 @@ class KneeboardPage: raise NotImplementedError -class AirfieldInfo: - def __init__(self, airfield: Airport) -> None: - self.airport = airfield - # TODO: Implement logic for picking preferred runway. - runway = airfield.runways[0] - runway_side = ["", "L", "R"][runway.leftright] - self.runway = f"{runway.heading}{runway_side}" - try: - extra_data = AIRFIELD_DATA[airfield.name] - self.atc = extra_data.atc.uhf or "" - self.tacan = extra_data.tacan or "" - self.ils = extra_data.ils_freq(self.runway) or "" - except KeyError: - self.atc = "" - self.ils = "" - self.tacan = "" - - @dataclass class CommInfo: """Communications information for the kneeboard.""" @@ -131,21 +113,15 @@ class JtacInfo: class BriefingPage(KneeboardPage): """A kneeboard page containing briefing information.""" - def __init__(self, flight: Flight, comms: List[CommInfo], + def __init__(self, flight: FlightData, comms: List[CommInfo], awacs: List[AwacsInfo], tankers: List[TankerInfo], jtacs: List[JtacInfo]) -> None: self.flight = flight - self.comms = comms + self.comms = list(comms) self.awacs = awacs self.tankers = tankers self.jtacs = jtacs - if self.flight.intra_flight_channel is not None: - self.comms.append( - CommInfo("Flight", self.flight.intra_flight_channel) - ) - self.departure = flight.from_cp.airport - self.arrival = flight.from_cp.airport - self.divert: Optional[Airport] = None + self.comms.append(CommInfo("Flight", self.flight.intra_flight_channel)) def write(self, path: Path) -> None: writer = KneeboardPageWriter() @@ -156,14 +132,14 @@ class BriefingPage(KneeboardPage): # TODO: Handle carriers. writer.heading("Airfield Info") writer.table([ - self.airfield_info_row("Departure", self.departure), - self.airfield_info_row("Arrival", self.arrival), - self.airfield_info_row("Divert", self.divert), + 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", "ILS", "RWY"]) writer.heading("Flight Plan") flight_plan = [] - for num, waypoint in enumerate(self.flight.points): + for num, waypoint in enumerate(self.flight.waypoints): alt = int(units.meters_to_feet(waypoint.alt)) flight_plan.append([num, waypoint.pretty_name, str(alt)]) writer.table(flight_plan, headers=["STPT", "Action", "Alt"]) @@ -171,13 +147,13 @@ class BriefingPage(KneeboardPage): writer.heading("Comm Ladder") comms = [] for comm in self.comms: - comms.append([comm.name, comm.freq]) + comms.append([comm.name, self.format_frequency(comm.freq)]) writer.table(comms, headers=["Name", "UHF"]) writer.heading("AWACS") awacs = [] for a in self.awacs: - awacs.append([a.callsign, a.freq]) + awacs.append([a.callsign, self.format_frequency(a.freq)]) writer.table(awacs, headers=["Callsign", "UHF"]) writer.heading("Tankers") @@ -187,7 +163,7 @@ class BriefingPage(KneeboardPage): tanker.callsign, tanker.variant, tanker.tacan, - tanker.freq, + self.format_frequency(tanker.freq), ]) writer.table(tankers, headers=["Callsign", "Type", "TACAN", "UHF"]) @@ -213,23 +189,42 @@ class BriefingPage(KneeboardPage): """ if airfield is None: return [row_title, "", "", "", "", ""] - info = AirfieldInfo(airfield) + + # TODO: Implement logic for picking preferred runway. + runway = airfield.runways[0] + runway_side = ["", "L", "R"][runway.leftright] + runway_text = f"{runway.heading}{runway_side}" + + try: + extra_data = AIRFIELD_DATA[airfield.name] + atc = self.format_frequency(extra_data.atc.uhf) + tacan = extra_data.tacan or "" + ils = extra_data.ils_freq(runway) or "" + except KeyError: + atc = "" + ils = "" + tacan = "" return [ row_title, airfield.name, - info.atc, - info.tacan, - info.ils, - info.runway, + atc, + tacan, + ils, + runway_text, ] + def format_frequency(self, frequency: RadioFrequency) -> str: + channel = self.flight.channel_for(frequency) + if channel is None: + return str(frequency) + return f"{channel.radio_name} Ch {channel.channel}" + class KneeboardGenerator: """Creates kneeboard pages for each client flight in the mission.""" - def __init__(self, mission: Mission, game) -> None: + def __init__(self, mission: Mission) -> None: self.mission = mission - self.game = game self.comms: List[CommInfo] = [] self.awacs: List[AwacsInfo] = [] self.tankers: List[TankerInfo] = [] @@ -271,11 +266,11 @@ class KneeboardGenerator: # TODO: Radio info? Type? self.jtacs.append(JtacInfo(callsign, region, code)) - def generate(self) -> None: + def generate(self, flights: List[FlightData]) -> 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(): + for aircraft, pages in self.pages_by_airframe(flights).items(): aircraft_dir = temp_dir / aircraft.id aircraft_dir.mkdir(exist_ok=True) for idx, page in enumerate(pages): @@ -283,7 +278,7 @@ class KneeboardGenerator: page.write(page_path) self.mission.add_aircraft_kneeboard(aircraft, page_path) - def pages_by_airframe(self) -> Dict[FlyingType, List[KneeboardPage]]: + def pages_by_airframe(self, flights: List[FlightData]) -> Dict[FlyingType, 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 @@ -295,15 +290,14 @@ class KneeboardGenerator: that aircraft. """ all_flights: Dict[FlyingType, List[KneeboardPage]] = defaultdict(list) - for cp in self.game.theater.controlpoints: - if cp.id in self.game.planners.keys(): - for flight in self.game.planners[cp.id].flights: - if flight.client_count > 0: - all_flights[flight.unit_type].extend( - self.generate_flight_kneeboard(flight)) + for flight in flights: + if not flight.client_units: + continue + all_flights[flight.aircraft_type].extend( + self.generate_flight_kneeboard(flight)) return all_flights - def generate_flight_kneeboard(self, flight: Flight) -> List[KneeboardPage]: + def generate_flight_kneeboard(self, flight: FlightData) -> List[KneeboardPage]: """Returns a list of kneeboard pages for the given flight.""" return [ BriefingPage(