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

Routing isolines python library

Browse Source
This commit is contained in:
Mario de Frutos
2016-02-10 15:11:45 +01:00
parent 9029e38216
commit 3b0f92d5e7
3 changed files with 320 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
server/lib/python/cartodb_services/cartodb_services/here/__init__.py
View File

@@ -1 +1,2 @@
from geocoder import HereMapsGeocoder
from routing import HereMapsRoutingIsoline

105
server/lib/python/cartodb_services/cartodb_services/here/routing.py Normal file
View File

@@ -0,0 +1,105 @@
import requests
import json
class HereMapsRoutingIsoline:
'A Here Maps Routing 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'
OPTIONAL_PARAMS = [
'departure',
'arrival',
'singlecomponent',
'resolution',
'maxpoints',
'quality'
]
def __init__(self, app_id, app_code, base_url=PRODUCTION_ROUTING_BASE_URL):
self._app_id = app_id
self._app_code = app_code
self._url = "{0}{1}".format(base_url, self.ISOLINE_PATH)
def calculate_isodistance(self, source, mode, data_range, options=[]):
return self.__calculate_isolines(source, mode, data_range, 'distance',
options)
def calculate_isochrone(self, source, mode, data_range, options=[]):
return self.__calculate_isolines(source, mode, data_range, 'time',
options)
def __calculate_isolines(self, source, mode, data_range, range_type,
options=[]):
parsed_options = self.__parse_options(options)
source_param = self.__parse_source_param(source, parsed_options)
mode_param = self.__parse_mode_param(mode, parsed_options)
request_params = self.__parse_request_parameters(source_param,
mode_param,
data_range,
range_type,
parsed_options)
response = requests.get(self._url, params=request_params)
if response.status_code == requests.codes.ok:
return self.__parse_isolines_response(response.text)
else:
response.raise_for_status()
def __parse_options(self, options):
return dict(option.split('=') for option in options)
def __parse_request_parameters(self, source, mode, data_range, range_type,
options):
filtered_options = {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': ",".join(data_range)})
filtered_options.update({'rangetype': range_type})
return filtered_options
def __parse_isolines_response(self, response):
parsed_response = json.loads(response)
isolines_response = parsed_response['response']['isoline']
isolines = []
for isoline in isolines_response:
isolines.append({'range': isoline['range'],
'geom': isoline['component'][0]['shape']})
return isolines
def __parse_source_param(self, source, options):
key = 'start'
if 'is_destination' in options and options['is_destination']:
key = 'destination'
return {key: source}
def __parse_mode_param(self, mode, options):
if 'mode_type' in options:
mode_type = options['mode_type']
else:
mode_type = 'shortest'
if 'mode_traffic' in options:
mode_traffic = "traffic:{0}".format(options['mode_traffic'])
else:
mode_traffic = None
if 'mode_feature' in options and 'mode_feature_weight' in options:
mode_feature = "{0}:{1}".format(options['mode_feature'],
options['mode_feature_weight'])
else:
mode_feature = None
mode_param = "{0};{1}".format(mode_type, mode)
if mode_traffic:
mode_param = "{0};{1}".format(mode_param, mode_traffic)
if mode_feature:
mode_param = "{0};{1}".format(mode_param, mode_feature)
return {'mode': mode_param}

214
server/lib/python/cartodb_services/test/test_heremapsrouting.py Normal file
View File

@@ -0,0 +1,214 @@
#!/usr/local/bin/python
# -*- coding: utf-8 -*-
import unittest
import requests_mock
from urlparse import urlparse, parse_qs
from cartodb_services.here import HereMapsRoutingIsoline
from cartodb_services.here.exceptions import BadGeocodingParams
from cartodb_services.here.exceptions import NoGeocodingParams
from cartodb_services.here.exceptions import MalformedResult
requests_mock.Mocker.TEST_PREFIX = 'test_'
@requests_mock.Mocker()
class HereMapsRoutingIsolineTestCase(unittest.TestCase):
EMPTY_RESPONSE = """{
"response": {
"metaInfo": {
"timestamp": "2016-02-10T10:42:21Z",
"mapVersion": "8.30.61.107",
"moduleVersion": "7.2.65.0-1222",
"interfaceVersion": "2.6.20"
},
"center": {
"latitude": 33,
"longitude": 0.9999999
},
"isoline": [
{
"range": 1000,
"component": [
{
"id": 0,
"shape": []
}
]
}
],
"start": {
"linkId": "+1025046831",
"mappedPosition": {
"latitude": 32.968725,
"longitude": 0.9993629
},
"originalPosition": {
"latitude": 33,
"longitude": 0.9999999
}
}
}
}"""
ERROR_RESPONSE = """{
"_type": "ns2:RoutingServiceErrorType",
"type": "ApplicationError",
"subtype": "InitIsolineSearchFailed",
"details": "Error is NGEO_ERROR_UNKNOWN",
"additionalData": [
{
"key": "error_code",
"value": "NGEO_ERROR_UNKNOWN"
}
],
"metaInfo": {
"timestamp": "2016-02-10T10:39:35Z",
"mapVersion": "8.30.61.107",
"moduleVersion": "7.2.65.0-1222",
"interfaceVersion": "2.6.20"
}
}"""
GOOD_RESPONSE = """{
"response": {
"metaInfo": {
"timestamp": "2016-02-10T10:42:21Z",
"mapVersion": "8.30.61.107",
"moduleVersion": "7.2.65.0-1222",
"interfaceVersion": "2.6.20"
},
"center": {
"latitude": 33,
"longitude": 0.9999999
},
"isoline": [
{
"range": 1000,
"component": [
{
"id": 0,
"shape": [
"32.9699707,0.9462833",
"32.9699707,0.9458542",
"32.9699707,0.9462833"
]
}
]
}, {
"range": 2000,
"component": [
{
"id": 0,
"shape": [
"32.9699707,0.9462833",
"32.9699707,0.9750366",
"32.9699707,0.9462833"
]
}
]
}
],
"start": {
"linkId": "+1025046831",
"mappedPosition": {
"latitude": 32.968725,
"longitude": 0.9993629
},
"originalPosition": {
"latitude": 33,
"longitude": 0.9999999
}
}
}
}"""
MALFORMED_RESPONSE = """{"manolo": "escobar"}"""
def setUp(self):
self.routing = HereMapsRoutingIsoline(None, None)
self.isoline_url = "{0}{1}".format(HereMapsRoutingIsoline.PRODUCTION_ROUTING_BASE_URL,
HereMapsRoutingIsoline.ISOLINE_PATH)
def test_calculate_isodistance_with_valid_params(self, req_mock):
print self.isoline_url
url = "{0}?start=geo%2133.0%2C1.0&mode=shortest%3Bcar".format(self.isoline_url)
req_mock.register_uri('GET', url, text=self.GOOD_RESPONSE)
response = self.routing.calculate_isodistance('geo!33.0,1.0', 'car',
['1000', '2000'])
self.assertEqual(len(response), 2)
self.assertEqual(response[0]['range'], 1000)
self.assertEqual(response[1]['range'], 2000)
self.assertEqual(response[0]['geom'], [u'32.9699707,0.9462833',
u'32.9699707,0.9458542',
u'32.9699707,0.9462833'])
self.assertEqual(response[1]['geom'], [u'32.9699707,0.9462833',
u'32.9699707,0.9750366',
u'32.9699707,0.9462833'])
def test_calculate_isochrone_with_valid_params(self, req_mock):
print self.isoline_url
url = "{0}?start=geo%2133.0%2C1.0&mode=shortest%3Bcar".format(self.isoline_url)
req_mock.register_uri('GET', url, text=self.GOOD_RESPONSE)
response = self.routing.calculate_isochrone('geo!33.0,1.0', 'car',
['1000', '2000'])
self.assertEqual(len(response), 2)
self.assertEqual(response[0]['range'], 1000)
self.assertEqual(response[1]['range'], 2000)
self.assertEqual(response[0]['geom'], [u'32.9699707,0.9462833',
u'32.9699707,0.9458542',
u'32.9699707,0.9462833'])
self.assertEqual(response[1]['geom'], [u'32.9699707,0.9462833',
u'32.9699707,0.9750366',
u'32.9699707,0.9462833'])
def test_calculate_isolines_empty_response(self, req_mock):
url = "{0}?start=geo%2133.0%2C1.0&mode=shortest%3Bcar".format(
self.isoline_url)
req_mock.register_uri('GET', url, text=self.EMPTY_RESPONSE)
response = self.routing.calculate_isochrone('geo!33.0,1.0', 'car',
['1000', '2000'])
self.assertEqual(len(response), 1)
self.assertEqual(response[0]['range'], 1000)
self.assertEqual(response[0]['geom'], [])
def test_non_listed_parameters_filter_works_properly(self, req_mock):
url = "{0}?start=geo%2133.0%2C1.0&mode=shortest%3Bcar".format(
self.isoline_url)
req_mock.register_uri('GET', url, text=self.GOOD_RESPONSE)
response = self.routing.calculate_isochrone('geo!33.0,1.0', 'car',
['1000', '2000'],
['singlecomponent=true',
'resolution=3',
'maxpoints=1000',
'quality=2',
'false_option=true'])
parsed_url = urlparse(req_mock.request_history[0].url)
url_params = parse_qs(parsed_url.query)
self.assertEqual(len(url_params), 8)
self.assertEqual('false_option' in url_params, False)
def test_mode_parameters_works_properly(self, req_mock):
req_mock.register_uri('GET', requests_mock.ANY,
text=self.GOOD_RESPONSE)
response = self.routing.calculate_isochrone('geo!33.0,1.0', 'car',
['1000', '2000'],
['mode_type=fastest',
'mode_feature=motorway',
'mode_feature_weight=-1',
'mode_traffic=false'])
parsed_url = urlparse(req_mock.request_history[0].url)
url_params = parse_qs(parsed_url.query)
self.assertEqual(url_params['mode'][0],
'fastest;car;traffic:false;motorway:-1')
def test_source_parameters_works_properly(self, req_mock):
req_mock.register_uri('GET', requests_mock.ANY,
text=self.GOOD_RESPONSE)
response = self.routing.calculate_isochrone('geo!33.0,1.0', 'car',
['1000', '2000'],
['is_destination=true'])
parsed_url = urlparse(req_mock.request_history[0].url)
url_params = parse_qs(parsed_url.query)
self.assertEqual(url_params['destination'][0], 'geo!33.0,1.0')