<s> """<STR_LIT>""" <EOL> from geopy . geocoders . base import Geocoder , DEFAULT_FORMAT_STRING , DEFAULT_TIMEOUT <EOL> from geopy . location import Location <EOL> from geopy . util import logger <EOL> from geopy . exc import GeocoderAuthenticationFailure , GeocoderQuotaExceeded , GeocoderServiceError <EOL> from geopy . compat import urlencode <EOL> __all__ = ( "<STR_LIT>" , ) <EOL> class GeocodeFarm ( Geocoder ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , <EOL> api_key = None , <EOL> format_string = DEFAULT_FORMAT_STRING , <EOL> timeout = DEFAULT_TIMEOUT , <EOL> proxies = None , <EOL> user_agent = None , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> super ( GeocodeFarm , self ) . __init__ ( <EOL> format_string , '<STR_LIT>' , timeout , proxies , user_agent = user_agent <EOL> ) <EOL> self . api_key = api_key <EOL> self . format_string = format_string <EOL> self . api = ( <EOL> "<STR_LIT>" % self . scheme <EOL> ) <EOL> self . reverse_api = ( <EOL> "<STR_LIT>" % self . scheme <EOL> ) <EOL> def geocode ( self , query , exactly_one = True , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : self . format_string % query , <EOL> } <EOL> if self . api_key : <EOL> params [ '<STR_LIT:key>' ] = self . api_key <EOL> url = "<STR_LIT:?>" . join ( ( self . api , urlencode ( params ) ) ) <EOL> logger . debug ( "<STR_LIT>" , self . __class__ . __name__ , url ) <EOL> return self . _parse_json ( <EOL> self . _call_geocoder ( url , timeout = timeout ) , exactly_one <EOL> ) <EOL> def reverse ( self , query , exactly_one = True , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> lat , lon = [ <EOL> x . strip ( ) for x in <EOL> self . _coerce_point_to_string ( query ) . split ( '<STR_LIT:U+002C>' ) <EOL> ] <EOL> except ValueError : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : lat , <EOL> '<STR_LIT>' : lon <EOL> } <EOL> if self . api_key : <EOL> params [ '<STR_LIT:key>' ] = self . api_key <EOL> url = "<STR_LIT:?>" . join ( ( self . reverse_api , urlencode ( params ) ) ) <EOL> logger . debug ( "<STR_LIT>" , self . __class__ . __name__ , url ) <EOL> return self . _parse_json ( <EOL> self . _call_geocoder ( url , timeout = timeout ) , exactly_one <EOL> ) <EOL> @ staticmethod <EOL> def parse_code ( results ) : <EOL> """<STR_LIT>""" <EOL> places = [ ] <EOL> for result in results . get ( '<STR_LIT>' ) : <EOL> coordinates = result . get ( '<STR_LIT>' , { } ) <EOL> address = result . get ( '<STR_LIT>' , { } ) <EOL> latitude = coordinates . get ( '<STR_LIT>' , None ) <EOL> longitude = coordinates . get ( '<STR_LIT>' , None ) <EOL> placename = address . get ( '<STR_LIT>' , None ) <EOL> if placename is None : <EOL> placename = address . get ( '<STR_LIT:address>' , None ) <EOL> if latitude and longitude : <EOL> latitude = float ( latitude ) <EOL> longitude = float ( longitude ) <EOL> places . append ( Location ( placename , ( latitude , longitude ) , result ) ) <EOL> return places <EOL> def _parse_json ( self , api_result , exactly_one ) : <EOL> if api_result is None : <EOL> return None <EOL> geocoding_results = api_result [ "<STR_LIT>" ] <EOL> self . _check_for_api_errors ( geocoding_results ) <EOL> places = self . parse_code ( geocoding_results ) <EOL> if exactly_one is True : <EOL> return places [ <NUM_LIT:0> ] <EOL> else : <EOL> return places <EOL> @ staticmethod <EOL> def _check_for_api_errors ( geocoding_results ) : <EOL> """<STR_LIT>""" <EOL> status_result = geocoding_results . get ( "<STR_LIT>" , { } ) <EOL> api_call_success = status_result . get ( "<STR_LIT:status>" , "<STR_LIT>" ) == "<STR_LIT>" <EOL> if not api_call_success : <EOL> access_error = status_result . get ( "<STR_LIT>" ) <EOL> access_error_to_exception = { <EOL> '<STR_LIT>' : GeocoderAuthenticationFailure , <EOL> '<STR_LIT>' : GeocoderQuotaExceeded , <EOL> } <EOL> exception_cls = access_error_to_exception . get ( <EOL> access_error , GeocoderServiceError <EOL> ) <EOL> raise exception_cls ( access_error ) </s>
