Use unique callsigns for each flight (#3445)

This PR partially addresses #1561 by automatically generating unique
callsigns for each flight.

changelog.md

@@ -4,12 +4,14 @@ Saves from 11.x are not compatible with 12.0.0.
 
 ## Features/Improvements
 
+* **[Engine]** Support for DCS 2.9.8.1214.
 * **[Campaign]** Removed deprecated settings for generating persistent and invulnerable AWACs and tankers.
-* **[Mods]** F/A-18 E/F/G Super Hornet mod version updated to 2.3.
 * **[Campaign]** Do not allow aircraft from a captured control point to retreat if the captured control point has a damaged runway.
+* **[Mods]** F/A-18 E/F/G Super Hornet mod version updated to 2.3.
 
 ## Fixes
 
+* **[Campaign]** Flights are assigned different callsigns appropriate to the faction.
 
 # 11.1.1
 

game/ato/flight.py

@@ -21,6 +21,7 @@
 )
 
 if TYPE_CHECKING:
+    from game.callsigns.callsigngenerator import Callsign
     from game.dcs.aircrafttype import AircraftType
     from game.sim.gameupdateevents import GameUpdateEvents
     from game.sim.simulationresults import SimulationResults
@@ -49,6 +50,7 @@ def __init__(
         custom_name: Optional[str] = None,
         cargo: Optional[TransferOrder] = None,
         roster: Optional[FlightRoster] = None,
+        callsign: Optional[Callsign] = None,
     ) -> None:
         self.id = uuid.uuid4()
         self.package = package
@@ -69,6 +71,8 @@ def __init__(
         # Only used by transport missions.
         self.cargo = cargo
 
+        self.callsign = callsign
+
         # Flight properties that can be set in the mission editor. This is used for
         # things like HMD selection, ripple quantity, etc. Any values set here will take
         # the place of the defaults defined by DCS.

game/callsigns/callsigngenerator.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+from abc import ABC
+from dataclasses import dataclass
+from enum import StrEnum
+
+from collections import deque
+from typing import Any, List, Optional
+
+from dcs.country import Country
+from dcs.countries import countries_by_name
+
+from game.ato.flight import Flight
+from game.ato.flighttype import FlightType
+
+
+MAX_GROUP_ID = 99
+
+
+class CallsignCategory(StrEnum):
+    AIR = "Air"
+    TANKERS = "Tankers"
+    AWACS = "AWACS"
+    GROUND_UNITS = "GroundUnits"
+    HELIPADS = "Helipad"
+    GRASS_AIRFIELDS = "GrassAirfield"
+
+
+@dataclass(frozen=True)
+class Callsign:
+    name: Optional[
+        str
+    ]  # Callsign name e.g. "Enfield" for western callsigns. None for eastern callsigns.
+    group_id: int  # ID of the group e.g. 2 in Enfield-2-3 for western callsigns. First two digits of eastern callsigns.
+    unit_id: int  # ID of the unit e.g. 3 in Enfield-2-3 for western callsigns. Last digit of eastern callsigns.
+
+    def __post_init__(self) -> None:
+        if self.group_id < 1 or self.group_id > MAX_GROUP_ID:
+            raise ValueError(
+                f"Invalid group ID {self.group_id}. Group IDs have to be between 1 and {MAX_GROUP_ID}."
+            )
+        if self.unit_id < 1 or self.unit_id > 9:
+            raise ValueError(
+                f"Invalid unit ID {self.unit_id}. Unit IDs have to be between 1 and 9."
+            )
+
+    def __str__(self) -> str:
+        if self.name is not None:
+            return f"{self.name}{self.group_id}{self.unit_id}"
+        else:
+            return str(self.group_id * 10 + self.unit_id)
+
+    def lead_callsign(self) -> Callsign:
+        return Callsign(self.name, self.group_id, 1)
+
+    def unit_callsign(self, unit_id: int) -> Callsign:
+        return Callsign(self.name, self.group_id, unit_id)
+
+    def group_name(self) -> str:
+        if self.name is not None:
+            return f"{self.name}-{self.group_id}"
+        else:
+            return str(self.lead_callsign())
+
+    def pydcs_dict(self, country: str) -> dict[Any, Any]:
+        country_obj = countries_by_name[country]()
+        for category in CallsignCategory:
+            if category in country_obj.callsign:
+                for index, name in enumerate(country_obj.callsign[category]):
+                    if name == self.name:
+                        return {
+                            "name": str(self),
+                            1: index + 1,
+                            2: self.group_id,
+                            3: self.unit_id,
+                        }
+        raise ValueError(f"Could not find callsign {name} in {country}.")
+
+
+class WesternGroupIdRegistry:
+
+    def __init__(self, country: Country, max_group_id: int = MAX_GROUP_ID):
+        self._names: dict[str, deque[int]] = {}
+        for category in CallsignCategory:
+            if category in country.callsign:
+                for name in country.callsign[category]:
+                    self._names[name] = deque()
+        self._max_group_id = max_group_id
+        self.reset()
+
+    def reset(self) -> None:
+        for name in self._names:
+            self._names[name] = deque()
+            for i in range(
+                self._max_group_id, 0, -1
+            ):  # Put group IDs on FIFO queue so 1 gets popped first
+                self._names[name].appendleft(i)
+
+    def alloc_group_id(self, name: str) -> int:
+        return self._names[name].popleft()
+
+    def release_group_id(self, callsign: Callsign) -> None:
+        if callsign.name is None:
+            raise ValueError("Releasing eastern callsign")
+        self._names[callsign.name].appendleft(callsign.group_id)
+
+
+class EasternGroupIdRegistry:
+
+    def __init__(self, max_group_id: int = MAX_GROUP_ID):
+        self._max_group_id = max_group_id
+        self._queue: deque[int] = deque()
+        self.reset()
+
+    def reset(self) -> None:
+        self._queue = deque()
+        for i in range(
+            self._max_group_id, 0, -1
+        ):  # Put group IDs on FIFO queue so 1 gets popped first
+            self._queue.appendleft(i)
+
+    def alloc_group_id(self) -> int:
+        return self._queue.popleft()
+
+    def release_group_id(self, callsign: Callsign) -> None:
+        self._queue.appendleft(callsign.group_id)
+
+
+class RoundRobinNameAllocator:
+
+    def __init__(self, names: List[str]):
+        self.names = names
+        self._index = 0
+
+    def allocate(self) -> str:
+        this_index = self._index
+        if this_index == len(self.names) - 1:
+            self._index = 0
+        else:
+            self._index += 1
+        return self.names[this_index]
+
+
+class FlightTypeNameAllocator:
+    def __init__(self, names: List[str]):
+        self.names = names
+
+    def allocate(self, flight: Flight) -> str:
+        index = self.FLIGHT_TYPE_LOOKUP.get(flight.flight_type, 0)
+        return self.names[index]
+
+    FLIGHT_TYPE_LOOKUP: dict[FlightType, int] = {
+        FlightType.TARCAP: 1,
+        FlightType.BARCAP: 1,
+        FlightType.INTERCEPTION: 1,
+        FlightType.SWEEP: 1,
+        FlightType.CAS: 2,
+        FlightType.ANTISHIP: 2,
+        FlightType.BAI: 2,
+        FlightType.STRIKE: 3,
+        FlightType.OCA_RUNWAY: 3,
+        FlightType.OCA_AIRCRAFT: 3,
+        FlightType.SEAD: 4,
+        FlightType.DEAD: 4,
+        FlightType.ESCORT: 5,
+        FlightType.AIR_ASSAULT: 6,
+        FlightType.TRANSPORT: 7,
+        FlightType.FERRY: 7,
+    }
+
+
+class WesternFlightCallsignGenerator:
+    """Generate western callsign for lead unit in a group"""
+
+    def __init__(self, country: str) -> None:
+        self._country = countries_by_name[country]()
+        self._group_id_registry = WesternGroupIdRegistry(self._country)
+        self._awacs_name_allocator = None
+        self._tankers_name_allocator = None
+
+        if CallsignCategory.AWACS in self._country.callsign:
+            self._awacs_name_allocator = RoundRobinNameAllocator(
+                self._country.callsign[CallsignCategory.AWACS]
+            )
+        if CallsignCategory.TANKERS in self._country.callsign:
+            self._tankers_name_allocator = RoundRobinNameAllocator(
+                self._country.callsign[CallsignCategory.TANKERS]
+            )
+        self._air_name_allocator = FlightTypeNameAllocator(
+            self._country.callsign[CallsignCategory.AIR]
+        )
+
+    def reset(self) -> None:
+        self._group_id_registry.reset()
+
+    def alloc_callsign(self, flight: Flight) -> Callsign:
+        if flight.flight_type == FlightType.AEWC:
+            if self._awacs_name_allocator is None:
+                raise ValueError(f"{self._country.name} does not have AWACs callsigns")
+            name = self._awacs_name_allocator.allocate()
+        elif flight.flight_type == FlightType.REFUELING:
+            if self._tankers_name_allocator is None:
+                raise ValueError(f"{self._country.name} does not have tanker callsigns")
+            name = self._tankers_name_allocator.allocate()
+        else:
+            name = self._air_name_allocator.allocate(flight)
+        group_id = self._group_id_registry.alloc_group_id(name)
+        return Callsign(name, group_id, 1)
+
+    def release_callsign(self, callsign: Callsign) -> None:
+        self._group_id_registry.release_group_id(callsign)
+
+
+class EasternFlightCallsignGenerator:
+    """Generate eastern callsign for lead unit in a group"""
+
+    def __init__(self) -> None:
+        self._group_id_registry = EasternGroupIdRegistry()
+
+    def reset(self) -> None:
+        self._group_id_registry.reset()
+
+    def alloc_callsign(self, flight: Flight) -> Callsign:
+        group_id = self._group_id_registry.alloc_group_id()
+        return Callsign(None, group_id, 1)
+
+    def release_callsign(self, callsign: Callsign) -> None:
+        self._group_id_registry.release_group_id(callsign)
+
+
+class FlightCallsignGenerator:
+
+    def __init__(self, country: str):
+        self._generators: dict[
+            bool, WesternFlightCallsignGenerator | EasternFlightCallsignGenerator
+        ] = {
+            True: WesternFlightCallsignGenerator(country),
+            False: EasternFlightCallsignGenerator(),
+        }
+        self._use_western_callsigns = countries_by_name[country]().use_western_callsigns
+
+    def reset(self) -> None:
+        self._generators[self._use_western_callsigns].reset()
+
+    def alloc_callsign(self, flight: Flight) -> Callsign:
+        return self._generators[self._use_western_callsigns].alloc_callsign(flight)
+
+    def release_callsign(self, callsign: Callsign) -> None:
+        self._generators[self._use_western_callsigns].release_callsign(callsign)

game/coalition.py

@@ -7,6 +7,7 @@
 
 from game.armedforces.armedforces import ArmedForces
 from game.ato.airtaaskingorder import AirTaskingOrder
+from game.callsigns.callsigngenerator import FlightCallsignGenerator
 from game.campaignloader.defaultsquadronassigner import DefaultSquadronAssigner
 from game.commander import TheaterCommander
 from game.commander.missionscheduler import MissionScheduler
@@ -46,6 +47,7 @@ def __init__(
         self.air_wing = AirWing(player, game, self.faction)
         self.armed_forces = ArmedForces(self.faction)
         self.transfers = PendingTransfers(game, player)
+        self.callsign_generator = FlightCallsignGenerator(faction.country)
 
         # Late initialized because the two coalitions in the game are mutually
         # dependent, so must be both constructed before this property can be set.
@@ -163,6 +165,8 @@ def end_turn(self) -> None:
         # is handled correctly.
         self.transfers.perform_transfers()
 
+        self.callsign_generator.reset()
+
     def preinit_turn_0(self) -> None:
         """Runs final Coalition initialization.
 

game/commander/packagebuilder.py

@@ -2,6 +2,7 @@
 
 from typing import Optional, TYPE_CHECKING
 
+from game.callsigns.callsigngenerator import FlightCallsignGenerator
 from game.theater import ControlPoint, MissionTarget, OffMapSpawn
 from game.utils import nautical_miles
 from ..ato.flight import Flight
@@ -26,6 +27,7 @@ def __init__(
         closest_airfields: ClosestAirfields,
         air_wing: AirWing,
         laser_code_registry: LaserCodeRegistry,
+        callsign_generator: FlightCallsignGenerator,
         flight_db: Database[Flight],
         is_player: bool,
         package_country: str,
@@ -38,6 +40,7 @@ def __init__(
         self.package = Package(location, flight_db, auto_asap=asap)
         self.air_wing = air_wing
         self.laser_code_registry = laser_code_registry
+        self.callsign_generator = callsign_generator
         self.start_type = start_type
 
     def plan_flight(self, plan: ProposedFlight) -> bool:
@@ -71,6 +74,7 @@ def plan_flight(self, plan: ProposedFlight) -> bool:
                 member.assign_tgp_laser_code(
                     self.laser_code_registry.alloc_laser_code()
                 )
+        flight.callsign = self.callsign_generator.alloc_callsign(flight)
         self.package.add_flight(flight)
         return True
 

game/commander/packagefulfiller.py

@@ -142,6 +142,7 @@ def plan_mission(
             ObjectiveDistanceCache.get_closest_airfields(mission.location),
             self.air_wing,
             self.coalition.laser_code_registry,
+            self.coalition.callsign_generator,
             self.flight_db,
             self.is_player,
             self.coalition.country_name,

game/missiongenerator/aircraft/flightdata.py

@@ -6,7 +6,7 @@
 
 from dcs.flyingunit import FlyingUnit
 
-from game.callsigns import create_group_callsign_from_unit
+from game.callsigns.callsign import create_group_callsign_from_unit
 
 if TYPE_CHECKING:
     from game.ato import FlightType, FlightWaypoint, Package