From 31afc82b566c2c25e5677428485425521a419c70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juan=20Ignacio=20S=C3=A1nchez=20Lara?= Date: Thu, 5 Jul 2018 17:12:22 +0200 Subject: [PATCH] TomTom bulk geocoding by bypassing to serial --- server/extension/Makefile | 3 ++ .../cdb_dataservices_server--0.32.0.sql | 24 +++++++++++-- .../extension/sql/21_bulk_geocode_street.sql | 24 +++++++++++-- .../cartodb_services/tomtom/__init__.py | 1 + .../cartodb_services/tomtom/bulk_geocoder.py | 36 +++++++++++++++++++ test/integration/test_street_functions.py | 15 ++++++-- 6 files changed, 97 insertions(+), 6 deletions(-) create mode 100644 server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py diff --git a/server/extension/Makefile b/server/extension/Makefile index 8f53576..968e51c 100644 --- a/server/extension/Makefile +++ b/server/extension/Makefile @@ -83,3 +83,6 @@ deploy: release_remove_parallel_deploy $(INSTALL_DATA) old_versions/*.sql *.sql '$(DESTDIR)$(datadir)/extension/' install: deploy + +reinstall: install + psql -U postgres -d dataservices_db -c "drop extension if exists cdb_dataservices_server; create extension cdb_dataservices_server;" diff --git a/server/extension/cdb_dataservices_server--0.32.0.sql b/server/extension/cdb_dataservices_server--0.32.0.sql index 849c688..4f66a15 100644 --- a/server/extension/cdb_dataservices_server--0.32.0.sql +++ b/server/extension/cdb_dataservices_server--0.32.0.sql @@ -2367,12 +2367,15 @@ RETURNS SETOF cdb_dataservices_server.geocoding AS $$ with metrics('cdb_bulk_geocode_street_point', user_geocoder_config, logger, params): if user_geocoder_config.google_geocoder: - plan = plpy.prepare("SELECT * FROM cdb_dataservices_server._cdb_bulk_google_geocode_street_point($1, $2, $3); ", ["text", "text", "jsonb"]) + provider_function = "_cdb_bulk_google_geocode_street_point"; elif user_geocoder_config.heremaps_geocoder: - plan = plpy.prepare("SELECT * FROM cdb_dataservices_server._cdb_bulk_heremaps_geocode_street_point($1, $2, $3); ", ["text", "text", "jsonb"]) + provider_function = "_cdb_bulk_heremaps_geocode_street_point"; + elif user_geocoder_config.tomtom_geocoder: + provider_function = "_cdb_bulk_tomtom_geocode_street_point"; else: raise Exception('Requested geocoder is not available') + plan = plpy.prepare("SELECT * FROM cdb_dataservices_server.{}($1, $2, $3); ".format(provider_function), ["text", "text", "jsonb"]) result = plpy.execute(plan, [username, orgname, searches]) return result @@ -2400,6 +2403,23 @@ RETURNS SETOF cdb_dataservices_server.geocoding AS $$ return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches) $$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED; +CREATE OR REPLACE FUNCTION cdb_dataservices_server._cdb_bulk_tomtom_geocode_street_point(username TEXT, orgname TEXT, searches jsonb) +RETURNS SETOF cdb_dataservices_server.geocoding AS $$ + from cartodb_services import run_street_point_geocoder + from cartodb_services.tools import ServiceManager + from cartodb_services.refactor.service.tomtom_geocoder_config import TomTomGeocoderConfigBuilder + from cartodb_services.tomtom import TomTomBulkGeocoder + from cartodb_services.tools import Logger + import cartodb_services + cartodb_services.init(plpy, GD) + + logger_config = GD["logger_config"] + logger = Logger(logger_config) + service_manager = ServiceManager('geocoder', TomTomGeocoderConfigBuilder, username, orgname, GD) + geocoder = TomTomBulkGeocoder(service_manager.config.tomtom_api_key, service_manager.logger, service_manager.config.service_params) + return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches) +$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED; + CREATE OR REPLACE FUNCTION cdb_dataservices_server.cdb_geocode_admin0_polygon(username text, orgname text, country_name text) RETURNS Geometry AS $$ from cartodb_services.metrics import QuotaService diff --git a/server/extension/sql/21_bulk_geocode_street.sql b/server/extension/sql/21_bulk_geocode_street.sql index 6732340..50d5008 100644 --- a/server/extension/sql/21_bulk_geocode_street.sql +++ b/server/extension/sql/21_bulk_geocode_street.sql @@ -24,12 +24,15 @@ RETURNS SETOF cdb_dataservices_server.geocoding AS $$ with metrics('cdb_bulk_geocode_street_point', user_geocoder_config, logger, params): if user_geocoder_config.google_geocoder: - plan = plpy.prepare("SELECT * FROM cdb_dataservices_server._cdb_bulk_google_geocode_street_point($1, $2, $3); ", ["text", "text", "jsonb"]) + provider_function = "_cdb_bulk_google_geocode_street_point"; elif user_geocoder_config.heremaps_geocoder: - plan = plpy.prepare("SELECT * FROM cdb_dataservices_server._cdb_bulk_heremaps_geocode_street_point($1, $2, $3); ", ["text", "text", "jsonb"]) + provider_function = "_cdb_bulk_heremaps_geocode_street_point"; + elif user_geocoder_config.tomtom_geocoder: + provider_function = "_cdb_bulk_tomtom_geocode_street_point"; else: raise Exception('Requested geocoder is not available') + plan = plpy.prepare("SELECT * FROM cdb_dataservices_server.{}($1, $2, $3); ".format(provider_function), ["text", "text", "jsonb"]) result = plpy.execute(plan, [username, orgname, searches]) return result @@ -57,3 +60,20 @@ RETURNS SETOF cdb_dataservices_server.geocoding AS $$ return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches) $$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED; +CREATE OR REPLACE FUNCTION cdb_dataservices_server._cdb_bulk_tomtom_geocode_street_point(username TEXT, orgname TEXT, searches jsonb) +RETURNS SETOF cdb_dataservices_server.geocoding AS $$ + from cartodb_services import run_street_point_geocoder + from cartodb_services.tools import ServiceManager + from cartodb_services.refactor.service.tomtom_geocoder_config import TomTomGeocoderConfigBuilder + from cartodb_services.tomtom import TomTomBulkGeocoder + from cartodb_services.tools import Logger + import cartodb_services + cartodb_services.init(plpy, GD) + + logger_config = GD["logger_config"] + logger = Logger(logger_config) + service_manager = ServiceManager('geocoder', TomTomGeocoderConfigBuilder, username, orgname, GD) + geocoder = TomTomBulkGeocoder(service_manager.config.tomtom_api_key, service_manager.logger, service_manager.config.service_params) + return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches) +$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED; + diff --git a/server/lib/python/cartodb_services/cartodb_services/tomtom/__init__.py b/server/lib/python/cartodb_services/cartodb_services/tomtom/__init__.py index 8c5df3c..21d5ced 100644 --- a/server/lib/python/cartodb_services/cartodb_services/tomtom/__init__.py +++ b/server/lib/python/cartodb_services/cartodb_services/tomtom/__init__.py @@ -1,3 +1,4 @@ from geocoder import TomTomGeocoder +from bulk_geocoder import TomTomBulkGeocoder from routing import TomTomRouting, TomTomRoutingResponse from isolines import TomTomIsolines, TomTomIsochronesResponse diff --git a/server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py b/server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py new file mode 100644 index 0000000..34d0974 --- /dev/null +++ b/server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py @@ -0,0 +1,36 @@ +from cartodb_services import StreetPointBulkGeocoder +from cartodb_services.tomtom import TomTomGeocoder + + +class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder): + # TODO: ? + MAX_BATCH_SIZE = 1000000 # From the docs + # TODO: ? + MIN_BATCHED_SEARCH = 100 # Under this, serial will be used + + def _bulk_geocode(self, searches): + if len(searches) > self.MAX_BATCH_SIZE: + raise Exception("Batch size can't be larger than {}".format(self.MAX_BATCH_SIZE)) + if self._should_use_batch(searches): + self._logger.debug('--> Batch geocode') + return self._batch_geocode(searches) + else: + self._logger.debug('--> Serial geocode') + return self._serial_geocode(searches) + + def _should_use_batch(self, searches): + return len(searches) >= self.MIN_BATCHED_SEARCH + + def _serial_geocode(self, searches): + results = [] + for search in searches: + (search_id, address, city, state, country) = search + self._logger.debug('--> Sending serial search: {}'.format(search)) + coordinates = self.geocode(searchtext=address.encode('utf-8'), + city=city.encode('utf-8'), + state_province=state.encode('utf-8'), + country=country.encode('utf-8')) + self._logger.debug('--> result sent') + results.append((search_id, coordinates, [])) + return results + diff --git a/test/integration/test_street_functions.py b/test/integration/test_street_functions.py index f4a83b2..2060385 100644 --- a/test/integration/test_street_functions.py +++ b/test/integration/test_street_functions.py @@ -71,9 +71,20 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp): 'Logroño, Argentina': [-61.69604, -29.50425] } + TOMTOM_POINTS = HERE_POINTS.copy() + TOMTOM_POINTS.update({ + 'Plaza Mayor, Valladolid': [-4.72183, 41.5826], + 'Paseo Zorrilla, Valladolid': [-4.74031, 41.63181], + 'Valladolid': [-4.72838, 41.6542], + 'Valladolid, Spain': [-4.72838, 41.6542], + 'Madrid': [-3.70035, 40.42028], + 'Logroño, Spain': [-2.44998, 42.46592], + }) + FIXTURE_POINTS = { - 'google': GOOGLE_POINTS, - 'heremaps': HERE_POINTS + 'google': GOOGLE_POINTS, + 'heremaps': HERE_POINTS, + 'tomtom': TOMTOM_POINTS } def setUp(self):