Use TACAN channels more selectively, use pytest (#1554)

* Use TACAN channels more selectively

* Increase tacan range to 126

* Use pytest and add workflow

* Skip faction tests due to outdated test data

* Run mypy on tests directory also

* Use iterators for bands AND usages, add tests

.github/workflows/build.yml

@@ -37,6 +37,11 @@ jobs:
         ./venv/scripts/activate
         mypy gen
 
+    - name: mypy tests
+      run: |
+        ./venv/scripts/activate
+        mypy tests
+      
     - name: update build number
       run: |
         [IO.File]::WriteAllLines($pwd.path + "\resources\buildnumber", $env:GITHUB_RUN_NUMBER)

.github/workflows/test.yml

@@ -0,0 +1,33 @@
+name: Test
+
+on: [push, pull_request]
+
+jobs:
+
+  build:
+    runs-on: windows-latest
+    steps:
+    - uses: actions/checkout@v2
+      with:
+        submodules: true
+
+    - name: Set up Python 3.9
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.9
+
+    - name: Install environment
+      run: |
+        python -m venv ./venv
+
+    - name: Install dependencies
+      run: |
+        ./venv/scripts/activate
+        python -m pip install -r requirements.txt
+        # For some reason the shiboken2.abi3.dll is not found properly, so I copy it instead
+        Copy-Item .\venv\Lib\site-packages\shiboken2\shiboken2.abi3.dll .\venv\Lib\site-packages\PySide2\ -Force
+
+    - name: run tests
+      run: |
+        ./venv/scripts/activate
+        pytest tests

game/operation/operation.py

@@ -34,7 +34,7 @@ from gen.kneeboard import KneeboardGenerator
 from gen.lasercoderegistry import LaserCodeRegistry
 from gen.naming import namegen
 from gen.radios import RadioFrequency, RadioRegistry
-from gen.tacan import TacanRegistry
+from gen.tacan import TacanRegistry, TacanUsage
 from gen.triggergen import TRIGGER_RADIUS_MEDIUM, TriggersGenerator
 from gen.visualgen import VisualGenerator
 from .. import db
@@ -228,7 +228,9 @@ class Operation:
                 if beacon.channel is None:
                     logging.error(f"TACAN beacon has no channel: {beacon.callsign}")
                 else:
-                    cls.tacan_registry.reserve(beacon.tacan_channel)
+                    cls.tacan_registry.reserve(
+                        beacon.tacan_channel, TacanUsage.TransmitReceive
+                    )
 
     @classmethod
     def _create_radio_registry(

gen/aircraft.py

@@ -92,7 +92,7 @@ from gen.flights.flight import (
 from gen.lasercoderegistry import LaserCodeRegistry
 from gen.radios import RadioFrequency, RadioRegistry
 from gen.runways import RunwayData
-from gen.tacan import TacanBand, TacanRegistry
+from gen.tacan import TacanBand, TacanRegistry, TacanUsage
 from .airsupport import AirSupport, AwacsInfo, TankerInfo
 from .callsigns import callsign_for_support_unit
 from .flights.flightplan import (
@@ -435,7 +435,7 @@ class AircraftConflictGenerator:
         if isinstance(flight.flight_plan, RefuelingFlightPlan):
             callsign = callsign_for_support_unit(group)
 
-            tacan = self.tacan_registy.alloc_for_band(TacanBand.Y)
+            tacan = self.tacan_registy.alloc_for_band(TacanBand.Y, TacanUsage.AirToAir)
             self.air_support.tankers.append(
                 TankerInfo(
                     group_name=str(group.name),

gen/airsupportgen.py

@@ -22,7 +22,7 @@ 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
+from .tacan import TacanBand, TacanRegistry, TacanUsage
 
 if TYPE_CHECKING:
     from game import Game
@@ -89,7 +89,9 @@ class AirSupportConflictGenerator:
                 # 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)
+                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

gen/groundobjectsgen.py

@@ -58,7 +58,7 @@ 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
+from .tacan import TacanBand, TacanChannel, TacanRegistry, TacanUsage
 
 if TYPE_CHECKING:
     from game import Game
@@ -377,7 +377,9 @@ class GenericCarrierGenerator(GenericGroundObjectGenerator[GenericCarrierGroundO
             for unit in group.units[1:]:
                 ship_group.add_unit(self.create_ship(unit, atc))
 
-            tacan = self.tacan_registry.alloc_for_band(TacanBand.X)
+            tacan = self.tacan_registry.alloc_for_band(
+                TacanBand.X, TacanUsage.TransmitReceive
+            )
             tacan_callsign = self.tacan_callsign()
             icls = next(self.icls_alloc)
 

gen/tacan.py

@@ -4,13 +4,37 @@ 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, 100))
+        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)
@@ -36,30 +60,42 @@ class TacanChannelInUseError(RuntimeError):
         super().__init__(f"{channel} is already in use")
 
 
+class TacanChannelForbiddenError(RuntimeError):
+    """Raised when attempting to reserve a, for technical reasons, forbidden channel."""
+
+    def __init__(self, channel: TacanChannel) -> None:
+        super().__init__(f"{channel} is forbidden")
+
+
 class TacanRegistry:
     """Manages allocation of TACAN channels."""
 
     def __init__(self) -> None:
         self.allocated_channels: Set[TacanChannel] = set()
-        self.band_allocators: Dict[TacanBand, Iterator[TacanChannel]] = {}
+        self.allocators: Dict[TacanBand, Dict[TacanUsage, Iterator[TacanChannel]]] = {}
 
         for band in TacanBand:
-            self.band_allocators[band] = band.range()
+            self.allocators[band] = {}
+            for usage in TacanUsage:
+                self.allocators[band][usage] = band.valid_channels(usage)
 
-    def alloc_for_band(self, band: TacanBand) -> TacanChannel:
+    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:
-            OutOfChannelsError: All channels compatible with the given radio are
+            OutOfTacanChannelsError: All channels compatible with the given radio are
                 already allocated.
         """
-        allocator = self.band_allocators[band]
+        allocator = self.allocators[band][intended_usage]
         try:
             while (channel := next(allocator)) in self.allocated_channels:
                 pass
@@ -67,17 +103,21 @@ class TacanRegistry:
         except StopIteration:
             raise OutOfTacanChannelsError(band)
 
-    def reserve(self, channel: TacanChannel) -> None:
+    def reserve(self, channel: TacanChannel, intended_usage: TacanUsage) -> None:
         """Reserves the given channel.
 
         Reserving a channel ensures that it will not be allocated in the future.
 
         Args:
             channel: The channel to reserve.
+            intended_usage: What the caller intends to use the tacan channel for.
 
         Raises:
-            ChannelInUseError: The given frequency is already in use.
+            TacanChannelInUseError: The given channel is already in use.
+            TacanChannelForbiddenError: The given channel is forbidden.
         """
+        if channel.number in UNAVAILABLE[intended_usage][channel.band]:
+            raise TacanChannelForbiddenError(channel)
         if channel in self.allocated_channels:
             raise TacanChannelInUseError(channel)
         self.allocated_channels.add(channel)

requirements.txt

@@ -1,5 +1,6 @@
 altgraph==0.17
 appdirs==1.4.4
+attrs==21.2.0
 black==21.4b0
 certifi==2020.12.5
 cfgv==3.2.0
@@ -9,7 +10,9 @@ Faker==8.2.1
 filelock==3.0.12
 future==0.18.2
 identify==1.5.13
+iniconfig==1.1.1
 Jinja2==2.11.3
+macholib==1.14
 MarkupSafe==1.1.1
 mypy==0.812
 mypy-extensions==0.4.3
@@ -18,13 +21,16 @@ packaging==20.9
 pathspec==0.8.1
 pefile==2019.4.18
 Pillow==8.2.0
+pluggy==0.13.1
 pre-commit==2.10.1
+py==1.10.0
 -e git://github.com/pydcs/dcs@eb0b9f2de660393ccd6ba17b2d82371d44e0d27b#egg=pydcs
 pyinstaller==4.3
 pyinstaller-hooks-contrib==2021.1
 pyparsing==2.4.7
 pyproj==3.0.1
 PySide2==5.15.2
+pytest==6.2.4
 python-dateutil==2.8.1
 pywin32-ctypes==0.2.0
 PyYAML==5.4.1

tests/test_factions.py

@@ -1,6 +1,7 @@
 import json
 from pathlib import Path
 import unittest
+import pytest
 
 from dcs.helicopters import UH_1H, AH_64A
 from dcs.planes import (
@@ -39,10 +40,11 @@ RESOURCES_DIR = THIS_DIR / "resources"
 
 
 class TestFactionLoader(unittest.TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         pass
 
-    def test_load_valid_faction(self):
+    @pytest.mark.skip(reason="Faction unit names in the json files are outdated")
+    def test_load_valid_faction(self) -> None:
         with (RESOURCES_DIR / "valid_faction.json").open("r") as data:
             faction = Faction.from_json(json.load(data))
 
@@ -112,7 +114,8 @@ class TestFactionLoader(unittest.TestCase):
             self.assertIn("OliverHazardPerryGroupGenerator", faction.navy_generators)
             self.assertIn("ArleighBurkeGroupGenerator", faction.navy_generators)
 
-    def test_load_valid_faction_with_invalid_country(self):
+    @pytest.mark.skip(reason="Faction unit names in the json files are outdated")
+    def test_load_valid_faction_with_invalid_country(self) -> None:
 
         with (RESOURCES_DIR / "invalid_faction_country.json").open("r") as data:
             try:

tests/test_tacan.py

@@ -0,0 +1,117 @@
+from gen.tacan import (
+    OutOfTacanChannelsError,
+    TacanBand,
+    TacanChannel,
+    TacanChannelForbiddenError,
+    TacanChannelInUseError,
+    TacanRegistry,
+    TacanUsage,
+)
+import pytest
+
+
+ALL_VALID_X_TR = [1, *range(31, 46 + 1), *range(64, 126 + 1)]
+ALL_VALID_X_A2A = [*range(37, 63 + 1), *range(100, 126 + 1)]
+
+
+def test_allocate_first_few_channels() -> None:
+    registry = TacanRegistry()
+    chan1 = registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+    chan2 = registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+    chan3 = registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+    assert chan1 == TacanChannel(1, TacanBand.X)
+    assert chan2 == TacanChannel(31, TacanBand.X)
+    assert chan3 == TacanChannel(32, TacanBand.X)
+
+
+def test_allocate_different_usages() -> None:
+    """Make sure unallocated channels for one use don't make channels unavailable for other usage"""
+    registry = TacanRegistry()
+
+    chanA2AX = registry.alloc_for_band(TacanBand.X, TacanUsage.AirToAir)
+    chanA2AY = registry.alloc_for_band(TacanBand.Y, TacanUsage.AirToAir)
+    assert chanA2AX == TacanChannel(37, TacanBand.X)
+    assert chanA2AY == TacanChannel(37, TacanBand.Y)
+
+    chanTRX = registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+    chanTRY = registry.alloc_for_band(TacanBand.Y, TacanUsage.TransmitReceive)
+    assert chanTRX == TacanChannel(1, TacanBand.X)
+    assert chanTRY == TacanChannel(1, TacanBand.Y)
+
+
+def test_reserve_all_valid_transmit_receive() -> None:
+    registry = TacanRegistry()
+    print("All valid x", ALL_VALID_X_TR)
+
+    for num in ALL_VALID_X_TR:
+        registry.reserve(TacanChannel(num, TacanBand.X), TacanUsage.TransmitReceive)
+
+    with pytest.raises(OutOfTacanChannelsError):
+        registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+
+    # Check that we still can allocate an a2a channel even
+    # though the T/R channels are used up
+    chanA2A = registry.alloc_for_band(TacanBand.X, TacanUsage.AirToAir)
+    assert chanA2A == TacanChannel(47, TacanBand.X)
+
+
+def test_reserve_all_valid_a2a() -> None:
+    registry = TacanRegistry()
+    print("All valid x", ALL_VALID_X_A2A)
+
+    for num in ALL_VALID_X_A2A:
+        registry.reserve(TacanChannel(num, TacanBand.X), TacanUsage.AirToAir)
+
+    with pytest.raises(OutOfTacanChannelsError):
+        registry.alloc_for_band(TacanBand.X, TacanUsage.AirToAir)
+
+    # Check that we still can allocate an a2a channel even
+    # though the T/R channels are used up
+    chanTR = registry.alloc_for_band(TacanBand.X, TacanUsage.TransmitReceive)
+    assert chanTR == TacanChannel(1, TacanBand.X)
+
+
+@pytest.mark.skip(reason="TODO")
+def test_allocate_all() -> None:
+    pass
+
+
+def test_reserve_invalid_tr_channels() -> None:
+    registry = TacanRegistry()
+    some_invalid_channels = [
+        TacanChannel(2, TacanBand.X),
+        TacanChannel(30, TacanBand.X),
+        TacanChannel(47, TacanBand.X),
+        TacanChannel(63, TacanBand.X),
+        TacanChannel(2, TacanBand.Y),
+        TacanChannel(30, TacanBand.Y),
+        TacanChannel(64, TacanBand.Y),
+        TacanChannel(92, TacanBand.Y),
+    ]
+    for chan in some_invalid_channels:
+        with pytest.raises(TacanChannelForbiddenError):
+            registry.reserve(chan, TacanUsage.TransmitReceive)
+
+
+def test_reserve_invalid_a2a_channels() -> None:
+    registry = TacanRegistry()
+    some_invalid_channels = [
+        TacanChannel(1, TacanBand.X),
+        TacanChannel(36, TacanBand.X),
+        TacanChannel(64, TacanBand.X),
+        TacanChannel(99, TacanBand.X),
+        TacanChannel(1, TacanBand.Y),
+        TacanChannel(36, TacanBand.Y),
+        TacanChannel(64, TacanBand.Y),
+        TacanChannel(99, TacanBand.Y),
+    ]
+    for chan in some_invalid_channels:
+        with pytest.raises(TacanChannelForbiddenError):
+            registry.reserve(chan, TacanUsage.AirToAir)
+
+
+def test_reserve_again() -> None:
+    registry = TacanRegistry()
+    with pytest.raises(TacanChannelInUseError):
+        registry.reserve(TacanChannel(1, TacanBand.X), TacanUsage.TransmitReceive)
+        registry.reserve(TacanChannel(1, TacanBand.X), TacanUsage.TransmitReceive)