Skip to content

Commit

Permalink
HereV7: move to the here.py module (#433)
Browse files Browse the repository at this point in the history
  • Loading branch information
KostyaEsmukov committed Apr 10, 2021
1 parent 6f8bcc6 commit 85b2352
Show file tree
Hide file tree
Showing 5 changed files with 398 additions and 410 deletions.
3 changes: 1 addition & 2 deletions geopy/geocoders/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,8 +233,7 @@
from geopy.geocoders.geolake import Geolake
from geopy.geocoders.geonames import GeoNames
from geopy.geocoders.googlev3 import GoogleV3
from geopy.geocoders.here import Here
from geopy.geocoders.herev7 import HereV7
from geopy.geocoders.here import Here, HereV7
from geopy.geocoders.ignfrance import IGNFrance
from geopy.geocoders.mapbox import MapBox
from geopy.geocoders.mapquest import MapQuest
Expand Down
271 changes: 269 additions & 2 deletions geopy/geocoders/here.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,24 @@
import collections.abc
import json
import warnings
from functools import partial
from urllib.parse import urlencode

from geopy.adapters import AdapterHTTPError
from geopy.exc import (
ConfigurationError,
GeocoderAuthenticationFailure,
GeocoderInsufficientPrivileges,
GeocoderQueryError,
GeocoderRateLimited,
GeocoderServiceError,
GeocoderUnavailable,
)
from geopy.geocoders.base import DEFAULT_SENTINEL, Geocoder
from geopy.geocoders.base import DEFAULT_SENTINEL, ERROR_CODE_MAP, Geocoder
from geopy.location import Location
from geopy.util import join_filter, logger

__all__ = ("Here", )
__all__ = ("Here", "HereV7")


class Here(Geocoder):
Expand Down Expand Up @@ -378,3 +381,267 @@ def parse_resource(resource):
return parse_resource(resources[0])
else:
return [parse_resource(resource) for resource in resources]


class HereV7(Geocoder):
"""Geocoder using the HERE Geocoding & Search v7 API.
Documentation at:
https://developer.here.com/documentation/geocoding-search-api/
Terms of Service at:
https://legal.here.com/en-gb/terms
"""

structured_query_params = {
'country',
'state',
'county',
'city',
'district',
'street',
'houseNumber',
'postalCode',
}

geocode_path = '/v1/geocode'
reverse_path = '/v1/revgeocode'

def __init__(
self,
apikey,
*,
scheme=None,
timeout=DEFAULT_SENTINEL,
proxies=DEFAULT_SENTINEL,
user_agent=None,
ssl_context=DEFAULT_SENTINEL,
adapter_factory=None
):
"""
:param str apikey: Should be a valid HERE Maps apikey.
A project can be created at
https://developer.here.com/projects.
:param str scheme:
See :attr:`geopy.geocoders.options.default_scheme`.
:param int timeout:
See :attr:`geopy.geocoders.options.default_timeout`.
:param dict proxies:
See :attr:`geopy.geocoders.options.default_proxies`.
:param str user_agent:
See :attr:`geopy.geocoders.options.default_user_agent`.
:type ssl_context: :class:`ssl.SSLContext`
:param ssl_context:
See :attr:`geopy.geocoders.options.default_ssl_context`.
:param callable adapter_factory:
See :attr:`geopy.geocoders.options.default_adapter_factory`.
"""
super().__init__(
scheme=scheme,
timeout=timeout,
proxies=proxies,
user_agent=user_agent,
ssl_context=ssl_context,
adapter_factory=adapter_factory,
)

domain = "search.hereapi.com"

self.apikey = apikey
self.api = "%s://geocode.%s%s" % (self.scheme, domain, self.geocode_path)
self.reverse_api = (
"%s://revgeocode.%s%s" % (self.scheme, domain, self.reverse_path)
)

def geocode(
self,
query=None,
*,
components=None,
at=None,
countries=None,
language=None,
limit=None,
exactly_one=True,
timeout=DEFAULT_SENTINEL
):
"""
Return a location point by address.
:param str query: The address or query you wish to geocode. Optional,
if ``components`` param is set.
:param dict components: A structured query. Can be used along with
the free-text ``query``. Should be a dictionary whose keys
are one of:
`country`, `state`, `county`, `city`, `district`, `street`,
`houseNumber`, `postalCode`.
:param at: The center of the search context.
:type at: :class:`geopy.point.Point`, list or tuple of ``(latitude,
longitude)``, or string as ``"%(latitude)s, %(longitude)s"``.
:param list countries: A list of country codes specified in
`ISO 3166-1 alpha-3 <https://en.wikipedia.org/wiki/ISO_3166-1_alpha-3>`_
format, e.g. ``['USA', 'CAN']``.
This is a hard filter.
:param str language: Affects the language of the response,
must be a BCP 47 compliant language code, e.g. ``en-US``.
:param int limit: Defines the maximum number of items in the
response structure. If not provided and there are multiple results
the HERE API will return 20 results by default. This will be reset
to one if ``exactly_one`` is True.
:param bool exactly_one: Return one result or a list of results, if
available.
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception. Set this only if you wish to override, on this call
only, the value set during the geocoder's initialization.
:rtype: ``None``, :class:`geopy.location.Location` or a list of them, if
``exactly_one=False``.
"""
params = {
'apiKey': self.apikey,
}

if query:
params['q'] = query

if components:
parts = [
"{}={}".format(key, val)
for key, val
in components.items()
if key in self.structured_query_params
]
if not parts:
raise GeocoderQueryError("`components` dict must not be empty")
for pair in parts:
if ';' in pair:
raise GeocoderQueryError(
"';' must not be used in values of the structured query. "
"Offending pair: {!r}".format(pair)
)
params['qq'] = ';'.join(parts)

if at:
point = self._coerce_point_to_string(at, output_format="%(lat)s,%(lon)s")
params['at'] = point

if countries:
params['in'] = 'countryCode:' + ','.join(countries)

if language:
params['lang'] = language

if limit:
params['limit'] = limit
if exactly_one:
params['limit'] = 1

url = "?".join((self.api, urlencode(params)))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
callback = partial(self._parse_json, exactly_one=exactly_one)
return self._call_geocoder(url, callback, timeout=timeout)

def reverse(
self,
query,
*,
language=None,
limit=None,
exactly_one=True,
timeout=DEFAULT_SENTINEL
):
"""
Return an address by location point.
:param query: The coordinates for which you wish to obtain the
closest human-readable addresses.
:type query: :class:`geopy.point.Point`, list or tuple of ``(latitude,
longitude)``, or string as ``"%(latitude)s, %(longitude)s"``.
:param str language: Affects the language of the response,
must be a BCP 47 compliant language code, e.g. ``en-US``.
:param int limit: Maximum number of results to be returned.
This will be reset to one if ``exactly_one`` is True.
:param bool exactly_one: Return one result or a list of results, if
available.
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception. Set this only if you wish to override, on this call
only, the value set during the geocoder's initialization.
:rtype: ``None``, :class:`geopy.location.Location` or a list of them, if
``exactly_one=False``.
"""

params = {
'at': self._coerce_point_to_string(query, output_format="%(lat)s,%(lon)s"),
'apiKey': self.apikey,
}

if language:
params['lang'] = language

if limit:
params['limit'] = limit
if exactly_one:
params['limit'] = 1

url = "%s?%s" % (self.reverse_api, urlencode(params))
logger.debug("%s.reverse: %s", self.__class__.__name__, url)
callback = partial(self._parse_json, exactly_one=exactly_one)
return self._call_geocoder(url, callback, timeout=timeout)

def _parse_json(self, doc, exactly_one=True):
resources = doc['items']
if not resources:
return None

def parse_resource(resource):
"""
Parse each return object.
"""
location = resource['title']
position = resource['position']

latitude, longitude = position['lat'], position['lng']

return Location(location, (latitude, longitude), resource)

if exactly_one:
return parse_resource(resources[0])
else:
return [parse_resource(resource) for resource in resources]

def _geocoder_exception_handler(self, error):
if not isinstance(error, AdapterHTTPError):
return
if error.status_code is None or error.text is None:
return
try:
body = json.loads(error.text)
except ValueError:
message = error.text
else:
# `title`: https://developer.here.com/documentation/geocoding-search-api/api-reference-swagger.html # noqa
# `error_description`: returned for queries without apiKey.
message = body.get('title') or body.get('error_description') or error.text
exc_cls = ERROR_CODE_MAP.get(error.status_code, GeocoderServiceError)
raise exc_cls(message) from error

0 comments on commit 85b2352

Please sign in to comment.