cartodb4
/
dataservices-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add HERE API V8 Routing Isolines support (draft)

Browse Source
remotes/origin/feature/ch115970/support-new-here-api-keys
Mmoncadaisla 3 years ago
parent fb8ffe8c3c
commit 9240a7ab71
1 changed files with 194 additions and 1 deletions
Show Stats Download Patch File Download Diff File

195
server/lib/python/cartodb_services/cartodb_services/here/routing.py
Unescape View File

@ -1,5 +1,6 @@
import requests
import json
import flexpolyline as fp
from cartodb_services.here.exceptions import WrongParams
from requests.adapters import HTTPAdapter
@ -7,11 +8,12 @@ from cartodb_services.metrics import Traceable
class HereMapsRoutingIsoline(Traceable):
'A Here Maps Routing wrapper for python'
'A Here Maps Routing v7 wrapper for python'
PRODUCTION_ROUTING_BASE_URL = 'https://isoline.route.api.here.com'
STAGING_ROUTING_BASE_URL = 'https://isoline.route.cit.api.here.com'
ISOLINE_PATH = '/routing/7.2/calculateisoline.json'
API_VERSION = 7
READ_TIMEOUT = 60
CONNECT_TIMEOUT = 10
MAX_RETRIES = 1
@ -41,6 +43,9 @@ class HereMapsRoutingIsoline(Traceable):
self.read_timeout = service_params.get('read_timeout', self.READ_TIMEOUT)
self.max_retries = service_params.get('max_retries', self.MAX_RETRIES)
self._url = "{0}{1}".format(base_url, isoline_path)
def get_api_version(self):
return self.API_VERSION
def calculate_isodistance(self, source, mode, data_range, options=[]):
return self.__calculate_isolines(source, mode, data_range, 'distance',
@ -150,3 +155,191 @@ class HereMapsRoutingIsoline(Traceable):
mode_param = "{0};{1}".format(mode_param, mode_feature)
return {'mode': mode_param}
class HereMapsRoutingIsolineV8(Traceable):
'A Here Maps Routing v8 wrapper for python'
PRODUCTION_ROUTING_BASE_URL = 'https://isoline.router.hereapi.com/v8/isolines'
API_VERSION = 8
READ_TIMEOUT = 60
CONNECT_TIMEOUT = 10
MAX_RETRIES = 1
QUALITY_PARAM_V7 = 'quality'
DEFAULT_OPTIMIZEFOR = 'quality'
DEFAULT_ROUTINGMODE = 'short'
ACCEPTED_MODES = {
"walk": "pedestrian",
"car": "car"
}
OPTIONAL_PARAMS = [
'departure',
'arrival',
'maxpoints',
'quality'
]
def __init__(self, apikey, logger, service_params=None):
service_params = service_params or {}
self._apikey = apikey
self._logger = logger
self._url = service_params.get('base_url', self.PRODUCTION_ROUTING_BASE_URL)
self.connect_timeout = service_params.get('connect_timeout', self.CONNECT_TIMEOUT)
self.read_timeout = service_params.get('read_timeout', self.READ_TIMEOUT)
self.max_retries = service_params.get('max_retries', self.MAX_RETRIES)
def get_api_version(self):
return self.API_VERSION
def calculate_isodistance(self, source, mode, data_range, options=None):
options = [] if options is None else options
return self.__calculate_isolines(source, mode, data_range, 'distance',
options)
def calculate_isochrone(self, source, mode, data_range, options=None):
options = [] if options is None else options
return self.__calculate_isolines(source, mode, data_range, 'time',
options)
def __calculate_isolines(self, source, mode, data_range, range_type,
options=None):
options = [] if options is None else options
parsed_options = self.__parse_options(options)
source_param = self.__parse_source_param(source, parsed_options)
mode_params = self.__get_mode_params(mode, parsed_options)
request_params = self.__parse_request_parameters(source_param,
mode_params,
data_range,
range_type,
parsed_options)
# TODO Extract HTTP client wrapper
session = requests.Session()
session.mount(self._url, HTTPAdapter(max_retries=self.max_retries))
response = requests.get(self._url, params=request_params,
timeout=(self.connect_timeout, self.read_timeout))
self.add_response_data(response, self._logger)
if response.status_code == requests.codes.ok:
return self.__parse_isolines_response(response.text)
elif response.status_code == requests.codes.bad_request:
return []
else:
self._logger.error('Error trying to calculate HERE isolines',
data={"response_status": response.status_code,
"response_reason": response.reason,
"response_content": response.text,
"reponse_url": response.url,
"response_headers": response.headers,
"source": source, "mode": mode,
"data_range": data_range,
"range_type": range_type,
"options": options})
raise Exception('Error trying to calculate HERE isolines')
def __parse_options(self, options):
return dict(option.split('=') for option in options)
def __get_v8_param(self, param, reverse=False):
mapping = {
'range': 'range[values]',
'rangetype': 'range[type]',
'mode': {
'type': 'routingmode',
'transportmodes': 'transportmode',
'trafficmode': None,
'feature': 'avoid[feature]'
},
'quality': 'optimizefor',
'maxpoints': 'shape[maxpoints]',
'start': 'origin',
'destination': 'destination',
'departure': 'departuretime',
'arrival': 'arrivaltime'
}
if reverse is True:
mapping = {v:k for k,v in mapping.items()}
return mapping.get(param)
def __get_v8_optimizefor_value(self, quality, reverse=False):
mapping = {
1: 'quality',
2: 'balanced',
3: 'performance'
}
if reverse is True:
mapping = {v:k for k,v in mapping.items()}
return mapping.get(quality, self.DEFAULT_OPTIMIZEFOR)
def __get_v8_routingmode_value(self, mode_type, reverse=False):
mapping = {
'fastest': 'fast',
'shortest': 'short'
}
if reverse is True:
mapping = {v:k for k,v in mapping.items()}
return mapping.get(mode_type, self.DEFAULT_ROUTINGMODE)
def __parse_request_parameters(self, source, mode, data_range, range_type,
options):
quality = self.QUALITY_PARAM_V7
if quality in options.keys():
options[quality] = self.__get_v8_optimizefor_value(options[quality])
filtered_options = {self.__get_v8_param(k): v for k, v in options.iteritems()
if k.lower() in self.OPTIONAL_PARAMS}
filtered_options.update(source)
filtered_options.update(mode)
filtered_options.update({'range[values]': ",".join(map(str, data_range))})
filtered_options.update({'range[type]': range_type})
filtered_options.update({'apikey': self._apikey})
return filtered_options
def __parse_isolines_response(self, response):
parsed_response = json.loads(response)
isolines_response = parsed_response['isolines']
isolines = []
for isoline in isolines_response:
if not isoline['polygons']:
geom_value = []
else:
geom_value = fp.decode(isoline['polygons'][0]['outer'])
isolines.append({'range': isoline['range']['value'],
'geom': geom_value})
return isolines
def __parse_source_param(self, source, options):
key = 'origin'
if 'is_destination' in options and options['is_destination'].lower() == 'true':
key = 'destination'
return {key: source}
def __get_mode_params(self, mode, options):
mode_params = {}
if mode in self.ACCEPTED_MODES:
mode_source = self.ACCEPTED_MODES[mode]
mode_params.update({'transportmode': mode_source})
else:
raise WrongParams("{0} is not an accepted mode type".format(mode))
if 'mode_type' in options:
mode_type = self.__get_v8_routingmode_value(options['mode_type'])
else:
mode_type = self.DEFAULT_ROUTINGMODE
mode_params.update({'routingmode': mode_type})
if not ('mode_traffic' in options and options['mode_traffic'] == 'enabled'):
mode_params.update({'departuretime': 'any'})
return mode_params
Write Preview
Loading…
Cancel
Save