Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
runtime = inputs.read_float("total_time")
dt = inputs.read_float("dt")
nt = int(runtime // dt)
uplift_per_step = uplift_rate * dt
mg = RasterModelGrid((nrows, ncols), xy_spacing=(dx, dx))
mg.add_zeros("topographic__elevation", at="node")
z = np.loadtxt(os.path.join(_THIS_DIR, "seddepinit.txt"))
mg["node"]["topographic__elevation"] = z
mg.set_closed_boundaries_at_grid_edges(True, False, True, False)
fr = FlowAccumulator(mg, flow_director="D8")
sde = SedDepEroder(mg, **inputs)
for i in range(nt):
mg.at_node["topographic__elevation"][mg.core_nodes] += uplift_per_step
mg = fr.run_one_step()
mg, _ = sde.erode(dt)
z_tg = np.loadtxt(os.path.join(_THIS_DIR, "seddepz_tg.txt"))
assert_array_almost_equal(
mg.at_node["topographic__elevation"][mg.core_nodes], z_tg[mg.core_nodes]
)
profile_IDs,
mg.at_node["flow__link_to_receiver_node"],
)
prf.plot_profiles(
dists_upstr, profile_IDs, mg.at_node["topographic__elevation"]
)
print("completed run to steady state...")
if show_figs_in_run:
show() # will cause a hang
# save a copy of the init conditions:
mg_init = deepcopy(mg)
# REinstantiate the components:
fr = FlowAccumulator(mg, flow_director="D8")
tle = TransportLimitedEroder(mg, input_file)
uplift_rate *= 10. # accelerate tenfold
runtime = 200000.
nt = int(runtime // dt)
uplift_per_step = uplift_rate * dt
print("uplift per step: {0}".format(uplift_per_step))
x_profiles = []
z_profiles = []
S_profiles = []
A_profiles = []
# plot init conds
if make_output_plots:
mg = fr.route_flow(grid=mg)
pylab.figure("long_profile_anim")
def test_phi_error_raised():
mg = RasterModelGrid((10, 10))
z = mg.add_zeros("topographic__elevation", at="node")
z += mg.x_of_node + mg.y_of_node
fa = FlowAccumulator(mg)
fa.run_one_step()
with pytest.raises(ValueError):
ErosionDeposition(mg, phi=0)
mg.nodes_at_left_edge,
mg.nodes_at_right_edge,
):
mg.status_at_node[edge] = CLOSED_BOUNDARY
for edge in mg.nodes_at_bottom_edge:
mg.status_at_node[edge] = FIXED_VALUE_BOUNDARY
z = mg.add_zeros("node", "topographic__elevation")
loading_vector = np.linspace(1, 4, num=nr)
ramp = np.repeat(loading_vector, nc)
# the tweaks to elevation below make lateral node at node 7
z += ramp
z[11] -= 0.9
z[12] -= 0.4
z[8] -= 0.001
fa = FlowAccumulator(
mg,
surface="topographic__elevation",
flow_director="FlowDirectorD8",
runoff_rate=None,
depression_finder=None,
)
latero = LateralEroder(mg, latero_mech="UC", Kv=0.1, Kl_ratio=1.5)
fa.accumulate_flow()
(mg, dzlat) = latero.run_one_step(dt=1.0)
vlname = mg["node"]["volume__lateral_erosion"]
# predicted volume of lateral eorsion
pred_vollat = 0.00045158164
# predicted elevation after 1 time step
pred_zafter = np.array(
[
ramp = np.repeat(loading_vector, nc)
ramp += np.random.random_sample(mg.number_of_nodes) * 0.8
z += ramp
# set inlet node = true, provide inlet node id, inlet drainage area, and inlet qs.
latero = LateralEroder(
mg,
latero_mech="UC",
Kv=0.001,
Kl_ratio=1.0,
inlet_on=True,
inlet_node=17,
inlet_area=500,
qsinlet=2.5,
)
fa = FlowAccumulator(
mg,
surface="topographic__elevation",
flow_director="FlowDirectorD8",
runoff_rate=None,
depression_finder=None,
)
for i in range(2000):
fa.run_one_step() # flow accumulator
# erode the landscape with lateral erosion
(mg, dzlat) = latero.run_one_step(dt)
mg.at_node["topographic__elevation"][mg.core_nodes] += (
U * dt
) # uplift the landscape
da = mg.at_node["surface_water__discharge"] / dx ** 2
num_sedflux = mg.at_node["qs"]
def test_erodep_slope_area_with_vs_unity():
"""Test steady state run with Vs = 1."""
# Set up a 5x5 grid with open boundaries and low initial elevations.
rg = RasterModelGrid((5, 5))
z = rg.add_zeros("topographic__elevation", at="node")
z[:] = 0.01 * rg.x_of_node
# Create a D8 flow handler
fa = FlowAccumulator(rg, flow_director="FlowDirectorD8")
# test: Vs = 1
K = 0.002
vs = 1.0
U = 0.001
dt = 10.0
# Create the ErosionDeposition component...
ed = ErosionDeposition(rg, K=K, v_s=vs, m_sp=0.5, n_sp=1.0, solver="adaptive")
# ... and run it to steady state.
for i in range(1000):
fa.run_one_step()
ed.run_one_step(dt=dt)
z[rg.core_nodes] += U * dt
from six.moves import range
from landlab import VoronoiDelaunayGrid
from landlab.components import FlowAccumulator, StreamPowerEroder
from landlab.plot.imshow import imshow_node_grid
import numpy as np
from matplotlib.pyplot import figure, show
nnodes = 10000
x, y = np.random.rand(nnodes), np.random.rand(nnodes)
mg = VoronoiDelaunayGrid(x,y)
z = mg.add_field('node', 'topographic__elevation', np.random.rand(nnodes)/10000., copy=False)
fr = FlowAccumulator(mg)
spe = StreamPowerEroder(mg, 'drive_sp_params_voronoi.txt')
for i in range(100):
z[mg.core_nodes] += 0.01
fr.run_one_step()
spe.erode(mg, 1.)
imshow_node_grid(mg, 'topographic__elevation')
show()
def test_bad_mask_size():
mg = RasterModelGrid((10, 10))
mg.add_zeros("node", "topographic__elevation")
noise = np.random.rand(mg.size("node"))
mg.at_node["topographic__elevation"] += noise
fr = FlowAccumulator(mg, flow_director="D8")
fr.run_one_step()
mask = np.zeros(20)
mask[np.where(mg.at_node["drainage_area"][:20] > 5)] = 1
with pytest.raises(ValueError):
DrainageDensity(mg, channel__mask=mask)
def test_can_run_with_hex():
"""Test that model can run with hex model grid."""
# Set up a 5x5 grid with open boundaries and low initial elevations.
mg = HexModelGrid((7, 7))
z = mg.add_zeros("topographic__elevation", at="node")
_ = mg.add_zeros("soil__depth", at="node")
z[:] = 0.01 * mg.x_of_node
# Create a D8 flow handler
fa = FlowAccumulator(mg, flow_director="FlowDirectorSteepest")
# Parameter values for test 1
U = 0.001
dt = 10.0
# Create the Space component...
sp = Space(
mg,
K_sed=0.00001,
K_br=0.00000000001,
F_f=0.5,
phi=0.1,
H_star=1.0,
v_s=0.001,
m_sp=0.5,
n_sp=1.0,
#create the fields in the grid
mg.add_zeros('topographic__elevation', at='node')
z = mg.zeros(at='node') + init_elev
mg['node'][ 'topographic__elevation'] = z + numpy.random.rand(len(z))/1000.
mg.add_zeros('node', 'water__unit_flux_in')
#make some K values in a field to test
#mg.at_node['K_values'] = 0.1+numpy.random.rand(nrows*ncols)/10.
mg.at_node['K_values'] = numpy.empty(nrows*ncols, dtype=float)
#mg.at_node['K_values'].fill(0.1+numpy.random.rand()/10.)
mg.at_node['K_values'].fill(0.001)
print( 'Running ...' )
#instantiate the components:
fr = FlowAccumulator(mg, flow_director='D8')
sp = StreamPowerEroder(mg, input_file_string)
#fsp = FastscapeEroder(mg, input_file_string)
precip = PrecipitationDistribution(input_file=input_file_string)
#load the Fastscape module too, to allow direct comparison
fsp = FastscapeEroder(mg, input_file_string)
try:
#raise NameError
mg = copy.deepcopy(mg_mature)
except NameError:
print('building a new grid...')
out_interval = 50000.
last_trunc = time_to_run #we use this to trigger taking an output plot
#run to a steady state:
#We're going to cheat by running Fastscape SP for the first part of the solution