radiosonde_modules.py

#
# Multi-Purpose APRS Daemon: radiosonde landing prediction
# Author: Joerg Schultze-Lutter, 2020
#
# Radiosonde landing prediction
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published byz
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
from utility_modules import read_program_config
import logging
from aprsdotfi_modules import get_position_on_aprsfi
from datetime import datetime, timedelta
import re
import requests
import xmltodict
import activesoup  # requires version 0.23 or greater
from bs4 import BeautifulSoup
from pprint import pformat

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s %(module)s -%(levelname)s- %(message)s"
)
logger = logging.getLogger(__name__)


def get_ascent_descent_burst(clmb: float, altitude: float):
    """
    Determines the ascent rate, descent rate and burst altitude
    based on both clmb rate and altitude from aprs.fi

    Parameters
    ==========
    clmb: 'float'
        Climb rate from the aprs.fi position report
    altitude: 'float'
        Altitude from the aprs.fi position report

    Returns
    =======
    ascent_rate: 'float'
        Ascent rate in meters
    descent_rate: 'float'
        Descent rate in meters
    burst_altitude: 'float'
        Burst altitude in meters
    """

    if clmb >= 0:
        ascent_rate = clmb
        descent_rate = 6
        if altitude < 25000:
            burst_altitude = 25000
        elif 25000 < altitude < 30000:
            burst_altitude = 30000
        elif 30000 < altitude < 35000:
            burst_altitude = 35000
        else:
            burst_altitude = 38000
    else:
        burst_altitude = altitude + 1
        ascent_rate = 0.01
        descent_rate = abs(clmb)
    return ascent_rate, descent_rate, burst_altitude


def get_clmb_from_comment(probe_comment: str):
    """
    Gets the 'clmb' rate from the aprs.fi position
    report (if present)

    Parameters
    ==========
    probe_comment: 'str'
        aprs.fi 'comment' from position report

    Returns
    =======
    clmb: 'float'
        Climb rate in meters
    """
    clmb = None
    matches = re.search(
        pattern=r"Clb=(-?[0-9]\d*(?:\.\d+)?)",
        string=probe_comment,
        flags=re.IGNORECASE,
    )

    if matches:
        try:
            clmb = float(matches[1])
        except ValueError:
            clmb = 0.0
    return clmb


def get_kml_data_from_habhub(
    latitude: float, longitude: float, altitude: float, clmb: float
):
    """
    Gets the 'clmb' rate from the aprs.fi position
    report (if present)

    Parameters
    ==========
    latitude: 'float'
        latitude from aprs.fi position report
    longitude: 'float'
        longitude from aprs.fi position report
    altitude: 'float'
        altitude from aprs.fi position report
    clmb: 'float'
        extracted clmb value from aprs.fi position report

    Returns
    =======
    success: 'bool'
        True if we were able to determine landing coordinates etc
    landing_latitude: 'float'
        Latitude of the predicted probe landing (if success = True)
    landing_longitude: 'float'
        Longitude of the predicted probe landing (if success = True)
    landing_timestamp: 'float'
        Timestamp of the predicted probe landing (if success = True)
    landing_url: 'str'
        habhub.org URL with the uuid which was generated by our query
    """

    landing_latitude = landing_longitude = 0.0
    landing_timestamp = datetime.min
    landing_url = None
    success = False

    ascent_rate, descent_rate, burst_altitude = get_ascent_descent_burst(
        clmb=clmb, altitude=altitude
    )

    # we can't use the aprs.fi timestamp as it is always in the past
    # the web site demands that we have a timestamp that is at least 1
    # minute in the future. So let's use the current UTC time and add
    # one minute to it
    timestamp = datetime.utcnow() + timedelta(minutes=1)

    # Create the payload item for the POST operation
    hubhab_payload = {
        "launchsite": "Other",
        "lat": f"{latitude}",
        "lon": f"{longitude}",
        "initial_alt": f"{altitude}",
        "hour": f"{timestamp.hour}",
        "min": f"{timestamp.minute}",
        "second": f"0",
        "day": f"{timestamp.day}",
        "month": f"{timestamp.month}",
        "year": f"{timestamp.year}",
        "ascent": f"{ascent_rate}",
        "burst": f"{burst_altitude}",
        "drag": f"{descent_rate}",
        "submit": "Run+Prediction",
    }
    # logger.info(hubhab_payload)

    # Send the payload to the site. If all goes well, Habhub responds
    # with a HTTP200 and provides us with a UUID
    url = "http://predict.habhub.org/ajax.php?action=submitForm"

    try:
        resp = requests.post(url=url, data=hubhab_payload)
    except:
        logger.info(f"Cannot connect to {url}")
        resp = None
    if resp:
        if resp.status_code == 200:
            json_content = resp.json()

            # Check if we have received a valid response from the  site
            valid = "false"
            if "valid" in json_content:
                valid = json_content["valid"]

            logger.info("Have received valid initial response from Habhub")

            # Everything seems to be okay so let's get the UUID (if present)
            if valid == "true":
                if "uuid" in json_content:
                    uuid = json_content["uuid"]

                    # We're going to download the KML file for this UUID
                    # Let's construct the respective URL
                    url = f"http://predict.habhub.org/kml.php?uuid={uuid}"
                    resp = requests.get(url=url)
                    if resp:
                        if resp.status_code == 200:

                            # We have received XML content. For better navigation throughout
                            # the data structure, let's convert the content to a 'dict' object
                            # (xmltodict converts to an OrderedDict)
                            try:
                                kml_dict = xmltodict.parse(resp.text)
                            except:
                                kml_dict = {}

                            logger.info("Have received valid uuid response from Habhub")

                            # Now navigate through the structure and get our data
                            # The stuff that we want is in the "Placemark" subsection
                            if "kml" in kml_dict:
                                if "Document" in kml_dict["kml"]:
                                    if "Placemark" in kml_dict["kml"]["Document"]:
                                        placemarks = kml_dict["kml"]["Document"][
                                            "Placemark"
                                        ]

                                        # Iterate through the available placemark objects and start
                                        # parsing once we have hit the "Predicted Balloon Landing" entry
                                        for placemark in placemarks:
                                            if "name" in placemark:
                                                name = placemark["name"]
                                                if name == "Predicted Balloon Landing":
                                                    # We have a winner! Our content is stored in the 'description' field
                                                    if "description" in placemark:
                                                        description = placemark[
                                                            "description"
                                                        ]

                                                        # run some regex magic for extracting what we want
                                                        regex_string = r"^Balloon landing at (-?\d*[.]\d*),\s*(-?\d*[.]\d*)\s*at\s*(\d*[:]\d* \d{2}\/\d{2}\/\d{4}).$"
                                                        matches = re.search(
                                                            pattern=regex_string,
                                                            string=description,
                                                            flags=re.IGNORECASE,
                                                        )
                                                        if matches:
                                                            # extract lat/lon/timestamp
                                                            # fmt: off
                                                            success = True
                                                            try:
                                                                landing_latitude = float(matches[1])
                                                                landing_longitude = float(matches[2])
                                                            except ValueError:
                                                                landing_latitude = landing_longitude = 0.0
                                                                success = False
                                                            ts_string = matches[3] + " UTC"  # timezone is UTC
                                                            # fmt: on
                                                            try:
                                                                landing_timestamp = datetime.strptime(
                                                                    ts_string,
                                                                    "%H:%M %d/%m/%Y %Z",
                                                                )
                                                                pass
                                                            except ValueError:
                                                                landing_latitude = (
                                                                    landing_longitude
                                                                ) = 0
                                                                landing_timestamp = (
                                                                    datetime.min
                                                                )
                                                                success = False
                                                            landing_url = f"https://predict.habhub.org/#!/uuid={uuid}"
                                                            break  # we have what we want so let's finish up
    return success, landing_latitude, landing_longitude, landing_timestamp, landing_url


def get_radiosonde_landing_prediction(aprsfi_callsign: str, aprsdotfi_api_key: str):
    """
    Provides a radiosonde landing prediction based on
    an aprs.fi call sign

    Parameters
    ==========
    aprsfi_callsign: 'str'
        aprs.fi callsign
    aprsdotfi_api_key: 'str'
        aprs.fi API access key

    Returns
    =======
    success: 'bool'
        True if we were able to determine landing coordinates etc
    landing_latitude: 'float'
        Latitude of the predicted probe landing (if success = True)
    landing_longitude: 'float'
        Longitude of the predicted probe landing (if success = True)
    landing_timestamp: 'float'
        Timestamp of the predicted probe landing (if success = True)
    landing_url: 'str'
        habhub.org URL with the uuid which was generated by our query
    """
    success = False
    landing_latitude = landing_longitude = 0.0
    landing_timestamp = datetime.min
    landing_url = None

    aprsfi_callsign = aprsfi_callsign.upper()

    # Try to get the position of the probe on aprs.fi
    # It seems that Radiosonde objects change their target
    # type during their live cycle; they all start with "l"
    # and end with "o". Therefore, limiting access to "o"
    # objects means that MPAD won't see them - so I removed that
    # restriction. It's not the most elegant way but seems
    # to work properly.
    (
        success,
        latitude,
        longitude,
        altitude,
        timestamp,
        comment,
        message_callsign,
    ) = get_position_on_aprsfi(
        aprsfi_callsign=aprsfi_callsign,
        aprsdotfi_api_key=aprsdotfi_api_key,
        #  aprs_target_type="o",
    )

    logger.info("Running query on aprs.fi")

    # We found the entry - so let's continue
    if success:
        if comment:
            # logger.info(comment)
            clmb = get_clmb_from_comment(probe_comment=comment)
            if clmb:
                logger.info("Getting KML data from Habhub")
                (
                    success,
                    landing_latitude,
                    landing_longitude,
                    landing_timestamp,
                    landing_url,
                ) = get_kml_data_from_habhub(
                    latitude=latitude, longitude=longitude, altitude=altitude, clmb=clmb
                )
            else:
                success = False
        else:
            success = False
            logger.info("Found on aprs.fi but does not contain a comment")
    else:
        logger.info("Not found on aprs.fi")
    return success, landing_latitude, landing_longitude, landing_timestamp, landing_url


def remove_trailing_content(source_string: str, trailing_content: str):
    if source_string:
        source_string = source_string.replace(trailing_content, "")
        source_string = None if len(source_string) == 0 else source_string
    return source_string


def parse_radiosondy_html_content(html_raw_content: str):
    _sonde_number = _launch_site = _probe_type = None
    _probe_aux = _probe_freq = _probe_status = None
    _max_speed = _max_speed_height = _avg_speed_kmh = None
    _max_altitude = _avg_ascent_speed = _avg_descent_speed = None

    # With the exception of the APRS data, the data that we want / need is stored as regular
    # text. We use the text's icons in order to identify the content
    regex_string = r"images\/balloon.png\"\> Number: ([\w\s]+)\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _sonde_number = matches[1]

    regex_string = r"images\/house.png\"\> Launch Site: (.*)\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _launch_site = matches[1]

    regex_string = r"images\/type.png\"\> Type: (.*)\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _probe_type = matches[1]

    regex_string = r"images\/aux.png\"\> AUX: (.*)\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _probe_aux = matches[1]

    regex_string = r"images\/freq.png\"\> Frequency: (.*) MHz\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _probe_freq = matches[1]

    regex_string = r"images\/found.png\"\> Status: (.*)\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _probe_status = matches[1]

    regex_string = r"images\/speed.png\"\> Max Speed: (.*) km\/h at (.*) m\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _max_speed = matches[1]
        _max_speed_height = matches[2]

    regex_string = r"images\/speed.png\"\> Average Speed: (.*) km\/h\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _avg_speed_kmh = matches[1]

    regex_string = r"images\/altitude.png\"\> Max Altitude: (.*) m\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _max_altitude = matches[1]

    regex_string = r"images\/up.png\"\> Average Ascent Speed: (.*) m\/s\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _avg_ascent_speed = matches[1]

    regex_string = r"images\/down.png\"\> Average Descent Speed: (.*) m\/s\<\/h4\>"
    matches = re.search(
        pattern=regex_string,
        string=html_raw_content,
        flags=re.IGNORECASE,
    )
    if matches:
        _avg_descent_speed = matches[1]

    response_dict = {
        "sonde_number": _sonde_number,
        "launch_site": _launch_site,
        "probe_type": _probe_type,
        "probe_aux": _probe_aux,
        "probe_freq": _probe_freq,
        "probe_status": _probe_status,
        "max_speed": _max_speed,
        "max_speed_height": _max_speed_height,
        "avg_speed_kmh": _avg_speed_kmh,
        "max_altitude": _max_altitude,
        "avg_ascent_speed": _avg_ascent_speed,
        "avg_descent_speed": _avg_descent_speed,
    }
    return response_dict


def get_radiosondy_data(sonde_id: str):
    """Get Radiosonde data from radiosondy.info
    Parameters
    ==========
    sonde_id : 'str'
        ID of the sonde whose data
    Returns
    =======
    success: 'bool'
        True if operation was successful
    radiosondy_response: 'dict'
        Dictionary with all possible response fields
        (value 'None' if not present)
    """

    sonde_id = sonde_id.upper()

    # start the communication
    d = activesoup.Driver()
    headers = {"User-Agent": "Mozilla"}

    # Init our target variables - this is the data that will be returned to the user
    launch_site = probe_type = probe_aux = probe_freq = None
    probe_status = probe_finder = landing_point = None
    landing_point_latitude = landing_point_longitude = 0.0
    landing_description = changes_made = None
    receiver = sonde_number = datetime_utc = None
    latitude = longitude = course_deg = speed_kmh = None
    altitude_m = aprs_comment = None
    climbing = temperature = pressure = humidity = aux_o3 = None
    max_speed = max_speed_height = avg_speed_kmh = max_altitude = None
    avg_ascent_speed = avg_descent_speed = None

    # general success / failure boolean
    success = False

    # We need to service up to two URLs:
    # main URL is of relevance if the probe has been archived (static content)
    # dyn URL will be used if the probeis still active (dynamic content)
    main_url = f"https://radiosondy.info/sonde.php?sondenumber={sonde_id}"
    dyn_url = f"https://radiosondy.info/dyn/get_sondeinfo.php?sondenumber={sonde_id}"

    # Get the main URL
    try:
        page = d.get(url=main_url, headers=headers)
    except:
        logger.info(f"Cannot access {main_url}")
        page = None

    if page:
        if page.last_response.status_code == 200:
            success = True
            # In case the response's URL indicates that the request got redirected to the archived data
            if "sonde_archive.php" in page.last_response.url:
                logger.info("Parsing static Radiosondy data")
                soup = BeautifulSoup(page.last_response.response.text, "html.parser")
                # Archived probe; we have proper tables and can parse them. Page has STATIC content
                # Parse Table "Status Changes"
                table = soup.find("table", attrs={"id": "Table2"})
                if table:
                    # get first row
                    rows = table.find("tr", attrs={"class": "bg_1"})
                    # There was at least one minimal change
                    if rows:
                        # There was at least one minimal change for this radiosonde
                        cols = rows.find_all("td")
                        if cols and len(cols) == 9:
                            launch_site = cols[0].string
                            probe_type = cols[1].string
                            probe_aux = cols[2].string
                            probe_freq = cols[3].string
                            probe_status = cols[4].string
                            probe_finder = cols[5].string
                            landing_point = cols[6].string
                            landing_description = cols[7].string
                            changes_made = cols[8].string

                            regex_string = r"^(-?\d*[.]\d*),\s*(-?\d*[.]\d*)$"
                            matches = re.search(
                                pattern=regex_string,
                                string=landing_point,
                                flags=re.IGNORECASE,
                            )
                            if matches:
                                try:
                                    landing_point_latitude = float(matches[1])
                                    landing_point_longitude = float(matches[2])
                                except ValueError:
                                    landing_point_longitude = (
                                        landing_point_longitude
                                    ) = 0.0
                    else:
                        # This branch gets executed in case the probe's status has never changed since its inception
                        # With the exception of the APRS data, the data that we want / need is stored as regular
                        # text. We use the text's icons in order to identify the content
                        html_response_dict = parse_radiosondy_html_content(
                            html_raw_content=page.last_response.response.text
                        )
                        sonde_number = html_response_dict["sonde_number"]
                        launch_site = html_response_dict["launch_site"]
                        probe_type = html_response_dict["probe_type"]
                        probe_freq = html_response_dict["probe_freq"]
                        probe_aux = html_response_dict["probe_aux"]
                        probe_status = html_response_dict["probe_status"]
                        max_speed = html_response_dict["max_speed"]
                        max_speed_height = html_response_dict["max_speed_height"]
                        avg_speed_kmh = html_response_dict["avg_speed_kmh"]
                        max_altitude = html_response_dict["max_altitude"]
                        avg_ascent_speed = html_response_dict["avg_ascent_speed"]
                        avg_descent_speed = html_response_dict["avg_descent_speed"]

                # pparse APRS data
                table = soup.find("table", attrs={"id": "Table1"})
                if table:
                    # get first row
                    rows = table.find("tr", attrs={"class": "bg_1"})
                    if rows:
                        cols = rows.find_all("td")
                        if cols and len(cols) == 9:
                            receiver = cols[0].string
                            sonde_number = cols[1].string
                            datetime_utc = cols[2].string
                            latitude = cols[3].string
                            longitude = cols[4].string
                            course_deg = cols[5].string
                            speed_kmh = cols[6].string
                            altitude_m = cols[7].string
                            aprs_comment = cols[8].string
            else:
                # Probe is either planned or still in process. We have DYNAMIC content and need to get this from a different URL
                logger.info("parsing dynamic URL")
                try:
                    page = d.get(url=dyn_url, headers=headers)
                except:
                    page = None
                    success = False
                if page:
                    if page.last_response.status_code == 200:

                        # With the exception of the APRS data, the data that we want / need is stored as regular
                        # text. We use the text's icons in order to identify the content
                        html_response_dict = parse_radiosondy_html_content(
                            html_raw_content=page.last_response.response.text
                        )
                        sonde_number = html_response_dict["sonde_number"]
                        launch_site = html_response_dict["launch_site"]
                        probe_type = html_response_dict["probe_type"]
                        probe_freq = html_response_dict["probe_freq"]
                        probe_aux = html_response_dict["probe_aux"]
                        probe_status = html_response_dict["probe_status"]
                        max_speed = html_response_dict["max_speed"]
                        max_speed_height = html_response_dict["max_speed_height"]
                        avg_speed_kmh = html_response_dict["avg_speed_kmh"]
                        max_altitude = html_response_dict["max_altitude"]
                        avg_ascent_speed = html_response_dict["avg_ascent_speed"]
                        avg_descent_speed = html_response_dict["avg_descent_speed"]

                        soup = BeautifulSoup(
                            page.last_response.response.text, "html.parser"
                        )
                        # Parse the APRS data
                        table = soup.find("table", attrs={"id": "Table1"})
                        if table:
                            # get first row
                            rows = table.find("tr", attrs={"class": "bg_1"})
                            if rows:
                                cols = rows.find_all("td")
                                if cols and len(cols) == 13:
                                    receiver = cols[0].string
                                    sonde_number = cols[1].string
                                    datetime_utc = cols[2].string
                                    latitude = cols[3].string
                                    longitude = cols[4].string
                                    course_deg = cols[5].string
                                    speed_kmh = cols[6].string
                                    altitude_m = cols[7].string
                                    climbing = cols[8].string
                                    temperature = cols[9].string
                                    pressure = cols[10].string
                                    humidity = cols[11].string
                                    aux_o3 = cols[12].string

                                    # Remove the additional content such as units of measure etc.
                                    # Yes, this is quick and dirty
                                    climbing = remove_trailing_content(
                                        source_string=climbing, trailing_content=" m/s"
                                    )
                                    altitude_m = remove_trailing_content(
                                        source_string=altitude_m, trailing_content=" m"
                                    )
                                    aux_o3 = remove_trailing_content(
                                        source_string=aux_o3, trailing_content=" mPa"
                                    )
                                    course_deg = remove_trailing_content(
                                        source_string=course_deg, trailing_content=" °"
                                    )
                                    humidity = remove_trailing_content(
                                        source_string=humidity, trailing_content=" %"
                                    )
                                    pressure = remove_trailing_content(
                                        source_string=pressure, trailing_content=" hPa"
                                    )
                                    speed_kmh = remove_trailing_content(
                                        source_string=speed_kmh,
                                        trailing_content=" km/h",
                                    )
                                    temperature = remove_trailing_content(
                                        source_string=temperature,
                                        trailing_content=" °C",
                                    )
                                    latitude = remove_trailing_content(
                                        source_string=latitude, trailing_content=" φ"
                                    )
                                    longitude = remove_trailing_content(
                                        source_string=longitude, trailing_content=" λ"
                                    )
                    else:
                        # We were unable to access the dynamic PHP data - return an error to the user
                        success = False

    radiosondy_response = {
        "launch_site": launch_site,
        "probe_type": probe_type,
        "probe_aux": probe_aux,
        "probe_freq": probe_freq,
        "probe_status": probe_status,
        "probe_finder": probe_finder,
        "landing_point": landing_point,
        "landing_point_latitude": landing_point_latitude,
        "landing_point_longitude": landing_point_longitude,
        "landing_description": landing_description,
        "changes_made": changes_made,
        "receiver": receiver,
        "sonde_number": sonde_number,
        "datetime_utc": datetime_utc,
        "latitude": latitude,
        "longitude": longitude,
        "course_deg": course_deg,
        "speed_kmh": speed_kmh,
        "altitude_m": altitude_m,
        "aprs_comment": aprs_comment,
        "climbing_meters_per_second": climbing,
        "temperature_celsius": temperature,
        "pressure_hpa": pressure,
        "humidity_percent": humidity,
        "aux_o3": aux_o3,
        "max_speed": max_speed,
        "max_speed_height": max_speed_height,
        "avg_speed_kmh": avg_speed_kmh,
        "max_altitude": max_altitude,
        "avg_ascent_speed": avg_ascent_speed,
        "avg_descent_speed": avg_descent_speed,
    }
    return success, radiosondy_response


if __name__ == "__main__":
    logger.info(pformat(get_radiosondy_data(sonde_id="S3130229")))