analysis/webservice/algorithms/doms/fetchedgeimpl.py

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import traceback
from datetime import datetime
from multiprocessing.pool import ThreadPool

import requests

from . import geo
from . import values
from webservice.webmodel import NexusProcessingException


def __parseDatetime(dtString):
    dt = datetime.strptime(dtString, "%Y-%m-%dT%H:%M:%SZ")
    epoch = datetime.utcfromtimestamp(0)
    time = (dt - epoch).total_seconds() * 1000.0
    return time


def __parseLocation(locString):
    if "Point" in locString:
        locString = locString[6:-1]

    if "," in locString:
        latitude = float(locString.split(",")[0])
        longitude = float(locString.split(",")[1])
    else:
        latitude = float(locString.split(" ")[1])
        longitude = float(locString.split(" ")[0])

    return (latitude, longitude)


def __resultRawToUsable(resultdict):
    resultdict["time"] = __parseDatetime(resultdict["time"])
    latitude, longitude = __parseLocation(resultdict["point"])

    resultdict["x"] = longitude
    resultdict["y"] = latitude

    if "id" not in resultdict and "metadata" in resultdict:
        resultdict["id"] = resultdict["metadata"]

    resultdict["id"] = "id-%s" % resultdict["id"]

    if "device" in resultdict:
        resultdict["device"] = values.getDeviceById(resultdict["device"])

    if "platform" in resultdict:
        resultdict["platform"] = values.getPlatformById(resultdict["platform"])

    if "mission" in resultdict:
        resultdict["mission"] = values.getMissionById(resultdict["mission"])

    if "sea_surface_temperature" in resultdict:
        resultdict["sea_water_temperature"] = resultdict["sea_surface_temperature"]
        del resultdict["sea_surface_temperature"]

    return resultdict


def __fetchJson(url, params, trycount=1, maxtries=5):
    if trycount > maxtries:
        raise Exception("Maximum retries attempted.")
    if trycount > 1:
        print("Retry #", trycount)
    r = requests.get(url, params=params, timeout=500.000)

    print(r.url)

    if r.status_code != 200:
        return __fetchJson(url, params, trycount + 1, maxtries)
    try:
        results = json.loads(r.text)
        return results
    except:
        return __fetchJson(url, params, trycount + 1, maxtries)


def __doQuery(endpoint, startTime, endTime, bbox, depth_min=None, depth_max=None, itemsPerPage=10, startIndex=0,
              platforms=None,
              pageCallback=None):
    params = {"startTime": startTime, "endTime": endTime, "bbox": bbox, "itemsPerPage": itemsPerPage,
              "startIndex": startIndex, "stats": "true"}

    if depth_min is not None:
        params['minDepth'] = depth_min
    if depth_max is not None:
        params['maxDepth'] = depth_max

    if platforms is not None:
        params["platform"] = platforms.split(",")

    resultsRaw = __fetchJson(endpoint["url"], params)
    boundsConstrainer = geo.BoundsConstrainer(north=-90, south=90, west=180, east=-180)

    if resultsRaw["totalResults"] == 0 or len(resultsRaw["results"]) == 0:  # Double-sanity check
        return [], resultsRaw["totalResults"], startIndex, itemsPerPage, boundsConstrainer

    try:
        results = []
        for resultdict in resultsRaw["results"]:
            result = __resultRawToUsable(resultdict)
            result["source"] = endpoint["name"]
            boundsConstrainer.testCoords(north=result["y"], south=result["y"], west=result["x"], east=result["x"])
            results.append(result)

        if "stats_fields" in resultsRaw and len(resultsRaw["results"]) == 0:
            stats = resultsRaw["stats_fields"]
            if "lat" in stats and "lon" in stats:
                boundsConstrainer.testCoords(north=stats['lat']['max'], south=stats['lat']['min'],
                                             west=stats['lon']['min'], east=stats['lon']['max'])

        if pageCallback is not None:
            pageCallback(results)

        '''
            If pageCallback was supplied, we assume this call to be asynchronous. Otherwise combine all the results data and return it.
        '''
        if pageCallback is None:
            return results, int(resultsRaw["totalResults"]), int(resultsRaw["startIndex"]), int(
                resultsRaw["itemsPerPage"]), boundsConstrainer
        else:
            return [], int(resultsRaw["totalResults"]), int(resultsRaw["startIndex"]), int(
                resultsRaw["itemsPerPage"]), boundsConstrainer
    except:
        print("Invalid or missing JSON in response.")
        traceback.print_exc()
        raise NexusProcessingException(reason="Invalid or missing JSON in response.")
        # return [], 0, startIndex, itemsPerPage, boundsConstrainer


def getCount(endpoint, startTime, endTime, bbox, depth_min, depth_max, platforms=None):
    startIndex = 0
    pageResults, totalResults, pageStartIndex, itemsPerPageR, boundsConstrainer = __doQuery(endpoint, startTime,
                                                                                            endTime, bbox,
                                                                                            depth_min, depth_max, 0,
                                                                                            startIndex, platforms)
    return totalResults, boundsConstrainer


def fetch(endpoint, startTime, endTime, bbox, depth_min, depth_max, platforms=None, pageCallback=None):
    results = []
    startIndex = 0

    mainBoundsConstrainer = geo.BoundsConstrainer(north=-90, south=90, west=180, east=-180)

    # First isn't parellel so we can get the ttl results, forced items per page, etc...
    pageResults, totalResults, pageStartIndex, itemsPerPageR, boundsConstrainer = __doQuery(endpoint, startTime,
                                                                                            endTime, bbox,
                                                                                            depth_min, depth_max,
                                                                                            endpoint["itemsPerPage"],
                                                                                            startIndex, platforms,
                                                                                            pageCallback)
    results = results + pageResults
    mainBoundsConstrainer.testOtherConstrainer(boundsConstrainer)

    pool = ThreadPool(processes=endpoint["fetchThreads"])
    mpResults = [pool.apply_async(__doQuery, args=(
        endpoint, startTime, endTime, bbox, depth_min, depth_max, itemsPerPageR, x, platforms, pageCallback)) for x in
                 range(len(pageResults), totalResults, itemsPerPageR)]
    pool.close()
    pool.join()

    '''
        If pageCallback was supplied, we assume this call to be asynchronous. Otherwise combine all the results data and return it.
    '''
    if pageCallback is None:
        mpResults = [p.get() for p in mpResults]
        for mpResult in mpResults:
            results = results + mpResult[0]
            mainBoundsConstrainer.testOtherConstrainer(mpResult[4])

        return results, mainBoundsConstrainer


def getValues(endpoint, startTime, endTime, bbox, depth_min, depth_max, platforms=None, placeholders=False):
    results, boundsConstrainer = fetch(endpoint, startTime, endTime, bbox, depth_min, depth_max, platforms)

    if placeholders:
        trimmedResults = []
        for item in results:
            depth = None
            if "depth" in item:
                depth = item["depth"]
            if "sea_water_temperature_depth" in item:
                depth = item["sea_water_temperature_depth"]

            trimmedItem = {
                "x": item["x"],
                "y": item["y"],
                "source": item["source"],
                "time": item["time"],
                "device": item["device"] if "device" in item else None,
                "platform": item["platform"],
                "depth": depth
            }
            trimmedResults.append(trimmedItem)

        results = trimmedResults

    return results, boundsConstrainer