aman-sys/aman/config/Airport.py

#!/usr/bin/env python

import configparser
import glob
import os
import sys

from aman.config.RHC import RHC
from aman.config.AirportSequencing import AirportSequencing
from aman.config.RunwaySequencing import RunwaySequencing
from aman.formats.SctEseFormat import SctEseFormat

class Airport:
    def findGngData(data, path):
        if None == data.get('gngwildcard'):
            return None, None

        # find the newest ESE file
        files = glob.glob(os.path.join(path, data['gngwildcard'] + '.ese'))
        latestEse = max(files, key=os.path.getctime)

        # search for the corresponding SCT file
        latestSct = os.path.splitext(latestEse)[0] + '.sct'

        # check if the files exist
        if False == os.path.isfile(latestEse) or False == os.path.isfile(latestSct):
            return None, None

        return latestSct, latestEse

    def parsePlanning(self, planning):
        if None == planning.get('routes'):
            return []
        return planning['routes'].split(':')

    def parseDefaultSequencingConfiguration(self, icao : str, planning):
        if None == planning.get('activearrivalrunwaydefault'):
            sys.stderr.write('No "activearrivalrunwaydefault" entry found!')
            sys.exit(-1)
        if None == planning.get('activearrivalmodedefault'):
            sys.stderr.write('No "activearrivalmodedefault" entry found!')
            sys.exit(-1)
        if None == planning.get('arrivalspacingdefault'):
            sys.stderr.write('No "arrivalspacingdefault" entry found!')
            sys.exit(-1)
        if not icao in self.GngData.runways:
            sys.stderr.write('Unable to find' + icao + 'in the SCT data!')
            sys.exit(-1)

        # parse the default arrival mode
        if 'STAGGERED' == planning['activearrivalmodedefault']:
            staggered = True
        elif 'IPA' == planning['activearrivalmodedefault']:
            staggered = False
        else:
            sys.stderr.write('Unknown arrival mode in "" found! (STAGGERED or IPA needs to be set)')
            sys.exit(-1)

        # translate the spacing into a map
        ident = ''
        spacings = {}
        spacingConfig = list(filter(None, planning['arrivalspacingdefault'].split(':')))
        for i in range(0, len(spacingConfig)):
            if 0 == i % 2:
                ident = spacingConfig[i]
            elif '' != ident:
                spacings[ident] = int(spacingConfig[i])
            else:
                sys.stderr.write('No runway defined in "arrivalspacingdefault"!')
                sys.exit(-1)

        # create the sequencing data per runway
        self.DefaultSequencing = AirportSequencing(icao)
        for ident in list(filter(None, planning['activearrivalrunwaydefault'].split(':'))):
            if not ident in spacings:
                sys.stderr.write('Unable to find sequencing data for ' + ident + ' of ' + icao)
                sys.exit(-1)

            found = False
            for runway in self.GngData.runways[icao]:
                if ident == runway.name:
                    sequence = RunwaySequencing(runway)
                    sequence.Spacing = spacings[ident]
                    self.DefaultSequencing.activateRunway(sequence)
                    found = True
                    break

            if False == found:
                sys.stderr.write('Unable to find the runway for ' + ident + ' of ' + icao + ' in SCT data!')
                sys.exit(-1)

        # create the dependencies, if needed
        if True == staggered:
            if None == planning.get('runwaydependenciesdefault'):
                sys.stderr.write('Unable to find the runway dependencies for staggered approaches of ' + icao + '!')
                sys.exit(-1)

            dependencies = list(filter(None, planning['runwaydependenciesdefault'].split(':')))
            if 0 != len(dependencies) % 2:
                sys.stderr.write('No valid set of runway dependencies found!')
                sys.exit(-1)

            for i in range(0, len(dependencies), 2):
                self.DefaultSequencing.addDependency(dependencies[i], dependencies[i + 1])

    def __init__(self, filepath : str, icao : str):
        self.arrivalRoutes = {}

        config = configparser.ConfigParser()
        config.read(filepath)

        dataConfig = None
        planningConfig = None
        rhcConfig = None

        # search the required sections
        for key in config:
            if 'DATA' == key:
                dataConfig = config['DATA']
            elif 'PLANNING' == key:
                planningConfig = config['PLANNING']
            elif 'RHC' == key:
                rhcConfig = config['RHC']

        # find the GNG-file data
        sctFile, eseFile = Airport.findGngData(dataConfig, os.path.dirname(filepath))
        if None == sctFile or None == eseFile:
            sys.stderr.write('No GNG-files found')
            sys.exit(-1)

        # parse the planning information
        if None == planningConfig or False == self.parsePlanning(planningConfig):
            sys.stderr.write('No planning configuration found')
            sys.exit(-1)
        requiredArrivalRoutes = self.parsePlanning(planningConfig)
        if 0 == len(requiredArrivalRoutes):
            sys.stderr.write('No valid planning configuration found')
            sys.exit(-1)

        # parse the RHC information
        if None == rhcConfig:
            sys.stderr.write('No RHC configuration found')
            sys.exit(-1)
        self.RecedingHorizonControl = RHC(rhcConfig)

        # parse the GNG data
        print('Used GNG-Data: ' + eseFile)
        self.GngData = SctEseFormat(sctFile, eseFile, icao, requiredArrivalRoutes)

        # get the GAFOR id
        if None == dataConfig.get('gaforid'):
            sys.stderr.write('No GAFOR-ID found!')
            sys.exit(-1)
        self.GaforId = dataConfig['gaforid']

        # get the default sequencing data
        self.parseDefaultSequencingConfiguration(icao, planningConfig)