Note

Last update 26/06/2020

Expand SuperflexPy: modify the existing components

Adding the routing to a node

The nodes in SuperflexPy are designed to provide the possibility of simulating the routing process that may happen in a catchment. The routing is a delay in the fluxes that comes from their propagation within the catchment (internal routing) and in the river network (external routing).

The default implementation of the node (Node class in superflexpy.framework.node) does not provide the routing functionality. Although the methods _internal_routing and external_routing exist and are integrate in the code, their implementation simply returns the incoming fluxes without any transformation.

The modeller that wants to implement the routing, therefore, has to implement a customized node that implements those two methods. The object-oriented design of SuperflexPy simplifies this operation since the new node class will inherit all the methods from the original class and has to overwrite only the two that are responsible of the routing.

We propose here an implementation of the routing that uses a lag function that has the shape of an isosceles triangle with base t_internal and t_external, for internal and external routing respectively. The implementation is similar to the case of the Half-triangular lag function.

The first step to do in the implementation is to import the Node component from SuperflexPy and implement the class RoutedNode with the following code

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 the methods receive, as input, a list of fluxes and return, as output, the same list of fluxes 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

Since, in this simple example, the two routing mechanisms are handled using the same lag function, the methods take advantage of the method _route (line 7 and 17) to handle the routing.

The method is implemented with the following code

 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 all the code in this block is highly similar to the one implemented in RoutedNode and that, for the implementation of the routing, the only two methods that are strictly necessary are _internal_routing and external_routing while all the others are only “support methods” to these two, needed only to make the code more organized and easier to maintain.