diff --git a/client/cdb_dataservices_client--0.25.0.sql b/client/cdb_dataservices_client--0.25.0.sql index e810dbc..b67c219 100644 --- a/client/cdb_dataservices_client--0.25.0.sql +++ b/client/cdb_dataservices_client--0.25.0.sql @@ -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; diff --git a/client/sql/05_utils.sql b/client/sql/05_utils.sql index d161c6a..31c8445 100644 --- a/client/sql/05_utils.sql +++ b/client/sql/05_utils.sql @@ -12,4 +12,9 @@ BEGIN RETURN ROWS; END -$func$ LANGUAGE plpgsql; \ No newline at end of file +$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; \ No newline at end of file diff --git a/client/sql/16_custom_types.sql b/client/sql/16_custom_types.sql index 1b74cac..7458ffa 100644 --- a/client/sql/16_custom_types.sql +++ b/client/sql/16_custom_types.sql @@ -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 ); diff --git a/client/sql/21_bulk_geocoding_functions.sql b/client/sql/21_bulk_geocoding_functions.sql index d0bcf5e..dfbe515 100644 --- a/client/sql/21_bulk_geocoding_functions.sql +++ b/client/sql/21_bulk_geocoding_functions.sql @@ -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; diff --git a/client/test/expected/21_bulk_geocoding_functions_test.out b/client/test/expected/21_bulk_geocoding_functions_test.out index 7ca70db..d2f268b 100644 --- a/client/test/expected/21_bulk_geocoding_functions_test.out +++ b/client/test/expected/21_bulk_geocoding_functions_test.out @@ -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; diff --git a/client/test/sql/21_bulk_geocoding_functions_test.sql b/client/test/sql/21_bulk_geocoding_functions_test.sql index 5433bd7..d17f470 100644 --- a/client/test/sql/21_bulk_geocoding_functions_test.sql +++ b/client/test/sql/21_bulk_geocoding_functions_test.sql @@ -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; diff --git a/server/extension/cdb_dataservices_server--0.32.0.sql b/server/extension/cdb_dataservices_server--0.32.0.sql index 8543e7f..dc82322 100644 --- a/server/extension/cdb_dataservices_server--0.32.0.sql +++ b/server/extension/cdb_dataservices_server--0.32.0.sql @@ -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; diff --git a/server/extension/sql/200_quotas.sql b/server/extension/sql/200_quotas.sql index 28ee724..6f34083 100644 --- a/server/extension/sql/200_quotas.sql +++ b/server/extension/sql/200_quotas.sql @@ -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; diff --git a/server/extension/sql/21_bulk_geocode_street.sql b/server/extension/sql/21_bulk_geocode_street.sql index adee498..309d323 100644 --- a/server/extension/sql/21_bulk_geocode_street.sql +++ b/server/extension/sql/21_bulk_geocode_street.sql @@ -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; diff --git a/server/lib/python/cartodb_services/cartodb_services/bulk_geocoders.py b/server/lib/python/cartodb_services/cartodb_services/bulk_geocoders.py new file mode 100644 index 0000000..6dfd555 --- /dev/null +++ b/server/lib/python/cartodb_services/cartodb_services/bulk_geocoders.py @@ -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 +} diff --git a/server/lib/python/cartodb_services/cartodb_services/geocoder.py b/server/lib/python/cartodb_services/cartodb_services/geocoder.py index 6a14594..54e80f1 100644 --- a/server/lib/python/cartodb_services/cartodb_services/geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/geocoder.py @@ -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') diff --git a/server/lib/python/cartodb_services/cartodb_services/google/__init__.py b/server/lib/python/cartodb_services/cartodb_services/google/__init__.py index 4e8cd6b..7570e31 100644 --- a/server/lib/python/cartodb_services/cartodb_services/google/__init__.py +++ b/server/lib/python/cartodb_services/cartodb_services/google/__init__.py @@ -1 +1,2 @@ from geocoder import GoogleMapsGeocoder +from bulk_geocoder import GoogleMapsBulkGeocoder diff --git a/server/lib/python/cartodb_services/cartodb_services/google/bulk_geocoder.py b/server/lib/python/cartodb_services/cartodb_services/google/bulk_geocoder.py new file mode 100644 index 0000000..0e1ab1f --- /dev/null +++ b/server/lib/python/cartodb_services/cartodb_services/google/bulk_geocoder.py @@ -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 diff --git a/server/lib/python/cartodb_services/cartodb_services/google/geocoder.py b/server/lib/python/cartodb_services/cartodb_services/google/geocoder.py index 1cc61c9..fb75c76 100644 --- a/server/lib/python/cartodb_services/cartodb_services/google/geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/google/geocoder.py @@ -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 = {} diff --git a/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py b/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py index d652891..8316c44 100644 --- a/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py @@ -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 diff --git a/server/lib/python/cartodb_services/cartodb_services/here/geocoder.py b/server/lib/python/cartodb_services/cartodb_services/here/geocoder.py index 6c3b81c..23dd871 100644 --- a/server/lib/python/cartodb_services/cartodb_services/here/geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/here/geocoder.py @@ -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 [] + ) diff --git a/server/lib/python/cartodb_services/cartodb_services/mapbox/bulk_geocoder.py b/server/lib/python/cartodb_services/cartodb_services/mapbox/bulk_geocoder.py index 5c502e3..2563d43 100644 --- a/server/lib/python/cartodb_services/cartodb_services/mapbox/bulk_geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/mapbox/bulk_geocoder.py @@ -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): diff --git a/server/lib/python/cartodb_services/cartodb_services/mapbox/geocoder.py b/server/lib/python/cartodb_services/cartodb_services/mapbox/geocoder.py index a3d6f93..d76937a 100644 --- a/server/lib/python/cartodb_services/cartodb_services/mapbox/geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/mapbox/geocoder.py @@ -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 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 index 8036667..8fc2e92 100644 --- a/server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/tomtom/bulk_geocoder.py @@ -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( diff --git a/server/lib/python/cartodb_services/cartodb_services/tomtom/geocoder.py b/server/lib/python/cartodb_services/cartodb_services/tomtom/geocoder.py index b4e82cb..46cdbda 100644 --- a/server/lib/python/cartodb_services/cartodb_services/tomtom/geocoder.py +++ b/server/lib/python/cartodb_services/cartodb_services/tomtom/geocoder.py @@ -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 diff --git a/test/integration/test_street_functions.py b/test/integration/test_street_functions.py index d1fdbd2..ebcc26f 100644 --- a/test/integration/test_street_functions.py +++ b/test/integration/test_street_functions.py @@ -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'])