116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
#!/usr/bin/env python
|
|
|
|
import json
|
|
import requests
|
|
import sys
|
|
|
|
from json import JSONEncoder
|
|
from requests.adapters import HTTPAdapter
|
|
from requests.packages.urllib3.util.retry import Retry
|
|
from requests.structures import CaseInsensitiveDict
|
|
|
|
from aman.config.AirportSequencing import AirportSequencing
|
|
from aman.config.RunwaySequencing import RunwaySequencing
|
|
from aman.config.Server import Server
|
|
|
|
class InboundEncoder(JSONEncoder):
|
|
def default(self, o):
|
|
pta = str(o.PlannedArrivalTime)
|
|
delimiter = pta.find('.')
|
|
if -1 == delimiter:
|
|
delimiter = pta.find('+')
|
|
return { 'callsign' : o.Callsign, 'runway' : o.PlannedRunway.Name, 'pta' : pta[0:delimiter] }
|
|
|
|
class WebUI:
|
|
def __init__(self):
|
|
self.Config = None
|
|
self.Aman = None
|
|
|
|
def __del__(self):
|
|
self.release()
|
|
|
|
def acquire(self, config : Server, aman):
|
|
self.Config = config
|
|
self.Aman = aman
|
|
|
|
def release(self):
|
|
return
|
|
|
|
def createSession():
|
|
# prepare the session
|
|
session = requests.Session()
|
|
retry = Retry(total=1, read=1, connect=1)
|
|
adapter = HTTPAdapter(max_retries=retry)
|
|
session.mount('http://', adapter)
|
|
session.mount('https://', adapter)
|
|
return session
|
|
|
|
def requestConfiguration(self, airport, navData):
|
|
if airport not in navData.Runways:
|
|
return None
|
|
|
|
# prepare the HTTP header and session
|
|
headers = CaseInsensitiveDict()
|
|
headers['Accept'] = 'application/json'
|
|
headers['Content-Type'] = 'application/json'
|
|
session = WebUI.createSession()
|
|
|
|
url = self.Config.WebUiUrl + self.Config.WebUiConfigurationReceiver
|
|
data = json.dumps({ 'airport': airport }, ensure_ascii=True)
|
|
|
|
try:
|
|
response = session.get(url, headers=headers, data=data, timeout=2)
|
|
if 200 != response.status_code:
|
|
return None
|
|
except requests.exceptions.ConnectTimeout:
|
|
return None
|
|
|
|
config = json.loads(response.json())
|
|
|
|
# WebUI sent the wrong airport
|
|
if config['airport'] != airport or 0 == len(config['activeRunways']):
|
|
return None
|
|
|
|
airportConfig = AirportSequencing(airport)
|
|
airportConfig.UseMustShallMay = config['useMustShallMay']
|
|
|
|
# build the sequencing information per runway
|
|
for activeRunway in config['activeRunways']:
|
|
gngRunway = None
|
|
for runway in navData.Runways[airport]:
|
|
if runway.Name == activeRunway['runway']:
|
|
gngRunway = runway
|
|
break
|
|
|
|
# could not find the runway
|
|
if None == gngRunway:
|
|
return None
|
|
|
|
runway = RunwaySequencing(gngRunway)
|
|
runway.Spacing = int(activeRunway['spacing'])
|
|
airportConfig.activateRunway(runway)
|
|
|
|
# resolve the dependencies
|
|
for dependency in config['dependentRunways']:
|
|
if 2 == len(dependency):
|
|
airportConfig.addDependency(dependency[0], dependency[1])
|
|
|
|
return airportConfig
|
|
|
|
def sendSequence(self, airport, inbounds):
|
|
# prepare the HTTP header and session
|
|
headers = CaseInsensitiveDict()
|
|
headers['Accept'] = 'application/json'
|
|
headers['Content-Type'] = 'application/json'
|
|
session = WebUI.createSession()
|
|
|
|
# prepare the relevant information
|
|
url = self.Config.WebUiUrl + self.Config.WebUiSequenceNotification
|
|
data = json.dumps({ 'airport': airport, 'sequence': inbounds }, ensure_ascii=True, cls=InboundEncoder)
|
|
|
|
# send to the server
|
|
try:
|
|
session.patch(url, headers=headers, data=data, timeout=2)
|
|
except requests.exceptions.ConnectTimeout:
|
|
return
|