forked from CSSEGISandData/COVID-19
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sat_downloader.py
53 lines (39 loc) · 2.34 KB
/
sat_downloader.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
from sentinelsat import SentinelAPI
from geopy.geocoders import Nominatim
import os
def sat_downloader(userid, password, allcities_list):
# Connect with the database
apiurl = 'https://s5phub.copernicus.eu/dhus'
testAPI = SentinelAPI(user=userid, password=password, api_url=apiurl)
dir_path = os.path.dirname(os.path.realpath(__file__))
# For all cities
for precity in allcities_list:
city = precity[0]
# Determine the GPS location needed to search
geolocator = Nominatim(user_agent="testApp")
targetcity = geolocator.geocode(city)
timeframes = ['beginposition:[NOW-1DAYS TO NOW]', 'beginposition:[2020-02-01T00:00:00.000Z TO 2020-02-02T00:00:00.000Z]', 'beginposition:[2020-01-05T00:00:00.000Z TO 2020-01-06T00:00:00.000Z]']
# Determine what files meet the criteria specified
#timeframe = 'beginposition:[NOW-1DAYS TO NOW]'
#timeframe = 'beginposition:[2020-02-01T00:00:00.000Z TO 2020-02-02T00:00:00.000Z]'
#timeframe = 'beginposition:[2020-01-05T00:00:00.000Z TO 2020-01-06T00:00:00.000Z]'
for timeframe in timeframes:
satquery_loc = 'footprint:"intersects(' + str(targetcity.latitude) + ',' + str(targetcity.longitude) + ')"'
products = testAPI.query(raw=satquery_loc + ' AND ' + timeframe + ' AND producttype:L2__NO2___')
if not products:
print('Dictionary empty')
else:
# Based on that, generate paths of available data
downloadedfile = products[next(iter(products))]['filename']
datafilesfolder = r'\\datafiles\\'
downloadedfile_full = dir_path + datafilesfolder + downloadedfile
firstdownload = dir_path + datafilesfolder.replace(r'datafiles\\', '') + downloadedfile.replace('.nc', '.zip')
# Check to see if you have already downloaded this file
if os.path.exists(downloadedfile_full):
#Exists!
print('File exists.. skipping')
else:
# Otherwise, download all results from the search
mypath = testAPI.download_all(products)
# Move the file to where its supposed to go
os.rename(firstdownload, downloadedfile_full)