-
-
Notifications
You must be signed in to change notification settings - Fork 139
/
localstorage.py
204 lines (165 loc) · 6.61 KB
/
localstorage.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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
"""
Methods for interacting with information Elodie caches about stored media.
"""
from builtins import map
from builtins import object
import hashlib
import json
import os
import sys
from math import radians, cos, sqrt
from shutil import copyfile
from time import strftime
from elodie import constants
class Db(object):
"""A class for interacting with the JSON files created by Elodie."""
def __init__(self):
# verify that the application directory (~/.elodie) exists,
# else create it
if not os.path.exists(constants.application_directory):
os.makedirs(constants.application_directory)
# If the hash db doesn't exist we create it.
# Otherwise we only open for reading
if not os.path.isfile(constants.hash_db):
with open(constants.hash_db, 'a'):
os.utime(constants.hash_db, None)
self.hash_db = {}
# We know from above that this file exists so we open it
# for reading only.
with open(constants.hash_db, 'r') as f:
try:
self.hash_db = json.load(f)
except ValueError:
pass
# If the location db doesn't exist we create it.
# Otherwise we only open for reading
if not os.path.isfile(constants.location_db):
with open(constants.location_db, 'a'):
os.utime(constants.location_db, None)
self.location_db = []
# We know from above that this file exists so we open it
# for reading only.
with open(constants.location_db, 'r') as f:
try:
self.location_db = json.load(f)
except ValueError:
pass
def add_hash(self, key, value, write=False):
"""Add a hash to the hash db.
:param str key:
:param str value:
:param bool write: If true, write the hash db to disk.
"""
self.hash_db[key] = value
if(write is True):
self.update_hash_db()
# Location database
# Currently quite simple just a list of long/lat pairs with a name
# If it gets many entries a lookup might take too long and a better
# structure might be needed. Some speed up ideas:
# - Sort it and inter-half method can be used
# - Use integer part of long or lat as key to get a lower search list
# - Cache a small number of lookups, photos are likely to be taken in
# clusters around a spot during import.
def add_location(self, latitude, longitude, place, write=False):
"""Add a location to the database.
:param float latitude: Latitude of the location.
:param float longitude: Longitude of the location.
:param str place: Name for the location.
:param bool write: If true, write the location db to disk.
"""
data = {}
data['lat'] = latitude
data['long'] = longitude
data['name'] = place
self.location_db.append(data)
if(write is True):
self.update_location_db()
def backup_hash_db(self):
"""Backs up the hash db."""
if os.path.isfile(constants.hash_db):
mask = strftime('%Y-%m-%d_%H-%M-%S')
backup_file_name = '%s-%s' % (constants.hash_db, mask)
copyfile(constants.hash_db, backup_file_name)
return backup_file_name
def check_hash(self, key):
"""Check whether a hash is present for the given key.
:param str key:
:returns: bool
"""
return key in self.hash_db
def checksum(self, file_path, blocksize=65536):
"""Create a hash value for the given file.
See http://stackoverflow.com/a/3431835/1318758.
:param str file_path: Path to the file to create a hash for.
:param int blocksize: Read blocks of this size from the file when
creating the hash.
:returns: str or None
"""
hasher = hashlib.sha256()
with open(file_path, 'rb') as f:
buf = f.read(blocksize)
while len(buf) > 0:
hasher.update(buf)
buf = f.read(blocksize)
return hasher.hexdigest()
def get_hash(self, key):
"""Get the hash value for a given key.
:param str key:
:returns: str or None
"""
if(self.check_hash(key) is True):
return self.hash_db[key]
return None
def get_location_name(self, latitude, longitude, threshold_m):
"""Find a name for a location in the database.
:param float latitude: Latitude of the location.
:param float longitude: Longitude of the location.
:param int threshold_m: Location in the database must be this close to
the given latitude and longitude.
:returns: str, or None if a matching location couldn't be found.
"""
last_d = sys.maxsize
name = None
for data in self.location_db:
# As threshold is quite small use simple math
# From http://stackoverflow.com/questions/15736995/how-can-i-quickly-estimate-the-distance-between-two-latitude-longitude-points # noqa
# convert decimal degrees to radians
lon1, lat1, lon2, lat2 = list(map(
radians,
[longitude, latitude, data['long'], data['lat']]
))
r = 6371000 # radius of the earth in m
x = (lon2 - lon1) * cos(0.5 * (lat2 + lat1))
y = lat2 - lat1
d = r * sqrt(x * x + y * y)
# Use if closer then threshold_km reuse lookup
if(d <= threshold_m and d < last_d):
name = data['name']
last_d = d
return name
def get_location_coordinates(self, name):
"""Get the latitude and longitude for a location.
:param str name: Name of the location.
:returns: tuple(float), or None if the location wasn't in the database.
"""
for data in self.location_db:
if data['name'] == name:
return (data['lat'], data['long'])
return None
def all(self):
"""Generator to get all entries from self.hash_db
:returns tuple(string)
"""
for checksum, path in self.hash_db.items():
yield (checksum, path)
def reset_hash_db(self):
self.hash_db = {}
def update_hash_db(self):
"""Write the hash db to disk."""
with open(constants.hash_db, 'w') as f:
json.dump(self.hash_db, f)
def update_location_db(self):
"""Write the location db to disk."""
with open(constants.location_db, 'w') as f:
json.dump(self.location_db, f)