Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
20.0,
0.0,
0.0,
32.0,
30.0,
0.0,
0.0,
0.0,
0.0,
0.0,
]
)
mg.add_field("node", "topographic__elevation", topographic__elevation)
mg.set_closed_boundaries_at_grid_edges(True, True, True, False)
fa = LossyFlowAccumulator(mg, topographic__elevation, runoff_rate=10.0)
assert_array_equal(
mg.at_node["water__unit_flux_in"], 10.0 * np.ones(mg.size("node"))
)
fa.run_one_step()
reciever = np.array(
[0, 1, 2, 3, 4, 1, 2, 7, 8, 10, 6, 11, 12, 14, 10, 15, 16, 17, 18, 19]
)
da = np.array(
[
0.0,
1.0,
5.0,
0.0,
0.0,
with pytest.raises(NotImplementedError):
LossyFlowAccumulator(
mg0, flow_director="MFD", depression_finder="DepressionFinderAndRouter"
)
with pytest.raises(NotImplementedError):
LossyFlowAccumulator(
mg0, flow_director="DINF", depression_finder="DepressionFinderAndRouter"
)
fa0 = LossyFlowAccumulator(mg0, flow_director="MFD")
fa0.run_one_step()
with pytest.raises(NotImplementedError):
DepressionFinderAndRouter(mg0)
fa1 = LossyFlowAccumulator(mg1, flow_director="DINF")
fa1.run_one_step()
with pytest.raises(NotImplementedError):
DepressionFinderAndRouter(mg1)
def test_water_discharge_in_supplied():
mg = RasterModelGrid((5, 5), xy_spacing=(1, 1))
mg.add_field("topographic__elevation", mg.node_x + mg.node_y, at="node")
mg.add_field("water__discharge_in", mg.node_x + mg.node_y, at="node")
with pytest.deprecated_call():
LossyFlowAccumulator(mg)
def test_flow_accumulator_properties():
mg = RasterModelGrid((5, 5), xy_spacing=(1, 1))
mg.add_field("topographic__elevation", mg.node_x + mg.node_y, at="node")
fa = LossyFlowAccumulator(mg)
fa.run_one_step()
node_drainage_area = np.array(
[
0.0,
3.0,
3.0,
3.0,
0.0,
0.0,
3.0,
3.0,
3.0,
0.0,
0.0,
2.0,
z = mg.add_field(
"topographic__elevation", mg.node_x ** 2 + mg.node_y ** 2, at="node"
)
LossyFlowAccumulator(mg)
assert_array_equal(z, mg.at_node["topographic__elevation"])
assert_array_equal(np.zeros(100), mg.at_node["drainage_area"])
assert_array_equal(np.ones(100), mg.at_node["water__unit_flux_in"])
assert_array_equal(np.zeros(100), mg.at_node["surface_water__discharge_loss"])
LossyFlowAccumulator(mg, runoff_rate=2.0)
assert_array_equal(np.full(100, 2.0), mg.at_node["water__unit_flux_in"])
# quick test that the component binds correctly to an existing field:
L = mg.at_node["surface_water__discharge_loss"]
fa = LossyFlowAccumulator(mg)
fa.run_one_step() # this line is padding to make flake8 happy
L[0] = 1.0
assert mg.at_node["surface_water__discharge_loss"] is L
)
fa0 = LossyFlowAccumulator(mg0, flow_director="D4")
fa0.run_one_step()
mg1 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg1.add_field(
"topographic__elevation", mg1.node_x ** 2 + mg1.node_y ** 2, at="node"
)
fa1 = LossyFlowAccumulator(mg1, flow_director="Steepest")
fa1.run_one_step()
mg2 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg2.add_field(
"topographic__elevation", mg2.node_x ** 2 + mg2.node_y ** 2, at="node"
)
fa2 = LossyFlowAccumulator(mg2, flow_director=FlowDirectorSteepest)
fa2.run_one_step()
mg3 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg3.add_field(
"topographic__elevation", mg3.node_x ** 2 + mg3.node_y ** 2, at="node"
)
fd = FlowDirectorSteepest(mg3)
fa3 = LossyFlowAccumulator(mg3, flow_director=fd)
fa3.run_one_step()
for key in mg0.at_node.keys():
assert_array_equal(mg0.at_node[key], mg1.at_node[key])
assert_array_equal(mg1.at_node[key], mg2.at_node[key])
assert_array_equal(mg2.at_node[key], mg3.at_node[key])
def test_director_adding_methods_are_equivalent_Steepest():
"""Check that different methods to specifying the director are the same."""
mg0 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg0.add_field(
"topographic__elevation", mg0.node_x ** 2 + mg0.node_y ** 2, at="node"
)
fa0 = LossyFlowAccumulator(mg0, flow_director="D4")
fa0.run_one_step()
mg1 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg1.add_field(
"topographic__elevation", mg1.node_x ** 2 + mg1.node_y ** 2, at="node"
)
fa1 = LossyFlowAccumulator(mg1, flow_director="Steepest")
fa1.run_one_step()
mg2 = RasterModelGrid((10, 10), xy_spacing=(1, 1))
mg2.add_field(
"topographic__elevation", mg2.node_x ** 2 + mg2.node_y ** 2, at="node"
)
fa2 = LossyFlowAccumulator(mg2, flow_director=FlowDirectorSteepest)
fa2.run_one_step()
def test_depression_finder_bad_uninstantiated_component():
mg = RasterModelGrid((5, 5), xy_spacing=(1, 1))
mg.add_field("topographic__elevation", mg.node_x + mg.node_y, at="node")
with pytest.raises(ValueError):
LossyFlowAccumulator(mg, flow_director="D8", depression_finder=LinearDiffuser)
3.0,
4.0,
4.0,
4.0,
4.0,
5.0,
5.0,
5.0,
5.0,
]
mg.add_field("node", "topographic__elevation", topographic__elevation)
mg.set_closed_boundaries_at_grid_edges(True, True, True, False)
fa = LossyFlowAccumulator(mg, "topographic__elevation", runoff_rate=runoff_rate)
fa.run_one_step()
reciever = np.array(
[0, 1, 2, 3, 4, 1, 2, 7, 8, 10, 6, 11, 12, 14, 10, 15, 16, 17, 18, 19]
)
da = np.array(
[
0.0,
1.0,
5.0,
0.0,
0.0,
1.0,
5.0,
0.0,
FlowDirectorMFD,
FlowDirectorSteepest,
FractureGridGenerator,
gFlex,
GroundwaterDupuitPercolator,
HackCalculator,
HeightAboveDrainageCalculator,
KinwaveImplicitOverlandFlow,
KinwaveOverlandFlowModel,
LakeMapperBarnes,
LandslideProbability,
LateralEroder,
LinearDiffuser,
LithoLayers,
Lithology,
LossyFlowAccumulator,
NetworkSedimentTransporter,
NormalFault,
OverlandFlow,
OverlandFlowBates,
PerronNLDiffuse,
PotentialEvapotranspiration,
PotentialityFlowRouter,
PrecipitationDistribution,
Profiler,
Radiation,
SedDepEroder,
SinkFiller,
SinkFillerBarnes,
SoilMoisture,
SoilInfiltrationGreenAmpt,
Space,