| # 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 |
