530 changes: 530 additions & 0 deletions geopy/distance.py
View file

Large diffs are not rendered by default.

75 changes: 75 additions & 0 deletions geopy/exc.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
"""
Exceptions raised by geopy.
"""

class GeopyError(Exception):
"""
Geopy-specific exceptions are all inherited from GeopyError.
"""

class ConfigurationError(GeopyError):
"""
When instantiating a geocoder, the arguments given were invalid. See
the documentation of each geocoder's `__init__` for more details.
"""

class GeocoderServiceError(GeopyError):
"""
There was an exception caused when calling the remote geocoding service,
and no more specific exception could be raised by geopy. When calling
geocoders' `geocode` or `reverse` methods, this is the most general
exception that can be raised, and any non-geopy exception will be caught
and turned into this. The exception's message will be that of the
original exception.
"""

class GeocoderQueryError(GeocoderServiceError):
"""
Either geopy detected input that would cause a request to fail,
or a request was made and the remote geocoding service responded
that the request was bad.
"""

class GeocoderQuotaExceeded(GeocoderServiceError):
"""
The remote geocoding service refused to fulfill the request
because the client has used its quota.
"""

class GeocoderAuthenticationFailure(GeocoderServiceError):
"""
The remote geocoding service rejects the API key or account
credentials this geocoder was instantiated with.
"""

class GeocoderInsufficientPrivileges(GeocoderServiceError):
"""
The remote geocoding service refused to fulfill a request using the
account credentials given.
"""

class GeocoderTimedOut(GeocoderServiceError):
"""
The call to the geocoding service was aborted because no response
was receiving within the `timeout` argument of either the geocoding class
or, if specified, the method call. Some services are just consistently
slow, and a higher timeout may be needed to use them.
"""

class GeocoderUnavailable(GeocoderServiceError):
"""
Either it was not possible to establish a connection to the remote
geocoding service, or the service responded with a code indicating
it was unavailable.
"""

class GeocoderParseError(GeocoderServiceError):
"""
Geopy could not parse the service's response. This is a bug in geopy.
"""

class GeocoderNotFound(GeopyError):
"""
Caller requested the geocoder matching a string, e.g.,
"google" > GoogleV3, but no geocoder could be found.
"""
126 changes: 126 additions & 0 deletions geopy/format.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
"""
Formatting...
"""

from geopy import units
from geopy.compat import py3k

if py3k:
unichr = chr # pylint: disable=W0622

# Unicode characters for symbols that appear in coordinate strings.
DEGREE = unichr(176)
PRIME = unichr(8242)
DOUBLE_PRIME = unichr(8243)
ASCII_DEGREE = ''
ASCII_PRIME = "'"
ASCII_DOUBLE_PRIME = '"'
LATIN1_DEGREE = chr(176)
HTML_DEGREE = '°'
HTML_PRIME = '′'
HTML_DOUBLE_PRIME = '″'
XML_DECIMAL_DEGREE = '°'
XML_DECIMAL_PRIME = '′'
XML_DECIMAL_DOUBLE_PRIME = '″'
XML_HEX_DEGREE = '&xB0;'
XML_HEX_PRIME = '&x2032;'
XML_HEX_DOUBLE_PRIME = '&x2033;'
ABBR_DEGREE = 'deg'
ABBR_ARCMIN = 'arcmin'
ABBR_ARCSEC = 'arcsec'

DEGREES_FORMAT = (
"%(degrees)d%(deg)s %(minutes)d%(arcmin)s %(seconds)g%(arcsec)s"
)

UNICODE_SYMBOLS = {
'deg': DEGREE,
'arcmin': PRIME,
'arcsec': DOUBLE_PRIME
}
ASCII_SYMBOLS = {
'deg': ASCII_DEGREE,
'arcmin': ASCII_PRIME,
'arcsec': ASCII_DOUBLE_PRIME
}
LATIN1_SYMBOLS = {
'deg': LATIN1_DEGREE,
'arcmin': ASCII_PRIME,
'arcsec': ASCII_DOUBLE_PRIME
}
HTML_SYMBOLS = {
'deg': HTML_DEGREE,
'arcmin': HTML_PRIME,
'arcsec': HTML_DOUBLE_PRIME
}
XML_SYMBOLS = {
'deg': XML_DECIMAL_DEGREE,
'arcmin': XML_DECIMAL_PRIME,
'arcsec': XML_DECIMAL_DOUBLE_PRIME
}
ABBR_SYMBOLS = {
'deg': ABBR_DEGREE,
'arcmin': ABBR_ARCMIN,
'arcsec': ABBR_ARCSEC
}

def format_degrees(degrees, fmt=DEGREES_FORMAT, symbols=None):
"""
TODO docs.
"""
symbols = symbols or ASCII_SYMBOLS
arcminutes = units.arcminutes(degrees=degrees - int(degrees))
arcseconds = units.arcseconds(arcminutes=arcminutes - int(arcminutes))
format_dict = dict(
symbols,
degrees=degrees,
minutes=abs(arcminutes),
seconds=abs(arcseconds)
)
return fmt % format_dict

DISTANCE_FORMAT = "%(magnitude)s%(unit)s"
DISTANCE_UNITS = {
'km': lambda d: d,
'm': lambda d: units.meters(kilometers=d),
'mi': lambda d: units.miles(kilometers=d),
'ft': lambda d: units.feet(kilometers=d),
'nm': lambda d: units.nautical(kilometers=d),
'nmi': lambda d: units.nautical(kilometers=d)
}

def format_distance(kilometers, fmt=DISTANCE_FORMAT, unit='km'):
"""
TODO docs.
"""
magnitude = DISTANCE_UNITS[unit](kilometers)
return fmt % {'magnitude': magnitude, 'unit': unit}

_DIRECTIONS = [
('north', 'N'),
('north by east', 'NbE'),
('north-northeast', 'NNE'),
('northeast by north', 'NEbN'),
('northeast', 'NE'),
('northeast by east', 'NEbE'),
('east-northeast', 'ENE'),
('east by north', 'EbN'),
('east', 'E'),
('east by south', 'EbS'),
('east-southeast', 'ESE'),
('southeast by east', 'SEbE'),
('southeast', 'SE'),
('southeast by south', 'SEbS'),
]

DIRECTIONS, DIRECTIONS_ABBR = zip(*_DIRECTIONS)
ANGLE_DIRECTIONS = {
n * 11.25: d
for n, d
in enumerate(DIRECTIONS)
}
ANGLE_DIRECTIONS_ABBR = {
n * 11.25: d
for n, d
in enumerate(DIRECTIONS_ABBR)
}
163 changes: 163 additions & 0 deletions geopy/geocoders/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,163 @@
"""
Each geolocation service you might use, such as Google Maps, Bing Maps, or
Yahoo BOSS, has its own class in ``geopy.geocoders`` abstracting the service's
API. Geocoders each define at least a ``geocode`` method, for resolving a
location from a string, and may define a ``reverse`` method, which resolves a
pair of coordinates to an address. Each Geocoder accepts any credentials
or settings needed to interact with its service, e.g., an API key or
locale, during its initialization.
To geolocate a query to an address and coordinates:
>>> from geopy.geocoders import Nominatim
>>> geolocator = Nominatim()
>>> location = geolocator.geocode("175 5th Avenue NYC")
>>> print(location.address)
Flatiron Building, 175, 5th Avenue, Flatiron, New York, NYC, New York, ...
>>> print((location.latitude, location.longitude))
(40.7410861, -73.9896297241625)
>>> print(location.raw)
{'place_id': '9167009604', 'type': 'attraction', ...}
To find the address corresponding to a set of coordinates:
>>> from geopy.geocoders import Nominatim
>>> geolocator = Nominatim()
>>> location = geolocator.reverse("52.509669, 13.376294")
>>> print(location.address)
Potsdamer Platz, Mitte, Berlin, 10117, Deutschland, European Union
>>> print((location.latitude, location.longitude))
(52.5094982, 13.3765983)
>>> print(location.raw)
{'place_id': '654513', 'osm_type': 'node', ...}
Locators' ``geolocate`` and ``reverse`` methods require the argument ``query``,
and also accept at least the argument ``exactly_one``, which is ``True``.
Geocoders may have additional attributes, e.g., Bing accepts ``user_location``,
the effect of which is to bias results near that location. ``geolocate``
and ``reverse`` methods may return three types of values:
- When there are no results found, returns ``None``.
- When the method's ``exactly_one`` argument is ``True`` and at least one
result is found, returns a :class:`geopy.location.Location` object, which
can be iterated over as:
(address<String>, (latitude<Float>, longitude<Float>))
Or can be accessed as `Location.address`, `Location.latitude`,
`Location.longitude`, `Location.altitude`, and `Location.raw`. The
last contains the geocoder's unparsed response for this result.
- When ``exactly_one`` is False, and there is at least one result, returns a
list of :class:`geopy.location.Location` objects, as above:
[Location, [...]]
If a service is unavailable or otherwise returns a non-OK response, or doesn't
receive a response in the allotted timeout, you will receive one of the
`Exceptions`_ detailed below.
Every geocoder accepts an argument ``format_string`` that defaults to '%s'
where the input string to geocode is interpolated. For example, if you only
need to geocode locations in Cleveland, Ohio, you could do::
>>> from geopy.geocoders import GeocoderDotUS
>>> geolocator = GeocoderDotUS(format_string="%s, Cleveland OH")
>>> address, (latitude, longitude) = geolocator.geocode("11111 Euclid Ave")
>>> print(address, latitude, longitude)
11111 Euclid Ave, Cleveland, OH 44106 41.506784 -81.608148
"""

__all__ = (
"get_geocoder_for_service",
"ArcGIS",
"Baidu",
"Bing",
"DataBC",
"GeocoderDotUS",
"GeocodeFarm",
"GeoNames",
"GoogleV3",
"IGNFrance",
"OpenCage",
"OpenMapQuest",
"NaviData",
"Nominatim",
"YahooPlaceFinder",
"LiveAddress",
'Yandex',
"What3Words",
"Photon",
)


from geopy.geocoders.arcgis import ArcGIS
from geopy.geocoders.baidu import Baidu
from geopy.geocoders.bing import Bing
from geopy.geocoders.databc import DataBC
from geopy.geocoders.dot_us import GeocoderDotUS
from geopy.geocoders.geocodefarm import GeocodeFarm
from geopy.geocoders.geonames import GeoNames
from geopy.geocoders.googlev3 import GoogleV3
from geopy.geocoders.opencage import OpenCage
from geopy.geocoders.openmapquest import OpenMapQuest
from geopy.geocoders.navidata import NaviData
from geopy.geocoders.osm import Nominatim
from geopy.geocoders.placefinder import YahooPlaceFinder
from geopy.geocoders.smartystreets import LiveAddress
from geopy.geocoders.what3words import What3Words
from geopy.geocoders.yandex import Yandex
from geopy.geocoders.ignfrance import IGNFrance
from geopy.geocoders.photon import Photon


from geopy.exc import GeocoderNotFound


SERVICE_TO_GEOCODER = {
"arcgis": ArcGIS,
"baidu": Baidu,
"bing": Bing,
"databc": DataBC,
"google": GoogleV3,
"googlev3": GoogleV3,
"geocoderdotus": GeocoderDotUS,
"geonames": GeoNames,
"yahoo": YahooPlaceFinder,
"placefinder": YahooPlaceFinder,
"opencage": OpenCage,
"openmapquest": OpenMapQuest,
"liveaddress": LiveAddress,
"navidata": NaviData,
"nominatim": Nominatim,
"geocodefarm": GeocodeFarm,
"what3words": What3Words,
"yandex": Yandex,
"ignfrance": IGNFrance,
"photon": Photon
}


def get_geocoder_for_service(service):
"""
For the service provided, try to return a geocoder class.
>>> from geopy.geocoders import get_geocoder_for_service
>>> get_geocoder_for_service("nominatim")
geopy.geocoders.osm.Nominatim
If the string given is not recognized, a
:class:`geopy.exc.GeocoderNotFound` exception is raised.
"""
try:
return SERVICE_TO_GEOCODER[service.lower()]
except KeyError:
raise GeocoderNotFound(
"Unknown geocoder '%s'; options are: %s" %
(service, SERVICE_TO_GEOCODER.keys())
)

249 changes: 249 additions & 0 deletions geopy/geocoders/arcgis.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,249 @@
"""
:class:`.ArcGIS` geocoder.
"""

import json
from time import time
from geopy.compat import urlencode, Request

from geopy.geocoders.base import Geocoder, DEFAULT_SCHEME, DEFAULT_TIMEOUT, \
DEFAULT_WKID
from geopy.exc import GeocoderServiceError, GeocoderAuthenticationFailure
from geopy.exc import ConfigurationError
from geopy.location import Location
from geopy.util import logger


__all__ = ("ArcGIS", )


class ArcGIS(Geocoder): # pylint: disable=R0921,R0902,W0223
"""
Geocoder using the ERSI ArcGIS API. Documentation at:
https://developers.arcgis.com/rest/geocode/api-reference/overview-world-geocoding-service.htm
"""

_TOKEN_EXPIRED = 498
_MAX_RETRIES = 3
auth_api = 'https://www.arcgis.com/sharing/generateToken'

def __init__(self, username=None, password=None, referer=None, # pylint: disable=R0913
token_lifetime=60, scheme=DEFAULT_SCHEME,
timeout=DEFAULT_TIMEOUT, proxies=None,
user_agent=None):
"""
Create a ArcGIS-based geocoder.
.. versionadded:: 0.97
:param string username: ArcGIS username. Required if authenticated
mode is desired.
:param string password: ArcGIS password. Required if authenticated
mode is desired.
:param string referer: Required if authenticated mode is desired.
'Referer' HTTP header to send with each request,
e.g., 'http://www.example.com'. This is tied to an issued token,
so fielding queries for multiple referrers should be handled by
having multiple ArcGIS geocoder instances.
:param int token_lifetime: Desired lifetime, in minutes, of an
ArcGIS-issued token.
:param string scheme: Desired scheme. If authenticated mode is in use,
it must be 'https'.
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(ArcGIS, self).__init__(
scheme=scheme, timeout=timeout, proxies=proxies, user_agent=user_agent
)
if username or password or referer:
if not (username and password and referer):
raise ConfigurationError(
"Authenticated mode requires username,"
" password, and referer"
)
if self.scheme != 'https':
raise ConfigurationError(
"Authenticated mode requires scheme of 'https'"
)
self._base_call_geocoder = self._call_geocoder
self._call_geocoder = self._authenticated_call_geocoder

self.username = username
self.password = password
self.referer = referer

self.token = None
self.token_lifetime = token_lifetime * 60 # store in seconds
self.token_expiry = None
self.retry = 1

self.api = (
'%s://geocode.arcgis.com/arcgis/rest/services/'
'World/GeocodeServer/find' % self.scheme
)
self.reverse_api = (
'%s://geocode.arcgis.com/arcgis/rest/services/'
'World/GeocodeServer/reverseGeocode' % self.scheme
)

def _authenticated_call_geocoder(self, url, timeout=None):
"""
Wrap self._call_geocoder, handling tokens.
"""
if self.token is None or int(time()) > self.token_expiry:
self._refresh_authentication_token()
request = Request(
"&token=".join((url, self.token)), # no urlencoding
headers={"Referer": self.referer}
)
return self._base_call_geocoder(request, timeout=timeout)

def geocode(self, query, exactly_one=True, timeout=None):
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
"""
params = {'text': query, 'f': 'json'}
if exactly_one is True:
params['maxLocations'] = 1
url = "?".join((self.api, urlencode(params)))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
response = self._call_geocoder(url, timeout=timeout)

# Handle any errors; recursing in the case of an expired token.
if 'error' in response:
if response['error']['code'] == self._TOKEN_EXPIRED:
self.retry += 1
self._refresh_authentication_token()
return self.geocode(
query, exactly_one=exactly_one, timeout=timeout
)
raise GeocoderServiceError(str(response['error']))

# Success; convert from the ArcGIS JSON format.
if not len(response['locations']):
return None
geocoded = []
for resource in response['locations']:
geometry = resource['feature']['geometry']
geocoded.append(
Location(
resource['name'], (geometry['y'], geometry['x']), resource
)
)
if exactly_one is True:
return geocoded[0]
return geocoded

def reverse(self, query, exactly_one=True, timeout=None, # pylint: disable=R0913,W0221
distance=None, wkid=DEFAULT_WKID):
"""
Given a point, find an address.
: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 bool exactly_one: Return one result, or a list?
: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.
:param int distance: Distance from the query location, in meters,
within which to search. ArcGIS has a default of 100 meters, if not
specified.
:param string wkid: WKID to use for both input and output coordinates.
"""
# ArcGIS is lon,lat; maintain lat,lon convention of geopy
point = self._coerce_point_to_string(query).split(",")
if wkid != DEFAULT_WKID:
location = {"x": point[1], "y": point[0], "spatialReference": wkid}
else:
location = ",".join((point[1], point[0]))
params = {'location': location, 'f': 'json', 'outSR': wkid}
if distance is not None:
params['distance'] = distance
url = "?".join((self.reverse_api, urlencode(params)))
logger.debug("%s.reverse: %s", self.__class__.__name__, url)
response = self._call_geocoder(url, timeout=timeout)
if not len(response):
return None
if 'error' in response:
if response['error']['code'] == self._TOKEN_EXPIRED:
self.retry += 1
self._refresh_authentication_token()
return self.reverse(query, exactly_one=exactly_one,
timeout=timeout, distance=distance,
wkid=wkid)
raise GeocoderServiceError(str(response['error']))
address = (
"%(Address)s, %(City)s, %(Region)s %(Postal)s,"
" %(CountryCode)s" % response['address']
)
return Location(
address,
(response['location']['y'], response['location']['x']),
response['address']
)

def _refresh_authentication_token(self):
"""
POST to ArcGIS requesting a new token.
"""
if self.retry == self._MAX_RETRIES:
raise GeocoderAuthenticationFailure(
'Too many retries for auth: %s' % self.retry
)
token_request_arguments = {
'username': self.username,
'password': self.password,
'expiration': self.token_lifetime,
'f': 'json'
}
token_request_arguments = "&".join([
"%s=%s" % (key, val)
for key, val
in token_request_arguments.items()
])
url = "&".join((
"?".join((self.auth_api, token_request_arguments)),
urlencode({'referer': self.referer})
))
logger.debug(
"%s._refresh_authentication_token: %s",
self.__class__.__name__, url
)
self.token_expiry = int(time()) + self.token_lifetime
response = self._base_call_geocoder(url)
if not 'token' in response:
raise GeocoderAuthenticationFailure(
'Missing token in auth request.'
'Request URL: %s; response JSON: %s' %
(url, json.dumps(response))
)
self.retry = 0
self.token = response['token']
215 changes: 215 additions & 0 deletions geopy/geocoders/baidu.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,215 @@
"""
:class:`.Baidu` is the Baidu Maps geocoder.
"""

from geopy.compat import urlencode
from geopy.geocoders.base import Geocoder, DEFAULT_TIMEOUT
from geopy.exc import (
GeocoderQueryError,
GeocoderQuotaExceeded,
GeocoderAuthenticationFailure,
)
from geopy.location import Location
from geopy.util import logger


__all__ = ("Baidu", )


class Baidu(Geocoder):
"""
Geocoder using the Baidu Maps v2 API. Documentation at:
http://developer.baidu.com/map/webservice-geocoding.htm
"""

def __init__(
self,
api_key,
scheme='http',
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None
):
"""
Initialize a customized Baidu geocoder using the v2 API.
.. versionadded:: 1.0.0
:param string api_key: The API key required by Baidu Map to perform
geocoding requests. API keys are managed through the Baidu APIs
console (http://lbsyun.baidu.com/apiconsole/key).
:param string scheme: Use 'https' or 'http' as the API URL's scheme.
Default is http and only http support.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(Baidu, self).__init__(
scheme=scheme, timeout=timeout, proxies=proxies, user_agent=user_agent
)
self.api_key = api_key
self.scheme = scheme
self.doc = {}
self.api = 'http://api.map.baidu.com/geocoder/v2/'


@staticmethod
def _format_components_param(components):
"""
Format the components dict to something Baidu understands.
"""
return "|".join(
(":".join(item)
for item in components.items()
)
)

def geocode(
self,
query,
exactly_one=True,
timeout=None
):
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
"""
params = {
'ak': self.api_key,
'output': 'json',
'address': self.format_string % query,
}

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

def reverse(self, query, timeout=None): # pylint: disable=W0221
"""
Given a point, find an address.
: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 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.
"""
params = {
'ak': self.api_key,
'output': 'json',
'location': self._coerce_point_to_string(query),
}

url = "?".join((self.api, urlencode(params)))

logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_reverse_json(
self._call_geocoder(url, timeout=timeout)
)


@staticmethod
def _parse_reverse_json(page):
"""
Parses a location from a single-result reverse API call.
"""
place = page.get('result')

location = place.get('formatted_address').encode('utf-8')
latitude = place['location']['lat']
longitude = place['location']['lng']

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


def _parse_json(self, page, exactly_one=True):
"""
Returns location, (latitude, longitude) from JSON feed.
"""

place = page.get('result', None)

if not place:
self._check_status(page.get('status'))
return None

def parse_place(place):
"""
Get the location, lat, lng from a single JSON place.
"""
location = place.get('level')
latitude = place['location']['lat']
longitude = place['location']['lng']
return Location(location, (latitude, longitude), place)

if exactly_one:
return parse_place(place)
else:
return [parse_place(item) for item in place]

@staticmethod
def _check_status(status):
"""
Validates error statuses.
"""
if status == '0':
# When there are no results, just return.
return
if status == '1':
raise GeocoderQueryError(
'Internal server error.'
)
elif status == '2':
raise GeocoderQueryError(
'Invalid request.'
)
elif status == '3':
raise GeocoderAuthenticationFailure(
'Authentication failure.'
)
elif status == '4':
raise GeocoderQuotaExceeded(
'Quota validate failure.'
)
elif status == '5':
raise GeocoderQueryError(
'AK Illegal or Not Exist.'
)
elif status == '101':
raise GeocoderQueryError(
'Your request was denied.'
)
elif status == '102':
raise GeocoderQueryError(
'IP/SN/SCODE/REFERER Illegal:'
)
elif status == '2xx':
raise GeocoderQueryError(
'Has No Privilleges.'
)
elif status == '3xx':
raise GeocoderQuotaExceeded(
'Quota Error.'
)
else:
raise GeocoderQueryError('Unknown error')
207 changes: 207 additions & 0 deletions geopy/geocoders/base.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,207 @@
"""
:class:`.GeoCoder` base object from which other geocoders are templated.
"""

from ssl import SSLError
from socket import timeout as SocketTimeout
import json

from geopy.compat import (
string_compare,
HTTPError,
py3k,
urlopen as urllib_urlopen,
build_opener,
ProxyHandler,
URLError,
install_opener,
Request,
)
from geopy.point import Point
from geopy.exc import (
GeocoderServiceError,
ConfigurationError,
GeocoderTimedOut,
GeocoderAuthenticationFailure,
GeocoderQuotaExceeded,
GeocoderQueryError,
GeocoderInsufficientPrivileges,
GeocoderUnavailable,
GeocoderParseError,
)
from geopy.util import decode_page, __version__


__all__ = (
"Geocoder",
"DEFAULT_FORMAT_STRING",
"DEFAULT_SCHEME",
"DEFAULT_TIMEOUT",
"DEFAULT_WKID",
)


DEFAULT_FORMAT_STRING = '%s'
DEFAULT_SCHEME = 'https'
DEFAULT_TIMEOUT = 1
DEFAULT_WKID = 4326
DEFAULT_USER_AGENT = "geopy/%s" % __version__


ERROR_CODE_MAP = {
400: GeocoderQueryError,
401: GeocoderAuthenticationFailure,
402: GeocoderQuotaExceeded,
403: GeocoderInsufficientPrivileges,
407: GeocoderAuthenticationFailure,
412: GeocoderQueryError,
413: GeocoderQueryError,
414: GeocoderQueryError,
502: GeocoderServiceError,
503: GeocoderTimedOut,
504: GeocoderTimedOut
}


class Geocoder(object): # pylint: disable=R0921
"""
Template object for geocoders.
"""

def __init__(
self,
format_string=DEFAULT_FORMAT_STRING,
scheme=DEFAULT_SCHEME,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None
):
"""
Mostly-common geocoder validation, proxies, &c. Not all geocoders
specify format_string and such.
"""
self.format_string = format_string
self.scheme = scheme
if self.scheme not in ('http', 'https'): # pragma: no cover
raise ConfigurationError(
'Supported schemes are `http` and `https`.'
)
self.proxies = proxies
self.timeout = timeout
self.headers = {'User-Agent': user_agent or DEFAULT_USER_AGENT}

if self.proxies:
install_opener(
build_opener(
ProxyHandler(self.proxies)
)
)
self.urlopen = urllib_urlopen

@staticmethod
def _coerce_point_to_string(point):
"""
Do the right thing on "point" input. For geocoders with reverse
methods.
"""
if isinstance(point, Point):
return ",".join((str(point.latitude), str(point.longitude)))
elif isinstance(point, (list, tuple)):
return ",".join((str(point[0]), str(point[1]))) # -altitude
elif isinstance(point, string_compare):
return point
else: # pragma: no cover
raise ValueError("Invalid point")

def _parse_json(self, page, exactly_one): # pragma: no cover
"""
Template for subclasses
"""
raise NotImplementedError()

def _call_geocoder(
self,
url,
timeout=None,
raw=False,
requester=None,
deserializer=json.loads,
**kwargs
):
"""
For a generated query URL, get the results.
"""
requester = requester or self.urlopen

if not requester:
req = Request(url=url, headers=self.headers)
else:
# work around for placefinder's use of requests
req = url

try:
page = requester(req, timeout=(timeout or self.timeout), **kwargs)
except Exception as error: # pylint: disable=W0703
message = (
str(error) if not py3k
else (
str(error.args[0])
if len(error.args)
else str(error)
)
)
if hasattr(self, '_geocoder_exception_handler'):
self._geocoder_exception_handler(error, message) # pylint: disable=E1101
if isinstance(error, HTTPError):
code = error.getcode()
try:
raise ERROR_CODE_MAP[code](message)
except KeyError:
raise GeocoderServiceError(message)
elif isinstance(error, URLError):
if "timed out" in message:
raise GeocoderTimedOut('Service timed out')
elif "unreachable" in message:
raise GeocoderUnavailable('Service not available')
elif isinstance(error, SocketTimeout):
raise GeocoderTimedOut('Service timed out')
elif isinstance(error, SSLError):
if "timed out" in message:
raise GeocoderTimedOut('Service timed out')
raise GeocoderServiceError(message)

if hasattr(page, 'getcode'):
status_code = page.getcode()
elif hasattr(page, 'status_code'):
status_code = page.status_code
else:
status_code = None
if status_code in ERROR_CODE_MAP:
raise ERROR_CODE_MAP[page.status_code]("\n%s" % decode_page(page))

if raw:
return page

page = decode_page(page)

if deserializer is not None:
try:
return deserializer(page)
except ValueError:
raise GeocoderParseError(
"Could not deserialize using deserializer:\n%s" % page
)
else:
return page

def geocode(self, query, exactly_one=True, timeout=None):
"""
Implemented in subclasses.
"""
raise NotImplementedError()

def reverse(self, query, exactly_one=True, timeout=None):
"""
Implemented in subclasses.
"""
raise NotImplementedError()
243 changes: 243 additions & 0 deletions geopy/geocoders/bing.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,243 @@
"""
:class:`.Bing` geocoder.
"""

from geopy.compat import urlencode
from geopy.geocoders.base import Geocoder, DEFAULT_FORMAT_STRING, \
DEFAULT_TIMEOUT, DEFAULT_SCHEME
from geopy.location import Location
from geopy.exc import (
GeocoderAuthenticationFailure,
GeocoderQuotaExceeded,
GeocoderInsufficientPrivileges,
GeocoderUnavailable,
GeocoderServiceError,
)
from geopy.util import logger, join_filter


__all__ = ("Bing", )


class Bing(Geocoder):
"""
Geocoder using the Bing Maps Locations API. Documentation at:
https://msdn.microsoft.com/en-us/library/ff701715.aspx
"""

structured_query_params = {
'addressLine',
'locality',
'adminDistrict',
'countryRegion',
'postalCode',
}

def __init__(
self,
api_key,
format_string=DEFAULT_FORMAT_STRING,
scheme=DEFAULT_SCHEME,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""Initialize a customized Bing geocoder with location-specific
address information and your Bing Maps API key.
:param string api_key: Should be a valid Bing Maps API key.
:param string format_string: String containing '%s' where the
string to geocode should be interpolated before querying the
geocoder. For example: '%s, Mountain View, CA'. The default
is just '%s'.
:param string scheme: Use 'https' or 'http' as the API URL's scheme.
Default is https. Note that SSL connections' certificates are not
verified.
.. versionadded:: 0.97
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception.
.. versionadded:: 0.97
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
.. versionadded:: 0.96
"""
super(Bing, self).__init__(format_string, scheme, timeout, proxies, user_agent=user_agent)
self.api_key = api_key
self.api = "%s://dev.virtualearth.net/REST/v1/Locations" % self.scheme

def geocode(
self,
query,
exactly_one=True,
user_location=None,
timeout=None,
culture=None,
include_neighborhood=None,
include_country_code=False
): # pylint: disable=W0221
"""
Geocode an address.
:param string query: The address or query you wish to geocode.
For a structured query, provide a dictionary whose keys
are one of: `addressLine`, `locality` (city), `adminDistrict` (state), `countryRegion`, or
`postalcode`.
:param bool exactly_one: Return one result or a list of results, if
available.
:param user_location: Prioritize results closer to
this location.
.. versionadded:: 0.96
:type user_location: :class:`geopy.point.Point`
: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.
.. versionadded:: 0.97
:param string culture: Affects the language of the response,
must be a two-letter country code.
.. versionadded:: 1.4.0
:param boolean include_neighborhood: Sets whether to include the
neighborhood field in the response.
.. versionadded:: 1.4.0
:param boolean include_country_code: Sets whether to include the
two-letter ISO code of the country in the response (field name
'countryRegionIso2').
.. versionadded:: 1.4.0
"""
if isinstance(query, dict):
params = {
key: val
for key, val
in query.items()
if key in self.structured_query_params
}
params['key'] = self.api_key
else:
params = {
'query': self.format_string % query,
'key': self.api_key
}
if user_location:
params['userLocation'] = ",".join(
(str(user_location.latitude), str(user_location.longitude))
)
if exactly_one is True:
params['maxResults'] = 1
if culture:
params['culture'] = culture
if include_neighborhood is not None:
params['includeNeighborhood'] = include_neighborhood
if include_country_code:
params['include'] = 'ciso2' # the only acceptable value

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

def reverse(self, query, exactly_one=True, timeout=None):
"""
Reverse geocode a 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 bool exactly_one: Return one result, or a list?
: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.
.. versionadded:: 0.97
"""
point = self._coerce_point_to_string(query)
params = {'key': self.api_key}
url = "%s/%s?%s" % (
self.api, point, urlencode(params))

logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout),
exactly_one
)

@staticmethod
def _parse_json(doc, exactly_one=True): # pylint: disable=W0221
"""
Parse a location name, latitude, and longitude from an JSON response.
"""
status_code = doc.get("statusCode", 200)
if status_code != 200:
err = doc.get("errorDetails", "")
if status_code == 401:
raise GeocoderAuthenticationFailure(err)
elif status_code == 403:
raise GeocoderInsufficientPrivileges(err)
elif status_code == 429:
raise GeocoderQuotaExceeded(err)
elif status_code == 503:
raise GeocoderUnavailable(err)
else:
raise GeocoderServiceError(err)

resources = doc['resourceSets'][0]['resources']
if resources is None or not len(resources): # pragma: no cover
return None

def parse_resource(resource):
"""
Parse each return object.
"""
stripchars = ", \n"
addr = resource['address']

address = addr.get('addressLine', '').strip(stripchars)
city = addr.get('locality', '').strip(stripchars)
state = addr.get('adminDistrict', '').strip(stripchars)
zipcode = addr.get('postalCode', '').strip(stripchars)
country = addr.get('countryRegion', '').strip(stripchars)

city_state = join_filter(", ", [city, state])
place = join_filter(" ", [city_state, zipcode])
location = join_filter(", ", [address, place, country])

latitude = resource['point']['coordinates'][0] or None
longitude = resource['point']['coordinates'][1] or None
if latitude and longitude:
latitude = float(latitude)
longitude = float(longitude)

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

if exactly_one:
return parse_resource(resources[0])
else:
return [parse_resource(resource) for resource in resources]
115 changes: 115 additions & 0 deletions geopy/geocoders/databc.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
"""
:class:`.DataBC` geocoder.
"""

from geopy.compat import urlencode

from geopy.geocoders.base import Geocoder, DEFAULT_SCHEME, DEFAULT_TIMEOUT
from geopy.exc import GeocoderQueryError
from geopy.location import Location
from geopy.util import logger


__all__ = ("DataBC", )


class DataBC(Geocoder):
"""
Geocoder using the Physical Address Geocoder from DataBC. Documentation at:
http://www.data.gov.bc.ca/dbc/geographic/locate/geocoding.page
"""

def __init__(self, scheme=DEFAULT_SCHEME, timeout=DEFAULT_TIMEOUT, proxies=None, user_agent=None):
"""
Create a DataBC-based geocoder.
:param string scheme: Desired scheme.
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(DataBC, self).__init__(
scheme=scheme, timeout=timeout, proxies=proxies, user_agent=user_agent
)
self.api = '%s://apps.gov.bc.ca/pub/geocoder/addresses.geojson' % self.scheme

def geocode(
self,
query,
max_results=25,
set_back=0,
location_descriptor='any',
exactly_one=True,
timeout=None,
):
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
:param int max_results: The maximum number of resutls to request.
:param float set_back: The distance to move the accessPoint away
from the curb (in meters) and towards the interior of the parcel.
location_descriptor must be set to accessPoint for set_back to
take effect.
:param string location_descriptor: The type of point requested. It
can be any, accessPoint, frontDoorPoint, parcelPoint,
rooftopPoint and routingPoint.
: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.
"""
params = {'addressString': query}
if set_back != 0:
params['setBack'] = set_back
if location_descriptor not in ['any',
'accessPoint',
'frontDoorPoint',
'parcelPoint',
'rooftopPoint',
'routingPoint']:
raise GeocoderQueryError(
"You did not provided a location_descriptor "
"the webservice can consume. It should be any, accessPoint, "
"frontDoorPoint, parcelPoint, rooftopPoint or routingPoint."
)
params['locationDescriptor'] = location_descriptor
if exactly_one is True:
max_results = 1
params['maxResults'] = max_results

url = "?".join((self.api, urlencode(params)))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
response = self._call_geocoder(url, timeout=timeout)

# Success; convert from GeoJSON
if not len(response['features']):
return None
geocoded = []
for feature in response['features']:
geocoded.append(self._parse_feature(feature))
if exactly_one is True:
return geocoded[0]
return geocoded

@staticmethod
def _parse_feature(feature):
properties = feature['properties']
coordinates = feature['geometry']['coordinates']
return Location(
properties['fullAddress'], (coordinates[1], coordinates[0]),
properties
)
159 changes: 159 additions & 0 deletions geopy/geocoders/dot_us.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
"""
:class:`GeocoderDotUS` geocoder.
"""

import csv
from base64 import encodestring
from geopy.compat import urlencode, py3k, Request
from geopy.geocoders.base import (
Geocoder,
DEFAULT_FORMAT_STRING,
DEFAULT_TIMEOUT,
)
from geopy.location import Location
from geopy.exc import ConfigurationError
from geopy.util import logger, join_filter


__all__ = ("GeocoderDotUS", )


class GeocoderDotUS(Geocoder): # pylint: disable=W0223
"""
GeocoderDotUS geocoder, documentation at:
http://geocoder.us/
Note that GeocoderDotUS does not support SSL.
"""

def __init__(
self,
username=None,
password=None,
format_string=DEFAULT_FORMAT_STRING,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""
:param string username:
:param string password:
:param string format_string: String containing '%s' where the
string to geocode should be interpolated before querying the
geocoder. For example: '%s, Mountain View, CA'. The default
is just '%s'.
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising an :class:`geopy.exc.GeocoderTimedOut`
exception.
.. versionadded:: 0.97
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
.. versionadded:: 0.96
"""
super(GeocoderDotUS, self).__init__(
format_string=format_string, timeout=timeout, proxies=proxies, user_agent=user_agent
)
if username or password:
if not (username and password):
raise ConfigurationError(
"Username and password must both specified"
)
self.authenticated = True
self.api = "http://geocoder.us/member/service/namedcsv"
else:
self.authenticated = False
self.api = "http://geocoder.us/service/namedcsv"
self.username = username
self.password = password

def geocode(self, query, exactly_one=True, timeout=None):
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
.. versionadded:: 0.97
"""
query_str = self.format_string % query

url = "?".join((self.api, urlencode({'address':query_str})))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
if self.authenticated is True:
auth = " ".join((
"Basic",
encodestring(":".join((self.username, self.password))\
.encode('utf-8')).strip().decode('utf-8')
))
url = Request(url, headers={"Authorization": auth})
page = self._call_geocoder(url, timeout=timeout, raw=True)
content = page.read().decode("utf-8") if py3k else page.read() # pylint: disable=E1101,E1103
places = [
r for r in csv.reader(
[content, ] if not isinstance(content, list)
else content
)
]
if not len(places):
return None
if exactly_one is True:
return self._parse_result(places[0])
else:
result = [self._parse_result(res) for res in places]
if None in result: # todo
return None
return result

@staticmethod
def _parse_result(result):
"""
Parse individual results. Different, but lazy actually, so... ok.
"""
# turn x=y pairs ("lat=47.6", "long=-117.426")
# into dict key/value pairs:
place = dict(
[x.split('=') for x in result if len(x.split('=')) > 1]
)
if 'error' in place:
if "couldn't find" in place['error']:
return None

address = [
place.get('number', None),
place.get('prefix', None),
place.get('street', None),
place.get('type', None),
place.get('suffix', None)
]
city = place.get('city', None)
state = place.get('state', None)
zip_code = place.get('zip', None)

name = join_filter(", ", [
join_filter(" ", address),
city,
join_filter(" ", [state, zip_code])
])

latitude = place.get('lat', None)
longitude = place.get('long', None)
if latitude and longitude:
latlon = float(latitude), float(longitude)
else:
return None
return Location(name, latlon, place)
172 changes: 172 additions & 0 deletions geopy/geocoders/geocodefarm.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
"""
:class:`.GeocodeFarm` geocoder.
"""

from geopy.geocoders.base import Geocoder, DEFAULT_FORMAT_STRING, \
DEFAULT_TIMEOUT
from geopy.location import Location
from geopy.util import logger
from geopy.exc import GeocoderAuthenticationFailure, GeocoderQuotaExceeded, \
GeocoderServiceError
from geopy.compat import urlencode


__all__ = ("GeocodeFarm", )


class GeocodeFarm(Geocoder):
"""
Geocoder using the GeocodeFarm API. Documentation at:
https://www.geocode.farm/geocoding/free-api-documentation/
"""

def __init__(
self,
api_key=None,
format_string=DEFAULT_FORMAT_STRING,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""
Create a geocoder for GeocodeFarm.
.. versionadded:: 0.99
:param string api_key: The API key required by GeocodeFarm to perform
geocoding requests.
:param string format_string: String containing '%s' where the
string to geocode should be interpolated before querying the
geocoder. For example: '%s, Mountain View, CA'. The default
is just '%s'.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(GeocodeFarm, self).__init__(
format_string, 'https', timeout, proxies, user_agent=user_agent
)
self.api_key = api_key
self.format_string = format_string
self.api = (
"%s://www.geocode.farm/v3/json/forward/" % self.scheme
)
self.reverse_api = (
"%s://www.geocode.farm/v3/json/reverse/" % self.scheme
)

def geocode(self, query, exactly_one=True, timeout=None):
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
"""
params = {
'addr': self.format_string % query,
}
if self.api_key:
params['key'] = self.api_key
url = "?".join((self.api, urlencode(params)))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout), exactly_one
)

def reverse(self, query, exactly_one=True, timeout=None):
"""
Returns a reverse geocoded location.
: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 bool exactly_one: Return one result or a list of results, if
available. GeocodeFarm's API will always return at most one
result.
: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.
"""
try:
lat, lon = [
x.strip() for x in
self._coerce_point_to_string(query).split(',')
]
except ValueError:
raise ValueError("Must be a coordinate pair or Point")
params = {
'lat': lat,
'lon': lon
}
if self.api_key:
params['key'] = self.api_key
url = "?".join((self.reverse_api, urlencode(params)))
logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout), exactly_one
)

@staticmethod
def parse_code(results):
"""
Parse each resource.
"""
places = []
for result in results.get('RESULTS'):
coordinates = result.get('COORDINATES', {})
address = result.get('ADDRESS', {})
latitude = coordinates.get('latitude', None)
longitude = coordinates.get('longitude', None)
placename = address.get('address_returned', None)
if placename is None:
placename = address.get('address', None)
if latitude and longitude:
latitude = float(latitude)
longitude = float(longitude)
places.append(Location(placename, (latitude, longitude), result))
return places

def _parse_json(self, api_result, exactly_one):
if api_result is None:
return None
geocoding_results = api_result["geocoding_results"]
self._check_for_api_errors(geocoding_results)

places = self.parse_code(geocoding_results)
if exactly_one is True:
return places[0]
else:
return places

@staticmethod
def _check_for_api_errors(geocoding_results):
"""
Raise any exceptions if there were problems reported
in the api response.
"""
status_result = geocoding_results.get("STATUS", {})
api_call_success = status_result.get("status", "") == "SUCCESS"
if not api_call_success:
access_error = status_result.get("access")
access_error_to_exception = {
'API_KEY_INVALID': GeocoderAuthenticationFailure,
'OVER_QUERY_LIMIT': GeocoderQuotaExceeded,
}
exception_cls = access_error_to_exception.get(
access_error, GeocoderServiceError
)
raise exception_cls(access_error)
184 changes: 184 additions & 0 deletions geopy/geocoders/geonames.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
"""
:class:`GeoNames` geocoder.
"""

from geopy.compat import urlencode

from geopy.geocoders.base import Geocoder, DEFAULT_TIMEOUT
from geopy.location import Location
from geopy.exc import (
GeocoderInsufficientPrivileges,
GeocoderServiceError,
ConfigurationError
)
from geopy.util import logger


__all__ = ("GeoNames", )


class GeoNames(Geocoder): # pylint: disable=W0223
"""
GeoNames geocoder, documentation at:
http://www.geonames.org/export/geonames-search.html
Reverse geocoding documentation at:
http://www.geonames.org/maps/us-reverse-geocoder.html
"""

def __init__(
self,
country_bias=None,
username=None,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
):
"""
:param string country_bias:
:param string username:
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception.
.. versionadded:: 0.97
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
.. versionadded:: 0.96
"""
super(GeoNames, self).__init__(
scheme='http', timeout=timeout, proxies=proxies, user_agent=user_agent
)
if username == None:
raise ConfigurationError(
'No username given, required for api access. If you do not '
'have a GeoNames username, sign up here: '
'http://www.geonames.org/login'
)
self.username = username
self.country_bias = country_bias
self.api = "%s://api.geonames.org/searchJSON" % self.scheme
self.api_reverse = (
"%s://api.geonames.org/findNearbyPlaceNameJSON" % self.scheme
)

def geocode(self, query, exactly_one=True, timeout=None): # pylint: disable=W0221
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
.. versionadded:: 0.97
"""
params = {
'q': query,
'username': self.username
}
if self.country_bias:
params['countryBias'] = self.country_bias
if exactly_one is True:
params['maxRows'] = 1
url = "?".join((self.api, urlencode(params)))
logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout),
exactly_one,
)

def reverse(
self,
query,
exactly_one=False,
timeout=None,
):
"""
Given a point, find an address.
.. versionadded:: 1.2.0
:param string 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 boolean 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.
"""
try:
lat, lng = [
x.strip() for x in
self._coerce_point_to_string(query).split(',')
]
except ValueError:
raise ValueError("Must be a coordinate pair or Point")
params = {
'lat': lat,
'lng': lng,
'username': self.username
}
url = "?".join((self.api_reverse, urlencode(params)))
logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout),
exactly_one
)

def _parse_json(self, doc, exactly_one):
"""
Parse JSON response body.
"""
places = doc.get('geonames', [])
err = doc.get('status', None)
if err and 'message' in err:
if err['message'].startswith("user account not enabled to use"):
raise GeocoderInsufficientPrivileges(err['message'])
else:
raise GeocoderServiceError(err['message'])
if not len(places):
return None

def parse_code(place):
"""
Parse each record.
"""
latitude = place.get('lat', None)
longitude = place.get('lng', None)
if latitude and longitude:
latitude = float(latitude)
longitude = float(longitude)
else:
return None

placename = place.get('name')
state = place.get('adminCode1', None)
country = place.get('countryCode', None)

location = ', '.join(
[x for x in [placename, state, country] if x]
)

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

if exactly_one:
return parse_code(places[0])
else:
return [parse_code(place) for place in places]
375 changes: 375 additions & 0 deletions geopy/geocoders/googlev3.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,375 @@
"""
:class:`.GoogleV3` is the Google Maps V3 geocoder.
"""

import base64
import hashlib
import hmac
from geopy.compat import urlencode
from geopy.geocoders.base import Geocoder, DEFAULT_TIMEOUT, DEFAULT_SCHEME
from geopy.exc import (
GeocoderQueryError,
GeocoderQuotaExceeded,
ConfigurationError,
GeocoderParseError,
GeocoderQueryError,
)
from geopy.location import Location
from geopy.util import logger

try:
from pytz import timezone, UnknownTimeZoneError
from calendar import timegm
from datetime import datetime
from numbers import Number
pytz_available = True
except ImportError:
pytz_available = False


__all__ = ("GoogleV3", )


class GoogleV3(Geocoder): # pylint: disable=R0902
"""
Geocoder using the Google Maps v3 API. Documentation at:
https://developers.google.com/maps/documentation/geocoding/
"""

def __init__(
self,
api_key=None,
domain='maps.googleapis.com',
scheme=DEFAULT_SCHEME,
client_id=None,
secret_key=None,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""
Initialize a customized Google geocoder.
API authentication is only required for Google Maps Premier customers.
:param string api_key: The API key required by Google to perform
geocoding requests. API keys are managed through the Google APIs
console (https://code.google.com/apis/console).
.. versionadded:: 0.98.2
:param string domain: Should be the localized Google Maps domain to
connect to. The default is 'maps.googleapis.com', but if you're
geocoding address in the UK (for example), you may want to set it
to 'maps.google.co.uk' to properly bias results.
:param string scheme: Use 'https' or 'http' as the API URL's scheme.
Default is https. Note that SSL connections' certificates are not
verified.
.. versionadded:: 0.97
:param string client_id: If using premier, the account client id.
:param string secret_key: If using premier, the account secret key.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
.. versionadded:: 0.96
"""
super(GoogleV3, self).__init__(
scheme=scheme, timeout=timeout, proxies=proxies, user_agent=user_agent
)
if client_id and not secret_key:
raise ConfigurationError('Must provide secret_key with client_id.')
if secret_key and not client_id:
raise ConfigurationError('Must provide client_id with secret_key.')

self.api_key = api_key
self.domain = domain.strip('/')
self.scheme = scheme
self.doc = {}

if client_id and secret_key:
self.premier = True
self.client_id = client_id
self.secret_key = secret_key
else:
self.premier = False
self.client_id = None
self.secret_key = None

self.api = '%s://%s/maps/api/geocode/json' % (self.scheme, self.domain)
self.tz_api = '%s://%s/maps/api/timezone/json' % (
self.scheme,
self.domain
)

def _get_signed_url(self, params):
"""
Returns a Premier account signed url. Docs on signature:
https://developers.google.com/maps/documentation/business/webservices/auth#digital_signatures
"""
params['client'] = self.client_id
path = "?".join(('/maps/api/geocode/json', urlencode(params)))
signature = hmac.new(
base64.urlsafe_b64decode(self.secret_key),
path.encode('utf-8'),
hashlib.sha1
)
signature = base64.urlsafe_b64encode(
signature.digest()
).decode('utf-8')
return '%s://%s%s&signature=%s' % (
self.scheme, self.domain, path, signature
)

@staticmethod
def _format_components_param(components):
"""
Format the components dict to something Google understands.
"""
return "|".join(
(":".join(item)
for item in components.items()
)
)

@staticmethod
def _format_bounds_param(bounds):
"""
Format the bounds to something Google understands.
"""
return '%f,%f|%f,%f' % (bounds[0], bounds[1], bounds[2], bounds[3])

def geocode(
self,
query,
exactly_one=True,
timeout=None,
bounds=None,
region=None,
components=None,
language=None,
sensor=False,
): # pylint: disable=W0221,R0913
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
.. versionadded:: 0.97
:param bounds: The bounding box of the viewport within which
to bias geocode results more prominently.
:type bounds: list or tuple
:param string region: The region code, specified as a ccTLD
("top-level domain") two-character value.
:param dict components: Restricts to an area. Can use any combination
of: route, locality, administrative_area, postal_code, country.
.. versionadded:: 0.97.1
:param string language: The language in which to return results.
:param bool sensor: Whether the geocoding request comes from a
device with a location sensor.
"""
params = {
'address': self.format_string % query,
'sensor': str(sensor).lower()
}
if self.api_key:
params['key'] = self.api_key
if bounds:
if len(bounds) != 4:
raise GeocoderQueryError(
"bounds must be a four-item iterable of lat,lon,lat,lon"
)
params['bounds'] = self._format_bounds_param(bounds)
if region:
params['region'] = region
if components:
params['components'] = self._format_components_param(components)
if language:
params['language'] = language

if self.premier is False:
url = "?".join((self.api, urlencode(params)))
else:
url = self._get_signed_url(params)

logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout), exactly_one
)

def reverse(
self,
query,
exactly_one=False,
timeout=None,
language=None,
sensor=False,
): # pylint: disable=W0221,R0913
"""
Given a point, find an address.
: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 boolean 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.
.. versionadded:: 0.97
:param string language: The language in which to return results.
:param boolean sensor: Whether the geocoding request comes from a
device with a location sensor.
"""
params = {
'latlng': self._coerce_point_to_string(query),
'sensor': str(sensor).lower()
}
if language:
params['language'] = language
if self.api_key:
params['key'] = self.api_key

if not self.premier:
url = "?".join((self.api, urlencode(params)))
else:
url = self._get_signed_url(params)

logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout), exactly_one
)

def timezone(self, location, at_time=None, timeout=None):
"""
**This is an unstable API.**
Finds the timezone a `location` was in for a specified `at_time`,
and returns a pytz timezone object.
.. versionadded:: 1.2.0
:param location: The coordinates for which you want a timezone.
:type location: :class:`geopy.point.Point`, list or tuple of (latitude,
longitude), or string as "%(latitude)s, %(longitude)s"
:param at_time: The time at which you want the timezone of this
location. This is optional, and defaults to the time that the
function is called in UTC.
:type at_time integer, long, float, datetime:
:rtype: pytz timezone
"""
if not pytz_available:
raise ImportError(
'pytz must be installed in order to locate timezones. '
' Install with `pip install geopy -e ".[timezone]"`.'
)
location = self._coerce_point_to_string(location)

if isinstance(at_time, Number):
timestamp = at_time
elif isinstance(at_time, datetime):
timestamp = timegm(at_time.utctimetuple())
elif at_time is None:
timestamp = timegm(datetime.utcnow().utctimetuple())
else:
raise GeocoderQueryError(
"`at_time` must be an epoch integer or "
"datetime.datetime object"
)

params = {
"location": location,
"timestamp": timestamp,
}
if self.api_key:
params['key'] = self.api_key
url = "?".join((self.tz_api, urlencode(params)))

logger.debug("%s.timezone: %s", self.__class__.__name__, url)
response = self._call_geocoder(url, timeout=timeout)

try:
tz = timezone(response["timeZoneId"])
except UnknownTimeZoneError:
raise GeocoderParseError(
"pytz could not parse the timezone identifier (%s) "
"returned by the service." % response["timeZoneId"]
)
except KeyError:
raise GeocoderParseError(
"geopy could not find a timezone in this response: %s" %
response
)
return tz

def _parse_json(self, page, exactly_one=True):
'''Returns location, (latitude, longitude) from json feed.'''

places = page.get('results', [])
if not len(places):
self._check_status(page.get('status'))
return None

def parse_place(place):
'''Get the location, lat, lng from a single json place.'''
location = place.get('formatted_address')
latitude = place['geometry']['location']['lat']
longitude = place['geometry']['location']['lng']
return Location(location, (latitude, longitude), place)

if exactly_one:
return parse_place(places[0])
else:
return [parse_place(place) for place in places]

@staticmethod
def _check_status(status):
"""
Validates error statuses.
"""
if status == 'ZERO_RESULTS':
# When there are no results, just return.
return
if status == 'OVER_QUERY_LIMIT':
raise GeocoderQuotaExceeded(
'The given key has gone over the requests limit in the 24'
' hour period or has submitted too many requests in too'
' short a period of time.'
)
elif status == 'REQUEST_DENIED':
raise GeocoderQueryError(
'Your request was denied.'
)
elif status == 'INVALID_REQUEST':
raise GeocoderQueryError('Probably missing address or latlng.')
else:
raise GeocoderQueryError('Unknown error.')

541 changes: 541 additions & 0 deletions geopy/geocoders/ignfrance.py
View file

Large diffs are not rendered by default.

197 changes: 197 additions & 0 deletions geopy/geocoders/navidata.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,197 @@
"""
:class:`.NaviData` is the NaviData.pl geocoder.
"""

from geopy.compat import urlencode
from geopy.location import Location
from geopy.util import logger
from geopy.geocoders.base import Geocoder, DEFAULT_TIMEOUT

from geopy.exc import (
GeocoderQueryError,
GeocoderQuotaExceeded,
)


__all__ = ("NaviData", )


class NaviData(Geocoder): # pylint: disable=W0223
"""
Geocoder using the NaviData API. Documentation at:
http://www.navidata.pl
"""

def __init__(
self,
api_key=None,
domain='api.navidata.pl',
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
):
"""
.. versionadded:: 1.8.0
Initialize NaviData geocoder. Please note that 'scheme' parameter is
not supported: at present state, all NaviData traffic use plain http.
:param string api_key: The commercial API key for service. None
required if you use the API for non-commercial purposes.
:param string domain: Currently it is 'api.navidata.pl', can
be changed for testing purposes.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(NaviData, self).__init__(
scheme="http", timeout=timeout, proxies=proxies, user_agent=user_agent
)

self.api_key = api_key
self.domain = domain.strip('/')
self.geocode_api = 'http://%s/geocode' % (self.domain)
self.reverse_geocode_api = 'http://%s/revGeo' % (self.domain)

def geocode(
self,
query,
exactly_one=True,
timeout=None,
):
"""
Geocode a location query.
:param string query: The query string to be geocoded; this must
be URL encoded.
: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.
"""
params = {
'q': self.format_string % query,
}

if self.api_key is not None:
params["api_key"] = self.api_key

url = "?".join((self.geocode_api, urlencode(params)))

logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json_geocode(
self._call_geocoder(url, timeout=timeout), exactly_one
)

def reverse(
self,
query,
exactly_one=True,
timeout=None,
):
"""
Given a point, find an address.
: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 boolean exactly_one: Return one result or a list of results, if
available. Currently this has no effect
(only one address is returned by API).
: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.
"""

(lat, lon) = self._coerce_point_to_string(query).split(',')

params = {
'lat': lat,
'lon': lon
}

if self.api_key is not None:
params["api_key"] = self.api_key

url = "?".join((self.reverse_geocode_api, urlencode(params)))
logger.debug("%s.reverse: %s", self.__class__.__name__, url)
return self._parse_json_revgeocode(
self._call_geocoder(url, timeout=timeout)
)

@staticmethod
def _parse_json_geocode(page, exactly_one=True):
'''Returns location, (latitude, longitude) from json feed.'''

places = page

if not len(places):
return None

def parse_place(place):
'''Get the location, lat, lon from a single json result.'''
location = place.get('description')
latitude = place.get('lat')
longitude = place.get('lon')
return Location(location, (latitude, longitude), place)

if exactly_one:
return parse_place(places[0])
else:
return [parse_place(place) for place in places]

@staticmethod
def _parse_json_revgeocode(page):
'''Returns location, (latitude, longitude) from json feed.'''
result = page

if result.get('description', None) is None:
return None

location = result.get('description')
latitude = result.get('lat')
longitude = result.get('lon')

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


@staticmethod
def _check_status(status):
"""
Validates error statuses.
"""
status_code = status['code']

if status_code == 200:
# When there are no results, just return.
return

elif status_code == 429:
# Rate limit exceeded
raise GeocoderQuotaExceeded(
'The given key has gone over the requests limit in the 24'
' hour period or has submitted too many requests in too'
' short a period of time.'
)

elif status_code == 403:
raise GeocoderQueryError(
'Your request was denied.'
)
else:
raise GeocoderQueryError('Unknown error: ' + str(status_code))
206 changes: 206 additions & 0 deletions geopy/geocoders/opencage.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,206 @@
"""
:class:`.OpenCage` is the Opencagedata geocoder.
"""

from geopy.compat import urlencode
from geopy.geocoders.base import Geocoder, DEFAULT_TIMEOUT, DEFAULT_SCHEME
from geopy.exc import (
GeocoderQueryError,
GeocoderQuotaExceeded,
)
from geopy.location import Location
from geopy.util import logger


__all__ = ("OpenCage", )


class OpenCage(Geocoder):
"""
Geocoder using the Open Cage Data API. Documentation at:
http://geocoder.opencagedata.com/api.html
..versionadded:: 1.1.0
"""

def __init__(
self,
api_key,
domain='api.opencagedata.com',
scheme=DEFAULT_SCHEME,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""
Initialize a customized Open Cage Data geocoder.
:param string api_key: The API key required by Open Cage Data
to perform geocoding requests. You can get your key here:
https://developer.opencagedata.com/
:param string domain: Currently it is 'api.opencagedata.com', can
be changed for testing purposes.
:param string scheme: Use 'https' or 'http' as the API URL's scheme.
Default is https. Note that SSL connections' certificates are not
verified.
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
"""
super(OpenCage, self).__init__(
scheme=scheme, timeout=timeout, proxies=proxies, user_agent=user_agent
)

self.api_key = api_key
self.domain = domain.strip('/')
self.scheme = scheme
self.api = '%s://%s/geocode/v1/json' % (self.scheme, self.domain)

def geocode(
self,
query,
bounds=None,
country=None,
language=None,
exactly_one=True,
timeout=None,
): # pylint: disable=W0221,R0913
"""
Geocode a location query.
:param string query: The query string to be geocoded; this must
be URL encoded.
:param string language: an IETF format language code (such as `es`
for Spanish or pt-BR for Brazilian Portuguese); if this is
omitted a code of `en` (English) will be assumed by the remote
service.
:param string bounds: Provides the geocoder with a hint to the region
that the query resides in. This value will help the geocoder
but will not restrict the possible results to the supplied
region. The bounds parameter should be specified as 4
coordinate points forming the south-west and north-east
corners of a bounding box. For example,
`bounds=-0.563160,51.280430,0.278970,51.683979`.
:param string country: Provides the geocoder with a hint to the
country that the query resides in. This value will help the
geocoder but will not restrict the possible results to the
supplied country. The country code is a 3 character code as
defined by the ISO 3166-1 Alpha 3 standard.
: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.
"""
params = {
'key': self.api_key,
'q': self.format_string % query,
}
if bounds:
params['bounds'] = bounds
if language:
params['language'] = language
if country:
params['country'] = country

url = "?".join((self.api, urlencode(params)))

logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout), exactly_one
)

def reverse(
self,
query,
language=None,
exactly_one=False,
timeout=None,
): # pylint: disable=W0221,R0913
"""
Given a point, find an address.
: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 string language: The language in which to return results.
:param boolean 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.
"""
params = {
'key': self.api_key,
'q': self._coerce_point_to_string(query),
}
if language:
params['language'] = language

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

def _parse_json(self, page, exactly_one=True):
'''Returns location, (latitude, longitude) from json feed.'''

places = page.get('results', [])
if not len(places):
self._check_status(page.get('status'))
return None

def parse_place(place):
'''Get the location, lat, lng from a single json place.'''
location = place.get('formatted')
latitude = place['geometry']['lat']
longitude = place['geometry']['lng']
return Location(location, (latitude, longitude), place)

if exactly_one:
return parse_place(places[0])
else:
return [parse_place(place) for place in places]

@staticmethod
def _check_status(status):
"""
Validates error statuses.
"""
status_code = status['code']
if status_code == 429:
# Rate limit exceeded
raise GeocoderQuotaExceeded(
'The given key has gone over the requests limit in the 24'
' hour period or has submitted too many requests in too'
' short a period of time.'
)
if status_code == 200:
# When there are no results, just return.
return

if status_code == 403:
raise GeocoderQueryError(
'Your request was denied.'
)
else:
raise GeocoderQueryError('Unknown error.')
124 changes: 124 additions & 0 deletions geopy/geocoders/openmapquest.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
"""
:class:`.OpenMapQuest` geocoder.
"""

from geopy.compat import urlencode
from geopy.geocoders.base import (
Geocoder,
DEFAULT_FORMAT_STRING,
DEFAULT_TIMEOUT,
DEFAULT_SCHEME
)
from geopy.location import Location
from geopy.util import logger


__all__ = ("OpenMapQuest", )


class OpenMapQuest(Geocoder): # pylint: disable=W0223
"""
Geocoder using MapQuest Open Platform Web Services. Documentation at:
http://developer.mapquest.com/web/products/open/geocoding-service
"""

def __init__(
self,
api_key=None,
format_string=DEFAULT_FORMAT_STRING,
scheme=DEFAULT_SCHEME,
timeout=DEFAULT_TIMEOUT,
proxies=None,
user_agent=None,
): # pylint: disable=R0913
"""
Initialize an Open MapQuest geocoder with location-specific
address information. No API Key is needed by the Nominatim based
platform.
:param string format_string: String containing '%s' where
the string to geocode should be interpolated before querying
the geocoder. For example: '%s, Mountain View, CA'. The default
is just '%s'.
:param string scheme: Use 'https' or 'http' as the API URL's scheme.
Default is https. Note that SSL connections' certificates are not
verified.
.. versionadded:: 0.97
:param int timeout: Time, in seconds, to wait for the geocoding service
to respond before raising a :class:`geopy.exc.GeocoderTimedOut`
exception.
.. versionadded:: 0.97
:param dict proxies: If specified, routes this geocoder's requests
through the specified proxy. E.g., {"https": "192.0.2.0"}. For
more information, see documentation on
:class:`urllib2.ProxyHandler`.
.. versionadded:: 0.96
"""
super(OpenMapQuest, self).__init__(
format_string, scheme, timeout, proxies, user_agent=user_agent
)
self.api_key = api_key or ''
self.api = "%s://open.mapquestapi.com/nominatim/v1/search" \
"?format=json" % self.scheme

def geocode(self, query, exactly_one=True, timeout=None): # pylint: disable=W0221
"""
Geocode a location query.
:param string query: The address or query you wish to geocode.
: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.
.. versionadded:: 0.97
"""
params = {
'q': self.format_string % query
}
if exactly_one:
params['maxResults'] = 1
url = "&".join((self.api, urlencode(params)))

logger.debug("%s.geocode: %s", self.__class__.__name__, url)
return self._parse_json(
self._call_geocoder(url, timeout=timeout),
exactly_one
)

@classmethod
def _parse_json(cls, resources, exactly_one=True):
"""
Parse display name, latitude, and longitude from an JSON response.
"""
if not len(resources): # pragma: no cover
return None
if exactly_one:
return cls.parse_resource(resources[0])
else:
return [cls.parse_resource(resource) for resource in resources]

@classmethod
def parse_resource(cls, resource):
"""
Return location and coordinates tuple from dict.
"""
location = resource['display_name']

latitude = resource['lat'] or None
longitude = resource['lon'] or None
if latitude and longitude:
latitude = float(latitude)
longitude = float(longitude)

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