# 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 io import csv import json import logging from datetime import datetime import requests from . import BaseDomsHandler from webservice.NexusHandler import nexus_handler from webservice.algorithms.doms import config as edge_endpoints from webservice.webmodel import NexusProcessingException, NoDataException ISO_8601 = '%Y-%m-%dT%H:%M:%S%z' @nexus_handler class DomsResultsRetrievalHandler(BaseDomsHandler.BaseDomsQueryCalcHandler): name = "DOMS In Situ Subsetter" path = "/domsinsitusubset" description = "Subset a DOMS in situ source given the search domain." params = [ { "name": "source", "type": "comma-delimited string", "description": "The in situ Dataset to be sub-setted", "required": "true", "sample": "spurs" }, { "name": "parameter", "type": "string", "description": "The parameter of interest. One of 'sst', 'sss', 'wind'", "required": "false", "default": "All", "sample": "sss" }, { "name": "startTime", "type": "string", "description": "Starting time in format YYYY-MM-DDTHH:mm:ssZ or seconds since EPOCH", "required": "true", "sample": "2013-10-21T00:00:00Z" }, { "name": "endTime", "type": "string", "description": "Ending time in format YYYY-MM-DDTHH:mm:ssZ or seconds since EPOCH", "required": "true", "sample": "2013-10-31T23:59:59Z" }, { "name": "b", "type": "comma-delimited float", "description": "Minimum (Western) Longitude, Minimum (Southern) Latitude, " "Maximum (Eastern) Longitude, Maximum (Northern) Latitude", "required": "true", "sample": "-30,15,-45,30" }, { "name": "depthMin", "type": "float", "description": "Minimum depth of measurements. Must be less than depthMax", "required": "false", "default": "No limit", "sample": "0" }, { "name": "depthMax", "type": "float", "description": "Maximum depth of measurements. Must be greater than depthMin", "required": "false", "default": "No limit", "sample": "5" }, { "name": "platforms", "type": "comma-delimited integer", "description": "Platforms to include for subset consideration", "required": "false", "default": "All", "sample": "1,2,3,4,5,6,7,8,9" }, { "name": "output", "type": "string", "description": "Output type. Only 'CSV' or 'JSON' is currently supported", "required": "false", "default": "JSON", "sample": "CSV" } ] singleton = True def __init__(self, tile_service_factory): BaseDomsHandler.BaseDomsQueryCalcHandler.__init__(self, tile_service_factory) self.log = logging.getLogger(__name__) def parse_arguments(self, request): # Parse input arguments self.log.debug("Parsing arguments") source_name = request.get_argument('source', None) if source_name is None or source_name.strip() == '': raise NexusProcessingException(reason="'source' argument is required", code=400) parameter_s = request.get_argument('parameter', None) if parameter_s not in ['sst', 'sss', 'wind', None]: raise NexusProcessingException( reason="Parameter %s not supported. Must be one of 'sst', 'sss', 'wind'." % parameter_s, code=400) try: start_time = request.get_start_datetime() start_time = start_time.strftime("%Y-%m-%dT%H:%M:%SZ") except: raise NexusProcessingException( reason="'startTime' argument is required. Can be int value seconds from epoch or string format YYYY-MM-DDTHH:mm:ssZ", code=400) try: end_time = request.get_end_datetime() end_time = end_time.strftime("%Y-%m-%dT%H:%M:%SZ") except: raise NexusProcessingException( reason="'endTime' argument is required. Can be int value seconds from epoch or string format YYYY-MM-DDTHH:mm:ssZ", code=400) if start_time > end_time: raise NexusProcessingException( reason="The starting time must be before the ending time. Received startTime: %s, endTime: %s" % ( request.get_start_datetime().strftime(ISO_8601), request.get_end_datetime().strftime(ISO_8601)), code=400) try: bounding_polygon = request.get_bounding_polygon() except: raise NexusProcessingException( reason="'b' argument is required. Must be comma-delimited float formatted as Minimum (Western) Longitude, Minimum (Southern) Latitude, Maximum (Eastern) Longitude, Maximum (Northern) Latitude", code=400) depth_min = request.get_decimal_arg('depthMin', default=None) depth_max = request.get_decimal_arg('depthMax', default=None) if depth_min is not None and depth_max is not None and depth_min >= depth_max: raise NexusProcessingException( reason="Depth Min should be less than Depth Max", code=400) platforms = request.get_argument('platforms', None) if platforms is not None: try: p_validation = platforms.split(',') p_validation = [int(p) for p in p_validation] del p_validation except: raise NexusProcessingException(reason="platforms must be a comma-delimited list of integers", code=400) return source_name, parameter_s, start_time, end_time, bounding_polygon, depth_min, depth_max, platforms def calc(self, request, **args): source_name, parameter_s, start_time, end_time, bounding_polygon, \ depth_min, depth_max, platforms = self.parse_arguments(request) with requests.session() as edge_session: edge_results = query_edge(source_name, parameter_s, start_time, end_time, ','.join([str(bound) for bound in bounding_polygon.bounds]), platforms, depth_min, depth_max, edge_session)['results'] if len(edge_results) == 0: raise NoDataException return InSituSubsetResult(results=edge_results) class InSituSubsetResult(object): def __init__(self, results): self.results = results def toJson(self): return json.dumps(self.results, indent=4) def toCSV(self): fieldnames = sorted(next(iter(self.results)).keys()) csv_mem_file = io.StringIO() try: writer = csv.DictWriter(csv_mem_file, fieldnames=fieldnames) writer.writeheader() writer.writerows(self.results) csv_out = csv_mem_file.getvalue() finally: csv_mem_file.close() return csv_out def query_edge(dataset, variable, startTime, endTime, bbox, platform, depth_min, depth_max, session, itemsPerPage=1000, startIndex=0, stats=True): log = logging.getLogger('webservice.algorithms.doms.insitusubset.query_edge') try: startTime = datetime.utcfromtimestamp(startTime).strftime('%Y-%m-%dT%H:%M:%SZ') except TypeError: # Assume we were passed a properly formatted string pass try: endTime = datetime.utcfromtimestamp(endTime).strftime('%Y-%m-%dT%H:%M:%SZ') except TypeError: # Assume we were passed a properly formatted string pass try: platform = platform.split(',') except AttributeError: # Assume we were passed a list pass params = {"startTime": startTime, "endTime": endTime, "bbox": bbox, "minDepth": depth_min, "maxDepth": depth_max, "itemsPerPage": itemsPerPage, "startIndex": startIndex, "stats": str(stats).lower()} if variable: params['variable'] = variable if platform: params['platform'] = platform edge_request = session.get(edge_endpoints.getEndpointByName(dataset)['url'], params=params) edge_request.raise_for_status() edge_response = json.loads(edge_request.text) # Get all edge results next_page_url = edge_response.get('next', None) while next_page_url is not None: log.debug("requesting %s" % next_page_url) edge_page_request = session.get(next_page_url) edge_page_request.raise_for_status() edge_page_response = json.loads(edge_page_request.text) edge_response['results'].extend(edge_page_response['results']) next_page_url = edge_page_response.get('next', None) return edge_response