-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstage_1.py
More file actions
143 lines (118 loc) · 5.63 KB
/
Copy pathstage_1.py
File metadata and controls
143 lines (118 loc) · 5.63 KB
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
import os
import shutil
from pathlib import Path
from typing import Union
from tqdm import tqdm
class PreProcessor:
def __init__(self,
data_dir: Union[str, os.PathLike],
source_name: str):
self.raw_data_dir = Path(data_dir) / 'raw'
self.preprocessed_dir = Path(data_dir) / 'preprocessed'
self.source_name = source_name
def process(self):
"""
This method is what is called to process existing eddy covariance data into
a format suitable for CarbonSense. The output directory should contain a
single subdirectory per data source. Each subdirectory will contain a single
CSV file with all the EC data, as well as a single `site_data.csv` file which
contains all metadata for the respective EC sites (lon, lat, IGBP, etc).
"""
raise NotImplementedError('Preprocessor does not have a process method!')
class AmerifluxPreprocessor(PreProcessor):
def process(self):
input_dir = self.raw_data_dir / self.source_name
output_dir = self.preprocessed_dir / self.source_name
def file_check(filename):
return 'FLUXNET_SUBSET_HH' in filename and 'VARINFO' not in filename
os.makedirs(output_dir, exist_ok=True)
shutil.copy(input_dir / 'site_data.csv', output_dir / 'site_data.csv')
for site in tqdm(os.listdir(input_dir / 'unzipped')):
site_dir = input_dir / 'unzipped' / site
files = os.listdir(site_dir)
valid_files = [f for f in files if file_check(f)]
if len(valid_files) == 1:
shutil.copy(site_dir / valid_files[0], output_dir / f'{site}.csv')
else:
#print(f'No valid file found for {site}')
pass
class FluxnetPreprocessorCO2_2015(PreProcessor):
def process(self):
input_dir = self.raw_data_dir / self.source_name
output_dir = self.preprocessed_dir / self.source_name
def file_check(filename):
return 'SUBSET_HH' in filename and 'VARINFO' not in filename
os.makedirs(output_dir, exist_ok=True)
shutil.copy(input_dir / 'site_data.csv', output_dir / 'site_data.csv')
for site in tqdm(os.listdir(input_dir / 'unzipped')):
site_dir = input_dir / 'unzipped' / site
files = os.listdir(site_dir)
valid_files = [f for f in files if file_check(f)]
if len(valid_files) == 1:
shutil.copy(site_dir / valid_files[0], output_dir / f'{site}.csv')
else:
#print(f'No valid file found for {site}')
pass
class FluxnetPreprocessorCH4_2015(PreProcessor):
def process(self):
input_dir = self.raw_data_dir / self.source_name
output_dir = self.preprocessed_dir / self.source_name
def file_check(filename):
return 'CH4_HH' in filename and 'VARINFO' not in filename
os.makedirs(output_dir, exist_ok=True)
shutil.copy(input_dir / 'site_data.csv', output_dir / 'site_data.csv')
for site in tqdm(os.listdir(input_dir / 'unzipped')):
site_dir = input_dir / 'unzipped' / site
site_name = site[4:10] # specific to this directory structure
files = os.listdir(site_dir)
valid_files = [f for f in files if file_check(f)]
if len(valid_files) == 1:
shutil.copy(site_dir / valid_files[0], output_dir / f'{site_name}.csv')
else:
#print(f'No valid file found for {site}')
pass
class ICOS2023Preprocessor(PreProcessor):
def process(self):
input_dir = self.raw_data_dir / self.source_name
output_dir = self.preprocessed_dir / self.source_name
def file_check(filename):
return 'FLUXNET_HH_L2' in filename and 'VARINFO' not in filename
os.makedirs(output_dir, exist_ok=True)
shutil.copy(input_dir / 'site_data.csv', output_dir / 'site_data.csv')
for site in tqdm(os.listdir(input_dir / 'unzipped')):
site_dir = input_dir / 'unzipped' / site
files = os.listdir(site_dir)
valid_files = [f for f in files if file_check(f)]
if len(valid_files) == 1:
shutil.copy(site_dir / valid_files[0], output_dir / f'{site}.csv')
else:
#print(f'No valid file found for {site}')
pass
class ICOSWarmWinterPreprocessor(PreProcessor):
def process(self):
input_dir = self.raw_data_dir / self.source_name
output_dir = self.preprocessed_dir / self.source_name
def file_check(filename):
return 'FULLSET_HH' in filename and 'VARINFO' not in filename
os.makedirs(output_dir, exist_ok=True)
shutil.copy(input_dir / 'site_data.csv', output_dir / 'site_data.csv')
for site in tqdm(os.listdir(input_dir / 'unzipped')):
site_dir = input_dir / 'unzipped' / site
files = os.listdir(site_dir)
valid_files = [f for f in files if file_check(f)]
if len(valid_files) == 1:
shutil.copy(site_dir / valid_files[0], output_dir / f'{site}.csv')
else:
#print(f'No valid file found for {site}')
pass
REGISTER = [
('ameriflux', AmerifluxPreprocessor),
('fluxnet', FluxnetPreprocessorCO2_2015),
('fluxnet-ch4', FluxnetPreprocessorCH4_2015),
('icos-2023', ICOS2023Preprocessor),
('icos-ww', ICOSWarmWinterPreprocessor),
]
def run_stage_1(data_dir):
for source_name, preprocessor in REGISTER:
print(f'Preprocessing initiated for {source_name}...')
preprocessor(data_dir, source_name).process()