Source code for pywatershed.hydrology.pass_through_flow_node
import numpy as np
from ..base.control import Control
from ..base.flow_graph import FlowNode, FlowNodeMaker
from ..constants import nan, zero
[docs]
class PassThroughFlowNode(FlowNode):
"""A FlowNode instance that gives what it takes and dosent store.
See :class:`FlowGraph` for a worked example using PassThroughFlowNode.
"""
[docs]
def __init__(self, control: Control):
"""Initialize a PassThroughFlowNode.
Args:
control: A control object.
"""
self.control = control
self._seg_outflow = np.array([nan])
self._inflow_subtimestep = np.array([nan])
return
[docs]
def prepare_timestep(self):
# self._simulation_time = simulation_time # add argument?
self._accum_inflow = zero
return
[docs]
def calculate_subtimestep(
self,
isubstep: int,
inflow_upstream: float,
inflow_lateral: float,
):
self._inflow_subtimestep = inflow_upstream + inflow_lateral
self._accum_inflow += self._inflow_subtimestep
self._seg_outflow = self._accum_inflow / (isubstep + 1)
return
[docs]
def finalize_timestep(self):
return
[docs]
def advance(self):
return
@property
def outflow(self):
return self._seg_outflow
@property
def outflow_substep(self):
return self._inflow_subtimestep
@property
def storage_change(self):
return zero
@property
def storage(self):
return nan
@property
def sink_source(self):
return zero
[docs]
class PassThroughFlowNodeMaker(FlowNodeMaker):
"""A FlowNodeMaker of PassThroughFlowNodes.
See :class:`FlowGraph` for a worked example using PassThroughFlowNode.
"""
[docs]
def __init__(
self,
) -> None:
"""Initialize a PassThroughFlowNodeMaker."""
self.name = "PassThroughFlowNodeMaker"
[docs]
def get_node(self, control: Control, index: int):
return PassThroughFlowNode(control)