Fix mypy regressions.

Mostly in the plugin system, which needed a handful of asserts that
shouldn't be necessary, but fixing them requires a refactor.
This commit is contained in:
Dan Albert 2020-10-24 00:40:02 -07:00
parent e9bfd58ee1
commit c06a855113
8 changed files with 129 additions and 96 deletions

View File

@ -14,7 +14,7 @@ from dcs.translation import String
from dcs.triggers import TriggerStart
from dcs.unittype import UnitType
from gen import Conflict, VisualGenerator, FlightType
from gen import Conflict, FlightType, VisualGenerator
from gen.aircraft import AIRCRAFT_DATA, AircraftConflictGenerator, FlightData
from gen.airfields import AIRFIELD_DATA
from gen.airsupportgen import AirSupport, AirSupportConflictGenerator
@ -28,10 +28,11 @@ 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 plugin import LuaPluginManager
from theater import ControlPoint
from .. import db
from ..debriefing import Debriefing
from plugin import LuaPluginManager
class Operation:
attackers_starting_position = None # type: db.StartingPosition
@ -74,7 +75,7 @@ class Operation:
self.departure_cp = departure_cp
self.to_cp = to_cp
self.is_quick = False
self.listOfPluginsScripts = []
self.plugin_scripts: List[str] = []
def units_of(self, country_name: str) -> List[UnitType]:
return []
@ -133,33 +134,37 @@ class Operation:
else:
self.defenders_starting_position = None
def injectLuaTrigger(self, luascript, comment = "LUA script"):
def inject_lua_trigger(self, contents: str, comment: str) -> None:
trigger = TriggerStart(comment=comment)
trigger.add_action(DoScript(String(luascript)))
trigger.add_action(DoScript(String(contents)))
self.current_mission.triggerrules.triggers.append(trigger)
def bypassPluginScript(self, pluginName, scriptFileMnemonic):
self.listOfPluginsScripts.append(scriptFileMnemonic)
def bypass_plugin_script(self, mnemonic: str) -> None:
self.plugin_scripts.append(mnemonic)
def injectPluginScript(self, pluginName, scriptFile, scriptFileMnemonic):
if not scriptFileMnemonic in self.listOfPluginsScripts:
self.listOfPluginsScripts.append(scriptFileMnemonic)
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}"
)
plugin_path = Path("./resources/plugins",pluginName)
self.plugin_scripts.append(script_mnemonic)
if scriptFile != None:
scriptFile_path = Path(plugin_path, scriptFile)
if scriptFile_path.exists():
trigger = TriggerStart(comment="Load " + scriptFileMnemonic)
filename = scriptFile_path.resolve()
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.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()
@ -334,7 +339,7 @@ class Operation:
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
flightTarget = flight.package.target
if flightTarget:
flightTargetName = None
flightTargetType = None
@ -453,8 +458,6 @@ dcsLiberation.TargetPoints = {
self.current_mission.triggerrules.triggers.append(trigger)
# Inject Plugins Lua Scripts and data
self.listOfPluginsScripts = []
for plugin in LuaPluginManager().getPlugins():
plugin.injectScripts(self)
plugin.injectConfiguration(self)

View File

@ -205,6 +205,9 @@ class ChannelAssignment:
class FlightData:
"""Details of a planned flight."""
#: The package that the flight belongs to.
package: Package
flight_type: FlightType
#: All units in the flight.
@ -237,14 +240,13 @@ class FlightData:
#: Map of radio frequencies to their assigned radio and channel, if any.
frequency_to_channel_map: Dict[RadioFrequency, ChannelAssignment]
#: Data concerning the target of a CAS/Strike/SEAD flight, or None else
targetPoint = None
def __init__(self, flight_type: FlightType, units: List[FlyingUnit],
size: int, friendly: bool, departure_delay: int,
departure: RunwayData, arrival: RunwayData,
divert: Optional[RunwayData], waypoints: List[FlightWaypoint],
intra_flight_channel: RadioFrequency, targetPoint: Optional) -> None:
def __init__(self, package: Package, flight_type: FlightType,
units: List[FlyingUnit], size: int, friendly: bool,
departure_delay: int, departure: RunwayData,
arrival: RunwayData, divert: Optional[RunwayData],
waypoints: List[FlightWaypoint],
intra_flight_channel: RadioFrequency) -> None:
self.package = package
self.flight_type = flight_type
self.units = units
self.size = size
@ -257,7 +259,6 @@ class FlightData:
self.intra_flight_channel = intra_flight_channel
self.frequency_to_channel_map = {}
self.callsign = create_group_callsign_from_unit(self.units[0])
self.targetPoint = targetPoint
@property
def client_units(self) -> List[FlyingUnit]:
@ -575,7 +576,8 @@ class AircraftConflictGenerator:
return StartType.Warm
def _setup_group(self, group: FlyingGroup, for_task: Type[Task],
flight: Flight, dynamic_runways: Dict[str, RunwayData]):
package: Package, flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
did_load_loadout = False
unit_type = group.units[0].unit_type
@ -635,6 +637,7 @@ class AircraftConflictGenerator:
departure_runway = fallback_runway
self.flights.append(FlightData(
package=package,
flight_type=flight.flight_type,
units=group.units,
size=len(group.units),
@ -646,8 +649,7 @@ class AircraftConflictGenerator:
divert=None,
# Waypoints are added later, after they've had their TOTs set.
waypoints=[],
intra_flight_channel=channel,
targetPoint=flight.targetPoint,
intra_flight_channel=channel
))
# Special case so Su 33 carrier take off
@ -789,7 +791,7 @@ class AircraftConflictGenerator:
logging.info(f"Generating flight: {flight.unit_type}")
group = self.generate_planned_flight(flight.from_cp, country,
flight)
self.setup_flight_group(group, flight, dynamic_runways)
self.setup_flight_group(group, package, flight, dynamic_runways)
self.create_waypoints(group, package, flight, timing)
def set_activation_time(self, flight: Flight, group: FlyingGroup,
@ -906,10 +908,11 @@ class AircraftConflictGenerator:
if flight.unit_type.eplrs:
group.points[0].tasks.append(EPLRS(group.id))
def configure_cap(self, group: FlyingGroup, flight: Flight,
def configure_cap(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
group.task = CAP.name
self._setup_group(group, CAP, flight, dynamic_runways)
self._setup_group(group, CAP, package, flight, dynamic_runways)
if flight.unit_type not in GUNFIGHTERS:
ammo_type = OptRTBOnOutOfAmmo.Values.AAM
@ -921,10 +924,11 @@ class AircraftConflictGenerator:
group.points[0].tasks.append(EngageTargets(max_distance=nm_to_meter(50),
targets=[Targets.All.Air]))
def configure_cas(self, group: FlyingGroup, flight: Flight,
def configure_cas(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
group.task = CAS.name
self._setup_group(group, CAS, flight, dynamic_runways)
self._setup_group(group, CAS, package, flight, dynamic_runways)
self.configure_behavior(
group,
react_on_threat=OptReactOnThreat.Values.EvadeFire,
@ -936,10 +940,11 @@ class AircraftConflictGenerator:
targets=[Targets.All.GroundUnits.GroundVehicles])
)
def configure_sead(self, group: FlyingGroup, flight: Flight,
def configure_sead(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
group.task = SEAD.name
self._setup_group(group, SEAD, flight, dynamic_runways)
self._setup_group(group, SEAD, package, flight, dynamic_runways)
self.configure_behavior(
group,
react_on_threat=OptReactOnThreat.Values.EvadeFire,
@ -947,33 +952,37 @@ class AircraftConflictGenerator:
rtb_winchester=OptRTBOnOutOfAmmo.Values.ASM,
restrict_jettison=True)
def configure_strike(self, group: FlyingGroup, flight: Flight,
def configure_strike(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
group.task = PinpointStrike.name
self._setup_group(group, GroundAttack, flight, dynamic_runways)
self._setup_group(group, GroundAttack, package, flight, dynamic_runways)
self.configure_behavior(
group,
react_on_threat=OptReactOnThreat.Values.EvadeFire,
roe=OptROE.Values.OpenFire,
restrict_jettison=True)
def configure_anti_ship(self, group: FlyingGroup, flight: Flight,
def configure_anti_ship(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
group.task = AntishipStrike.name
self._setup_group(group, AntishipStrike, flight, dynamic_runways)
self._setup_group(group, AntishipStrike, package, flight,
dynamic_runways)
self.configure_behavior(
group,
react_on_threat=OptReactOnThreat.Values.EvadeFire,
roe=OptROE.Values.OpenFire,
restrict_jettison=True)
def configure_escort(self, group: FlyingGroup, flight: Flight,
def configure_escort(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
# Escort groups are actually given the CAP task so they can perform the
# Search Then Engage task, which we have to use instead of the Escort
# task for the reasons explained in JoinPointBuilder.
group.task = CAP.name
self._setup_group(group, CAP, flight, dynamic_runways)
self._setup_group(group, CAP, package, flight, dynamic_runways)
self.configure_behavior(group, roe=OptROE.Values.OpenFire,
restrict_jettison=True)
@ -982,22 +991,23 @@ class AircraftConflictGenerator:
logging.error(f"Unhandled flight type: {flight.flight_type.name}")
self.configure_behavior(group)
def setup_flight_group(self, group: FlyingGroup, flight: Flight,
def setup_flight_group(self, group: FlyingGroup, package: Package,
flight: Flight,
dynamic_runways: Dict[str, RunwayData]) -> None:
flight_type = flight.flight_type
if flight_type in [FlightType.BARCAP, FlightType.TARCAP,
FlightType.INTERCEPTION]:
self.configure_cap(group, flight, dynamic_runways)
self.configure_cap(group, package, flight, dynamic_runways)
elif flight_type in [FlightType.CAS, FlightType.BAI]:
self.configure_cas(group, flight, dynamic_runways)
self.configure_cas(group, package, flight, dynamic_runways)
elif flight_type in [FlightType.SEAD, FlightType.DEAD]:
self.configure_sead(group, flight, dynamic_runways)
self.configure_sead(group, package, flight, dynamic_runways)
elif flight_type in [FlightType.STRIKE]:
self.configure_strike(group, flight, dynamic_runways)
self.configure_strike(group, package, flight, dynamic_runways)
elif flight_type in [FlightType.ANTISHIP]:
self.configure_anti_ship(group, flight, dynamic_runways)
self.configure_anti_ship(group, package, flight, dynamic_runways)
elif flight_type == FlightType.ESCORT:
self.configure_escort(group, flight, dynamic_runways)
self.configure_escort(group, package, flight, dynamic_runways)
else:
self.configure_unknown_task(group, flight)

View File

@ -206,10 +206,9 @@ class PackageBuilder:
if assignment is None:
return False
airfield, aircraft = assignment
flight = Flight(aircraft, plan.num_aircraft, airfield, plan.task,
self.start_type)
flight = Flight(self.package, aircraft, plan.num_aircraft, airfield,
plan.task, self.start_type)
self.package.add_flight(flight)
flight.targetPoint = self.package.target
return True
def build(self) -> Package:
@ -222,7 +221,7 @@ class PackageBuilder:
for flight in flights:
self.global_inventory.return_from_flight(flight)
self.package.remove_flight(flight)
flight.targetPoint = None
class ObjectiveFinder:
"""Identifies potential objectives for the mission planner."""

View File

@ -1,5 +1,7 @@
from __future__ import annotations
from enum import Enum
from typing import Dict, Iterable, List, Optional
from typing import Dict, Iterable, List, Optional, TYPE_CHECKING
from dcs.mapping import Point
from dcs.point import MovingPoint, PointAction
@ -8,6 +10,9 @@ from dcs.unittype import UnitType
from game import db
from theater.controlpoint import ControlPoint, MissionTarget
if TYPE_CHECKING:
from gen.ato import Package
class FlightType(Enum):
CAP = 0 # Do not use. Use BARCAP or TARCAP.
@ -138,10 +143,11 @@ class Flight:
use_custom_loadout = False
preset_loadout_name = ""
group = False # Contains DCS Mission group data after mission has been generated
targetPoint = None # Contains either None or a Strike/SEAD target point location
def __init__(self, unit_type: UnitType, count: int, from_cp: ControlPoint,
flight_type: FlightType, start_type: str) -> None:
def __init__(self, package: Package, unit_type: UnitType, count: int,
from_cp: ControlPoint, flight_type: FlightType,
start_type: str) -> None:
self.package = package
self.unit_type = unit_type
self.count = count
self.from_cp = from_cp

View File

@ -186,6 +186,12 @@ class TotEstimator:
time_to_target = self.estimate_waypoints_to_target(flight, {
FlightWaypointType.PATROL_TRACK
})
if time_to_target is None:
logging.warning(
f"Found no race track. Cannot estimate TOT for {flight}")
# Return 0 so this flight's travel time does not affect the rest
# of the package.
return 0
else:
time_to_ingress = self.estimate_waypoints_to_target(
flight, INGRESS_TYPES

View File

@ -1,44 +1,48 @@
from typing import List
from pathlib import Path
from PySide2.QtCore import QSize, Qt, QItemSelectionModel, QPoint
from PySide2.QtWidgets import QLabel, QDialog, QGridLayout, QListView, QStackedLayout, QComboBox, QWidget, \
QAbstractItemView, QPushButton, QGroupBox, QCheckBox, QVBoxLayout, QSpinBox
import json
from pathlib import Path
from typing import List, Optional
class LuaPluginWorkOrder():
from PySide2.QtCore import Qt
from PySide2.QtWidgets import QCheckBox, QGridLayout, QGroupBox, QLabel
def __init__(self, parent, filename:str, mnemonic:str, disable:bool):
class LuaPluginWorkOrder:
def __init__(self, parent, filename: str, mnemonic: str,
disable: bool) -> None:
self.filename = filename
self.mnemonic = mnemonic
self.disable = disable
self.parent = parent
def work(self, mnemonic:str, operation):
def work(self, operation):
if self.disable:
operation.bypassPluginScript(self.parent.mnemonic, self.mnemonic)
operation.bypass_plugin_script(self.mnemonic)
else:
operation.injectPluginScript(self.parent.mnemonic, self.filename, self.mnemonic)
operation.inject_plugin_script(self.parent.mnemonic, self.filename,
self.mnemonic)
class LuaPluginSpecificOption():
class LuaPluginSpecificOption:
def __init__(self, parent, mnemonic:str, nameInUI:str, defaultValue:bool):
def __init__(self, parent, mnemonic: str, nameInUI: str,
defaultValue: bool) -> None:
self.mnemonic = mnemonic
self.nameInUI = nameInUI
self.defaultValue = defaultValue
self.parent = parent
class LuaPlugin():
class LuaPlugin:
NAME_IN_SETTINGS_BASE:str = "plugins."
def __init__(self, jsonFilename:str):
self.mnemonic:str = None
def __init__(self, jsonFilename: str) -> None:
self.mnemonic: Optional[str] = None
self.skipUI: bool = False
self.nameInUI:str = None
self.nameInSettings:str = None
self.nameInUI: Optional[str] = None
self.nameInSettings: Optional[str] = None
self.defaultValue: bool = False
self.specificOptions = []
self.scriptsWorkOrders: List[LuaPluginWorkOrder] = None
self.configurationWorkOrders: List[LuaPluginWorkOrder] = None
self.specificOptions: List[LuaPluginSpecificOption] = []
self.scriptsWorkOrders: List[LuaPluginWorkOrder] = []
self.configurationWorkOrders: List[LuaPluginWorkOrder] = []
self.initFromJson(jsonFilename)
self.enabled = self.defaultValue
self.settings = None
@ -50,6 +54,7 @@ class LuaPlugin():
self.mnemonic = jsonData.get("mnemonic")
self.skipUI = jsonData.get("skipUI", False)
self.nameInUI = jsonData.get("nameInUI")
assert self.mnemonic is not None
self.nameInSettings = LuaPlugin.NAME_IN_SETTINGS_BASE + self.mnemonic
self.defaultValue = jsonData.get("defaultValue", False)
self.specificOptions = []
@ -76,6 +81,9 @@ class LuaPlugin():
self.setSettings(settingsWindow.game.settings)
if not self.skipUI:
assert self.nameInSettings is not None
assert self.settings is not None
# create the plugin choice checkbox interface
self.uiWidget: QCheckBox = QCheckBox()
self.uiWidget.setChecked(self.isEnabled())
@ -95,6 +103,7 @@ class LuaPlugin():
# browse each option in the specific options list
row = 0
for specificOption in self.specificOptions:
assert specificOption.mnemonic is not None
nameInSettings = self.nameInSettings + "." + specificOption.mnemonic
if not nameInSettings in self.settings.plugins:
self.settings.plugins[nameInSettings] = specificOption.defaultValue
@ -149,7 +158,7 @@ class LuaPlugin():
# execute the work order
if self.scriptsWorkOrders != None:
for workOrder in self.scriptsWorkOrders:
workOrder.work(self.mnemonic, operation)
workOrder.work(operation)
# serves for subclasses
return self.isEnabled()
@ -177,12 +186,12 @@ class LuaPlugin():
lua += defineAllOptions
lua += "end"
operation.injectLuaTrigger(lua, f"{self.mnemonic} plugin configuration")
operation.inject_lua_trigger(lua, f"{self.mnemonic} plugin configuration")
# execute the work order
if self.configurationWorkOrders != None:
for workOrder in self.configurationWorkOrders:
workOrder.work(self.mnemonic, operation)
workOrder.work(operation)
# serves for subclasses
return self.isEnabled()

View File

@ -107,7 +107,7 @@ class QFlightCreator(QDialog):
start_type = "Cold"
else:
start_type = "Warm"
flight = Flight(aircraft, size, origin, task, start_type)
flight = Flight(self.package, aircraft, size, origin, task, start_type)
flight.scheduled_in = self.package.delay
flight.client_count = self.client_slots_spinner.value()

View File

@ -2,7 +2,7 @@ import pickle
from typing import Collection, Optional, Tuple
Zone = Collection[Tuple[float, float]]
Landmap = Tuple[Collection[Zone], Collection[Zone]]
Landmap = Tuple[Collection[Zone], Collection[Zone], Collection[Zone]]
def load_landmap(filename: str) -> Optional[Landmap]: