Skip to content

Commit

Permalink
Re-create tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@mdbartos
mdbartos committed May 31, 2024
1 parent c51df7e commit 04057af
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 161 deletions.
2 changes: 1 addition & 1 deletion pipedream_solver/nsuperlink.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1558,7 +1558,7 @@ def reposition_junctions(self, reposition=None):
_H_dk = H_j[_J_dk]
# Handle which superlinks to reposition
if reposition is None:
reposition = np.ones(NK, dtype=np.bool8)
reposition = np.ones(NK, dtype=np.bool_)
# Reposition junctions
numba_reposition_junctions(_x_Ik, _z_inv_Ik, _h_Ik, _dx_ik, _Q_ik, _H_dk,
_b0, _zc, _xc, _m, _elem_pos, _i_1k, _I_1k,
Expand Down
192 changes: 32 additions & 160 deletions test/test_pipedream.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,7 @@

internal_links = 24

hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=internal_links)
hillslope_superlink_wq_params[['No_of_obs_point', 'KF_obs_point', 'KF_process_sigma', 'KF_measure_sigma']] = [1, 1, 1, 1]

hillslope_nsuperlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
Expand All @@ -28,75 +26,46 @@
hillslope_greenampt_model = GreenAmpt(hillslope_soil_params)
hillslope_ngreenampt_model = nGreenAmpt(hillslope_soil_params)

hillslope_water_quality_model = QualityBuilder(hillslope_nsuperlink_model,
superjunction_params=hillslope_superjunction_wq_params,
superlink_params=hillslope_superlink_wq_params)
# hillslope_water_quality_model = QualityBuilder(hillslope_nsuperlink_model,
# superjunction_params=hillslope_superjunction_wq_params,
# superlink_params=hillslope_superlink_wq_params)

initial_nsuperlink_states = copy.deepcopy(hillslope_nsuperlink_model.states)

def test_superlink_step():
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4)
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_superlink_model.NIk)
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
hillslope_superlink_model.reposition_junctions()

def test_nsuperlink_step():
hillslope_nsuperlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4,
mobile_elements=True)
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_nsuperlink_model.NIk)
hillslope_nsuperlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
hillslope_nsuperlink_model.reposition_junctions()

def test_superlink_spinup():
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4)
hillslope_superlink_model.spinup(n_steps=100)

def test_superlink_convergence():
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4)
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_superlink_model.NIk)
Q_0Ik = 1e-3 * np.ones(hillslope_nsuperlink_model.NIk)
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, num_iter=8)

def test_superlink_banded_step():
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4, auto_permute=True)
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_superlink_model.NIk)
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)

def test_superlink_recurrence_method():
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4, method='f')
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_superlink_model.NIk)
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4, method='nnls')
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4, method='lsq')
hillslope_superlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)

def test_simulation_manager():
dt = 10
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=24)
hillslope_superlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=24)
Q_in = pd.DataFrame.from_dict(
{
0 : np.zeros(hillslope_superlink_model.M),
Expand All @@ -106,7 +75,6 @@ def test_simulation_manager():
18001 : np.zeros(hillslope_superlink_model.M),
28000 : np.zeros(hillslope_superlink_model.M)
}, orient='index')

Q_Ik = pd.DataFrame.from_dict(
{
0 : np.zeros(hillslope_superlink_model.NIk),
Expand All @@ -130,9 +98,9 @@ def test_simulation_manager():

def test_adaptive_timestep():
dt = 10
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=24)
hillslope_superlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=24)
Q_in = pd.DataFrame.from_dict(
{
0 : np.zeros(hillslope_superlink_model.M),
Expand All @@ -142,7 +110,6 @@ def test_adaptive_timestep():
18001 : np.zeros(hillslope_superlink_model.M),
28000 : np.zeros(hillslope_superlink_model.M)
}, orient='index')

Q_Ik = pd.DataFrame.from_dict(
{
0 : np.zeros(hillslope_superlink_model.NIk),
Expand Down Expand Up @@ -176,9 +143,9 @@ def test_superlink_geometry():
hillslope_superlinks['g3'] = 1
for geom in geoms:
hillslope_superlinks['shape'] = geom
hillslope_superlink_model = SuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4)
hillslope_superlink_model = nSuperLink(hillslope_superlinks,
hillslope_superjunctions,
internal_links=4)
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_superlink_model.NIk)
Expand All @@ -188,23 +155,14 @@ def test_superlink_geometry():
hillslope_superlinks['g2'] = 5

def test_plot_profile():
hillslope_superlink_model.plot_profile([0, 1], width=100)
hillslope_nsuperlink_model.plot_profile([0, 1], width=100)

def test_plot_network_2d():
hillslope_superlink_model.plot_network_2d(junction_kwargs={'s' : 4})
hillslope_nsuperlink_model.plot_network_2d(junction_kwargs={'s' : 4})

def test_greenampt_step():
dt = 120
i = 50 / 1000 / 3600 * np.ones(hillslope_superlink_model.NIk)
infiltration_model = hillslope_greenampt_model
for _ in range(100):
infiltration_model.step(dt=dt, i=i)

def test_ngreenampt_step():
dt = 10
i = 50 / 1000 / 3600 * np.ones(hillslope_superlink_model.NIk)
i = 50 / 1000 / 3600 * np.ones(hillslope_nsuperlink_model.NIk)
hydraulic_model = hillslope_nsuperlink_model
infiltration_model = hillslope_ngreenampt_model
hydraulic_model.load_state(initial_nsuperlink_states)
Expand All @@ -214,42 +172,14 @@ def test_ngreenampt_step():
Q_0Ik = infiltration_model.Q
hydraulic_model.step(dt=dt, Q_0Ik=Q_0Ik)

def test_water_quality_step():
dt = 10
Q_in = 1e-2 * np.asarray([1., 0.])
Q_0Ik = 1e-3 * np.ones(hillslope_nsuperlink_model.NIk)
c_0j = 10. * np.asarray([1., 0.])
for _ in range(100):
hillslope_nsuperlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
hillslope_water_quality_model.step(dt=dt, c_0j=c_0j)

def test_orifice():
superjunctions = copy.deepcopy(hillslope_superjunctions)
superlinks = hillslope_superlinks
superjunctions.loc[2] = superjunctions.loc[0]
superjunctions.loc[2, ['name', 'id']] = 2
superjunctions.loc[2, 'h_0'] = 2.0
orifices = {
'id' : 0,
'sj_0' : 2,
'sj_1' : 0,
'A' : 0.3048**2,
'orientation' : 'side',
'z_o' : 0,
'y_max' : 0.3048,
'C' : 0.67}
orifices = pd.DataFrame(orifices, index=[0])
hydraulic_model = SuperLink(superlinks, superjunctions, orifices=orifices,
internal_links=internal_links)
dt = 10
Q_in = 1e-2 * np.asarray([0., 0., 0.])
Q_0Ik = 1e-3 * np.ones(hydraulic_model.NIk)
for _ in range(100):
if (hydraulic_model.t > 200):
u_o = 0.5 * np.ones(1)
else:
u_o = np.zeros(1)
hydraulic_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, u_o=u_o)
# def test_water_quality_step():
# dt = 10
# Q_in = 1e-2 * np.asarray([1., 0.])
# Q_0Ik = 1e-3 * np.ones(hillslope_nsuperlink_model.NIk)
# c_0j = 10. * np.asarray([1., 0.])
# for _ in range(100):
# hillslope_nsuperlink_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik)
# hillslope_water_quality_model.step(dt=dt, c_0j=c_0j)

def test_norifice():
superjunctions = copy.deepcopy(hillslope_superjunctions)
Expand Down Expand Up @@ -279,36 +209,6 @@ def test_norifice():
u_o = np.zeros(1)
hydraulic_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, u_o=u_o)

def test_weir():
superjunctions = copy.deepcopy(hillslope_superjunctions)
superlinks = hillslope_superlinks
superjunctions.loc[2] = superjunctions.loc[0]
superjunctions.loc[2, ['name', 'id']] = 2
superjunctions.loc[2, 'h_0'] = 2.0
weirs = {
'id' : 0,
'sj_0' : 2,
'sj_1' : 0,
'z_w' : 0,
'y_max' : 0.3048,
'Cr' : 0.67,
'Ct' : 0.67,
'L' : 0.3048,
's' : 0.01
}
weirs = pd.DataFrame(weirs, index=[0])
hydraulic_model = SuperLink(superlinks, superjunctions, weirs=weirs,
internal_links=internal_links)
dt = 10
Q_in = 1e-2 * np.asarray([0., 0., 0.])
Q_0Ik = 1e-3 * np.ones(hydraulic_model.NIk)
for _ in range(100):
if (hydraulic_model.t > 200):
u_w = 0.5 * np.ones(1)
else:
u_w = np.zeros(1)
hydraulic_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, u_w=u_w)

def test_nweir():
superjunctions = copy.deepcopy(hillslope_superjunctions)
superlinks = hillslope_superlinks
Expand Down Expand Up @@ -339,35 +239,6 @@ def test_nweir():
u_w = np.zeros(1)
hydraulic_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, u_w=u_w)

def test_pump():
superjunctions = copy.deepcopy(hillslope_superjunctions)
superlinks = hillslope_superlinks
superjunctions.loc[2] = superjunctions.loc[0]
superjunctions.loc[2, ['name', 'id']] = 2
superjunctions.loc[2, 'h_0'] = 2.0
pumps = {
'id' : 0,
'sj_0' : 0,
'sj_1' : 2,
'z_p' : 0,
'a_q' : 2.0,
'a_h' : 0.1,
'dH_min' : 0.5,
'dH_max' : 2.0
}
pumps = pd.DataFrame(pumps, index=[0])
hydraulic_model = SuperLink(superlinks, superjunctions, pumps=pumps,
internal_links=internal_links)
dt = 10
Q_in = 1e-2 * np.asarray([0., 0., 0.])
Q_0Ik = 1e-3 * np.ones(hydraulic_model.NIk)
for _ in range(100):
if (hydraulic_model.t > 200):
u_p = 0.5 * np.ones(1)
else:
u_p = np.zeros(1)
hydraulic_model.step(dt=dt, Q_in=Q_in, Q_0Ik=Q_0Ik, u_p=u_p)

def test_npump():
superjunctions = copy.deepcopy(hillslope_superjunctions)
superlinks = hillslope_superlinks
Expand All @@ -379,8 +250,9 @@ def test_npump():
'sj_0' : 0,
'sj_1' : 2,
'z_p' : 0,
'a_q' : 2.0,
'a_h' : 0.1,
'a_p' : 1.0,
'b_p' : 1.0,
'c_p' : 1.0,
'dH_min' : 0.5,
'dH_max' : 2.0
}
Expand Down

0 comments on commit 04057af

Please sign in to comment.