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

Added geocoding and routing services

Browse Source
This commit is contained in:
Antonio
2018-02-16 17:43:22 +01:00
parent c14fb057d3
commit 90d8d207eb
6 changed files with 348 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/lib/python/cartodb_services/cartodb_services/tomtom/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from geocoder import TomTomGeocoder
from routing import TomTomRouting, TomTomRoutingResponse

87
server/lib/python/cartodb_services/cartodb_services/tomtom/geocoder.py Normal file
View File

@@ -0,0 +1,87 @@
import json
import requests
from uritemplate import URITemplate
from cartodb_services.metrics import Traceable
from cartodb_services.tools.exceptions import ServiceException
from cartodb_services.tools.qps import qps_retry
from cartodb_services.tools.normalize import normalize
BASEURI = ('https://api.tomtom.com/search/2/geocode/'
'{searchtext}.JSON'
'?key={apiKey}'
'&limit=1')
ENTRY_RESULTS = 'results'
ENTRY_POSITION = 'position'
ENTRY_LON = 'lon'
ENTRY_LAT = 'lat'
class TomTomGeocoder(Traceable):
'''
Python wrapper for the TomTom Geocoder service.
'''
def __init__(self, apikey, logger, service_params=None):
service_params = service_params or {}
self._apikey = apikey
self._logger = logger
def _uri(self, searchtext, countries=None):
baseuri = BASEURI + '&countrySet={}'.format(countries) \
if countries else BASEURI
uri = URITemplate(baseuri).expand(apiKey=self._apikey,
searchtext=searchtext.encode('utf-8'))
return uri
def _parse_geocoder_response(self, response):
json_response = json.loads(response)
if json_response and json_response[ENTRY_RESULTS]:
result = json_response[ENTRY_RESULTS][0]
return self._extract_lng_lat_from_feature(result)
else:
return []
def _extract_lng_lat_from_feature(self, result):
position = result[ENTRY_POSITION]
longitude = position[ENTRY_LON]
latitude = position[ENTRY_LAT]
return [longitude, latitude]
@qps_retry(qps=5)
def geocode(self, searchtext, city=None, state_province=None,
country=None):
if searchtext and searchtext.strip():
address = [normalize(searchtext)]
if city:
address.append(normalize(city))
if state_province:
address.append(normalize(state_province))
else:
return []
uri = self._uri(searchtext=', '.join(address), countries=country)
try:
response = requests.get(uri)
if response.status_code == requests.codes.ok:
return self._parse_geocoder_response(response.text)
elif response.status_code == requests.codes.bad_request:
return []
elif response.status_code == requests.codes.unprocessable_entity:
return []
else:
raise ServiceException(response.status_code, response)
except requests.Timeout as te:
# In case of timeout we want to stop the job because the server
# could be down
self._logger.error('Timeout connecting to TomTom geocoding server',
te)
raise ServiceException('Error geocoding {0} using TomTom'.format(
searchtext), None)
except requests.ConnectionError as ce:
# Don't raise the exception to continue with the geocoding job
self._logger.error('Error connecting to TomTom geocoding server',
exception=ce)
return []

148
server/lib/python/cartodb_services/cartodb_services/tomtom/routing.py Normal file
View File

@@ -0,0 +1,148 @@
'''
Python client for the TomTom Routing service.
'''
import json
import requests
from uritemplate import URITemplate
from cartodb_services.metrics import Traceable
from cartodb_services.tools import PolyLine
from cartodb_services.tools.coordinates import (validate_coordinates,
marshall_coordinates)
from cartodb_services.tools.exceptions import ServiceException
from cartodb_services.tools.qps import qps_retry
BASEURI = ('https://api.tomtom.com/routing/1/calculateRoute/'
'{coordinates}'
'/json'
'?key={apikey}'
'&travelMode={travelmode}'
'&computeBestOrder=true')
NUM_WAYPOINTS_MIN = 2
NUM_WAYPOINTS_MAX = 20
PROFILE_DRIVING = 'car'
PROFILE_CYCLING = 'bicycle'
PROFILE_WALKING = 'pedestrian'
DEFAULT_PROFILE = PROFILE_DRIVING
VALID_PROFILES = [PROFILE_DRIVING,
PROFILE_CYCLING,
PROFILE_WALKING]
ENTRY_ROUTES = 'routes'
ENTRY_SUMMARY = 'summary'
ENTRY_LENGTH = 'lengthInMeters'
ENTRY_TIME = 'travelTimeInSeconds'
ENTRY_LEGS = 'legs'
ENTRY_POINTS = 'points'
ENTRY_LATITUDE = 'latitude'
ENTRY_LONGITUDE = 'longitude'
class TomTomRouting(Traceable):
'''
Python wrapper for the TomTom Routing service.
'''
def __init__(self, apikey, logger, service_params=None):
service_params = service_params or {}
self._apikey = apikey
self._logger = logger
def _uri(self, coordinates, profile=DEFAULT_PROFILE):
uri = URITemplate(BASEURI).expand(apikey=self._apikey,
coordinates=coordinates,
travelmode=profile)
return uri
def _validate_profile(self, profile):
if profile not in VALID_PROFILES:
raise ValueError('{profile} is not a valid profile. '
'Valid profiles are: {valid_profiles}'.format(
profile=profile,
valid_profiles=', '.join(
[x for x in VALID_PROFILES])))
def _marshall_coordinates(self, coordinates):
return ':'.join(['{lat},{lon}'.format(lat=coordinate.latitude,
lon=coordinate.longitude)
for coordinate in coordinates])
def _parse_routing_response(self, response):
json_response = json.loads(response)
if json_response:
route = json_response[ENTRY_ROUTES][0] # Force the first route
geometry = self._parse_legs(route[ENTRY_LEGS])
summary = route[ENTRY_SUMMARY]
distance = summary[ENTRY_LENGTH]
duration = summary[ENTRY_TIME]
return TomTomRoutingResponse(geometry, distance, duration)
else:
return TomTomRoutingResponse(None, None, None)
def _parse_legs(self, legs):
geometry = []
for leg in legs:
points = leg[ENTRY_POINTS]
for point in points:
geometry.append((point[ENTRY_LATITUDE],
point[ENTRY_LONGITUDE]))
return geometry
@qps_retry(qps=5)
def directions(self, waypoints, profile=DEFAULT_PROFILE):
self._validate_profile(profile)
validate_coordinates(waypoints, NUM_WAYPOINTS_MIN, NUM_WAYPOINTS_MAX)
coordinates = self._marshall_coordinates(waypoints)
uri = self._uri(coordinates, profile)
try:
response = requests.get(uri)
if response.status_code == requests.codes.ok:
return self._parse_routing_response(response.text)
elif response.status_code == requests.codes.bad_request:
return TomTomRoutingResponse(None, None, None)
elif response.status_code == requests.codes.unprocessable_entity:
return TomTomRoutingResponse(None, None, None)
else:
raise ServiceException(response.status_code, response)
except requests.Timeout as te:
# In case of timeout we want to stop the job because the server
# could be down
self._logger.error('Timeout connecting to TomTom routing service',
te)
raise ServiceException('Error getting routing data from TomTom',
None)
except requests.ConnectionError as ce:
# Don't raise the exception to continue with the geocoding job
self._logger.error('Error connecting to TomTom routing service',
exception=ce)
return TomTomRoutingResponse(None, None, None)
class TomTomRoutingResponse:
def __init__(self, shape, length, duration):
self._shape = shape
self._length = length
self._duration = duration
@property
def shape(self):
return self._shape
@property
def length(self):
return self._length
@property
def duration(self):
return self._duration

3
server/lib/python/cartodb_services/test/credentials.py
View File

@@ -4,3 +4,6 @@ def mapbox_api_key():
"""Returns Mapbox API key. Requires setting MAPBOX_API_KEY environment variable."""
return os.environ['MAPBOX_API_KEY']
def tomtom_api_key():
"""Returns TomTom API key. Requires setting TOMTOM_API_KEY environment variable."""
return os.environ['TOMTOM_API_KEY']

56
server/lib/python/cartodb_services/test/test_tomtomgeocoder.py Normal file
View File

@@ -0,0 +1,56 @@
import unittest
from mock import Mock
from cartodb_services.tomtom import TomTomGeocoder
from cartodb_services.tools.exceptions import ServiceException
from credentials import tomtom_api_key
INVALID_APIKEY = 'invalid_apikey'
VALID_ADDRESS = 'Plaza Mayor 3, Valladolid'
WELL_KNOWN_LONGITUDE = -4.728
WELL_KNOWN_LATITUDE = 41.653
class TomTomGeocoderTestCase(unittest.TestCase):
def setUp(self):
self.geocoder = TomTomGeocoder(apikey=tomtom_api_key(), logger=Mock())
def test_invalid_token(self):
invalid_geocoder = TomTomGeocoder(apikey=INVALID_APIKEY, logger=Mock())
with self.assertRaises(ServiceException):
invalid_geocoder.geocode(VALID_ADDRESS)
def test_valid_request(self):
place = self.geocoder.geocode(VALID_ADDRESS)
self.assertEqual('%.3f' % place[0], '%.3f' % WELL_KNOWN_LONGITUDE)
self.assertEqual('%.3f' % place[1], '%.3f' % WELL_KNOWN_LATITUDE)
def test_valid_request_namedplace(self):
place = self.geocoder.geocode(searchtext='Barcelona')
assert place
def test_valid_request_namedplace2(self):
place = self.geocoder.geocode(searchtext='New York', country='us')
assert place
def test_odd_characters(self):
place = self.geocoder.geocode(searchtext='Barcelona; "Spain"')
assert place
def test_empty_request(self):
place = self.geocoder.geocode(searchtext='', country=None, city=None, state_province=None)
assert place == []
def test_empty_search_text_request(self):
place = self.geocoder.geocode(searchtext=' ', country='us', city=None, state_province="")
assert place == []
def test_unknown_place_request(self):
place = self.geocoder.geocode(searchtext='[unknown]', country='ch', state_province=None, city=None)
assert place == []

52
server/lib/python/cartodb_services/test/test_tomtomrouting.py Normal file
View File

@@ -0,0 +1,52 @@
import unittest
from mock import Mock
from cartodb_services.tomtom import TomTomRouting
from cartodb_services.tomtom.routing import DEFAULT_PROFILE
from cartodb_services.tools.exceptions import ServiceException
from cartodb_services.tools import Coordinate
from credentials import tomtom_api_key
INVALID_APIKEY = 'invalid_apikey'
VALID_WAYPOINTS = [Coordinate(13.42936, 52.50931),
Coordinate(13.43872, 52.50274)]
NUM_WAYPOINTS_MAX = 20
INVALID_WAYPOINTS_EMPTY = []
INVALID_WAYPOINTS_MIN = [Coordinate(13.42936, 52.50931)]
INVALID_WAYPOINTS_MAX = [Coordinate(13.42936, 52.50931)
for x in range(0, NUM_WAYPOINTS_MAX + 2)]
VALID_PROFILE = DEFAULT_PROFILE
INVALID_PROFILE = 'invalid_profile'
class TomTomRoutingTestCase(unittest.TestCase):
def setUp(self):
self.routing = TomTomRouting(apikey=tomtom_api_key(), logger=Mock())
def test_invalid_profile(self):
with self.assertRaises(ValueError):
self.routing.directions(VALID_WAYPOINTS, INVALID_PROFILE)
def test_invalid_waypoints_empty(self):
with self.assertRaises(ValueError):
self.routing.directions(INVALID_WAYPOINTS_EMPTY, VALID_PROFILE)
def test_invalid_waypoints_min(self):
with self.assertRaises(ValueError):
self.routing.directions(INVALID_WAYPOINTS_MIN, VALID_PROFILE)
def test_invalid_waypoints_max(self):
with self.assertRaises(ValueError):
self.routing.directions(INVALID_WAYPOINTS_MAX, VALID_PROFILE)
def test_invalid_token(self):
invalid_routing = TomTomRouting(apikey=INVALID_APIKEY, logger=Mock())
with self.assertRaises(ServiceException):
invalid_routing.directions(VALID_WAYPOINTS,
VALID_PROFILE)
def test_valid_request(self):
route = self.routing.directions(VALID_WAYPOINTS, VALID_PROFILE)
assert route.shape
assert route.length
assert route.duration