Note

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. The methods _internal_routing and 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 the methods _internal_routing and external_routing 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_internal and 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 RoutedNode

 1 2 3 from superflexpy.framework.node import Node class RoutedNone(Node): 

We then need to implement the methods _internal_routing and 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 delay applied.

  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 _internal_routing and external_routing take advantage of the method _route (line 7 and 17).

The method _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[0]) state[0] = 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 _internal_routing and external_routing, though the resulting code would be less modular.