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
1from superflexpy.framework.node import Node
2
3class 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 def _internal_routing(self, flux):
2
3 t_internal = self.get_parameters(names=[self._prefix_local_parameters + 't_internal'])
4 flux_out = []
5
6 for f in flux:
7 flux_out.append(self._route(f, t_internal))
8
9 return flux_out
10
11 def external_routing(self, flux):
12
13 t_external = self.get_parameters(names=[self._prefix_local_parameters + 't_external'])
14 flux_out = []
15
16 for f in flux:
17 flux_out.append(self._route(f, t_external))
18
19 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 def _route(self, flux, time):
2
3 state = np.zeros(int(np.ceil(time)))
4 weight = self._calculate_weight(time)
5
6 out = []
7 for value in flux:
8 state = state + weight * value
9 out.append(state[0])
10 state[0] = 0
11 state = np.roll(state, shift=-1)
12
13 return np.array(out)
14
15 def _calculate_weight(self, time):
16
17 weight = []
18
19 array_length = np.ceil(time)
20
21 for i in range(int(array_length)):
22 weight.append(self._calculate_lag_area(i + 1, time)
23 - self._calculate_lag_area(i, time))
24
25 return weight
26
27 @staticmethod
28 def _calculate_lag_area(portion, time):
29
30 half_time = time / 2
31
32 if portion <= 0:
33 value = 0
34 elif portion < half_time:
35 value = 2 * (portion / time) ** 2
36 elif portion < time:
37 value = 1 - 2 * ((time - portion) / time)**2
38 else:
39 value = 1
40
41 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.