Last update 04/05/2021
Expand SuperflexPy: Build customized components¶
Adding routing to a node¶
Nodes in SuperflexPy have the capability to apply a lag to the fluxes simulated by the units. Such lags can represent routing delays in the fluxes as they propagate through the catchment (“internal” routing), or routing delays associated with the river network (“external” routing). Both types of routing can be implemented within a SuperflexPy node.
The default implementation of the node (
Node class in
superflexpy.framework.node) does not provide the routing functionality.
external_routing exist but are
set to simply return the incoming fluxes without any transformation.
To support routing within a node, we need to create a customized node that implements
for given lag functions. The object-oriented design of
SuperflexPy simplifies this operation, because the new node class inherits all
the methods from the original class, and has to overwrite only the two methods
that are responsible for the routing.
In this example, we illustrate an implementation of routing with a lag function
in the shape of an isosceles triangle with base
t_external, for internal and external routing respectively. This new
implementation is similar to the implementation of the Half-triangular lag function.
The first step is to import the
Node component from SuperflexPy and
define the class
1 2 3
from superflexpy.framework.node import Node class RoutedNone(Node):
We then need to implement the methods
external_routing. Both methods receive as input a list of fluxes,
and return as output the fluxes (in the same order of the inputs) with the
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
def _internal_routing(self, flux): t_internal = self.get_parameters(names=[self._prefix_local_parameters + 't_internal']) flux_out =  for f in flux: flux_out.append(self._route(f, t_internal)) return flux_out def external_routing(self, flux): t_external = self.get_parameters(names=[self._prefix_local_parameters + 't_external']) flux_out =  for f in flux: flux_out.append(self._route(f, t_external)) return flux_out
In this simple example, the two routing mechanisms are handled using the same
lag functional form. Hence, the methods
take advantage of the method
(line 7 and 17).
_route is implemented as follows
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41
def _route(self, flux, time): state = np.zeros(int(np.ceil(time))) weight = self._calculate_weight(time) out =  for value in flux: state = state + weight * value out.append(state) state = 0 state = np.roll(state, shift=-1) return np.array(out) def _calculate_weight(self, time): weight =  array_length = np.ceil(time) for i in range(int(array_length)): weight.append(self._calculate_lag_area(i + 1, time) - self._calculate_lag_area(i, time)) return weight @staticmethod def _calculate_lag_area(portion, time): half_time = time / 2 if portion <= 0: value = 0 elif portion < half_time: value = 2 * (portion / time) ** 2 elif portion < time: value = 1 - 2 * ((time - portion) / time)**2 else: value = 1 return value
Note that the code in this block is similar to the code implemented in
Half-triangular lag function. The methods in this last code block are “support” methods that
make the code more organized and easier to maintain. The same numerical results
can be obtained by moving the functionality of these methods directly into
external_routing, though the resulting code would be less modular.