aman-sys/aman/com/Euroscope.py

#!/usr/bin/env python

import ctypes
import glob
import os
import sys
import time

import zmq
import zmq.auth

from aman.com import AircraftReport_pb2
from aman.config.Server import Server
from threading import Thread, _active

class ReceiverThread(Thread):
    def __init__(self, socket, aman):
        Thread.__init__(self)
        self.socket = socket
        self.aman = aman

    def run(self):
        while True:
            try:
                msg = self.socket.recv(zmq.NOBLOCK)

                # parse the received message
                report = AircraftReport_pb2.AircraftReport()
                report.ParseFromString(msg)

                # try to associate the received aircraft to an airport
                self.aman.updateAircraftReport(report)

            except zmq.ZMQError as error:
                if zmq.EAGAIN == error.errno:
                    time.sleep(0.5)
                    continue
                else:
                    return

    def threadId(self):
        if hasattr(self, '_thread_id'):
            return self._thread_id
        for id, thread in _active.items():
            if thread is self:
                return id

    def stopThread(self):
        id = self.threadId()
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(id, ctypes.py_object(SystemExit))
        if 1 < res:
            ctypes.pythonapi.PyThreadState_SetAsyncExc(id, 0)

# @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 acquire(self, configPath : str, config : Server, aman):
        self.context = zmq.Context()

        # find the key directories
        serverKeyPath = os.path.join(os.path.join(configPath, 'keys'), 'server')
        if False == os.path.isdir(serverKeyPath):
            sys.stderr.write('No directory for the server key found')
            sys.exit(-1)
        print('Path to the server key: ' + serverKeyPath)

        clientKeyPath = os.path.join(os.path.join(configPath, 'keys'), 'clients')
        if False == os.path.isdir(clientKeyPath):
            sys.stderr.write('No directory for the client keys found')
            sys.exit(-1)
        print('Path to the client keys: ' + clientKeyPath)

        # read the certificates
        keyPairPath = glob.glob(os.path.join(serverKeyPath, '*.key_secret'))
        if 1 != len(keyPairPath):
            sys.stderr.write('No public-private keypair found for the server certificate')
            sys.exit(-1)
        keyPair = zmq.auth.load_certificate(keyPairPath[0])

        # initialize the receiver
        self.receiverSocket = zmq.Socket(self.context, zmq.SUB)
        self.receiverSocket.setsockopt(zmq.CURVE_PUBLICKEY, keyPair[0])
        self.receiverSocket.setsockopt(zmq.CURVE_SECRETKEY, keyPair[1])
        self.receiverSocket.setsockopt(zmq.CURVE_SERVER, True)
        self.receiverSocket.bind('tcp://' + config.Address + ':' + str(config.PortReceiver))
        self.receiverSocket.setsockopt(zmq.SUBSCRIBE, b'')
        self.receiverThread = ReceiverThread(self.receiverSocket, aman)
        self.receiverThread.start()
        print('Listening to tcp://' + config.Address + ':' + str(config.PortReceiver))

        # initialize the notification
        self.notificationSocket = zmq.Socket(self.context, zmq.PUB)
        self.notificationSocket.setsockopt(zmq.CURVE_PUBLICKEY, keyPair[0])
        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 to tcp://' + config.Address + ':' + str(config.PortNotification))

    def release(self):
        if None != self.receiverThread:
            self.receiverThread.stopThread()
            self.receiverThread.join()
        self.receiverThread = None

        if None != self.receiverSocket:
            self.receiverSocket.close()
        self.receiverSocket = None

        if None != self.notificationSocket:
            self.notificationSocket.close()
        self.notificationSocket = None