Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_field_name_array_float_case6():
"""Topography as array, runoff rate as array"""
mg = RasterModelGrid((5, 4), xy_spacing=(1, 1))
topographic__elevation = np.array(
[
0.0,
0.0,
0.0,
0.0,
0.0,
21.0,
10.0,
0.0,
0.0,
31.0,
20.0,
0.0,
0.0,
32.0,
def test_D_infinity_flat_closed_lower():
mg = RasterModelGrid((5, 4), xy_spacing=(1, 1))
z = mg.add_zeros("node", "topographic__elevation")
z[mg.core_nodes] += 1
mg.set_closed_boundaries_at_grid_edges(
bottom_is_closed=True,
left_is_closed=True,
right_is_closed=True,
top_is_closed=True,
)
fd = FlowDirectorDINF(mg)
fd.run_one_step()
node_ids = np.arange(mg.number_of_nodes)
true_recievers = -1 * np.ones(fd.receivers.shape)
true_recievers[:, 0] = node_ids
def test_netcdf_write_uint8_field_netcdf4(tmpdir):
"""Test write_netcdf with a grid that has an uint8 field."""
field = RasterModelGrid((4, 3))
field.add_field("node", "topographic__elevation", np.arange(12, dtype=np.uint8))
with tmpdir.as_cwd():
write_netcdf("test.nc", field, format="NETCDF4")
root = nc.Dataset("test.nc", "r", format="NETCDF4")
for name in ["topographic__elevation"]:
assert name in root.variables
assert_array_equal(root.variables[name][:].flatten(), field.at_node[name])
assert root.variables[name][:].dtype == "uint8"
root.close()
def test_providing_array_and_kwargs():
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(len(mg.at_node["topographic__elevation"]), dtype=np.uint8)
mask[np.where(mg.at_node["drainage_area"] > 5)] = 1
with pytest.warns(UserWarning):
DrainageDensity(mg, channel__mask=mask, area_coefficient=1)
with pytest.warns(UserWarning):
DrainageDensity(mg, channel__mask=mask, slope_coefficient=1)
with pytest.warns(UserWarning):
DrainageDensity(mg, channel__mask=mask, area_exponent=1)
with pytest.warns(UserWarning):
DrainageDensity(mg, channel__mask=mask, slope_exponent=1)
with pytest.warns(UserWarning):
def test_diagonal_list_with_scalar_arg():
rmg = RasterModelGrid((5, 4))
assert_array_equal(rmg.diagonal_adjacent_nodes_at_node[6], np.array([11, 9, 1, 3]))
assert_array_equal(rmg.diagonal_adjacent_nodes_at_node[-1], np.array([X, X, 14, X]))
assert_array_equal(
rmg.diagonal_adjacent_nodes_at_node[-2], np.array([X, X, 13, 15])
)
params = {
"shape": (10, 20),
"xy_spacing": (25, 45),
"bc": {
"right": "closed",
"top": "closed",
"left": "closed",
"bottom": "closed",
},
"xy_of_lower_left": (35, 55),
"axis_name": ("spam", "eggs"),
"axis_units": ("smoot", "parsec"),
"xy_of_reference": (12345, 678910),
}
mg = RasterModelGrid.from_dict(params)
# assert things.
assert mg.shape == (10, 20)
assert mg.dx == 25
assert mg.dy == 45
assert (mg.x_of_node.min(), mg.y_of_node.min()) == (35, 55)
assert np.all(mg.status_at_node[mg.boundary_nodes] == CLOSED_BOUNDARY)
assert mg.axis_units == ("smoot", "parsec")
assert mg.axis_name == ("spam", "eggs")
assert mg.xy_of_reference == (12345, 678910)
def test_add_field_from_function_constant(at, to_list):
grid = RasterModelGrid((20, 30))
function = to_list(("constant", dict(value=1.0)))
add_field_from_function(grid, "z", function, at=at)
assert_array_almost_equal(grid[at]["z"], 1.0)
def test_getters_and_setters():
grid = RasterModelGrid((3, 5))
grid.add_zeros("topographic__elevation", at="node")
tfc = TidalFlowCalculator(grid)
tfc.roughness = 0.01
assert_array_equal(tfc.roughness, 0.01 + np.zeros(grid.number_of_links))
tfc.tidal_range = 4.0
assert_equal(tfc.tidal_range, 4.0)
assert_equal(tfc._tidal_half_range, 2.0)
tfc.tidal_period = 40000.0
assert_equal(tfc.tidal_period, 40000.0)
assert_equal(tfc._tidal_half_period, 20000.0)
tfc.mean_sea_level = 1.0
assert_equal(tfc.mean_sea_level, 1.0)
fa.run_one_step()
assert sorted(list(mg.at_node.keys())) == [
"drainage_area",
"flow__data_structure_delta",
"flow__link_to_receiver_node",
"flow__receiver_node",
"flow__sink_flag",
"flow__upstream_node_order",
"surface_water__discharge",
"topographic__elevation",
"topographic__steepest_slope",
"water__unit_flux_in",
]
mg2 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg2.add_field("topographic__elevation", mg2.node_x + mg2.node_y, at="node")
fa2 = FlowAccumulator(mg2, flow_director="MFD")
fa2.run_one_step()
assert sorted(list(mg2.at_node.keys())) == [
"drainage_area",
"flow__data_structure_delta",
"flow__link_to_receiver_node",
"flow__receiver_node",
"flow__receiver_proportions",
"flow__sink_flag",
"flow__upstream_node_order",
"surface_water__discharge",
"topographic__elevation",
"topographic__steepest_slope",
"water__unit_flux_in",
]
def main():
"""
do some stuff
"""
# User-defined parameter values
nr = 5
nc = 6
nnodes = nr*nc
dx=1
#instantiate grid
rg = landlab.RasterModelGrid(nr, nc, dx)
#rg.set_inactive_boundaries(False, False, True, True)
nodata_val=0
elevations = nodata_val*np.ones( nnodes )
#set-up interior elevations with random numbers
#for i in range(0, nnodes):
# if rg.is_interior(i):
# elevations[i]=random.random_sample()
#set-up with prescribed elevations to test drainage area calcualtion
helper = [7,8,9,10,13,14,15,16]
elevations[helper]=2
helper = [19,20,21,22]
elevations[helper]=3
elevations[7]=1