Merge pull request #506 from CartoDB/geocoder_boost_refactor

Geocoder boost refactor
This commit is contained in:
Juan Ignacio Sánchez Lara
2018-07-17 17:59:30 +02:00
committed by GitHub
21 changed files with 654 additions and 306 deletions

View File

@@ -18,7 +18,12 @@ BEGIN
RETURN ROWS;
END
$func$ LANGUAGE plpgsql;--
$func$ LANGUAGE plpgsql;
-- Taken from https://stackoverflow.com/a/48013356/351721
CREATE OR REPLACE FUNCTION jsonb_array_casttext(jsonb) RETURNS text[] AS $f$
SELECT array_agg(x) || ARRAY[]::text[] FROM jsonb_array_elements_text($1) t(x);
$f$ LANGUAGE sql IMMUTABLE;--
-- Geocoder server connection config
--
-- The purpose of this function is provide to the PL/Proxy functions
@@ -112,7 +117,8 @@ CREATE TYPE cdb_dataservices_client.service_quota_info AS (
monthly_quota NUMERIC,
used_quota NUMERIC,
soft_limit BOOLEAN,
provider TEXT
provider TEXT,
max_batch_size NUMERIC
);
--
-- Public dataservices API function
@@ -1987,42 +1993,48 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_client._DST_DisconnectUserTable(
TARGET cdb_dataservices_server._DST_DisconnectUserTable;
$$ LANGUAGE plproxy VOLATILE PARALLEL UNSAFE;
CREATE OR REPLACE FUNCTION cdb_dataservices_client.cdb_bulk_geocode_street_point (query text,
street_column text, city_column text default null, state_column text default null, country_column text default null, batch_size integer DEFAULT 100)
street_column text, city_column text default null, state_column text default null, country_column text default null, batch_size integer DEFAULT NULL)
RETURNS SETOF cdb_dataservices_client.geocoding AS $$
DECLARE
query_row_count integer;
enough_quota boolean;
remaining_quota integer;
max_batch_size integer;
cartodb_id_batch integer;
batches_n integer;
DEFAULT_BATCH_SIZE CONSTANT numeric := 100;
MAX_BATCH_SIZE CONSTANT numeric := 10000;
MAX_SAFE_BATCH_SIZE CONSTANT numeric := 5000;
current_row_count integer ;
temp_table_name text;
BEGIN
SELECT csqi.monthly_quota - csqi.used_quota AS remaining_quota, csqi.max_batch_size
INTO remaining_quota, max_batch_size
FROM cdb_dataservices_client.cdb_service_quota_info() csqi
WHERE service = 'hires_geocoder';
RAISE DEBUG 'remaining_quota: %; max_batch_size: %', remaining_quota, max_batch_size;
IF batch_size IS NULL THEN
RAISE EXCEPTION 'batch_size can''t be null';
ELSIF batch_size > MAX_BATCH_SIZE THEN
RAISE EXCEPTION 'batch_size must be lower than %', MAX_BATCH_SIZE + 1;
batch_size := max_batch_size;
ELSIF batch_size > max_batch_size THEN
RAISE EXCEPTION 'batch_size must be lower than %', max_batch_size + 1;
END IF;
EXECUTE format('SELECT COUNT(1) from (%s) _x', query) INTO query_row_count;
IF batch_size > MAX_SAFE_BATCH_SIZE THEN
batch_size := MAX_SAFE_BATCH_SIZE;
END IF;
EXECUTE format('SELECT count(1), ceil(count(1)::float/%s) FROM (%s) _x', batch_size, query)
INTO query_row_count, batches_n;
RAISE DEBUG 'cdb_bulk_geocode_street_point --> query_row_count: %; query: %; country: %; state: %; city: %; street: %',
query_row_count, query, country_column, state_column, city_column, street_column;
SELECT cdb_dataservices_client.cdb_enough_quota('hires_geocoder', query_row_count) INTO enough_quota;
IF enough_quota IS NOT NULL AND NOT enough_quota THEN
SELECT csqi.monthly_quota - csqi.used_quota AS remaining_quota
INTO remaining_quota
FROM cdb_dataservices_client.cdb_service_quota_info() csqi
WHERE service = 'hires_geocoder';
IF remaining_quota < query_row_count THEN
RAISE EXCEPTION 'Remaining quota: %. Estimated cost: %', remaining_quota, query_row_count;
END IF;
EXECUTE format('SELECT ceil(max(cartodb_id)::float/%s) FROM (%s) _x', batch_size, query) INTO batches_n;
RAISE DEBUG 'batches_n: %', batches_n;
temp_table_name := 'bulk_geocode_street_' || md5(random()::text);
@@ -2036,25 +2048,27 @@ BEGIN
coalesce(state_column, ''''''), coalesce(country_column, '''''')
into street_column, city_column, state_column, country_column;
FOR cartodb_id_batch in 0..(batches_n - 1)
LOOP
IF batches_n > 0 THEN
FOR cartodb_id_batch in 0..(batches_n - 1)
LOOP
EXECUTE format(
'WITH geocoding_data as (' ||
' SELECT ' ||
' json_build_object(''id'', cartodb_id, ''address'', %s, ''city'', %s, ''state'', %s, ''country'', %s) as data , ' ||
' floor((cartodb_id-1)::float/$1) as batch' ||
' FROM (%s) _x' ||
') ' ||
'INSERT INTO %s SELECT (cdb_dataservices_client._cdb_bulk_geocode_street_point(jsonb_agg(data))).* ' ||
'FROM geocoding_data ' ||
'WHERE batch = $2', street_column, city_column, state_column, country_column, query, temp_table_name)
USING batch_size, cartodb_id_batch;
EXECUTE format(
'WITH geocoding_data as (' ||
' SELECT ' ||
' json_build_object(''id'', cartodb_id, ''address'', %s, ''city'', %s, ''state'', %s, ''country'', %s) as data , ' ||
' floor((row_number() over ())::float/$1) as batch' ||
' FROM (%s) _x' ||
') ' ||
'INSERT INTO %s SELECT (cdb_dataservices_client._cdb_bulk_geocode_street_point(jsonb_agg(data))).* ' ||
'FROM geocoding_data ' ||
'WHERE batch = $2', street_column, city_column, state_column, country_column, query, temp_table_name)
USING batch_size, cartodb_id_batch;
GET DIAGNOSTICS current_row_count = ROW_COUNT;
RAISE DEBUG 'Batch % --> %', cartodb_id_batch, current_row_count;
GET DIAGNOSTICS current_row_count = ROW_COUNT;
RAISE DEBUG 'Batch % --> %', cartodb_id_batch, current_row_count;
END LOOP;
END LOOP;
END IF;
RETURN QUERY EXECUTE 'SELECT * FROM ' || quote_ident(temp_table_name);
END;

View File

@@ -12,4 +12,9 @@ BEGIN
RETURN ROWS;
END
$func$ LANGUAGE plpgsql;
$func$ LANGUAGE plpgsql;
-- Taken from https://stackoverflow.com/a/48013356/351721
CREATE OR REPLACE FUNCTION jsonb_array_casttext(jsonb) RETURNS text[] AS $f$
SELECT array_agg(x) || ARRAY[]::text[] FROM jsonb_array_elements_text($1) t(x);
$f$ LANGUAGE sql IMMUTABLE;

View File

@@ -39,5 +39,6 @@ CREATE TYPE cdb_dataservices_client.service_quota_info AS (
monthly_quota NUMERIC,
used_quota NUMERIC,
soft_limit BOOLEAN,
provider TEXT
provider TEXT,
max_batch_size NUMERIC
);

View File

@@ -1,40 +1,46 @@
CREATE OR REPLACE FUNCTION cdb_dataservices_client.cdb_bulk_geocode_street_point (query text,
street_column text, city_column text default null, state_column text default null, country_column text default null, batch_size integer DEFAULT 100)
street_column text, city_column text default null, state_column text default null, country_column text default null, batch_size integer DEFAULT NULL)
RETURNS SETOF cdb_dataservices_client.geocoding AS $$
DECLARE
query_row_count integer;
enough_quota boolean;
remaining_quota integer;
max_batch_size integer;
cartodb_id_batch integer;
batches_n integer;
DEFAULT_BATCH_SIZE CONSTANT numeric := 100;
MAX_BATCH_SIZE CONSTANT numeric := 10000;
MAX_SAFE_BATCH_SIZE CONSTANT numeric := 5000;
current_row_count integer ;
temp_table_name text;
BEGIN
SELECT csqi.monthly_quota - csqi.used_quota AS remaining_quota, csqi.max_batch_size
INTO remaining_quota, max_batch_size
FROM cdb_dataservices_client.cdb_service_quota_info() csqi
WHERE service = 'hires_geocoder';
RAISE DEBUG 'remaining_quota: %; max_batch_size: %', remaining_quota, max_batch_size;
IF batch_size IS NULL THEN
RAISE EXCEPTION 'batch_size can''t be null';
ELSIF batch_size > MAX_BATCH_SIZE THEN
RAISE EXCEPTION 'batch_size must be lower than %', MAX_BATCH_SIZE + 1;
batch_size := max_batch_size;
ELSIF batch_size > max_batch_size THEN
RAISE EXCEPTION 'batch_size must be lower than %', max_batch_size + 1;
END IF;
EXECUTE format('SELECT COUNT(1) from (%s) _x', query) INTO query_row_count;
IF batch_size > MAX_SAFE_BATCH_SIZE THEN
batch_size := MAX_SAFE_BATCH_SIZE;
END IF;
EXECUTE format('SELECT count(1), ceil(count(1)::float/%s) FROM (%s) _x', batch_size, query)
INTO query_row_count, batches_n;
RAISE DEBUG 'cdb_bulk_geocode_street_point --> query_row_count: %; query: %; country: %; state: %; city: %; street: %',
query_row_count, query, country_column, state_column, city_column, street_column;
SELECT cdb_dataservices_client.cdb_enough_quota('hires_geocoder', query_row_count) INTO enough_quota;
IF enough_quota IS NOT NULL AND NOT enough_quota THEN
SELECT csqi.monthly_quota - csqi.used_quota AS remaining_quota
INTO remaining_quota
FROM cdb_dataservices_client.cdb_service_quota_info() csqi
WHERE service = 'hires_geocoder';
IF remaining_quota < query_row_count THEN
RAISE EXCEPTION 'Remaining quota: %. Estimated cost: %', remaining_quota, query_row_count;
END IF;
EXECUTE format('SELECT ceil(max(cartodb_id)::float/%s) FROM (%s) _x', batch_size, query) INTO batches_n;
RAISE DEBUG 'batches_n: %', batches_n;
temp_table_name := 'bulk_geocode_street_' || md5(random()::text);
@@ -48,25 +54,27 @@ BEGIN
coalesce(state_column, ''''''), coalesce(country_column, '''''')
into street_column, city_column, state_column, country_column;
FOR cartodb_id_batch in 0..(batches_n - 1)
LOOP
IF batches_n > 0 THEN
FOR cartodb_id_batch in 0..(batches_n - 1)
LOOP
EXECUTE format(
'WITH geocoding_data as (' ||
' SELECT ' ||
' json_build_object(''id'', cartodb_id, ''address'', %s, ''city'', %s, ''state'', %s, ''country'', %s) as data , ' ||
' floor((cartodb_id-1)::float/$1) as batch' ||
' FROM (%s) _x' ||
') ' ||
'INSERT INTO %s SELECT (cdb_dataservices_client._cdb_bulk_geocode_street_point(jsonb_agg(data))).* ' ||
'FROM geocoding_data ' ||
'WHERE batch = $2', street_column, city_column, state_column, country_column, query, temp_table_name)
USING batch_size, cartodb_id_batch;
EXECUTE format(
'WITH geocoding_data as (' ||
' SELECT ' ||
' json_build_object(''id'', cartodb_id, ''address'', %s, ''city'', %s, ''state'', %s, ''country'', %s) as data , ' ||
' floor((row_number() over ())::float/$1) as batch' ||
' FROM (%s) _x' ||
') ' ||
'INSERT INTO %s SELECT (cdb_dataservices_client._cdb_bulk_geocode_street_point(jsonb_agg(data))).* ' ||
'FROM geocoding_data ' ||
'WHERE batch = $2', street_column, city_column, state_column, country_column, query, temp_table_name)
USING batch_size, cartodb_id_batch;
GET DIAGNOSTICS current_row_count = ROW_COUNT;
RAISE DEBUG 'Batch % --> %', cartodb_id_batch, current_row_count;
GET DIAGNOSTICS current_row_count = ROW_COUNT;
RAISE DEBUG 'Batch % --> %', cartodb_id_batch, current_row_count;
END LOOP;
END LOOP;
END IF;
RETURN QUERY EXECUTE 'SELECT * FROM ' || quote_ident(temp_table_name);
END;

View File

@@ -1,21 +1,21 @@
\set VERBOSITY terse
-- Test bulk size mandatory
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''', null, null, null, null);
ERROR: batch_size can't be null
-- Test quota check by mocking quota 0
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info() RENAME TO cdb_service_quota_info_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_service_quota_info ()
RETURNS SETOF cdb_dataservices_client.service_quota_info AS $$
SELECT 'hires_geocoder'::cdb_dataservices_client.service_type AS service, 0::NUMERIC AS monthly_quota, 0::NUMERIC AS used_quota, FALSE AS soft_limit, 'google' AS provider, 1::NUMERIC AS max_batch_size;
$$ LANGUAGE SQL;
ALTER FUNCTION cdb_dataservices_client.cdb_enough_quota (service TEXT ,input_size NUMERIC) RENAME TO cdb_enough_quota_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_enough_quota (service TEXT ,input_size NUMERIC)
RETURNS BOOLEAN as $$
SELECT FALSE;
$$ LANGUAGE SQL;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info() RENAME TO cdb_service_quota_info_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_service_quota_info ()
RETURNS SETOF cdb_dataservices_client.service_quota_info AS $$
SELECT 'hires_geocoder'::cdb_dataservices_client.service_type AS service, 0::NUMERIC AS monthly_quota, 0::NUMERIC AS used_quota, FALSE AS soft_limit, 'google' AS provider;
$$ LANGUAGE SQL;
-- Test bulk size not mandatory (it will get the optimal)
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''', null, null, null, null);
ERROR: Remaining quota: 0. Estimated cost: 1
-- Test quota check by mocking quota 0
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''');
ERROR: Remaining quota: 0. Estimated cost: 1
DROP FUNCTION cdb_dataservices_client.cdb_service_quota_info;
DROP FUNCTION cdb_dataservices_client.cdb_enough_quota;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info_mocked() RENAME TO cdb_service_quota_info;
ALTER FUNCTION cdb_dataservices_client.cdb_enough_quota_mocked (service TEXT ,input_size NUMERIC) RENAME TO cdb_enough_quota;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info_mocked() RENAME TO cdb_service_quota_info;

View File

@@ -1,26 +1,26 @@
\set VERBOSITY terse
-- Test bulk size mandatory
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''', null, null, null, null);
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info() RENAME TO cdb_service_quota_info_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_service_quota_info ()
RETURNS SETOF cdb_dataservices_client.service_quota_info AS $$
SELECT 'hires_geocoder'::cdb_dataservices_client.service_type AS service, 0::NUMERIC AS monthly_quota, 0::NUMERIC AS used_quota, FALSE AS soft_limit, 'google' AS provider, 1::NUMERIC AS max_batch_size;
$$ LANGUAGE SQL;
-- Test quota check by mocking quota 0
ALTER FUNCTION cdb_dataservices_client.cdb_enough_quota (service TEXT ,input_size NUMERIC) RENAME TO cdb_enough_quota_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_enough_quota (service TEXT ,input_size NUMERIC)
RETURNS BOOLEAN as $$
SELECT FALSE;
$$ LANGUAGE SQL;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info() RENAME TO cdb_service_quota_info_mocked;
CREATE FUNCTION cdb_dataservices_client.cdb_service_quota_info ()
RETURNS SETOF cdb_dataservices_client.service_quota_info AS $$
SELECT 'hires_geocoder'::cdb_dataservices_client.service_type AS service, 0::NUMERIC AS monthly_quota, 0::NUMERIC AS used_quota, FALSE AS soft_limit, 'google' AS provider;
$$ LANGUAGE SQL;
-- Test bulk size not mandatory (it will get the optimal)
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''', null, null, null, null);
-- Test quota check by mocking quota 0
SELECT cdb_dataservices_client.cdb_bulk_geocode_street_point('select 1 as cartodb_id', '''Valladolid, Spain''');
DROP FUNCTION cdb_dataservices_client.cdb_service_quota_info;
DROP FUNCTION cdb_dataservices_client.cdb_enough_quota;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info_mocked() RENAME TO cdb_service_quota_info;
ALTER FUNCTION cdb_dataservices_client.cdb_enough_quota_mocked (service TEXT ,input_size NUMERIC) RENAME TO cdb_enough_quota;
ALTER FUNCTION cdb_dataservices_client.cdb_service_quota_info_mocked() RENAME TO cdb_service_quota_info;

View File

@@ -1861,7 +1861,8 @@ BEGIN
monthly_quota NUMERIC,
used_quota NUMERIC,
soft_limit BOOLEAN,
provider TEXT
provider TEXT,
max_batch_size NUMERIC
);
END IF;
END $$;
@@ -1872,6 +1873,7 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server.cdb_service_quota_info(
RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
from cartodb_services.metrics.user import UserMetricsService
from datetime import date
from cartodb_services.bulk_geocoders import BATCH_GEOCODER_CLASS_BY_PROVIDER
plpy.execute("SELECT cdb_dataservices_server._connect_to_redis('{0}')".format(username))
redis_conn = GD["redis_connection_{0}".format(username)]['redis_metrics_connection']
@@ -1889,7 +1891,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_isolines_config.service_type, today)
soft_limit = user_isolines_config.soft_isolines_limit
provider = user_isolines_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
#-- Hires Geocoder
service = 'hires_geocoder'
@@ -1901,7 +1903,12 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_geocoder_config.service_type, today)
soft_limit = user_geocoder_config.soft_geocoding_limit
provider = user_geocoder_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
batch_geocoder_class = BATCH_GEOCODER_CLASS_BY_PROVIDER.get(provider, None)
if batch_geocoder_class and hasattr(batch_geocoder_class, 'MAX_BATCH_SIZE'):
max_batch_size = batch_geocoder_class.MAX_BATCH_SIZE
else:
max_batch_size = 1
ret += [[service, monthly_quota, used_quota, soft_limit, provider, max_batch_size]]
#-- Routing
service = 'routing'
@@ -1913,7 +1920,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_routing_config.service_type, today)
soft_limit = user_routing_config.soft_limit
provider = user_routing_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
#-- Observatory
service = 'observatory'
@@ -1925,7 +1932,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_obs_config.service_type, today)
soft_limit = user_obs_config.soft_limit
provider = user_obs_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
return ret
$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;
@@ -2387,10 +2394,10 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server._cdb_bulk_google_geocode_stre
RETURNS SETOF cdb_dataservices_server.geocoding AS $$
from cartodb_services import run_street_point_geocoder
from cartodb_services.tools import LegacyServiceManager
from cartodb_services.google import GoogleMapsGeocoder
from cartodb_services.google import GoogleMapsBulkGeocoder
service_manager = LegacyServiceManager('geocoder', username, orgname, GD)
geocoder = GoogleMapsGeocoder(service_manager.config.google_client_id, service_manager.config.google_api_key, service_manager.logger)
geocoder = GoogleMapsBulkGeocoder(service_manager.config.google_client_id, service_manager.config.google_api_key, service_manager.logger)
return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches)
$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;

View File

@@ -22,7 +22,8 @@ BEGIN
monthly_quota NUMERIC,
used_quota NUMERIC,
soft_limit BOOLEAN,
provider TEXT
provider TEXT,
max_batch_size NUMERIC
);
END IF;
END $$;
@@ -33,6 +34,7 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server.cdb_service_quota_info(
RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
from cartodb_services.metrics.user import UserMetricsService
from datetime import date
from cartodb_services.bulk_geocoders import BATCH_GEOCODER_CLASS_BY_PROVIDER
plpy.execute("SELECT cdb_dataservices_server._connect_to_redis('{0}')".format(username))
redis_conn = GD["redis_connection_{0}".format(username)]['redis_metrics_connection']
@@ -50,7 +52,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_isolines_config.service_type, today)
soft_limit = user_isolines_config.soft_isolines_limit
provider = user_isolines_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
#-- Hires Geocoder
service = 'hires_geocoder'
@@ -62,7 +64,12 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_geocoder_config.service_type, today)
soft_limit = user_geocoder_config.soft_geocoding_limit
provider = user_geocoder_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
batch_geocoder_class = BATCH_GEOCODER_CLASS_BY_PROVIDER.get(provider, None)
if batch_geocoder_class and hasattr(batch_geocoder_class, 'MAX_BATCH_SIZE'):
max_batch_size = batch_geocoder_class.MAX_BATCH_SIZE
else:
max_batch_size = 1
ret += [[service, monthly_quota, used_quota, soft_limit, provider, max_batch_size]]
#-- Routing
service = 'routing'
@@ -74,7 +81,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_routing_config.service_type, today)
soft_limit = user_routing_config.soft_limit
provider = user_routing_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
#-- Observatory
service = 'observatory'
@@ -86,7 +93,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
used_quota = user_service.used_quota(user_obs_config.service_type, today)
soft_limit = user_obs_config.soft_limit
provider = user_obs_config.provider
ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]
return ret
$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;

View File

@@ -44,10 +44,10 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server._cdb_bulk_google_geocode_stre
RETURNS SETOF cdb_dataservices_server.geocoding AS $$
from cartodb_services import run_street_point_geocoder
from cartodb_services.tools import LegacyServiceManager
from cartodb_services.google import GoogleMapsGeocoder
from cartodb_services.google import GoogleMapsBulkGeocoder
service_manager = LegacyServiceManager('geocoder', username, orgname, GD)
geocoder = GoogleMapsGeocoder(service_manager.config.google_client_id, service_manager.config.google_api_key, service_manager.logger)
geocoder = GoogleMapsBulkGeocoder(service_manager.config.google_client_id, service_manager.config.google_api_key, service_manager.logger)
return run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches)
$$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;

View File

@@ -0,0 +1,11 @@
from google import GoogleMapsBulkGeocoder
from here import HereMapsBulkGeocoder
from tomtom import TomTomBulkGeocoder
from mapbox import MapboxBulkGeocoder
BATCH_GEOCODER_CLASS_BY_PROVIDER = {
'google': GoogleMapsBulkGeocoder,
'heremaps': HereMapsBulkGeocoder,
'tomtom': TomTomBulkGeocoder,
'mapbox': MapboxBulkGeocoder
}

View File

@@ -6,6 +6,22 @@ from collections import namedtuple
import json
PRECISION_PRECISE = 'precise'
PRECISION_INTERPOLATED = 'interpolated'
def geocoder_metadata(relevance, precision, match_types):
return {
'relevance': round(relevance, 2),
'precision': precision,
'match_types': match_types
}
def compose_address(street, city=None, state=None, country=None):
return ', '.join(filter(None, [street, city, state, country]))
def run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, orgname, searches):
plpy.execute("SELECT cdb_dataservices_server._get_logger_config()")
logger_config = GD["logger_config"]
@@ -18,12 +34,18 @@ def run_street_point_geocoder(plpy, GD, geocoder, service_manager, username, org
if geocode_results:
results = []
for result in geocode_results:
if len(result) > 2:
metadata = json.dumps(result[2])
else:
logger.warning('Geocoding for {} without metadata'.format(username))
metadata = '{}'
if result[1] and len(result[1]) == 2:
plan = plpy.prepare("SELECT ST_SetSRID(ST_MakePoint($1, $2), 4326) as the_geom; ", ["double precision", "double precision"])
point = plpy.execute(plan, result[1], 1)[0]
results.append([result[0], point['the_geom'], None])
results.append([result[0], point['the_geom'], metadata])
else:
results.append([result[0], None, None])
results.append([result[0], None, metadata])
service_manager.quota_service.increment_success_service_use(len(results))
return results
else:
@@ -47,7 +69,12 @@ StreetGeocoderSearch = namedtuple('StreetGeocoderSearch', 'id address city state
class StreetPointBulkGeocoder:
"""
Classes extending StreetPointBulkGeocoder should implement:
* _bulk_geocode(street_geocoder_searches)
* _batch_geocode(street_geocoder_searches)
* MAX_BATCH_SIZE
If they want to provide an alternative serial (for small batches):
* _should_use_batch(street_geocoder_searches)
* _serial_geocode(street_geocoder_searches)
"""
SEARCH_KEYS = ['id', 'address', 'city', 'state', 'country']
@@ -73,10 +100,20 @@ class StreetPointBulkGeocoder:
street_geocoder_searches.append(
StreetGeocoderSearch(search_id, address, city, state, country))
return self._bulk_geocode(street_geocoder_searches)
if len(street_geocoder_searches) > self.MAX_BATCH_SIZE:
raise Exception("Batch size can't be larger than {}".format(self.MAX_BATCH_SIZE))
if self._should_use_batch(street_geocoder_searches):
return self._batch_geocode(street_geocoder_searches)
else:
return self._serial_geocode(street_geocoder_searches)
def _batch_geocode(self, street_geocoder_searches):
raise NotImplementedError('Subclasses must implement _batch_geocode')
def _serial_geocode(self, street_geocoder_searches):
raise NotImplementedError('Subclasses must implement _serial_geocode')
def _should_use_batch(self, street_geocoder_searches):
return True
def _bulk_geocode(self, street_geocoder_searches):
"""
Subclasses must implement _bulk_geocode
"""
raise NotImplementedError('Subclasses must implement _bulk_geocode')

View File

@@ -1 +1,2 @@
from geocoder import GoogleMapsGeocoder
from bulk_geocoder import GoogleMapsBulkGeocoder

View File

@@ -0,0 +1,62 @@
from multiprocessing import Pool
from exceptions import MalformedResult
from cartodb_services import StreetPointBulkGeocoder
from cartodb_services.geocoder import compose_address
from cartodb_services.google import GoogleMapsGeocoder
def async_geocoder(geocoder, address, components):
return geocoder.geocode(address=address, components=components)
class GoogleMapsBulkGeocoder(GoogleMapsGeocoder, StreetPointBulkGeocoder):
"""A Google Maps Geocoder wrapper for python"""
MAX_BATCH_SIZE = 1000
MIN_BATCHED_SEARCH = 2 # Batched is a parallelization
PARALLEL_PROCESSES = 13
def __init__(self, client_id, client_secret, logger):
GoogleMapsGeocoder.__init__(self, client_id, client_secret, logger)
def _should_use_batch(self, searches):
return len(searches) >= self.MIN_BATCHED_SEARCH
def _serial_geocode(self, searches):
results = []
for search in searches:
(cartodb_id, street, city, state, country) = search
lng_lat, metadata = self.geocode_meta(street, city, state, country)
results.append((cartodb_id, lng_lat, metadata))
return results
def _batch_geocode(self, searches):
bulk_results = {}
pool = Pool(processes=self.PARALLEL_PROCESSES)
for search in searches:
(cartodb_id, street, city, state, country) = search
address = compose_address(street, city, state, country)
if address:
components = self._build_optional_parameters(city, state, country)
result = pool.apply_async(async_geocoder,
(self.geocoder, address, components))
bulk_results[cartodb_id] = result
pool.close()
pool.join()
try:
results = []
for cartodb_id, bulk_result in bulk_results.items():
try:
lng_lat, metadata = self._process_results(bulk_result.get())
except Exception as e:
self._logger.error('Error at Google async_geocoder', e)
lng_lat, metadata = [[], {}]
results.append((cartodb_id, lng_lat, metadata))
return results
except KeyError as e:
self._logger.error('KeyError error', exception=e)
raise MalformedResult()
except Exception as e:
self._logger.error('General error', exception=e)
raise e

View File

@@ -1,29 +1,42 @@
#!/usr/local/bin/python
# -*- coding: utf-8 -*-
import googlemaps
from urlparse import parse_qs
from exceptions import MalformedResult
from cartodb_services import StreetPointBulkGeocoder
from cartodb_services.geocoder import compose_address, geocoder_metadata, PRECISION_PRECISE, PRECISION_INTERPOLATED
from cartodb_services.google.exceptions import InvalidGoogleCredentials
from client_factory import GoogleMapsClientFactory
from multiprocessing import Pool, TimeoutError
EMPTY_RESPONSE = [[], {}]
PARTIAL_FACTOR = 0.8
RELEVANCE_BY_LOCATION_TYPE = {
'ROOFTOP': 1,
'GEOMETRIC_CENTER': 0.9,
'RANGE_INTERPOLATED': 0.8,
'APPROXIMATE': 0.7
}
PRECISION_BY_LOCATION_TYPE = {
'ROOFTOP': PRECISION_PRECISE,
'GEOMETRIC_CENTER': PRECISION_PRECISE,
'RANGE_INTERPOLATED': PRECISION_INTERPOLATED,
'APPROXIMATE': PRECISION_INTERPOLATED
}
MATCH_TYPE_BY_MATCH_LEVEL = {
'point_of_interest': 'point_of_interest',
'country': 'country',
'administrative_area_level_1': 'state',
'administrative_area_level_2': 'county',
'locality': 'locality',
'sublocality': 'district',
'street_address': 'street',
'intersection': 'intersection',
'street_number': 'street_number',
'postal_code': 'postal_code'
}
import time, random
def async_geocoder(geocoder, address, components):
# TODO: clean this and previous import
# time.sleep(.3 + random.random())
# return [{ 'geometry': { 'location': { 'lng': 1, 'lat': 2 } } }]
results = geocoder.geocode(address=address, components=components)
return results if results else []
class GoogleMapsGeocoder(StreetPointBulkGeocoder):
"""A Google Maps Geocoder wrapper for python"""
PARALLEL_PROCESSES = 13
class GoogleMapsGeocoder():
def __init__(self, client_id, client_secret, logger):
if client_id is None:
@@ -33,55 +46,28 @@ class GoogleMapsGeocoder(StreetPointBulkGeocoder):
self.geocoder = GoogleMapsClientFactory.get(self.client_id, self.client_secret, self.channel)
self._logger = logger
def geocode(self, searchtext, city=None, state=None,
country=None):
def geocode(self, searchtext, city=None, state=None, country=None):
return self.geocode_meta(searchtext, city, state, country)[0]
def geocode_meta(self, searchtext, city=None, state=None, country=None):
address = compose_address(searchtext, city, state, country)
try:
opt_params = self._build_optional_parameters(city, state, country)
results = self.geocoder.geocode(address=searchtext,
results = self.geocoder.geocode(address=address,
components=opt_params)
if results:
return self._extract_lng_lat_from_result(results[0])
else:
return []
except KeyError:
raise MalformedResult()
def _bulk_geocode(self, searches):
bulk_results = {}
pool = Pool(processes=self.PARALLEL_PROCESSES)
for search in searches:
(search_id, address, city, state, country) = search
opt_params = self._build_optional_parameters(city, state, country)
# Geocoding works better if components are also inside the address
address = ', '.join(filter(None, [address, city, state, country]))
if address:
self._logger.debug('async geocoding --> {} {}'.format(address.encode('utf-8'), opt_params))
result = pool.apply_async(async_geocoder,
(self.geocoder, address, opt_params))
else:
result = []
bulk_results[search_id] = result
pool.close()
pool.join()
try:
results = []
for search_id, bulk_result in bulk_results.items():
try:
result = bulk_result.get()
except Exception as e:
self._logger.error('Error at Google async_geocoder', e)
result = []
lng_lat = self._extract_lng_lat_from_result(result[0]) if result else []
results.append((search_id, lng_lat, []))
return results
return self._process_results(results)
except KeyError as e:
self._logger.error('KeyError error', exception=e)
self._logger.error('address: {}'.format(address), e)
raise MalformedResult()
except Exception as e:
self._logger.error('General error', exception=e)
raise e
def _process_results(self, results):
if results:
return [
self._extract_lng_lat_from_result(results[0]),
self._extract_metadata_from_result(results[0])
]
else:
return EMPTY_RESPONSE
def _extract_lng_lat_from_result(self, result):
location = result['geometry']['location']
@@ -89,6 +75,20 @@ class GoogleMapsGeocoder(StreetPointBulkGeocoder):
latitude = location['lat']
return [longitude, latitude]
def _extract_metadata_from_result(self, result):
location_type = result['geometry']['location_type']
base_relevance = RELEVANCE_BY_LOCATION_TYPE[location_type]
partial_match = result.get('partial_match', False)
partial_factor = PARTIAL_FACTOR if partial_match else 1
match_types = [MATCH_TYPE_BY_MATCH_LEVEL.get(match_level, None)
for match_level in result['types']]
return geocoder_metadata(
base_relevance * partial_factor,
PRECISION_BY_LOCATION_TYPE[location_type],
filter(None, match_types)
)
def _build_optional_parameters(self, city=None, state=None,
country=None):
optional_params = {}

View File

@@ -8,6 +8,7 @@ from collections import namedtuple
from requests.adapters import HTTPAdapter
from cartodb_services import StreetPointBulkGeocoder
from cartodb_services.here import HereMapsGeocoder
from cartodb_services.geocoder import geocoder_metadata
from cartodb_services.metrics import Traceable
from cartodb_services.tools.exceptions import ServiceException
@@ -15,16 +16,16 @@ from cartodb_services.tools.exceptions import ServiceException
HereJobStatus = namedtuple('HereJobStatus', 'total_count processed_count status')
class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
BATCH_URL = 'https://batch.geocoder.cit.api.here.com/6.2/jobs'
MAX_BATCH_SIZE = 1000000 # From the docs
MIN_BATCHED_SEARCH = 100 # Under this, serial will be used
BATCH_URL = 'https://batch.geocoder.cit.api.here.com/6.2/jobs'
# https://developer.here.com/documentation/batch-geocoder/topics/read-batch-request-output.html
META_COLS = ['relevance', 'matchType', 'matchCode', 'matchLevel', 'matchQualityStreet']
MAX_STALLED_RETRIES = 100
BATCH_RETRY_SLEEP_S = 5
MIN_BATCHED_SEARCH = 100 # Under this, serial will be used
JOB_FINAL_STATES = ['completed', 'cancelled', 'deleted', 'failed']
def __init__(self, app_id, app_code, logger, service_params=None, maxresults=MAX_BATCH_SIZE):
def __init__(self, app_id, app_code, logger, service_params=None, maxresults=HereMapsGeocoder.DEFAULT_MAXRESULTS):
HereMapsGeocoder.__init__(self, app_id, app_code, logger, service_params, maxresults)
self.session = requests.Session()
self.session.mount(self.BATCH_URL,
@@ -34,16 +35,6 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
'app_code': self.app_code,
}
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
@@ -51,13 +42,12 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
results = []
for search in searches:
(search_id, address, city, state, country) = search
coordinates = self.geocode(searchtext=address, city=city, state=state, country=country)
results.append((search_id, coordinates, []))
result = self.geocode_meta(searchtext=address, city=city, state=state, country=country)
results.append((search_id, result[0], result[1]))
return results
def _batch_geocode(self, searches):
request_id = self._send_batch(self._searches_to_csv(searches))
self._logger.debug('--> Sent batch {}'.format(request_id))
last_processed = 0
stalled_retries = 0
@@ -72,16 +62,12 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
stalled_retries = 0
last_processed = job_info.processed_count
self._logger.debug('--> Job poll check: {}'.format(job_info))
if job_info.status in self.JOB_FINAL_STATES:
break
else:
time.sleep(self.BATCH_RETRY_SLEEP_S)
self._logger.debug('--> Job complete: {}'.format(job_info))
results = self._download_results(request_id)
self._logger.debug('--> Results: {} rows; {}'.format(len(results), results))
return results
@@ -105,7 +91,7 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
request_params.update({
'gen': 8,
'action': 'run',
#'mailto': 'juanignaciosl@carto.com',
# 'mailto': 'juanignaciosl@carto.com',
'header': 'true',
'inDelim': '|',
'outDelim': '|',
@@ -131,8 +117,8 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
timeout=(self.connect_timeout, self.read_timeout))
polling_root = ET.fromstring(polling_r.text)
return HereJobStatus(
total_count=polling_root.find('./Response/TotalCount').text,
processed_count=polling_root.find('./Response/ProcessedCount').text,
total_count=int(polling_root.find('./Response/TotalCount').text),
processed_count=int(polling_root.find('./Response/ProcessedCount').text),
status=polling_root.find('./Response/Status').text)
def _download_results(self, job_id):
@@ -147,7 +133,16 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
reader = csv.DictReader(root_zip.open(name), delimiter='|')
for row in reader:
if row['SeqNumber'] == '1': # First per requested data
results.append((row['recId'], [row['displayLongitude'], row['displayLatitude']]))
precision = self.PRECISION_BY_MATCH_TYPE[
row.get('matchType', 'pointAddress')]
match_type = self.MATCH_TYPE_BY_MATCH_LEVEL.get(row['matchLevel'], None)
results.append((row['recId'],
[row['displayLongitude'], row['displayLatitude']],
geocoder_metadata(
float(row['relevance']),
precision,
[match_type] if match_type else []
)))
return results

View File

@@ -6,9 +6,9 @@ import requests
from requests.adapters import HTTPAdapter
from exceptions import *
from cartodb_services.geocoder import PRECISION_PRECISE, PRECISION_INTERPOLATED, geocoder_metadata
from cartodb_services.metrics import Traceable
class HereMapsGeocoder(Traceable):
'A Here Maps Geocoder wrapper for python'
@@ -52,6 +52,23 @@ class HereMapsGeocoder(Traceable):
'strictlanguagemode'
] + ADDRESS_PARAMS
PRECISION_BY_MATCH_TYPE = {
'pointAddress': PRECISION_PRECISE,
'interpolated': PRECISION_INTERPOLATED
}
MATCH_TYPE_BY_MATCH_LEVEL = {
'landmark': 'point_of_interest',
'country': 'country',
'state': 'state',
'county': 'county',
'city': 'locality',
'district': 'district',
'street': 'street',
'intersection': 'intersection',
'houseNumber': 'street_number',
'postalCode': 'postal_code'
}
def __init__(self, app_id, app_code, logger, service_params=None, maxresults=DEFAULT_MAXRESULTS):
service_params = service_params or {}
self.app_id = app_id
@@ -65,12 +82,15 @@ class HereMapsGeocoder(Traceable):
self.max_retries = service_params.get('max_retries', self.MAX_RETRIES)
def geocode(self, **kwargs):
return self.geocode_meta(**kwargs)[0]
def geocode_meta(self, **kwargs):
params = {}
for key, value in kwargs.iteritems():
if value and value.strip():
params[key] = value
if not params:
return []
return [[], {}]
return self._execute_geocode(params)
def _execute_geocode(self, params):
@@ -78,11 +98,13 @@ class HereMapsGeocoder(Traceable):
raise BadGeocodingParams(params)
try:
response = self._perform_request(params)
results = response['Response']['View'][0]['Result'][0]
return self._extract_lng_lat_from_result(results)
result = response['Response']['View'][0]['Result'][0]
return [self._extract_lng_lat_from_result(result),
self._extract_metadata_from_result(result)]
except IndexError:
return []
except KeyError:
return [[], {}]
except KeyError as e:
self._logger.error('params: {}'.format(params), e)
raise MalformedResult()
def _perform_request(self, params):
@@ -118,3 +140,14 @@ class HereMapsGeocoder(Traceable):
latitude = location['DisplayPosition']['Latitude']
return [longitude, latitude]
def _extract_metadata_from_result(self, result):
# See https://stackoverflow.com/questions/51285622/missing-matchtype-at-here-geocoding-responses
precision = self.PRECISION_BY_MATCH_TYPE[
result.get('MatchType', 'pointAddress')]
match_type = self.MATCH_TYPE_BY_MATCH_LEVEL.get(result['MatchLevel'], None)
return geocoder_metadata(
result['Relevance'],
precision,
[match_type] if match_type else []
)

View File

@@ -21,16 +21,6 @@ class MapboxBulkGeocoder(MapboxGeocoder, StreetPointBulkGeocoder):
self.max_retries = service_params.get('max_retries', self.MAX_RETRIES)
self.session = requests.Session()
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
@@ -38,16 +28,10 @@ class MapboxBulkGeocoder(MapboxGeocoder, StreetPointBulkGeocoder):
results = []
for search in searches:
elements = self._encoded_elements(search)
self._logger.debug('--> Sending serial search: {}'.format(search))
coordinates = self._geocode_search(*elements)
results.append((search[0], coordinates, []))
return results
result = self.geocode_meta(*elements)
def _geocode_search(self, address, city, state, country):
coordinates = self.geocode(searchtext=address, city=city,
state_province=state, country=country)
self._logger.debug('--> result sent')
return coordinates
results.append((search[0], result[0], result[1]))
return results
def _encoded_elements(self, search):
(search_id, address, city, state, country) = search
@@ -67,13 +51,10 @@ class MapboxBulkGeocoder(MapboxGeocoder, StreetPointBulkGeocoder):
free = ', '.join([elem for elem in elements if elem])
frees.append(free)
self._logger.debug('--> sending free search: {}'.format(frees))
xy_results = self.geocode_free_text(frees)
full_results = self.geocode_free_text_meta(frees)
results = []
self._logger.debug('--> searches: {}; xy: {}'.format(searches, xy_results))
for s, r in zip(searches, xy_results):
results.append((s[0], r, []))
self._logger.debug('--> results: {}'.format(results))
for s, r in zip(searches, full_results):
results.append((s[0], r[0], r[1]))
return results
def _country_code(self, country):

View File

@@ -5,6 +5,7 @@ Python client for the Mapbox Geocoder service.
import json
import requests
from mapbox import Geocoder
from cartodb_services.geocoder import PRECISION_PRECISE, PRECISION_INTERPOLATED, geocoder_metadata
from cartodb_services.metrics import Traceable
from cartodb_services.tools.exceptions import ServiceException
from cartodb_services.tools.qps import qps_retry
@@ -22,6 +23,19 @@ ENTRY_COORDINATES = 'coordinates'
ENTRY_TYPE = 'type'
TYPE_POINT = 'Point'
EMPTY_RESPONSE = [[], {}]
MATCH_TYPE_BY_MATCH_LEVEL = {
'poi': 'point_of_interest',
'poi.landmark': 'point_of_interest',
'place': 'point_of_interest',
'country': 'country',
'region': 'state',
'locality': 'locality',
'district': 'district',
'address': 'street'
}
class MapboxGeocoder(Traceable):
'''
@@ -39,7 +53,6 @@ class MapboxGeocoder(Traceable):
def _parse_geocoder_response(self, response):
json_response = json.loads(response)
self._logger.debug('--> json response: {}'.format(json_response))
if json_response:
if type(json_response) != list:
@@ -49,12 +62,16 @@ class MapboxGeocoder(Traceable):
for a_json_response in json_response:
if a_json_response[ENTRY_FEATURES]:
feature = a_json_response[ENTRY_FEATURES][0]
result.append(self._extract_lng_lat_from_feature(feature))
result.append([
self._extract_lng_lat_from_feature(feature),
self._extract_metadata_from_result(feature)
]
)
else:
result.append([])
result.append(EMPTY_RESPONSE)
return result
else:
return []
return EMPTY_RESPONSE
def _extract_lng_lat_from_feature(self, feature):
geometry = feature[ENTRY_GEOMETRY]
@@ -67,6 +84,23 @@ class MapboxGeocoder(Traceable):
latitude = location[1]
return [longitude, latitude]
def _extract_metadata_from_result(self, result):
if result[ENTRY_GEOMETRY].get('interpolated', False):
precision = PRECISION_INTERPOLATED
else:
precision = PRECISION_PRECISE
match_types = [MATCH_TYPE_BY_MATCH_LEVEL.get(match_level, None)
for match_level in result['place_type']]
return geocoder_metadata(
self._normalize_relevance(float(result['relevance'])),
precision,
filter(None, match_types)
)
def _normalize_relevance(self, relevance):
return 1 if relevance >= 0.99 else relevance
def _validate_input(self, searchtext, city=None, state_province=None,
country=None):
if searchtext and searchtext.strip():
@@ -88,8 +122,13 @@ class MapboxGeocoder(Traceable):
:param country: Country ISO 3166 code
:return: [x, y] on success, [] on error
"""
return self.geocode_meta(searchtext, city, state_province, country)[0]
@qps_retry(qps=10)
def geocode_meta(self, searchtext, city=None, state_province=None,
country=None):
if not self._validate_input(searchtext, city, state_province, country):
return []
return EMPTY_RESPONSE
address = []
if searchtext and searchtext.strip():
@@ -99,32 +138,30 @@ class MapboxGeocoder(Traceable):
if state_province:
address.append(normalize(state_province))
country = [country] if country else None
free_search = ', '.join(address)
return self.geocode_free_text([free_search], country)[0]
return self.geocode_free_text_meta([free_search], country)[0]
@qps_retry(qps=10)
def geocode_free_text(self, free_searches, country=None):
def geocode_free_text_meta(self, free_searches, country=None):
"""
:param free_searches: Free text searches
:param country: Country ISO 3166 code
:return: list of [x, y] on success, [] on error
"""
country = [country] if country else None
try:
free_search = ';'.join([self._escape(fs) for fs in free_searches])
self._logger.debug('--> free search: {}'.format(free_search))
response = self._geocoder.forward(address=free_search.decode('utf-8'),
country=country,
limit=1)
country=country)
if response.status_code == requests.codes.ok:
return self._parse_geocoder_response(response.text)
elif response.status_code == requests.codes.bad_request:
return []
return EMPTY_RESPONSE
elif response.status_code == requests.codes.unprocessable_entity:
return []
return EMPTY_RESPONSE
else:
raise ServiceException(response.status_code, response)
except requests.Timeout as te:
@@ -138,7 +175,7 @@ class MapboxGeocoder(Traceable):
# Don't raise the exception to continue with the geocoding job
self._logger.error('Error connecting to Mapbox geocoding server',
exception=ce)
return []
return EMPTY_RESPONSE
def _escape(self, free_search):
# Semicolon is used to separate batch geocoding; there's no documented

View File

@@ -6,12 +6,12 @@ from cartodb_services.tools.exceptions import ServiceException
class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder):
MAX_BATCH_SIZE = 1000000 # From the docs
MIN_BATCHED_SEARCH = 10 # Batch API is really fast
BASE_URL = 'https://api.tomtom.com'
BATCH_URL = BASE_URL + '/search/2/batch.json'
MAX_BATCH_SIZE = 1000000 # From the docs
MAX_STALLED_RETRIES = 100
BATCH_RETRY_SLEEP_S = 5
MIN_BATCHED_SEARCH = 10 # Batch API is really fast
READ_TIMEOUT = 60
CONNECT_TIMEOUT = 10
MAX_RETRIES = 1
@@ -26,16 +26,6 @@ class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder):
self.session.mount(self.BATCH_URL,
HTTPAdapter(max_retries=self.max_retries))
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
@@ -47,25 +37,21 @@ class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder):
city = city.encode('utf-8') if city else None
state = state.encode('utf-8') if state else None
country = country.encode('utf-8') if country else None
self._logger.debug('--> Sending serial search: {}'.format(search))
coordinates = self.geocode(searchtext=address, city=city,
result = self.geocode_meta(searchtext=address, city=city,
state_province=state, country=country)
self._logger.debug('--> result sent')
results.append((search_id, coordinates, []))
results.append((search_id, result[0], result[1]))
return results
def _batch_geocode(self, searches):
location = self._send_batch(searches)
xy_results = self._download_results(location)
full_results = self._download_results(location)
results = []
for s, r in zip(searches, xy_results):
results.append((s[0], r, []))
self._logger.debug('--> results: {}'.format(results))
for s, r in zip(searches, full_results):
results.append((s[0], r[0], r[1]))
return results
def _send_batch(self, searches):
body = {'batchItems': [{'query': self._query(s)} for s in searches]}
self._logger.debug('--> {}; Body: {}'.format(self.BATCH_URL, body))
request_params = {
'key': self._apikey
}
@@ -73,9 +59,7 @@ class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder):
allow_redirects=False,
params=request_params,
timeout=(self.connect_timeout, self.read_timeout))
self._logger.debug('--> response: {}'.format(response.status_code))
if response.status_code == 303:
self._logger.debug(response.headers)
return response.headers['Location']
else:
msg = "Error sending batch: {}; Headers: {}".format(
@@ -88,14 +72,12 @@ class TomTomBulkGeocoder(TomTomGeocoder, StreetPointBulkGeocoder):
while True:
response = self.session.get(self.BASE_URL + location)
if response.status_code == 200:
self._logger.debug('--> Results ready {}'.format(location))
return self._parse_results(response.json())
elif response.status_code == 202:
stalled_retries += 1
if stalled_retries > self.MAX_STALLED_RETRIES:
raise Exception('Too many retries for job {}'.format(location))
location = response.headers['Location']
self._logger.debug('--> Waiting for {}'.format(location))
time.sleep(self.BATCH_RETRY_SLEEP_S)
else:
msg = "Error downloading batch: {}; Headers: {}".format(

View File

@@ -4,6 +4,8 @@
import json
import requests
from uritemplate import URITemplate
from math import tanh
from cartodb_services.geocoder import PRECISION_PRECISE, PRECISION_INTERPOLATED, geocoder_metadata
from cartodb_services.metrics import Traceable
from cartodb_services.tools.exceptions import ServiceException
from cartodb_services.tools.qps import qps_retry
@@ -18,7 +20,17 @@ ENTRY_RESULTS = 'results'
ENTRY_POSITION = 'position'
ENTRY_LON = 'lon'
ENTRY_LAT = 'lat'
EMPTY_RESPONSE = [[], {}]
SCORE_NORMALIZATION_FACTOR = 0.15
PRECISION_SCORE_THRESHOLD = 0.5
MATCH_TYPE_BY_MATCH_LEVEL = {
'POI': 'point_of_interest',
'Street': 'street',
'Address Range': 'street',
'Cross Street': 'intersection',
'Point Address': 'street_number'
}
class TomTomGeocoder(Traceable):
'''
@@ -62,6 +74,11 @@ class TomTomGeocoder(Traceable):
@qps_retry(qps=5)
def geocode(self, searchtext, city=None, state_province=None,
country=None):
return self.geocode_meta(searchtext, city, state_province, country)[0]
@qps_retry(qps=5)
def geocode_meta(self, searchtext, city=None, state_province=None,
country=None):
if searchtext:
searchtext = searchtext.decode('utf-8')
if city:
@@ -72,7 +89,7 @@ class TomTomGeocoder(Traceable):
country = country.decode('utf-8')
if not self._validate_input(searchtext, city, state_province, country):
return []
return EMPTY_RESPONSE
address = []
if searchtext and searchtext.strip():
@@ -98,15 +115,15 @@ class TomTomGeocoder(Traceable):
# Don't raise the exception to continue with the geocoding job
self._logger.error('Error connecting to TomTom geocoding server',
exception=ce)
return []
return EMPTY_RESPONSE
def _parse_response(self, status_code, text):
if status_code == requests.codes.ok:
return self._parse_geocoder_response(text)
elif status_code == requests.codes.bad_request:
return []
return EMPTY_RESPONSE
elif status_code == requests.codes.unprocessable_entity:
return []
return EMPTY_RESPONSE
else:
msg = 'Unknown response {}: {}'.format(str(status_code), text)
raise ServiceException(msg, None)
@@ -117,7 +134,25 @@ class TomTomGeocoder(Traceable):
if json_response and json_response[ENTRY_RESULTS]:
result = json_response[ENTRY_RESULTS][0]
return self._extract_lng_lat_from_feature(result)
return [
self._extract_lng_lat_from_feature(result),
self._extract_metadata_from_result(result)
]
else:
return []
return EMPTY_RESPONSE
def _extract_metadata_from_result(self, result):
score = self._normalize_score(result['score'])
match_type = MATCH_TYPE_BY_MATCH_LEVEL.get(result['type'], None)
return geocoder_metadata(
score,
self._precision_from_score(score),
[match_type] if match_type else []
)
def _normalize_score(self, score):
return tanh(score * SCORE_NORMALIZATION_FACTOR)
def _precision_from_score(self, score):
return PRECISION_PRECISE \
if score > PRECISION_SCORE_THRESHOLD else PRECISION_INTERPOLATED

View File

@@ -2,44 +2,11 @@
# -*- coding: utf-8 -*-
from unittest import TestCase
from nose.tools import assert_raises
from nose.tools import assert_not_equal, assert_equal
from nose.tools import assert_not_equal, assert_equal, assert_true
from ..helpers.integration_test_helper import IntegrationTestHelper
from ..helpers.integration_test_helper import assert_close_enough
from ..helpers.integration_test_helper import assert_close_enough, isclose
class TestStreetFunctionsSetUp(TestCase):
def setUp(self):
self.env_variables = IntegrationTestHelper.get_environment_variables()
self.sql_api_url = "{0}://{1}.{2}/api/v1/sql".format(
self.env_variables['schema'],
self.env_variables['username'],
self.env_variables['host'],
self.env_variables['api_key']
)
class TestStreetFunctions(TestStreetFunctionsSetUp):
def test_if_select_with_street_point_is_ok(self):
query = "SELECT cdb_geocode_street_point(street) " \
"as geometry FROM {0} LIMIT 1&api_key={1}".format(
self.env_variables['table_name'],
self.env_variables['api_key'])
geometry = IntegrationTestHelper.execute_query(self.sql_api_url, query)
assert_not_equal(geometry['geometry'], None)
def test_if_select_with_street_without_api_key_raise_error(self):
query = "SELECT cdb_geocode_street_point(street) " \
"as geometry FROM {0} LIMIT 1".format(
self.env_variables['table_name'])
try:
IntegrationTestHelper.execute_query(self.sql_api_url, query)
except Exception as e:
assert_equal(e.message[0], "The api_key must be provided")
class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
provider = None
fixture_points = None
@@ -54,7 +21,8 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
'Valladolid, Mexico': [-88.2022488, 20.68964],
'Madrid': [-3.7037902, 40.4167754],
'Logroño, Spain': [-2.4449852, 42.4627195],
'Logroño, Argentina': [-61.6961807, -29.5031057]
'Logroño, Argentina': [-61.6961807, -29.5031057],
'Plaza España, Barcelona': [2.1482563, 41.375485]
}
HERE_POINTS = {
@@ -68,7 +36,8 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
'Valladolid, Mexico': [-88.20117, 20.69021],
'Madrid': [-3.70578, 40.42028],
'Logroño, Spain': [-2.45194, 42.46592],
'Logroño, Argentina': [-61.69604, -29.50425]
'Logroño, Argentina': [-61.69604, -29.50425],
'Plaza España, Barcelona': [2.14834, 41.37494]
}
TOMTOM_POINTS = HERE_POINTS.copy()
@@ -79,6 +48,7 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
'Valladolid, Spain': [-4.72838, 41.6542],
'Madrid': [-3.70035, 40.42028],
'Logroño, Spain': [-2.44998, 42.46592],
'Plaza España, Barcelona': [2.1497, 41.37516]
})
MAPBOX_POINTS = GOOGLE_POINTS.copy()
@@ -89,6 +59,7 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
'Valladolid, Spain': [-4.72856, 41.652251],
'1902 amphitheatre parkway': [-118.03, 34.06], # TODO: huge mismatch
'Madrid': [-3.69194, 40.4167754],
'Plaza España, Barcelona': [2.342231, 41.50677] # TODO: not ideal
})
FIXTURE_POINTS = {
@@ -98,8 +69,49 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
'mapbox': MAPBOX_POINTS
}
GOOGLE_METADATAS = {
'Plaza España, Barcelona':
{'relevance': 0.9, 'precision': 'precise', 'match_types': ['point_of_interest']},
'Santiago Rusiñol 123, Valladolid':
{'relevance': 0.56, 'precision': 'interpolated', 'match_types': ['locality']}
}
HERE_METADATAS = {
'Plaza España, Barcelona':
{'relevance': 1, 'precision': 'precise', 'match_types': ['street']},
'Santiago Rusiñol 123, Valladolid':
{'relevance': 0.89, 'precision': 'precise', 'match_types': ['street']} # Wrong. See https://stackoverflow.com/questions/51285622/missing-matchtype-at-here-geocoding-responses
}
TOMTOM_METADATAS = {
'Plaza España, Barcelona':
{'relevance': 0.85, 'precision': 'precise', 'match_types': ['street']},
'Santiago Rusiñol 123, Valladolid':
{'relevance': 0.45, 'precision': 'interpolated', 'match_types': ['street']}
}
MAPBOX_METADATAS = {
'Plaza España, Barcelona':
{'relevance': 0.67, 'precision': 'precise', 'match_types': ['point_of_interest']},
'Santiago Rusiñol 123, Valladolid':
{'relevance': 0.67, 'precision': 'precise', 'match_types': ['point_of_interest']} # TODO: wrong
}
METADATAS = {
'google': GOOGLE_METADATAS,
'heremaps': HERE_METADATAS,
'tomtom': TOMTOM_METADATAS,
'mapbox': MAPBOX_METADATAS
}
def setUp(self):
TestStreetFunctionsSetUp.setUp(self)
self.env_variables = IntegrationTestHelper.get_environment_variables()
self.sql_api_url = "{0}://{1}.{2}/api/v1/sql".format(
self.env_variables['schema'],
self.env_variables['username'],
self.env_variables['host'],
self.env_variables['api_key']
)
if not self.fixture_points:
query = "select provider from " \
@@ -109,6 +121,54 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
provider = response['rows'][0]['provider']
self.fixture_points = self.FIXTURE_POINTS[provider]
self.metadata = self.METADATAS[provider]
def _run_authenticated(self, query):
authenticated_query = "{}&api_key={}".format(query,
self.env_variables[
'api_key'])
return IntegrationTestHelper.execute_query_raw(self.sql_api_url,
authenticated_query)
def _used_quota(self):
query = "select used_quota " \
"from cdb_dataservices_client.cdb_service_quota_info() " \
"where service = 'hires_geocoder'"
return self._run_authenticated(query)['rows'][0]['used_quota']
class TestStreetFunctions(TestStreetFunctionsSetUp):
def test_if_select_with_street_point_is_ok(self):
query = "SELECT cdb_dataservices_client.cdb_geocode_street_point(street) " \
"as geometry FROM {0} LIMIT 1&api_key={1}".format(
self.env_variables['table_name'],
self.env_variables['api_key'])
geometry = IntegrationTestHelper.execute_query(self.sql_api_url, query)
assert_not_equal(geometry['geometry'], None)
def test_if_select_with_street_without_api_key_raise_error(self):
table = self.env_variables['table_name']
query = "SELECT cdb_dataservices_client.cdb_geocode_street_point(street) " \
"as geometry FROM {0} LIMIT 1".format(table)
try:
IntegrationTestHelper.execute_query(self.sql_api_url, query)
except Exception as e:
assert_equal(e.message[0],
"permission denied for relation {}".format(table))
def test_component_aggregation(self):
query = "select st_x(the_geom), st_y(the_geom) from (" \
"select cdb_dataservices_client.cdb_geocode_street_point( " \
"'Plaza España', 'Barcelona', null, 'Spain') as the_geom) _x"
response = self._run_authenticated(query)
row = response['rows'][0]
x_y = [row['st_x'], row['st_y']]
# Wrong coordinates (Plaza España, Madrid): [-3.7138975, 40.4256762]
assert_close_enough(x_y, self.fixture_points['Plaza España, Barcelona'])
class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
def test_full_spec(self):
query = "select cartodb_id, st_x(the_geom), st_y(the_geom) " \
"FROM cdb_dataservices_client.cdb_bulk_geocode_street_point(" \
@@ -240,20 +300,35 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
"""
Useful just to test a good batch size
"""
n = 10
n = 110
first_cartodb_id = -1
first_street_number = 1
batch_size = 'NULL' # NULL for optimal
streets = []
for i in range(0, n):
streets.append('{{"cartodb_id": {}, "address": "{} Yonge Street, ' \
'Toronto, Canada"}}'.format(i, i))
'Toronto, Canada"}}'.format(first_cartodb_id + i,
first_street_number + i))
used_quota = self._used_quota()
query = "select *, st_x(the_geom), st_y(the_geom) " \
"FROM cdb_dataservices_client.cdb_bulk_geocode_street_point( " \
"'select * from jsonb_to_recordset(''[" \
"{}" \
"]''::jsonb) as (cartodb_id integer, address text)', " \
"'address', null, null, null, {})".format(','.join(streets), n)
"'address', null, null, null, {})".format(','.join(streets), batch_size)
response = self._run_authenticated(query)
assert_equal(n - 1, len(response['rows']))
assert_equal(n, len(response['rows']))
for row in response['rows']:
assert_not_equal(row['st_x'], None)
assert_not_equal(row['metadata'], {})
metadata = row['metadata']
assert_not_equal(metadata['relevance'], None)
assert_not_equal(metadata['precision'], None)
assert_not_equal(metadata['match_types'], None)
assert_equal(self._used_quota(), used_quota + n)
def test_missing_components_on_private_function(self):
query = "SELECT _cdb_bulk_geocode_street_point(" \
@@ -275,6 +350,52 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
x_y_by_cartodb_id = self._x_y_by_cartodb_id(response)
assert_equal(x_y_by_cartodb_id[1], x_y_by_cartodb_id[2])
def test_component_aggregation(self):
query = "select cartodb_id, st_x(the_geom), st_y(the_geom) " \
"FROM cdb_dataservices_client.cdb_bulk_geocode_street_point(" \
"'select 1 as cartodb_id, ''Spain'' as country, " \
"''Barcelona'' as city, " \
"''Plaza España'' as street' " \
", 'street', 'city', NULL, 'country')"
response = self._run_authenticated(query)
assert_close_enough(self._x_y_by_cartodb_id(response)[1],
self.fixture_points['Plaza España, Barcelona'])
def _test_known_table(self):
subquery = 'select * from unknown_table where cartodb_id < 1100'
subquery_count = 'select count(1) from ({}) _x'.format(subquery)
count = self._run_authenticated(subquery_count)['rows'][0]['count']
query = "select cartodb_id, st_x(the_geom), st_y(the_geom) " \
"FROM cdb_dataservices_client.cdb_bulk_geocode_street_point(" \
"'{}' " \
", 'street', 'city', NULL, 'country')".format(subquery)
response = self._run_authenticated(query)
assert_equal(len(response['rows']), count)
assert_not_equal(response['rows'][0]['st_x'], None)
def test_metadata(self):
query = "select metadata " \
"FROM cdb_dataservices_client.cdb_bulk_geocode_street_point(" \
"'select 1 as cartodb_id, ''Spain'' as country, " \
"''Barcelona'' as city, " \
"''Plaza España'' as street " \
"UNION " \
"select 2 as cartodb_id, ''Spain'' as country, " \
"''Valladolid'' as city, " \
"''Santiago Rusiñol 123'' as street' " \
", 'street', 'city', NULL, 'country')"
response = self._run_authenticated(query)
expected = [
self.metadata['Plaza España, Barcelona'],
self.metadata['Santiago Rusiñol 123, Valladolid']
]
assert_equal(len(response['rows']), len(expected))
for r, e in zip(response['rows'], expected):
self.assert_metadata(r['metadata'], e)
def _run_authenticated(self, query):
authenticated_query = "{}&api_key={}".format(query,
self.env_variables[
@@ -292,3 +413,14 @@ class TestBulkStreetFunctions(TestStreetFunctionsSetUp):
assert_equal(len(points_a_by_cartodb_id), len(points_b_by_cartodb_id))
for cartodb_id, point in points_a_by_cartodb_id.iteritems():
assert_close_enough(point, points_b_by_cartodb_id[cartodb_id])
@staticmethod
def assert_metadata(metadata, expected):
relevance = metadata['relevance']
expected_relevance = expected['relevance']
assert_true(isclose(relevance, expected_relevance, 0.02),
'{} not close to {}'.format(relevance, expected_relevance))
assert_equal(metadata['precision'], expected['precision'])
assert_equal(metadata['match_types'], expected['match_types'])