-
Notifications
You must be signed in to change notification settings - Fork 1
/
landslide_data_sync.py
177 lines (138 loc) · 5.66 KB
/
landslide_data_sync.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
import csv
from constants import *
from utilities.convertor import *
from utilities.parser import *
from collections import OrderedDict
from abc import ABC, abstractmethod
import os
class AInputCsvDataBuilder(ABC):
"""
Extract necessary fields from csv scene file and move it into another input csv file for downloading scene
"""
_inputf = INPUT_FILE_PATH
_csv_file = ""
def __init__(self, csv_file):
self._csv_file = csv_file
def __init__(self, csv_file, inputf):
self._csv_file = csv_file
self._inputf = inputf
def get_csv_file_path(self):
return self._csv_file
def get_input_file(self):
return self._inputf
def build_csv_header(self, inputf):
with open(inputf, 'r') as f:
input_csv = csv.DictReader(f, delimiter=',')
inputs = list(input_csv)
os.remove(inputf)
# create csv header
with open(inputf, 'a') as a:
writer = csv.writer(a)
writer.writerow(['id','lat','lng','start_date','end_date','size','cloudcover','satellite','station','downloaded_path'])
def build_input_csv_file(self):
input_file = self.get_input_file()
self.build_csv_header(input_file)
with open(inputf, "a") as inpf:
for satellite in ['LC8', 'LE7', 'LT5']:
line = OrderedDict()
line['id'] = count
line['lat'] = data['latitude']
line['lng'] = data['longitude']
if not(data['date_']):
continue
date_ = utc_to_normal_date(data['date_'])
start_date = parse_date(date_)
end_date = start_date + datetime.timedelta(days=91)
line['start_date'] = date_
line['end_date'] = convert_date_to_normal_date_str(end_date)
line['size'] = data['landslide1']
line['cloudcover'] = '100'
line['satellite'] = satellite
line['station'] = 'ALL'
line["downloaded_path"] = ''
print("All necessary fields are extracted")
def build_csv_data(self, source_data, input_file):
count = 0
for data in source_data:
with open(input_file, "a") as inpf:
for satellite in ['LC8']:
line = self.extract_necessary_fields_for_download(data, satellite, count)
if line == None:
continue
# print(line)
writer = csv.writer(inpf)
writer.writerow(line.values())
count += 1
@abstractmethod
def extract_necessary_fields_for_download(self, source_data, satellite, count):
pass
class LandslidedataCSVBuilder(AInputCsvDataBuilder):
def extract_necessary_fields_for_download(self, source_data, satellite, count):
line = OrderedDict()
line['id'] = count
line['lat'] = source_data['latitude']
line['lng'] = source_data['longitude']
if not(source_data['date_']):
return None
date_ = utc_to_normal_date(source_data['date_'])
start_date = parse_date(date_)
end_date = start_date + datetime.timedelta(days=180)
line['start_date'] = date_
line['end_date'] = convert_date_to_normal_date_str(end_date)
line['size'] = source_data['landslide1']
line['cloudcover'] = '10'
line['satellite'] = satellite
line['station'] = 'ALL'
line["downloaded_path"] = ''
return line
class NasaGlobalLandslideCSVBuilder(AInputCsvDataBuilder):
def extract_necessary_fields_for_download(self, source_data, satellite, count):
line = OrderedDict()
line['id'] = count
line['lat'] = source_data['latitude']
line['lng'] = source_data['longitude']
if not(source_data['event_date']):
return None
date_ = convert_datestr_to_normal_date(source_data['event_date'], "%Y-%m-%d %H:%M:%S")
start_date = parse_date(date_)
end_date = start_date + datetime.timedelta(days=180)
line['start_date'] = date_
line['end_date'] = convert_date_to_normal_date_str(end_date)
line['size'] = source_data['landslide_size']
line['cloudcover'] = '10'
line['satellite'] = satellite
line['station'] = 'ALL'
line["downloaded_path"] = ''
return line
class InputCSVFactory():
_inputf = INPUT_FILE_PATH
_file_type = ""
_csv_file = ""
def __init__(self, csv_file, file_type):
self._csv_file = csv_file
self._file_type = file_type
def __init__(self, csv_file, inputf, file_type):
self._csv_file = csv_file
self._inputf = inputf
self._file_type = file_type
def get_csv_file_path(self):
return self._csv_file
def get_input_file(self):
return self._inputf
def get_file_type(self):
return self._file_type
def create_csv_builder(self):
file_type = self.get_file_type()
source_csv_path = self.get_csv_file_path()
if file_type == "LANDSLIDE":
return LandslidedataCSVBuilder(source_csv_path, self.get_input_file())
elif file_type == "NASA_GLOBAL_LANDSLIDE":
return NasaGlobalLandslideCSVBuilder(source_csv_path, self.get_input_file())
else:
print("Wrong type")
return None
if __name__ == "__main__":
sync_landslidedata_to_input_csv(True)
factory = InputCSVFactory(LANDSLIDE_DATA_FILE_PATH, INPUT_FILE_PATH, "NASA_GLOBAL_LANDSLIDE")
builder = factory.create_csv_builder()
builder.build_input_csv_file()