radiosonde_auto_rx/auto_rx/autorx/ozimux.py

#!/usr/bin/env python
#
#   radiosonde_auto_rx - OziMux Output
#
#   Copyright (C) 2018  Mark Jessop <vk5qi@rfhead.net>
#   Released under GNU GPL v3 or later
#

import json
import logging
import socket
import time
from threading import Thread
try:
    # Python 2
    from Queue import Queue
except ImportError:
    # Python 3
    from queue import Queue


class OziUploader(object):
    """ Push radiosonde telemetry out via UDP broadcast to the Horus Chase-Car Utilities

    Uploads to:
        - OziMux / OziPlotter (UDP Broadcast, port 8942 by default) - Refer here for information on the data format:
            https://github.com/projecthorus/oziplotter/wiki/3---Data-Sources
        - "Payload Summary" (UDP Broadcast, on port 55672 by default)
            Refer here for information: https://github.com/projecthorus/horus_utils/wiki/5.-UDP-Broadcast-Messages#payload-summary-payload_summary

    NOTE: This class is designed to only handle telemetry from a single radiosonde at a time. It should only be operated in a
    single-SDR configuration, where only one sonde is tracked at a time.

    """

    # We require the following fields to be present in the incoming telemetry dictionary data
    REQUIRED_FIELDS = ['frame', 'id', 'datetime', 'lat', 'lon', 'alt', 'temp', 'type', 'freq', 'freq_float', 'datetime_dt']

    # Extra fields we can pass on to other programs.
    EXTRA_FIELDS = ['bt', 'humidity', 'sats']


    def __init__(self,
        ozimux_port = None,
        payload_summary_port = None,
        update_rate = 5
        ):
        """ Initialise an OziUploader Object.

        Args:
            ozimux_port (int): UDP port to push ozimux/oziplotter messages to. Set to None to disable.
            payload_summary_port (int): UDP port to push payload summary messages to. Set to None to disable.
            update_rate (int): Time in seconds between updates.
        """

        self.ozimux_port = ozimux_port
        self.payload_summary_port = payload_summary_port
        self.update_rate = update_rate

        # Input Queue. 
        self.input_queue = Queue()

        # Start the input queue processing thread.
        self.input_processing_running = True
        self.input_thread = Thread(target=self.process_queue)
        self.input_thread.start()

        self.log_info("Started OziMux / Payload Summary Exporter")


    def send_ozimux_telemetry(self, telemetry):
        """ Send a packet of telemetry into the network in OziMux/OziPlotter-compatible format.

        Args:
            telemetry (dict): Telemetry dictionary to send.

        """

        _short_time = telemetry['datetime_dt'].strftime("%H:%M:%S")
        _sentence = "TELEMETRY,%s,%.5f,%.5f,%d\n" % (_short_time, telemetry['lat'], telemetry['lon'], telemetry['alt'])

        try:
            _ozisock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)

            # Set up socket for broadcast, and allow re-use of the address
            _ozisock.setsockopt(socket.SOL_SOCKET,socket.SO_BROADCAST,1)
            _ozisock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            # Under OSX we also need to set SO_REUSEPORT to 1
            try:
                _ozisock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
            except:
                pass

            try:
                _ozisock.sendto(_sentence.encode('ascii'),('<broadcast>',self.ozimux_port))
            # Catch any socket errors, that may occur when attempting to send to a broadcast address
            # when there is no network connected. In this case, re-try and send to localhost instead.
            except socket.error as e:
                self.log_debug("Send to broadcast address failed, sending to localhost instead.")
                _ozisock.sendto(_sentence.encode('ascii'),('127.0.0.1',self.ozimux_port))

            _ozisock.close()

        except Exception as e:
            self.log_error("Failed to send OziMux packet: %s" % str(e))


    def send_payload_summary(self, telemetry):
        """ Send a payload summary message into the network via UDP broadcast.

        Args:
        telemetry (dict): Telemetry dictionary to send.

        """

        try:
            # Prepare heading & speed fields, if they are provided in the incoming telemetry blob.
            if 'heading' in telemetry.keys():
                _heading = telemetry['heading']
            else:
                _heading = -1

            if 'vel_h' in telemetry.keys():
                _speed = telemetry['vel_h']*3.6
            else:
                _speed = -1

            # Generate 'short' time field.
            _short_time = telemetry['datetime_dt'].strftime("%H:%M:%S")


            packet = {
                'type' : 'PAYLOAD_SUMMARY',
                'callsign' : telemetry['id'],
                'latitude' : telemetry['lat'],
                'longitude' : telemetry['lon'],
                'altitude' : telemetry['alt'],
                'speed' : _speed,
                'heading': _heading,
                'time' : _short_time,
                'comment' : 'Radiosonde',
                # Additional fields specifically for radiosondes
                'model': telemetry['type'],
                'freq': telemetry['freq'],
                'temp': telemetry['temp']
            }

            # Add in any extra fields we may care about.
            for _field in self.EXTRA_FIELDS:
                if _field in telemetry:
                    packet[_field] = telemetry[_field]



            # Set up our UDP socket
            _s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
            _s.settimeout(1)
            # Set up socket for broadcast, and allow re-use of the address
            _s.setsockopt(socket.SOL_SOCKET,socket.SO_BROADCAST,1)
            _s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            # Under OSX we also need to set SO_REUSEPORT to 1
            try:
                _s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
            except:
                pass

            try:
                _s.sendto(json.dumps(packet).encode('ascii'), ('<broadcast>', self.payload_summary_port))
            # Catch any socket errors, that may occur when attempting to send to a broadcast address
            # when there is no network connected. In this case, re-try and send to localhost instead.
            except socket.error as e:
                self.log_debug("Send to broadcast address failed, sending to localhost instead.")
                _s.sendto(json.dumps(packet).encode('ascii'), ('127.0.0.1', self.payload_summary_port))

            _s.close()

        except Exception as e:
            self.log_error("Error sending Payload Summary: %s" % str(e))


    def process_queue(self):
        """ Process packets from the input queue.

        This thread handles packets from the input queue (provided by the decoders)
        Packets are sorted by ID, and a dictionary entry is created. 

        """

        while self.input_processing_running:

            if self.input_queue.qsize() > 0:
                # Dump the queue, keeping the most recent element.
                while not self.input_queue.empty():
                    _telem = self.input_queue.get()

                # Send!
                if self.ozimux_port != None:
                    self.send_ozimux_telemetry(_telem)

                if self.payload_summary_port != None:
                    self.send_payload_summary(_telem)

            time.sleep(self.update_rate)




    def add(self, telemetry):
        """ Add a dictionary of telemetry to the input queue. 

        Args:
            telemetry (dict): Telemetry dictionary to add to the input queue.

        """

        # Check the telemetry dictionary contains the required fields.
        for _field in self.REQUIRED_FIELDS:
            if _field not in telemetry:
                self.log_error("JSON object missing required field %s" % _field)
                return

        # Add it to the queue if we are running.
        if self.input_processing_running:
            self.input_queue.put(telemetry)
        else:
            self.log_error("Processing not running, discarding.")


    def close(self):
        """ Shutdown processing thread. """
        self.log_debug("Waiting for processing thread to close...")
        self.input_processing_running = False

        if self.input_thread is not None:
            self.input_thread.join()


    def log_debug(self, line):
        """ Helper function to log a debug message with a descriptive heading. 
        Args:
            line (str): Message to be logged.
        """
        logging.debug("OziMux - %s" % line)


    def log_info(self, line):
        """ Helper function to log an informational message with a descriptive heading. 
        Args:
            line (str): Message to be logged.
        """
        logging.info("OziMux - %s" % line)


    def log_error(self, line):
        """ Helper function to log an error message with a descriptive heading. 
        Args:
            line (str): Message to be logged.
        """
        logging.error("OziMux - %s" % line)