fimbox/tests/test_nwmstreamflow.py at main · sdmlua/fimbox · GitHub

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
"""
Author: Supath Dhital
Date Created: June 2026

Streamflow retrieval / plot / statistics — minimal, call-the-function tests.

Point AOI_DIR at a working directory whose feature_id.csv exists (or pass a
feature_ids list / CSV per call). Edit the dates and USGS site to your basin,
then run the functions you want.
"""

from __future__ import annotations

from pathlib import Path

from fimbox import (
    getNWManalysisassim,
    getNWMretrospective,
)

AOI_DIR = Path(__file__).resolve().parents[2] / "out" / "test_smallB"
# AOI_DIR = Path(__file__).resolve().parents[2] / "out" / "HUC08060202"

START = "2016-10-05"
END = "2016-10-20"
EVENT = "2020-10-10 21:00:00"


# # retrospective — different extraction combinations
# def test_retrospective_event_date():
#     getNWMretrospective(AOI_DIR, date=EVENT)


# def test_retrospective_range_continuous():
#     # start + end, nothing else -> one CSV per hour
#     getNWMretrospective(AOI_DIR, start=START, end=END)


def test_retrospective_range_sortby():
    # start + end + sortby -> one aggregated CSV
    getNWMretrospective(AOI_DIR, start=START, end=END, sortby="maximum")


# def test_retrospective_feature_ids_list():
#     # pass feature_ids directly instead of relying on the AOI's feature_id.csv
#     getNWMretrospective(AOI_DIR, feature_ids=[FEATURE_ID], date=EVENT)


# # analysis and assimilation (AnA) — different extraction combinations
# def test_analysisassim_event_date():
#     getNWManalysisassim(AOI_DIR, date=EVENT)


# def test_analysisassim_range_continuous():
#     # start + end, nothing else -> one CSV per hour
#     getNWManalysisassim(AOI_DIR, start=START, end=END)


# def test_analysisassim_range_sortby():
#     # start + end + sortby -> one aggregated CSV
#     getNWManalysisassim(AOI_DIR, start=START, end=END, sortby="maximum")


# # forecast — different combinations
# def test_forecast_shortrange():
#     getNWMforecast(AOI_DIR, "shortrange")


# def test_forecast_mediumrange_maxsort():
#     getNWMforecast(AOI_DIR, "mediumrange", sort_by="maximum")


# def test_forecast_specific_cycle():
#     getNWMforecast(AOI_DIR, "shortrange", forecast_date="2024-06-01", hour=12)


# # USGS observations
# def test_usgs_fetch():
#     USGSData(AOI_DIR).fetch([USGS_SITE], START, END)


# def test_usgs_feature_id_pairs():
#     # which USGS gage falls on which reach (feature_id) within the AOI
#     pairs = get_usgs_fid_pairs(AOI_DIR)
#     print(pairs)


# # plots
# def test_plot_feature_id():
#     plot_nwm(AOI_DIR, [FEATURE_ID], START, END)


# def test_plot_usgs():
#     plot_usgs(AOI_DIR, [USGS_SITE], START, END)


# def test_plot_usgs_and_feature_id():
#     # time series overlay of USGS and the NWM feature_id together
#     plot_comparison(AOI_DIR, FEATURE_ID, USGS_SITE, START, END)


# # statistics
# def test_statistics_usgs_vs_nwm():
#     calculate_statistics(AOI_DIR, FEATURE_ID, USGS_SITE, START, END)