forked from tesladdicts/testatus
-
Notifications
You must be signed in to change notification settings - Fork 0
/
locator.py
117 lines (100 loc) · 3.93 KB
/
locator.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import geopy
import sqlite3
import pprint
import pickle
import math
import logging
logger = logging.getLogger(__name__)
class Locate(object):
""" Tools for determining location from lat/lon data including
sqlite3 cache for reverse geocode lookups
Thanks:
https://stackoverflow.com/users/95810/alex-martelli
"""
def __init__(self, fn='location_cache.db'):
# Initilize Geolocator
# g = geopy.geocoders.GoogleV3()
self.g = geopy.geocoders.Nominatim(user_agent=__name__, timeout=20)
self.conn = conn = sqlite3.connect(fn)
cur = conn.cursor()
cur.execute('CREATE TABLE IF NOT EXISTS '
'Geo ( '
'location STRING PRIMARY KEY, '
'address BLOB '
')')
conn.commit()
def _address_cached(self, location):
cur = self.conn.cursor()
cur.execute('SELECT address FROM Geo WHERE location=?', (location,))
res = cur.fetchone()
if res is None:
return False
return pickle.loads(res[0])
def _save_to_cache(self, location, address):
cur = self.conn.cursor()
cur.execute('INSERT INTO Geo(location, address) VALUES(?, ?)',
(location, sqlite3.Binary(pickle.dumps(address))))
self.conn.commit()
def _latlon_to_tile(self, location, zoom=18):
""" Identify a the 'tile' a partciular latitude and longitude is in
from: https://wiki.openstreetmap.org/wiki/
Slippy_map_tilenames#Lon..2Flat._to_tile_numbers_2
Args:
location (list): [latitude, longitude]
"""
lat_deg, lon_deg = location
lat_rad = math.radians(lat_deg)
n = 2.0 ** zoom
xtile = int((lon_deg + 180.0) / 360.0 * n)
ytile = int(
(1.0 - math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad))) /
math.pi) / 2.0 * n)
return str(xtile) + ',' + str(ytile)
def _geolocate(self, location):
""" Look up location data from lat/lon """
tilehash = self._latlon_to_tile(location)
logger.debug('Looking up location {} with {}'.format(location, self.g))
address_record = self._address_cached(tilehash)
if address_record is False:
latitude, longitude = location
lat_lon = str(latitude) + ',' + str(longitude)
try:
address_record = self.g.reverse(lat_lon)
except geopy.exc.GeocoderTimedOut as e:
return "Skipped"
except Exception as e:
logger.error('Exception: {}'.format(e))
raise
self._save_to_cache(tilehash, address_record)
logger.debug('{} Address: {}'.format(
location, address_record.address))
return address_record
def get_town(self, location):
""" Return town from lat,lon
Args:
location (list): [latitude, longitude]
Returns:
string: Town/City the coordinates are in or:
"Skipped": if there was an error retrieving data
"NotInData: got data, but couldn't find a town/city
"""
record = self._geolocate(location)
if type(record) != geopy.location.Location:
return record
for key in ['hamlet', 'village', 'town', 'city']:
if key in record.raw['address']:
return record.raw['address'][key]
return "NotInData"
def get_address(self, location):
""" Return town from lat,lon
Args:
location (list): [latitude, longitude]
Returns:
string: Town/City the coordinates are in or:
"Skipped": if there was an error retrieving data
"NotInData: got data, but couldn't find a town/city
"""
record = self._geolocate(location)
if type(record) != geopy.location.Location:
return record
return record.address