code: import numpy as np
  import pytest
  from numpy import testing
  
  from landlab import RasterModelGrid
  from landlab.components import ErosionDeposition, FlowAccumulator
  
  
  def test_Ff_too_high_vals():
      """
      Test that instantiating ErosionDeposition with a F_f value > 1 throws a
      ValueError.
      """
  
      # set up a 5x5 grid with one open outlet node and low initial elevations.
      nr = 5
      nc = 5
      mg = RasterModelGrid((nr, nc), xy_spacing=10.0)
  
      mg.add_zeros("topographic__elevation", at="node")
  
      mg["node"]["topographic__elevation"] += (
          mg.node_y / 100000 + mg.node_x / 100000 + np.random.rand(len(mg.node_y)) / 10000
      )
      mg.set_closed_boundaries_at_grid_edges(
          bottom_is_closed=True,
          left_is_closed=True,
          right_is_closed=True,
          top_is_closed=True,
      )
      mg.set_watershed_boundary_condition_outlet_id(
          0, mg["node"]["topographic__elevation"], -9999.0
      )
  
      # Create a D8 flow handler
      FlowAccumulator(
          mg, flow_director="D8", depression_finder="DepressionFinderAndRouter"
      )
  
      # Instantiate the ErosionDeposition component...
      with pytest.raises(ValueError):
          ErosionDeposition(
              mg,
              K=0.01,
              F_f=2.0,
              v_s=0.001,
              m_sp=0.5,
              n_sp=1.0,
              sp_crit=0.0,
              solver="basic",
          )
  
  
  def test_Ff_too_low_vals():
      """
      Test that instantiating ErosionDeposition with a F_f value < 0 throws a
      ValueError.
      """
  
      # set up a 5x5 grid with one open outlet node and low initial elevations.
      nr = 5
      nc = 5
      mg = RasterModelGrid((nr, nc), xy_spacing=10.0)
  
      mg.add_zeros("topographic__elevation", at="node")
  
      mg["node"]["topographic__elevation"] += (
          mg.node_y / 100000 + mg.node_x / 100000 + np.random.rand(len(mg.node_y)) / 10000
      )
      mg.set_closed_boundaries_at_grid_edges(
          bottom_is_closed=True,
          left_is_closed=True,
          right_is_closed=True,
          top_is_closed=True,
      )
      mg.set_watershed_boundary_condition_outlet_id(
          0, mg["node"]["topographic__elevation"], -9999.0
      )
  
      # Create a D8 flow handler
      FlowAccumulator(
          mg, flow_director="D8", depression_finder="DepressionFinderAndRouter"
      )
  
      # Instantiate the ErosionDeposition component...
      with pytest.raises(ValueError):
          ErosionDeposition(
              mg,
              K=0.01,
              F_f=-0.5,
              v_s=0.001,
              m_sp=0.5,
              n_sp=1.0,
              sp_crit=0.0,
              solver="basic",
          )
  
  
  def test_q_as_field():
      """
      Test that passing in water discharge as a grid field results in self.q
      holding correct values
      """
  
      # set up a 5x5 grid with one open outlet node and low initial elevations.
      nr = 5
      nc = 5
      mg = RasterModelGrid((nr, nc), xy_spacing=10.0)
  
      mg.add_zeros("topographic__elevation", at="node")
      q = mg.add_zeros("user_imposed_discharge", at="node")
      q[:] += 1.0  # add 1.0 m3/yr of water
  
      mg["node"]["topographic__elevation"] += (
          mg.node_y / 100000 + mg.node_x / 100000 + np.random.rand(len(mg.node_y)) / 10000
      )
      mg.set_closed_boundaries_at_grid_edges(
          bottom_is_closed=True,
          left_is_closed=True,
          right_is_closed=True,
          top_is_closed=True,
      )
      mg.set_watershed_boundary_condition_outlet_id(
          0, mg["node"]["topographic__elevation"], -9999.0
      )
  
      # Create a D8 flow handler
      FlowAccumulator(
          mg, flow_director="D8", depression_finder="DepressionFinderAndRouter"
      )
  
      # Instantiate the ErosionDeposition component...
      ed = ErosionDeposition(
          mg,
          K=0.01,
          F_f=0.0,
          v_s=0.001,
          m_sp=0.5,
          n_sp=1.0,
          sp_crit=0.0,
          discharge_field="user_imposed_discharge",
          solver="basic",
      )
  
      # ensure that ed._q is everywhere equal to 1.0 m3/yr.
      testing.assert_array_equal(
          np.ones(mg.number_of_nodes),
          ed._q,
          err_msg="E/D discharge field test failed",
          verbose=True,
      )
  
  
  def test_q_as_array():
      """
      Test that passing in water discharge as an array results in self.q
      holding correct values
      """
  
      # set up a 5x5 grid with one open outlet node and low initial elevations.
      nr = 5
      nc = 5
      mg = RasterModelGrid((nr, nc), xy_spacing=10.0)
  
      mg.add_zeros("topographic__elevation", at="node")
      q = np.zeros(mg.number_of_nodes)
      q[:] += 1.0  # add 1.0 m3/yr of water
  
      mg["node"]["topographic__elevation"] += (
          mg.node_y / 100000 + mg.node_x / 100000 + np.random.rand(len(mg.node_y)) / 10000
      )
      mg.set_closed_boundaries_at_grid_edges(
          bottom_is_closed=True,
          left_is_closed=True,
          right_is_closed=True,
          top_is_closed=True,
      )
      mg.set_watershed_boundary_condition_outlet_id(
          0, mg["node"]["topographic__elevation"], -9999.0
      )
  
      # Create a D8 flow handler
      FlowAccumulator(
          mg, flow_director="D8", depression_finder="DepressionFinderAndRouter"
      )
  
      # Instantiate the ErosionDeposition component...
      ed = ErosionDeposition(
          mg,
          K=0.01,
          F_f=0.0,
          v_s=0.001,
          m_sp=0.5,
          n_sp=1.0,
          sp_crit=0.0,
          discharge_field=q,
          solver="basic",
      )
  
      # ensure that ed._q is everywhere equal to 1.0 m3/yr.
      testing.assert_array_equal(
          np.ones(mg.number_of_nodes),
          ed._q,
          err_msg="E/D discharge array test failed",
          verbose=True,
      )
  
  
  def test_sediment__flux_already_created():
      """
      Test that an existing sediment flux grid field is not changed by
      instantiating ErosionDeposition.
      """
  
      # set up a 5x5 grid with one open outlet node and low initial elevations.
      nr = 5
      nc = 5
      mg = RasterModelGrid((nr, nc), xy_spacing=10.0)
  
      mg.add_zeros("topographic__elevation", at="node")
      qs = mg.add_zeros("sediment__flux", at="node")
      qs[:] += 1.0  # add 1.0 m3/yr of flux
  
      mg["node"]["topographic__elevation"] += (
          mg.node_y / 100000 + mg.node_x / 100000 + np.random.rand(len(mg.node_y)) / 10000
      )
      mg.set_closed_boundaries_at_grid_edges(
          bottom_is_closed=True,
          left_is_closed=True,
          right_is_closed=True,
          top_is_closed=True,
      )
      mg.set_watershed_boundary_condition_outlet_id(
          0, mg["node"]["topographic__elevation"], -9999.0
      )
  
      # Create a D8 flow handler
      FlowAccumulator(
          mg, flow_director="D8", depression_finder="DepressionFinderAndRouter"
      )
  
      # Instantiate the ErosionDeposition component...
      ed = ErosionDeposition(
          mg,
          K=0.01,
          F_f=0.0,
          v_s=0.001,
          m_sp=0.5,
          n_sp=1.0,
          sp_crit=0.0,
          solver="basic",
      )
  
      # ensure that 'sediment__flux' field is everywhere equal to 1.0 m3/yr.
      testing.assert_array_equal(
          np.ones(mg.number_of_nodes),
          ed._qs,
          err_msg="E/D sediment flux field test failed",
          verbose=True,
      )
   # coding: utf8
  # ! /usr/env/python
  """Base class for profile constructors."""
  
  from abc import ABC, abstractmethod
  
  import matplotlib.pyplot as plt
  from matplotlib.collections import LineCollection
  
  from landlab import Component
  from landlab.plot import imshow_grid
  from landlab.utils.return_array import return_array_at_node
  
  
  def _recursive_max(jagged):
      """
      Examples
      --------
      from landlab.components.profiler.base_profiler import _recursive_max
      >>> struct = [[1, 2, 3, 4],
      ...           [[2, 3, 4, 5],
      ...            [3, 4, 5, 6]],
      ...           [4, 5, 6, 7]]
      >>> _recursive_max(struct)
      7
      >>> _recursive_max([100])
      100
      """
      return max(_recursive_max(j) if hasattr(j, "__iter__") else j for j in jagged)
  
  
  def _recursive_min(jagged):
      """
      Examples
      --------
      from landlab.components.profiler.base_profiler import _recursive_min
      >>> struct = [[1, 2, 3, 4],
      ...           [[2, 3, 4, 5],
      ...            [3, 4, 5, 6]],
      ...           [4, 5, 6, 7]]
      >>> _recursive_min(struct)
      1
      >>> _recursive_min([100])
      100
      """
      return min(_recursive_min(j) if hasattr(j, "__iter__") else j for j in jagged)
  
  
  class _BaseProfiler(ABC, Component):
      """Base class to handle profilers.
  
      Primarily exists to handle plotting.
      """
  
      _name = "_BaseProfiler"
  
      _unit_agnostic = True
  
      _info = {}
  
      def __init__(self, grid):
          super().__init__(grid)
  
      def run_one_step(self):
          """Calculate the profile data structure and distances along it."""
          # calculate the profile IDs data structure.
          self._create_profile_structure()
  
      @abstractmethod
      def _create_profile_structure(self):
          """Private class for creating profile structure.
  
          Expectation is that this will be overridden to create the following
          three private attributes:
  
          self._nodes
          self._distance_along_profile
  
          are each lists of numpy arrays, one array per segment.
  
          self._colors
  
          is a list of RGBA tuples, one tuple per segment.
  
          The order of segments is expected to be consistent between each of the
          three data structures.
          """
          ...  # pragma: no cover
  
      @property
      def distance_along_profile(self):
          """List of distances along profile for each segment.
  
          Examples
          --------
          >>> import numpy as np
          >>> from landlab import RasterModelGrid
          >>> from landlab.components import (
          ...     FastscapeEroder,
          ...     FlowAccumulator,
          ...     ChannelProfiler
          ...     )
          >>> mg = RasterModelGrid((10, 10), xy_spacing=10)
          >>> np.random.seed(42)
          >>> z = mg.add_zeros('topographic__elevation', at='node')
          >>> z[mg.core_nodes] += np.random.randn(mg.core_nodes.size)
          >>> fa = FlowAccumulator(mg)
          >>> sp = FastscapeEroder(mg, K_sp=0.0001)
          >>> dt = 1000
          >>> for i in range(200):
          ...     fa.run_one_step()
          ...     sp.run_one_step(dt=dt)
          ...     z[mg.core_nodes] += 0.001 * dt
          >>> profiler = ChannelProfiler(mg)
          >>> profiler.run_one_step()
          >>> profiler.distance_along_profile
          [array([  0.,  10.,  20.,  30.,  40.,  50.])]
          """
          return self._distance_along_profile
  
      @property
      def nodes(self):
          """List of node ids for each segment.
  
          Examples
          --------
          >>> import numpy as np
          >>> from landlab import RasterModelGrid
          >>> from landlab.components import (
          ...     FastscapeEroder,
          ...     FlowAccumulator,
          ...     ChannelProfiler
          ...     )
          >>> mg = RasterModelGrid((10, 10), xy_spacing=10)
          >>> np.random.seed(42)
          >>> z = mg.add_zeros('topographic__elevation', at='node')
          >>> z[mg.core_nodes] += np.random.randn(mg.core_nodes.size)
          >>> fa = FlowAccumulator(mg)
          >>> sp = FastscapeEroder(mg, K_sp=0.0001)
          >>> dt = 1000
          >>> for i in range(200):
          ...     fa.run_one_step()
          ...     sp.run_one_step(dt=dt)
          ...     z[mg.core_nodes] += 0.001 * dt
          >>> profiler = ChannelProfiler(mg)
          >>> profiler.run_one_step()
          >>> profiler.nodes
          [array([59, 58, 57, 56, 46, 45])]
          """
          return self._nodes
  
      @property
      def colors(self):
          """List of colors for each segment.
  
          Examples
          --------
          >>> import numpy as np
          >>> from landlab import RasterModelGrid
          >>> from landlab.components import (
          ...     FastscapeEroder,
          ...     FlowAccumulator,
          ...     ChannelProfiler
          ...     )
          >>> mg = RasterModelGrid((10, 10), xy_spacing=10)
          >>> np.random.seed(42)
          >>> z = mg.add_zeros('topographic__elevation', at='node')
          >>> z[mg.core_nodes] += np.random.randn(mg.core_nodes.size)
          >>> fa = FlowAccumulator(mg)
          >>> sp = FastscapeEroder(mg, K_sp=0.0001)
          >>> dt = 1000
          >>> for i in range(200):
          ...     fa.run_one_step()
          ...     sp.run_one_step(dt=dt)
          ...     z[mg.core_nodes] += 0.001 * dt
          >>> profiler = ChannelProfiler(mg)
          >>> profiler.run_one_step()
          >>> np.round(profiler.colors, decimals=2)
          array([[ 0.27,  0.  ,  0.33,  1.  ]])
          """
          return self._colors
  
      def plot_profiles(
          self,
          field="topographic__elevation",
          xlabel="Distance Along Profile",
          ylabel="Plotted Quantity",
          title="Extracted Profiles",
          color=None,
      ):
          """Plot distance-upstream vs at at-node or size (nnodes,) quantity.
  
          Parameters
          ----------
          field : field name or nnode array
              Array of the at-node-field to plot against distance upstream.
              Default value is the at-node field 'topographic__elevation'.
          xlabel : str, optional
              X-axis label, default is "Distance Along Profile".
          ylabel : str, optional
              Y-axis label, default value is "Plotted Quantity".
          title : str, optional
              Plot title, default value is "Extracted Profiles".
          color : RGBA tuple or color string
              Color to use in order to plot all profiles the same color. Default
              is None, and the colors assigned to each profile are used.
          """
          quantity = return_array_at_node(self._grid, field)
  
          # create segments the way that line collection likes them.
          segments = []
          qmin = []
          qmax = []
          for idx, nodes in enumerate(self._nodes):
              segments.append(
                  list(zip(self._distance_along_profile[idx], quantity[nodes]))
              )
              qmin.append(min(quantity[nodes]))
              qmax.append(max(quantity[nodes]))
  
          # We need to set the plot limits.
          ax = plt.gca()
          ax.set_xlim(
              _recursive_min(self._distance_along_profile),
              _recursive_max(self._distance_along_profile),
          )
          ax.set_ylim(min(qmin), max(qmax))
  
          line_segments = LineCollection(segments)
          colors = color or self._colors
          line_segments.set_color(colors)
          ax.add_collection(line_segments)
          ax.set_xlabel(xlabel)
          ax.set_ylabel(ylabel)
          ax.set_title(title)
  
      def plot_profiles_in_map_view(
          self, field="topographic__elevation", endpoints_only=False, color=None, **kwds
      ):
          """Plot profile locations in map view.
  
          Parameters
          ----------
          field : field name or nnode array
              Array of the at-node-field to plot as the 2D map values.
              Default value is the at-node field 'topographic__elevation'.
          endpoints_only : boolean
              Boolean where False (default) indicates every node along the
              profile is plotted, or True indicating only segment endpoints are
              plotted.
          color : RGBA tuple or color string
              Color to use in order to plot all profiles the same color. Default
              is None, and the colors assigned to each profile are used.
          **kwds : dictionary
              Keyword arguments to pass to imshow_grid.
          """
          # make imshow_grid background
          imshow_grid(self._grid, field, **kwds)
          ax = plt.gca()
  
          # create segments the way that line collection likes them.
          segments = []
          for idx, nodes in enumerate(self._nodes):
              if endpoints_only:
                  select_nodes = [nodes[0], nodes[-1]]
                  segments.append(
                      list(
                          zip(
                              self._grid.x_of_node[select_nodes],
                              self._grid.y_of_node[select_nodes],
                          )
                      )
                  )
  
              else:
                  segments.append(
                      list(zip(self._grid.x_of_node[nodes], self._grid.y_of_node[nodes]))
                  )
  
          line_segments = LineCollection(segments)
          colors = color or self._colors
          line_segments.set_color(colors)
          ax.add_collection(line_segments)
  
apis: numpy.ones numpy.zeros
apis: matplotlib.pyplot.gca matplotlib.collections.LineCollection