-
Notifications
You must be signed in to change notification settings - Fork 0
/
flood_pipeline.py
54 lines (50 loc) · 1.75 KB
/
flood_pipeline.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
from floodpipeline.pipeline import Pipeline
from floodpipeline.secrets import Secrets
from floodpipeline.settings import Settings
from datetime import date, datetime, timedelta
import click
@click.command()
@click.option("--country", help="country ISO3", default="UGA")
@click.option("--prepare", help="prepare discharge data", default=False, is_flag=True)
@click.option("--extract", help="extract discharge data", default=False, is_flag=True)
@click.option("--forecast", help="forecast floods", default=False, is_flag=True)
@click.option("--send", help="send to IBF", default=False, is_flag=True)
@click.option("--save", help="save to storage", default=False, is_flag=True)
@click.option(
"--datetimestart",
help="datetime start ISO 8601",
default=date.today().strftime("%Y-%m-%dT%H:%M:%S"),
)
@click.option(
"--datetimeend",
help="datetime end ISO 8601",
default=(date.today() + timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S"),
)
@click.option(
"--debug",
help="debug mode: process only one ensemble member from yesterday",
default=False,
is_flag=True,
)
def run_river_flood_pipeline(
country, prepare, extract, forecast, send, save, datetimestart, datetimeend, debug
):
datetimestart = datetime.strptime(datetimestart, "%Y-%m-%dT%H:%M:%S")
datetimeend = datetime.strptime(datetimeend, "%Y-%m-%dT%H:%M:%S")
pipe = Pipeline(
country=country,
settings=Settings("config/config.yaml"),
secrets=Secrets(".env"),
)
pipe.run_pipeline(
prepare=prepare,
extract=extract,
forecast=forecast,
send=send,
save=save,
debug=debug,
datetimestart=datetimestart,
datetimeend=datetimeend,
)
if __name__ == "__main__":
run_river_flood_pipeline()