Passed
Pull Request — master (#357)
by Thomas
02:02
created

elodie.geolocation.get_key()   A

Complexity

Conditions 5

Size

Total Lines 19
Code Lines 15

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 5
eloc 15
nop 0
dl 0
loc 19
rs 9.1832
c 0
b 0
f 0
1
"""Look up geolocation information for media objects."""
2
from __future__ import print_function
3
from __future__ import division
4
from future import standard_library
5
from past.utils import old_div
6
7
standard_library.install_aliases()  # noqa
8
9
from os import path
10
11
import requests
12
import urllib.request
13
import urllib.parse
14
import urllib.error
15
16
from elodie.config import load_config
17
from elodie import constants
18
from elodie import log
19
from elodie.localstorage import Db
20
21
__KEY__ = None
22
__DEFAULT_LOCATION__ = 'Unknown Location'
23
__PREFER_ENGLISH_NAMES__ = None
24
__PREFER_LANGUAGE__ = None
25
26
def coordinates_by_name(name):
27
    # Try to get cached location first
28
    db = Db()
29
    cached_coordinates = db.get_location_coordinates(name)
30
    if(cached_coordinates is not None):
31
        return {
32
            'latitude': cached_coordinates[0],
33
            'longitude': cached_coordinates[1]
34
        }
35
36
    # If the name is not cached then we go ahead with an API lookup
37
    geolocation_info = lookup(location=name)
38
39
    if(geolocation_info is not None):
40
        if(
41
            'results' in geolocation_info and
42
            len(geolocation_info['results']) != 0 and
43
            'locations' in geolocation_info['results'][0] and
44
            len(geolocation_info['results'][0]['locations']) != 0
45
        ):
46
47
            # By default we use the first entry unless we find one with
48
            #   geocodeQuality=city.
49
            geolocation_result = geolocation_info['results'][0]
50
            use_location = geolocation_result['locations'][0]['latLng']
51
            # Loop over the locations to see if we come accross a
52
            #   geocodeQuality=city.
53
            # If we find a city we set that to the use_location and break
54
            for location in geolocation_result['locations']:
55
                if(
56
                    'latLng' in location and
57
                    'lat' in location['latLng'] and
58
                    'lng' in location['latLng'] and
59
                    location['geocodeQuality'].lower() == 'city'
60
                ):
61
                    use_location = location['latLng']
62
                    break
63
64
            return {
65
                'latitude': use_location['lat'],
66
                'longitude': use_location['lng']
67
            }
68
69
    return None
70
71
72
def decimal_to_dms(decimal):
73
    decimal = float(decimal)
74
    decimal_abs = abs(decimal)
75
    minutes, seconds = divmod(decimal_abs*3600, 60)
76
    degrees, minutes = divmod(minutes, 60)
77
    degrees = degrees
78
    sign = 1 if decimal >= 0 else -1
79
    return (degrees, minutes, seconds, sign)
80
81
82
def dms_to_decimal(degrees, minutes, seconds, direction=' '):
83
    sign = 1
84
    if(direction[0] in 'WSws'):
85
        sign = -1
86
    return (
87
        float(degrees) + old_div(float(minutes), 60) +
88
        old_div(float(seconds), 3600)
89
    ) * sign
90
91
92
def dms_string(decimal, type='latitude'):
93
    # Example string -> 38 deg 14' 27.82" S
94
    dms = decimal_to_dms(decimal)
95
    if type == 'latitude':
96
        direction = 'N' if decimal >= 0 else 'S'
97
    elif type == 'longitude':
98
        direction = 'E' if decimal >= 0 else 'W'
99
    return '{} deg {}\' {}" {}'.format(dms[0], dms[1], dms[2], direction)
100
101
102
def get_key():
103
    global __KEY__
104
    if __KEY__ is not None:
105
        return __KEY__
106
107
    if constants.mapquest_key is not None:
108
        __KEY__ = constants.mapquest_key
109
        return __KEY__
110
111
    config_file = '%s/config.ini' % constants.application_directory
112
    if not path.exists(config_file):
113
        return None
114
115
    config = load_config()
116
    if('MapQuest' not in config):
117
        return None
118
119
    __KEY__ = config['MapQuest']['key']
120
    return __KEY__
121
122
def get_prefer_language():
123
    global __PREFER_LANGUAGE__
124
125
    config_file = '%s/config.ini' % constants.application_directory
126
    if not path.exists(config_file):
127
        return constants.accepted_language
128
129
    config = load_config()
130
    if('MapQuest' not in config):
131
        return constants.accepted_language
132
133
    if('prefer_language' not in config['MapQuest']):
134
        return constants.accepted_language
135
136
    __PREFER_LANGUAGE__ = config['MapQuest']['prefer_language']
137
    return __PREFER_LANGUAGE__
138
139
def get_prefer_english_names():
140
    global __PREFER_ENGLISH_NAMES__
141
    if __PREFER_ENGLISH_NAMES__ is not None:
142
        return __PREFER_ENGLISH_NAMES__
143
144
    config_file = '%s/config.ini' % constants.application_directory
145
    if not path.exists(config_file):
146
        return False
147
148
    config = load_config()
149
    if('MapQuest' not in config):
150
        return False
151
152
    if('prefer_english_names' not in config['MapQuest']):
153
        return False
154
155
    __PREFER_ENGLISH_NAMES__ = bool(config['MapQuest']['prefer_english_names'])
156
    return __PREFER_ENGLISH_NAMES__
157
158
def place_name(lat, lon):
159
    lookup_place_name_default = {'default': __DEFAULT_LOCATION__}
160
    if(lat is None or lon is None):
161
        return lookup_place_name_default
162
163
    # Convert lat/lon to floats
164
    if(not isinstance(lat, float)):
165
        lat = float(lat)
166
    if(not isinstance(lon, float)):
167
        lon = float(lon)
168
169
    # Try to get cached location first
170
    db = Db()
171
    # 3km distace radious for a match
172
    cached_place_name = db.get_location_name(lat, lon, 3000)
173
    # We check that it's a dict to coerce an upgrade of the location
174
    #  db from a string location to a dictionary. See gh-160.
175
    if(isinstance(cached_place_name, dict)):
176
        return cached_place_name
177
178
    lookup_place_name = {}
179
    geolocation_info = lookup(lat=lat, lon=lon)
180
    if(geolocation_info is not None and 'address' in geolocation_info):
181
        address = geolocation_info['address']
182
        for loc in ['city', 'state', 'country']:
183
            if(loc in address):
184
                lookup_place_name[loc] = address[loc]
185
                # In many cases the desired key is not available so we
186
                #  set the most specific as the default.
187
                if('default' not in lookup_place_name):
188
                    lookup_place_name['default'] = address[loc]
189
190
    if(lookup_place_name):
191
        db.add_location(lat, lon, lookup_place_name)
192
        # TODO: Maybe this should only be done on exit and not for every write.
193
        db.update_location_db()
194
195
    if('default' not in lookup_place_name):
196
        lookup_place_name = lookup_place_name_default
197
198
    return lookup_place_name
199
200
201
def lookup(**kwargs):
202
    if(
203
        'location' not in kwargs and
204
        'lat' not in kwargs and
205
        'lon' not in kwargs
206
    ):
207
        return None
208
209
    prefer_language = get_prefer_language()
210
    key = get_key()
211
    prefer_english_names = get_prefer_english_names()
212
    params = {'format': 'json', 'key': key}
213
214
    if(prefer_language is not None)
215
       params = {'format': 'json', 'accept-language': prefer_language, 'key': key}
216
217
    if(key is None):
218
        return None
219
220
    try:
221
        params.update(kwargs)
222
        path = '/geocoding/v1/address'
223
        if('lat' in kwargs and 'lon' in kwargs):
224
            path = '/nominatim/v1/reverse.php'
225
        url = '%s%s?%s' % (
226
                    constants.mapquest_base_url,
227
                    path,
228
                    urllib.parse.urlencode(params)
229
              )
230
        headers = {}
231
        if(prefer_english_names):
232
            headers = {'Accept-Language':'en-EN,en;q=0.8'}
233
        r = requests.get(url, headers=headers)
234
        return parse_result(r.json())
235
    except requests.exceptions.RequestException as e:
236
        log.error(e)
237
        return None
238
    except ValueError as e:
239
        log.error(r.text)
240
        log.error(e)
241
        return None
242
243
244
def parse_result(result):
245
    if('error' in result):
246
        return None
247
248
    if(
249
        'results' in result and
250
        len(result['results']) > 0 and
251
        'locations' in result['results'][0]
252
        and len(result['results'][0]['locations']) > 0 and
253
        'latLng' in result['results'][0]['locations'][0]
254
    ):
255
        latLng = result['results'][0]['locations'][0]['latLng']
256
        if(latLng['lat'] == 39.78373 and latLng['lng'] == -100.445882):
257
            return None
258
259
    return result
260