Merge branch 'feature/weather' into 'develop'

Feature/weather

See merge request nav/aman/aman-sys!2

algorithm/rhcacsass.py

@@ -1,18 +0,0 @@
-# RHC-ACS-ASS Algorithm
-# This class is written to contain an implementation
-# of the Ant Colony System based upon the Receding Horizon
-# for Aircraft Arrival Sequencing
-
-class RhcAcsAss:
-    k = 1  # Receding horizon counter
-    N_rhc = 4  # The number of receding horizons
-    T_ti = 1  # The scheduling window
-
-    def __init__(self, n, t):
-        self.N_rhc = n
-        self.T_ti = t
-
-    def find_aircraft_for_horizon(self, ki):
-        # Omega(k) = [(k-1) * T_ti, (k+N_rhc-1) * T_ti]
-        pass
-    

aman/AMAN.py

@@ -6,6 +6,7 @@ import sys
 
 from aman.com import AircraftReport_pb2
 from aman.com.Euroscope import Euroscope
+from aman.com.Weather import Weather
 from aman.config.AircraftPerformance import AircraftPerformance
 from aman.config.Airport import Airport
 from aman.config.System import System
@@ -29,9 +30,14 @@ class AMAN:
         self.systemConfig = None
         self.aircraftPerformance = None
         self.receiver = None
+        self.weather = None
         self.workers = []
         self.inbounds = {}
 
+    def __del__(self):
+        self.release()
+
+    def aquire(self):
         configPath = AMAN.findConfigPath()
 
         # read all system relevant configuration files
@@ -46,6 +52,9 @@ class AMAN:
         else:
             print('Parsed PerformanceData.ini. Extracted ' + str(len(self.aircraftPerformance.aircrafts)) + ' aircrafts')
 
+        self.weather = Weather()
+        self.weather.acquire(self.systemConfig.Weather)
+
         # find the airport configurations and create the workers
         airportsPath = os.path.join(os.path.join(configPath, 'airports'), '*.ini')
         for file in glob.glob(airportsPath):
@@ -55,28 +64,34 @@ class AMAN:
             airportConfig = Airport(file, icao)
 
             # initialize the worker thread
-            worker = Worker(icao, airportConfig)
+            worker = Worker()
-            worker.start()
+            worker.acquire(icao, airportConfig)
             self.workers.append(worker)
-            print('Starter worker for ' + icao)
+            print('Started worker for ' + icao)
 
         # create the EuroScope receiver
-        self.receiver = Euroscope(configPath, self.systemConfig.Server, self)
+        self.receiver = Euroscope()
+        self.receiver.acquire(configPath, self.systemConfig.Server, self)
 
-    def __del__(self):
+    def release(self):
         if None != self.receiver:
-            del self.receiver
+            self.receiver.release()
         self.receiver = None
 
-        for worker in self.workers:
+        if None != self.weather:
-            worker.stop()
+            self.weather.release()
+        self.weather = None
+
+        if None != self.workers:
+            for worker in self.workers:
+                worker.release()
+        self.workers = None
 
     def updateAircraftReport(self, report : AircraftReport_pb2.AircraftReport):
         # find the correct worker for the inbound
         for worker in self.workers:
             if worker.icao == report.destination:
-                print('Updated ' + report.aircraft.callsign + ' for ' + worker.icao)
+                worker.acquireLock()
-                worker.acquire()
                 worker.reportQueue[report.aircraft.callsign] = report
-                worker.release()
+                worker.releaseLock()
                 break

aman/__init__.py

@@ -1,2 +0,0 @@
-import com
-import tools

aman/com/DwdCrawler.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python
+
+import datetime
+import time
+import urllib.request
+
+from bs4 import BeautifulSoup
+from datetime import datetime as dt
+
+# @brief Checks the DWD pages for wind information
+# Format:
+#   Provides next update tine (updateTime) of the DWD page in UTC
+#   Provides a list of wind information (windData)
+#       - organized as a list of tuples
+#           - first element of tuple: GAFOR-IDs for the following wind information
+#           - second element of tuple: list of tuples of wind data
+#               - first element of wind data tuple: minimum altitude AMSL for this wind information
+#               - second element of wind data tuple: wind direction
+#               - third  element of wind data tuple: wind speed (KT)
+class DwdCrawler():
+    def __init__(self):
+        self.updateTime = None
+        self.windData = None
+
+    def parseGaforAreas(areas : str):
+        areas = areas.replace(':', '')
+        areas = areas.split(' ')[1]
+        areaIds = []
+
+        # some IDs are lists
+        for segment in areas.split(','):
+            # check if we have range definitions or single IDs
+            borders = segment.split('-')
+            if 2 == len(borders):
+                areaIds.extend(range(int(borders[0]), int(borders[1]) + 1))
+            else:
+                areaIds.append(int(borders[0]))
+
+        return areaIds
+
+    def parseWindTableRow(row : str, table):
+        # get the columns
+        entries = row.split('|')
+
+        # check if the line is invalid or we have the header
+        if 2 > len(entries) or 'AMSL' in entries[0]:
+            return table
+
+        # parse the wind data
+        windData = entries[1].strip().split(' ')[0].split('/')
+        if 2 != len(windData):
+            return table
+
+        # extend the table
+        altitude = entries[0].strip()
+        if 'FL' in altitude:
+            altitude = int(altitude.replace('FL', '')) * 100
+        else:
+            altitude = int(altitude.replace('FT', ''))
+        row = ( altitude, int(windData[0]), int(windData[1].replace('KT', '')) )
+        table.append(row)
+
+        return table
+
+    def parseNextUpdateTime(line : str):
+        entries = line.split(' ')
+        if 4 <= len(entries):
+            utcIndex = 2
+        if 'UTC' in entries[len(entries) - 2]:
+            utcIndex = len(entries) - 3
+        elif 'UTC' in entries[len(entries) - 1]:
+            utcIndex = len(entries - 2)
+
+        currentUtc = dt.utcfromtimestamp(int(time.time()))
+        currentHour = int(currentUtc.strftime('%H'))
+
+        # check if we have a day overlap
+        if currentHour > int(entries[utcIndex].split('.')[0]):
+            nextDay = currentUtc + datetime.timedelta(days=1)
+            date = nextDay.strftime('%Y-%m-%d')
+        else:
+            date = currentUtc.strftime('%Y-%m-%d')
+
+        # create the new UTC update time
+        return dt.strptime(date + ' ' + entries[utcIndex] + '+0000', '%Y-%m-%d %H.%M%z')
+
+    def parseGaforPage(self, url : str):
+        with urllib.request.urlopen(url) as site:
+            data = site.read().decode('utf-8')
+            site.close()
+
+            parsed = BeautifulSoup(data, features='lxml')
+
+            # search the info about the GAFOR areas
+            content = None
+            for element in parsed.body.find_all('pre'):
+                content = element.text
+
+            # analyze the received data
+            if None != content:
+                windInformation = []
+                nextUpdate = None
+                windTable = []
+                areaIds = None
+
+                # find all relevant information
+                for line in content.splitlines():
+                    if '' == line:
+                        if 0 != len(windTable):
+                            windInformation.append(( areaIds, windTable ))
+                        areaIds = None
+                        windTable = []
+                    elif line.startswith('GAFOR-Gebiete'):
+                        areaIds = DwdCrawler.parseGaforAreas(line)
+                        windTable = []
+                    elif None != areaIds:
+                        windTable = DwdCrawler.parseWindTableRow(line, windTable)
+                    elif 'Aktualisierung erfolgt um ' in line:
+                        nextUpdate = DwdCrawler.parseNextUpdateTime(line)
+
+                # return the collected information
+                if 0 == len(windInformation) or None == nextUpdate:
+                    return None, None
+                else:
+                    return nextUpdate, windInformation
+
+    def receiveWindData(self):
+        self.updateTime = None
+        self.windData = None
+
+        with urllib.request.urlopen('https://www.dwd.de/DE/fachnutzer/luftfahrt/teaser/luftsportberichte/luftsportberichte_node.html') as site:
+            data = site.read().decode('utf-8')
+            site.close()
+
+            # find the pages of the GAFOR reports
+            pages = []
+            parsed = BeautifulSoup(data, features='lxml')
+            for link in parsed.body.find_all('a', title=True):
+                if 'node' in link['href'] and 'Flugwetterprognose' in link['title']:
+                    # remove the jsession from the link
+                    pages.append('https://www.dwd.de/' + link['href'].split(';')[0])
+
+            # receive the wind data
+            self.updateTime = None
+            self.windData = []
+            for page in pages:
+                next, wind = self.parseGaforPage(page)
+                if None != next:
+                    if None == self.updateTime or self.updateTime > next:
+                        self.updateTime = next
+                    self.windData.extend(wind)
+
+            # indicate that new wind data is available
+            if None != self.updateTime:
+                return True
+            else:
+                return False

aman/com/Euroscope.py

@@ -4,7 +4,6 @@ import ctypes
 import glob
 import os
 import sys
-import threading
 import time
 
 import zmq
@@ -12,10 +11,11 @@ import zmq.auth
 
 from aman.com import AircraftReport_pb2
 from aman.config.Server import Server
+from threading import Thread, _active
 
-class ReceiverThread(threading.Thread):
+class ReceiverThread(Thread):
     def __init__(self, socket, aman):
-        threading.Thread.__init__(self)
+        Thread.__init__(self)
         self.socket = socket
         self.aman = aman
 
@@ -41,7 +41,7 @@ class ReceiverThread(threading.Thread):
     def threadId(self):
         if hasattr(self, '_thread_id'):
             return self._thread_id
-        for id, thread in threading._active.items():
+        for id, thread in _active.items():
             if thread is self:
                 return id
 
@@ -53,9 +53,18 @@ class ReceiverThread(threading.Thread):
 
 # @brief Receives and sends messages to EuroScope plugins
 class Euroscope:
+    def __init__(self):
+        self.context = None
+        self.receiverSocket = None
+        self.receiverThread = None
+        self.notificationSocket = None
+
+    def __del__(self):
+        self.release()
+
     # @brief Initializes the ZMQ socket
     # @param[in] config The server configuration
-    def __init__(self, configPath : str, config : Server, aman):
+    def acquire(self, configPath : str, config : Server, aman):
         self.context = zmq.Context()
 
         # find the key directories
@@ -87,7 +96,7 @@ class Euroscope:
         self.receiverSocket.setsockopt(zmq.SUBSCRIBE, b'')
         self.receiverThread = ReceiverThread(self.receiverSocket, aman)
         self.receiverThread.start()
-        print('Listening at tcp://' + config.Address + ':' + str(config.PortReceiver))
+        print('Listening to tcp://' + config.Address + ':' + str(config.PortReceiver))
 
         # initialize the notification
         self.notificationSocket = zmq.Socket(self.context, zmq.PUB)
@@ -95,10 +104,18 @@ class Euroscope:
         self.notificationSocket.setsockopt(zmq.CURVE_SECRETKEY, keyPair[1])
         self.notificationSocket.setsockopt(zmq.CURVE_SERVER, True)
         self.notificationSocket.bind('tcp://' + config.Address + ':' + str(config.PortNotification))
-        print('Publishing at tcp://' + config.Address + ':' + str(config.PortNotification))
+        print('Publishing to tcp://' + config.Address + ':' + str(config.PortNotification))
 
-    def __del__(self):
+    def release(self):
-        self.receiverThread.stopThread()
+        if None != self.receiverThread:
-        self.receiverThread.join()
+            self.receiverThread.stopThread()
-        self.receiverSocket.close()
+            self.receiverThread.join()
-        self.notificationSocket.close()
+        self.receiverThread = None
+
+        if None != self.receiverSocket:
+            self.receiverSocket.close()
+        self.receiverSocket = None
+
+        if None != self.notificationSocket:
+            self.notificationSocket.close()
+        self.notificationSocket = None

aman/com/Weather.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+
+import pytz
+import sys
+import time
+
+from datetime import datetime as dt
+from threading import Thread
+
+from aman.com.DwdCrawler import DwdCrawler
+import aman.config.Weather
+
+class Weather(Thread):
+    def __init__(self):
+        Thread.__init__(self)
+
+        self.nextUpdate = None
+        self.lastUpdateTried = None
+        self.stopThread = False
+        self.provider = None
+
+    def acquire(self, config : aman.config.Weather.Weather):
+        self.nextUpdate = dt.utcfromtimestamp(int(time.time()))
+        self.lastUpdateTried = None
+        self.stopThread = False
+        self.provider = None
+
+        if 'DWD' == config.Provider.upper():
+            self.provider = DwdCrawler()
+        else:
+            sys.stderr.write('Invalid or unknown weather-provider defined')
+            sys.exit(-1)
+
+        self.start()
+
+    def release(self):
+        self.stopThread = True
+        self.join()
+
+    def currentClock():
+        clock = dt.utcfromtimestamp(int(time.time())).replace(tzinfo = pytz.UTC)
+        return clock
+
+    def run(self):
+        while False == self.stopThread and None != self.provider:
+            now = Weather.currentClock()
+
+            # check if an update is required
+            if None != self.provider.updateTime and self.provider.updateTime > now:
+                time.sleep(1)
+                continue
+
+            if None == self.lastUpdateTried or self.lastUpdateTried <= now:
+                if True == self.provider.receiveWindData():
+                    self.nextUpdate = self.provider.updateTime
+                    print('Received new wind data')

aman/config/System.py

@@ -4,6 +4,7 @@ import configparser
 import sys
 
 from aman.config.Server import Server
+from aman.config.Weather import Weather
 
 class System:
     def __init__(self, filepath : str):
@@ -15,9 +16,15 @@ class System:
         for key in config:
             if 'SERVER' == key:
                 serverSectionAvailable = True
+            elif 'WEATHER' == key:
+                weatherSectionAvailable = True
 
         if not serverSectionAvailable:
             sys.stderr.write('No server-configuration section found!')
             sys.exit(-1)
+        if not weatherSectionAvailable:
+            sys.stderr.write('No weather-configuration section found!')
+            sys.exit(1)
 
         self.Server = Server(config['SERVER'])
+        self.Weather = Weather(config['WEATHER'])

aman/config/Weather.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python
+
+import configparser;
+import sys
+
+class Weather():
+    def __init__(self, config : configparser.ConfigParser):
+        self.Provider = None
+        self.PortReceiver = None
+        self.PortNotification = None
+
+        # search the required sections
+        for key in config:
+            if 'provider' == key:
+                self.Provider = config['provider']
+
+        if self.Provider is None:
+            sys.stderr.write('No weather-provider configuration found!')
+            sys.exit(-1)

aman/sys/WeatherModel.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python
+
+from aman.com.Weather import Weather
+
+import math
+import scipy.interpolate
+
+class WeatherModel:
+    def __init__(self, gaforId, weather : Weather):
+        self.gafor = gaforId
+        self.weather = weather
+        self.windDirectionModel = None
+        self.windSpeedModel = None
+        self.lastWeatherUpdate = None
+        self.minimumAltitude = 1000000
+        self.maximumAltitude = -1
+
+        # create the density interpolation model
+        # the density model is based on https://aerotoolbox.com/atmcalc/
+        altitudes = [
+            50000,
+            45000,
+            40000,
+            38000,
+            36000,
+            34000,
+            32000,
+            30000,
+            28000,
+            26000,
+            24000,
+            22000,
+            20000,
+            18000,
+            16000,
+            15000,
+            14000,
+            13000,
+            12000,
+            11000,
+            10000,
+            9000,
+            8000,
+            7000,
+            6000,
+            5000,
+            4000,
+            3000,
+            2000,
+            1000,
+            0
+        ]
+        densities = [
+            0.18648,
+            0.23714,
+            0.24617,
+            0.33199,
+            0.36518,
+            0.39444,
+            0.42546,
+            0.45831,
+            0.402506,
+            0.432497,
+            0.464169,
+            0.60954,
+            0.65269,
+            0.69815,
+            0.74598,
+            0.77082,
+            0.79628,
+            0.82238,
+            0.84914,
+            0.87655,
+            0.90464,
+            0.93341,
+            0.96287,
+            0.99304,
+            1.02393,
+            1.05555,
+            1.08791,
+            1.12102,
+            1.1549,
+            1.18955,
+            1.225
+        ]
+        self.densityModel = scipy.interpolate.interp1d(altitudes, densities)
+
+    def calculateTAS(self, altitude : int, ias : int):
+        if altitude >= 50000:
+            altitude = 49999
+        if altitude <= 0:
+            altitude = 1
+
+        # calculation based on https://aerotoolbox.com/airspeed-conversions/
+        return ias * math.sqrt(1.225 / self.densityModel(altitude).item())
+
+    def updateWindModel(self):
+        if None == self.lastWeatherUpdate or self.lastWeatherUpdate != self.weather.provider.updateTime:
+            self.lastWeatherUpdate = self.weather.provider.updateTime
+
+            self.minimumAltitude = 1000000
+            self.maximumAltitude = -1
+            self.windDirectionModel = None
+            self.windSpeedModel = None
+
+            if None != self.weather.provider.windData and self.gafor in self.weather.provider.windData:
+                altitudes = []
+                directions = []
+                speeds = []
+
+                # collect the data for the wind model
+                for level in self.weather.provider.windData[self.gafor]:
+                    altitudes.append(level[0])
+                    directions.append(level[1])
+                    speeds.append(level[2])
+
+                    # define the thresholds for later boundary checks
+                    if self.minimumAltitude > level[0]:
+                        self.minimumAltitude = level[0]
+                    if self.maximumAltitude < level[0]:
+                        self.maximumAltitude = level[0]
+
+                # calculate the models
+                if 1 < len(altitudes):
+                    self.windDirectionModel = scipy.interpolate.interp1d(altitudes, directions)
+                    self.windSpeedModel = scipy.interpolate.interp1d(altitudes, speeds)
+
+    def calculateGS(self, altitude : int, ias : int, heading : int):
+        self.updateWindModel()
+        tas = self.calculateTAS(altitude, ias)
+
+        # initialize the wind data
+        if None != self.windDirectionModel and None != self.windSpeedModel:
+            direction = 0.0
+            speed = 0.0
+            if None != self.windSpeedModel and None != self.windDirectionModel:
+                if self.maximumAltitude <= altitude:
+                    altitude = self.maximumAltitude - 1
+                if self.minimumAltitude >= altitude:
+                    altitude = self.minimumAltitude + 1
+                direction = self.windDirectionModel(altitude).item()
+                speed = self.windSpeedModel(altitude).item()
+        else:
+            speed = 0
+            direction = 0
+
+        # calculate the ground speed based on the headwind component
+        return tas + speed * math.cos(math.radians(direction) - math.radians(heading))

aman/sys/Worker.py

@@ -6,18 +6,38 @@ import time
 from aman.config.Airport import Airport
 
 class Worker(Thread):
-    def __init__(self, icao : str, configuration : Airport):
+    def __init__(self):
         Thread.__init__(self)
+        self.stopThread = None
+        self.icao = None
+        self.configuration = None
+        self.arrivalRoutes = None
+        self.updateLock = None
+        self.reportQueue = {}
 
+    def __del__(self):
+        self.release()
+
+    def acquire(self, icao : str, configuration : Airport):
         self.stopThread = None
         self.icao = icao
         self.configuration = configuration
         self.arrivalRoutes = configuration.gngData.arrivalRoutes
         self.updateLock = Lock()
         self.reportQueue = {}
+        self.start()
 
-    def stop(self):
+    def acquireLock(self):
+        if None != self.updateLock:
+            self.updateLock.acquire()
+
+    def release(self):
         self.stopThread = True
+        self.join()
+
+    def releaseLock(self):
+        if None != self.updateLock:
+            self.updateLock.release()
 
     def run(self):
         counter = 0

tcp/TCPServer.py

@@ -1,51 +0,0 @@
-import socket
-import _thread
-
-
-class TCPServer(socket.socket):
-    clients = []
-
-    def __init__(self):
-        socket.socket.__init__(self)
-        self.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.bind(('0.0.0.0', 8765))
-        self.listen(5)
-
-    def run(self):
-        print("Starting TCP Server")
-        try:
-            self.accept_clients()
-        except Exception as ex:
-            print(ex)
-        finally:
-            print("Server shutdown")
-            for client in self.clients:
-                client.close()
-            self.close()
-
-    def accept_clients(self):
-        while True:
-            (client_socket, address) = self.accept()
-            self.clients.append(client_socket)
-            self.on_open(client_socket)
-            _thread.start_new_thread(self.receive, (client_socket,))
-
-    def receive(self, client):
-        while True:
-            data = client.recv(1024)
-            if data == '':
-                break
-            self.on_message(client, data)
-        self.clients.remove(client)
-        self.on_close(client)
-        client.close()
-        _thread.exit()
-
-    def on_open(self, client):
-        pass
-
-    def on_message(self, client, message):
-        pass
-
-    def on_close(self, client):
-        pass