Skip to content

API Reference - Other Components

This section of the documentation provides a reference for the API of the nodes.catchment, nodes.demand, nodes.distribution, and nodes.waste modules.

Created on Mon Nov 15 14:20:36 2021.

@author: bdobson

Converted to totals on 2022-05-03

Catchment

Bases: Node

Source code in wsimod\nodes\catchment.py 
 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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class Catchment(Node):
    """"""

    def __init__(
        self,
        name,
        data_input_dict={},
    ):
        """Node that reads input data to create VQIPs that are pushed downstream and
        tracks abstractions made from the node, adjusting pushes accordingly.

        Args:
            name (str): Node name
            data_input_dict (dict, optional): Dictionary of data inputs relevant for
                the node. Keys are tuples where first value is the name of the
                variable to read from the dict and the second value is the time.
                Defaults to {}.

        Functions intended to call in orchestration:
            route

        Key assumptions:
            - Flows from `Catchment` nodes are simply read from data, thus
                assumed to be unresponsive to wider water cycle changes.

        Input data and parameter requirements:
            - Flow data in the `data_input_dict` at the model timestep.
                _Units_: cubic metres/timestep
            - Values for each variable defined in `constants.POLLUTANTS` also
                stored in `data_input_dict` at the model timestep.
                _Units_: kg/m3/timestep (additive pollutants)
        """
        # Update args
        super().__init__(name)
        self.data_input_dict = data_input_dict

        # Update handlers
        self.pull_set_handler["default"] = self.pull_set_abstraction
        self.pull_check_handler["default"] = self.pull_check_abstraction
        self.push_set_handler["default"] = self.push_set_deny
        self.push_check_handler["default"] = self.push_set_deny
        self.unrouted_water = self.empty_vqip()
        # Mass balance
        self.mass_balance_in.append(lambda: self.get_flow())
        self.mass_balance_out.append(lambda: self.unrouted_water)
        self.end_timestep = self.end_timestep_

    def get_flow(self):
        """Read volume data, read pollutant data, convert additibve pollutants from
        kg/m3 to kg.

        Returns:
            vqip (dict): Return read data as a VQIP
        """
        # TODO (if used) - note that if flow is < float accuracy then it won't
        # get pushed, and the pollutants will 'disappear', causing a mass balance error
        vqip = {"volume": self.data_input_dict[("flow", self.t)]}
        for pollutant in constants.POLLUTANTS:
            vqip[pollutant] = self.data_input_dict[(pollutant, self.t)]
        for pollutant in constants.ADDITIVE_POLLUTANTS:
            vqip[pollutant] *= vqip["volume"]

        return vqip

    def route(self):
        """Send any water that has not already been abstracted downstream."""
        # Get amount of water
        avail = self.get_avail()
        # Route excess flow onwards
        reply = self.push_distributed(avail, of_type=["Node", "River", "Waste"])
        self.unrouted_water = self.sum_vqip(self.unrouted_water, reply)
        if reply["volume"] > constants.FLOAT_ACCURACY:
            pass
            # print('Catchment unable to route')

    def get_avail(self):
        """Water available for abstraction (Read data and subtract pre-existing
        abstractions).

        Returns:
            avail (dict): A VQIP of water available for abstraction
        """
        # Get available vqip
        avail = self.get_flow()

        # Remove abstractions already made
        for name, arc in self.out_arcs.items():
            avail = self.v_change_vqip(avail, avail["volume"] - arc.vqip_in["volume"])

        return avail

    def pull_check_abstraction(self, vqip=None):
        """Check wrapper for get_avail that updates response if VQIP is given.

        Args:
            vqip (dict, optional): A VQIP that is compared with get_avail and the
                minimum is returned. Only the 'volume' key is used. Defaults to None.

        Returns:
            avail (dict): A VQIP of water available for abstraction
        """
        # Respond to abstraction check request
        avail = self.get_avail()

        if vqip:
            avail = self.v_change_vqip(avail, min(avail["volume"], vqip["volume"]))

        return avail

    def pull_set_abstraction(self, vqip):
        """Request set wrapper for get_avail where VQIP is specified.

        Args:
            vqip (dict): A VQIP of water to pull. Only the 'volume' key is used.

        Returns:
            avail (dict): A VQIP of water abstracted
        """
        # Respond to abstraction set request
        avail = self.get_avail()
        avail = self.v_change_vqip(avail, min(avail["volume"], vqip["volume"]))

        return avail

    def end_timestep_(self):
        """Reset unrouted water."""
        self.unrouted_water = self.empty_vqip()

__init__(name, data_input_dict={})

Node that reads input data to create VQIPs that are pushed downstream and tracks abstractions made from the node, adjusting pushes accordingly.

Parameters:

Name Type Description Default
name str

Node name

 required
data_input_dict dict

Dictionary of data inputs relevant for the node. Keys are tuples where first value is the name of the variable to read from the dict and the second value is the time. Defaults to {}.

 {}
Functions intended to call in orchestration

route

Key assumptions
  • Flows from Catchment nodes are simply read from data, thus assumed to be unresponsive to wider water cycle changes.
Input data and parameter requirements
  • Flow data in the data_input_dict at the model timestep. Units: cubic metres/timestep
  • Values for each variable defined in constants.POLLUTANTS also stored in data_input_dict at the model timestep. Units: kg/m3/timestep (additive pollutants)
Source code in wsimod\nodes\catchment.py 
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    name,
    data_input_dict={},
):
    """Node that reads input data to create VQIPs that are pushed downstream and
    tracks abstractions made from the node, adjusting pushes accordingly.

    Args:
        name (str): Node name
        data_input_dict (dict, optional): Dictionary of data inputs relevant for
            the node. Keys are tuples where first value is the name of the
            variable to read from the dict and the second value is the time.
            Defaults to {}.

    Functions intended to call in orchestration:
        route

    Key assumptions:
        - Flows from `Catchment` nodes are simply read from data, thus
            assumed to be unresponsive to wider water cycle changes.

    Input data and parameter requirements:
        - Flow data in the `data_input_dict` at the model timestep.
            _Units_: cubic metres/timestep
        - Values for each variable defined in `constants.POLLUTANTS` also
            stored in `data_input_dict` at the model timestep.
            _Units_: kg/m3/timestep (additive pollutants)
    """
    # Update args
    super().__init__(name)
    self.data_input_dict = data_input_dict

    # Update handlers
    self.pull_set_handler["default"] = self.pull_set_abstraction
    self.pull_check_handler["default"] = self.pull_check_abstraction
    self.push_set_handler["default"] = self.push_set_deny
    self.push_check_handler["default"] = self.push_set_deny
    self.unrouted_water = self.empty_vqip()
    # Mass balance
    self.mass_balance_in.append(lambda: self.get_flow())
    self.mass_balance_out.append(lambda: self.unrouted_water)
    self.end_timestep = self.end_timestep_

end_timestep_()

Reset unrouted water.

Source code in wsimod\nodes\catchment.py 
136
137
138
def end_timestep_(self):
    """Reset unrouted water."""
    self.unrouted_water = self.empty_vqip()

get_avail()

Water available for abstraction (Read data and subtract pre-existing abstractions).

Returns:

Name Type Description
avail dict

A VQIP of water available for abstraction
Source code in wsimod\nodes\catchment.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_avail(self):
    """Water available for abstraction (Read data and subtract pre-existing
    abstractions).

    Returns:
        avail (dict): A VQIP of water available for abstraction
    """
    # Get available vqip
    avail = self.get_flow()

    # Remove abstractions already made
    for name, arc in self.out_arcs.items():
        avail = self.v_change_vqip(avail, avail["volume"] - arc.vqip_in["volume"])

    return avail

get_flow()

Read volume data, read pollutant data, convert additibve pollutants from kg/m3 to kg.

Returns:

Name Type Description
vqip dict

Return read data as a VQIP
Source code in wsimod\nodes\catchment.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_flow(self):
    """Read volume data, read pollutant data, convert additibve pollutants from
    kg/m3 to kg.

    Returns:
        vqip (dict): Return read data as a VQIP
    """
    # TODO (if used) - note that if flow is < float accuracy then it won't
    # get pushed, and the pollutants will 'disappear', causing a mass balance error
    vqip = {"volume": self.data_input_dict[("flow", self.t)]}
    for pollutant in constants.POLLUTANTS:
        vqip[pollutant] = self.data_input_dict[(pollutant, self.t)]
    for pollutant in constants.ADDITIVE_POLLUTANTS:
        vqip[pollutant] *= vqip["volume"]

    return vqip

pull_check_abstraction(vqip=None)

Check wrapper for get_avail that updates response if VQIP is given.

Parameters:

Name Type Description Default
vqip dict

A VQIP that is compared with get_avail and the minimum is returned. Only the 'volume' key is used. Defaults to None.

 None

Returns:

Name Type Description
avail dict

A VQIP of water available for abstraction
Source code in wsimod\nodes\catchment.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def pull_check_abstraction(self, vqip=None):
    """Check wrapper for get_avail that updates response if VQIP is given.

    Args:
        vqip (dict, optional): A VQIP that is compared with get_avail and the
            minimum is returned. Only the 'volume' key is used. Defaults to None.

    Returns:
        avail (dict): A VQIP of water available for abstraction
    """
    # Respond to abstraction check request
    avail = self.get_avail()

    if vqip:
        avail = self.v_change_vqip(avail, min(avail["volume"], vqip["volume"]))

    return avail

pull_set_abstraction(vqip)

Request set wrapper for get_avail where VQIP is specified.

Parameters:

Name Type Description Default
vqip dict

A VQIP of water to pull. Only the 'volume' key is used.

 required

Returns:

Name Type Description
avail dict

A VQIP of water abstracted
Source code in wsimod\nodes\catchment.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def pull_set_abstraction(self, vqip):
    """Request set wrapper for get_avail where VQIP is specified.

    Args:
        vqip (dict): A VQIP of water to pull. Only the 'volume' key is used.

    Returns:
        avail (dict): A VQIP of water abstracted
    """
    # Respond to abstraction set request
    avail = self.get_avail()
    avail = self.v_change_vqip(avail, min(avail["volume"], vqip["volume"]))

    return avail

route()

Send any water that has not already been abstracted downstream.

Source code in wsimod\nodes\catchment.py 
76
77
78
79
80
81
82
83
84
def route(self):
    """Send any water that has not already been abstracted downstream."""
    # Get amount of water
    avail = self.get_avail()
    # Route excess flow onwards
    reply = self.push_distributed(avail, of_type=["Node", "River", "Waste"])
    self.unrouted_water = self.sum_vqip(self.unrouted_water, reply)
    if reply["volume"] > constants.FLOAT_ACCURACY:
        pass

Created on Mon Nov 15 14:20:36 2021.

@author: bdobson

Converted to totals BD 2022-05-03

Demand

Bases: Node

Source code in wsimod\nodes\demand.py 
 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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class Demand(Node):
    """"""

    def __init__(
        self,
        name,
        constant_demand=0,
        pollutant_load={},
        data_input_dict={},
    ):
        """Node that generates and moves water. Currently only subclass
        ResidentialDemand is in use.

        Args:
            name (str): node name constant_demand (float, optional): A constant portion
            of demand if no subclass
                is used. Defaults to 0.
            pollutant_load (dict, optional): Pollutant mass per timestep of
            constant_demand.
                Defaults to 0.
            data_input_dict (dict, optional):  Dictionary of data inputs relevant for
                the node (temperature). Keys are tuples where first value is the name of
                the variable to read from the dict and the second value is the time.
                Defaults to {}

        Functions intended to call in orchestration:
            create_demand
        """
        # TODO should temperature be defined in pollutant dict? TODO a lot of this
        # should be moved to ResidentialDemand Assign parameters
        self.constant_demand = constant_demand
        self.pollutant_load = pollutant_load
        # Update args
        super().__init__(name, data_input_dict=data_input_dict)
        # Update handlers
        self.push_set_handler["default"] = self.push_set_deny
        self.push_check_handler["default"] = self.push_check_deny
        self.pull_set_handler["default"] = self.pull_set_deny
        self.pull_check_handler["default"] = self.pull_check_deny

        # Initialise states
        self.total_demand = self.empty_vqip()
        self.total_backup = self.empty_vqip()  # ew
        self.total_received = self.empty_vqip()

        # Mass balance Because we assume demand is always satisfied received water
        # 'disappears' for mass balance and consumed water 'appears' (this makes)
        # introduction of pollutants easy
        self.mass_balance_in.append(lambda: self.total_demand)
        self.mass_balance_out.append(lambda: self.total_backup)
        self.mass_balance_out.append(lambda: self.total_received)

    def apply_overrides(self, overrides: Dict[str, Any] = {}):
        """Apply overrides to the sewer.

        Enables a user to override any of the following parameters:
        constant_demand, pollutant_load.

        Args:
            overrides (dict, optional): Dictionary of overrides. Defaults to {}.
        """
        self.constant_demand = overrides.pop("constant_demand", self.constant_demand)
        self.pollutant_load.update(overrides.pop("pollutant_load", {}))
        super().apply_overrides(overrides)

    def create_demand(self):
        """Function to call get_demand, which should return a dict with keys that match
        the keys in directions.

        A dict that determines how to push_distributed the generated wastewater/garden
        irrigation. Water is drawn from attached nodes.
        """
        demand = self.get_demand()
        total_requested = 0
        for dem in demand.values():
            total_requested += dem["volume"]

        self.total_received = self.pull_distributed({"volume": total_requested})

        # TODO Currently just assume all water is received and then pushed onwards
        if (total_requested - self.total_received["volume"]) > constants.FLOAT_ACCURACY:
            print(
                "demand deficit of {2} at {0} on {1}".format(
                    self.name, self.t, total_requested - self.total_received["volume"]
                )
            )

        directions = {
            "garden": {"tag": ("Demand", "Garden"), "of_type": "Land"},
            "house": {"tag": "Demand", "of_type": "Sewer"},
            "default": {"tag": "default", "of_type": None},
        }

        # Send water where it needs to go
        for key, item in demand.items():
            # Distribute
            remaining = self.push_distributed(
                item, of_type=directions[key]["of_type"], tag=directions[key]["tag"]
            )
            self.total_backup = self.sum_vqip(self.total_backup, remaining)
            if remaining["volume"] > constants.FLOAT_ACCURACY:
                print("Demand not able to push")

        # Update for mass balance
        for dem in demand.values():
            self.total_demand = self.sum_vqip(self.total_demand, dem)

    def get_demand(self):
        """Holder function to enable constant demand generation.

        Returns:
            (dict): A VQIP that will contain constant demand
        """
        # TODO read/gen demand
        pol = self.v_change_vqip(self.empty_vqip(), self.constant_demand)
        for key, item in self.pollutant_load.items():
            pol[key] = item
        return {"default": pol}

    def end_timestep(self):
        """Reset state variable trackers."""
        self.total_demand = self.empty_vqip()
        self.total_backup = self.empty_vqip()
        self.total_received = self.empty_vqip()

__init__(name, constant_demand=0, pollutant_load={}, data_input_dict={})

Node that generates and moves water. Currently only subclass ResidentialDemand is in use.

Parameters:

Name Type Description Default
name str

node name constant_demand (float, optional): A constant portion

 required
pollutant_load dict

Pollutant mass per timestep of

 {}
data_input_dict dict

Dictionary of data inputs relevant for the node (temperature). Keys are tuples where first value is the name of the variable to read from the dict and the second value is the time. Defaults to {}

 {}
Functions intended to call in orchestration

create_demand

Source code in wsimod\nodes\demand.py 
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    name,
    constant_demand=0,
    pollutant_load={},
    data_input_dict={},
):
    """Node that generates and moves water. Currently only subclass
    ResidentialDemand is in use.

    Args:
        name (str): node name constant_demand (float, optional): A constant portion
        of demand if no subclass
            is used. Defaults to 0.
        pollutant_load (dict, optional): Pollutant mass per timestep of
        constant_demand.
            Defaults to 0.
        data_input_dict (dict, optional):  Dictionary of data inputs relevant for
            the node (temperature). Keys are tuples where first value is the name of
            the variable to read from the dict and the second value is the time.
            Defaults to {}

    Functions intended to call in orchestration:
        create_demand
    """
    # TODO should temperature be defined in pollutant dict? TODO a lot of this
    # should be moved to ResidentialDemand Assign parameters
    self.constant_demand = constant_demand
    self.pollutant_load = pollutant_load
    # Update args
    super().__init__(name, data_input_dict=data_input_dict)
    # Update handlers
    self.push_set_handler["default"] = self.push_set_deny
    self.push_check_handler["default"] = self.push_check_deny
    self.pull_set_handler["default"] = self.pull_set_deny
    self.pull_check_handler["default"] = self.pull_check_deny

    # Initialise states
    self.total_demand = self.empty_vqip()
    self.total_backup = self.empty_vqip()  # ew
    self.total_received = self.empty_vqip()

    # Mass balance Because we assume demand is always satisfied received water
    # 'disappears' for mass balance and consumed water 'appears' (this makes)
    # introduction of pollutants easy
    self.mass_balance_in.append(lambda: self.total_demand)
    self.mass_balance_out.append(lambda: self.total_backup)
    self.mass_balance_out.append(lambda: self.total_received)

apply_overrides(overrides={})

Apply overrides to the sewer.

Enables a user to override any of the following parameters: constant_demand, pollutant_load.

Parameters:

Name Type Description Default
overrides dict

Dictionary of overrides. Defaults to {}.

 {}
Source code in wsimod\nodes\demand.py 
66
67
68
69
70
71
72
73
74
75
76
77
def apply_overrides(self, overrides: Dict[str, Any] = {}):
    """Apply overrides to the sewer.

    Enables a user to override any of the following parameters:
    constant_demand, pollutant_load.

    Args:
        overrides (dict, optional): Dictionary of overrides. Defaults to {}.
    """
    self.constant_demand = overrides.pop("constant_demand", self.constant_demand)
    self.pollutant_load.update(overrides.pop("pollutant_load", {}))
    super().apply_overrides(overrides)

create_demand()

Function to call get_demand, which should return a dict with keys that match the keys in directions.

A dict that determines how to push_distributed the generated wastewater/garden irrigation. Water is drawn from attached nodes.

Source code in wsimod\nodes\demand.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def create_demand(self):
    """Function to call get_demand, which should return a dict with keys that match
    the keys in directions.

    A dict that determines how to push_distributed the generated wastewater/garden
    irrigation. Water is drawn from attached nodes.
    """
    demand = self.get_demand()
    total_requested = 0
    for dem in demand.values():
        total_requested += dem["volume"]

    self.total_received = self.pull_distributed({"volume": total_requested})

    # TODO Currently just assume all water is received and then pushed onwards
    if (total_requested - self.total_received["volume"]) > constants.FLOAT_ACCURACY:
        print(
            "demand deficit of {2} at {0} on {1}".format(
                self.name, self.t, total_requested - self.total_received["volume"]
            )
        )

    directions = {
        "garden": {"tag": ("Demand", "Garden"), "of_type": "Land"},
        "house": {"tag": "Demand", "of_type": "Sewer"},
        "default": {"tag": "default", "of_type": None},
    }

    # Send water where it needs to go
    for key, item in demand.items():
        # Distribute
        remaining = self.push_distributed(
            item, of_type=directions[key]["of_type"], tag=directions[key]["tag"]
        )
        self.total_backup = self.sum_vqip(self.total_backup, remaining)
        if remaining["volume"] > constants.FLOAT_ACCURACY:
            print("Demand not able to push")

    # Update for mass balance
    for dem in demand.values():
        self.total_demand = self.sum_vqip(self.total_demand, dem)

end_timestep()

Reset state variable trackers.

Source code in wsimod\nodes\demand.py 
133
134
135
136
137
def end_timestep(self):
    """Reset state variable trackers."""
    self.total_demand = self.empty_vqip()
    self.total_backup = self.empty_vqip()
    self.total_received = self.empty_vqip()

get_demand()

Holder function to enable constant demand generation.

Returns:

Type Description
dict

A VQIP that will contain constant demand
Source code in wsimod\nodes\demand.py 
121
122
123
124
125
126
127
128
129
130
131
def get_demand(self):
    """Holder function to enable constant demand generation.

    Returns:
        (dict): A VQIP that will contain constant demand
    """
    # TODO read/gen demand
    pol = self.v_change_vqip(self.empty_vqip(), self.constant_demand)
    for key, item in self.pollutant_load.items():
        pol[key] = item
    return {"default": pol}

NonResidentialDemand

Bases: Demand

Holder class to enable non-residential demand generation.

Source code in wsimod\nodes\demand.py 
140
141
142
143
144
145
146
147
148
149
150
class NonResidentialDemand(Demand):
    """Holder class to enable non-residential demand generation."""

    def get_demand(self):
        """Holder function.

        Returns:
            (dict): A dict of VQIPs, where the keys match with directions
                in Demand/create_demand
        """
        return {"house": self.get_demand()}

get_demand()

Holder function.

Returns:

Type Description
dict

A dict of VQIPs, where the keys match with directions in Demand/create_demand
Source code in wsimod\nodes\demand.py 
143
144
145
146
147
148
149
150
def get_demand(self):
    """Holder function.

    Returns:
        (dict): A dict of VQIPs, where the keys match with directions
            in Demand/create_demand
    """
    return {"house": self.get_demand()}

ResidentialDemand

Bases: Demand

Source code in wsimod\nodes\demand.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class ResidentialDemand(Demand):
    """"""

    def __init__(
        self,
        population=1,
        pollutant_load={},
        per_capita=0.12,
        gardening_efficiency=0.6 * 0.7,  # Watering efficiency by irrigated area
        data_input_dict={},  # For temperature
        constant_temp=30,
        constant_weighting=0.2,
        **kwargs,
    ):
        """Subclass of demand with functions to handle internal and external water use.

        Args:
            population (float, optional): population of node. Defaults to 1. per_capita
            (float, optional): Volume per person per timestep of water
                used. Defaults to 0.12.
            pollutant_load (dict, optional): Mass per person per timestep of
                different pollutants generated. Defaults to {}.
            gardening_efficiency (float, optional): Value between 0 and 1 that
                translates irrigation demand from GardenSurface into water requested
                from the distribution network. Should account for percent of garden that
                is irrigated and the efficacy of people in meeting their garden water
                demand. Defaults to 0.6*0.7.
            data_input_dict (dict, optional):  Dictionary of data inputs relevant for
                the node (temperature). Keys are tuples where first value is the name of
                the variable to read from the dict and the second value is the time.
                Defaults to {}
            constant_temp (float, optional): A constant temperature associated with
                generated water. Defaults to 30
            constant_weighting (float, optional): Proportion of temperature that is
                made up from by constant_temp. Defaults to 0.2.

        Key assumptions:
            - Per capita calculations to generate demand based on population.
            - Pollutant concentration of generated demand uses a fixed mass per person
              per timestep.
            - Temperature of generated wastewater is based partially on air temperature
              and partially on a constant.
            - Can interact with `land.py/GardenSurface` to simulate garden water use.

        Input data and parameter requirements:
            - `population`.
                _Units_: n
            - `per_capita`.
                _Units_: m3/timestep
            - `data_input_dict` should contain air temperature at model timestep.
                _Units_: C
        """
        self.gardening_efficiency = gardening_efficiency
        self.population = population
        self.per_capita = per_capita
        self.constant_weighting = constant_weighting
        self.constant_temp = constant_temp
        super().__init__(
            data_input_dict=data_input_dict, pollutant_load=pollutant_load, **kwargs
        )
        # Label as Demand class so that other nodes treat it the same
        self.__class__.__name__ = "Demand"

    def apply_overrides(self, overrides: Dict[str, Any] = {}):
        """Apply overrides to the sewer.

        Enables a user to override any of the following parameters:
        gardening_efficiency, population, per_capita, constant_weighting, constant_temp.

        Args:
            overrides (dict, optional): Dictionary of overrides. Defaults to {}.
        """
        self.gardening_efficiency = overrides.pop(
            "gardening_efficiency", self.gardening_efficiency
        )
        self.population = overrides.pop("population", self.population)
        self.per_capita = overrides.pop("per_capita", self.per_capita)
        self.constant_weighting = overrides.pop(
            "constant_weighting", self.constant_weighting
        )
        self.constant_temp = overrides.pop("constant_temp", self.constant_temp)
        super().apply_overrides(overrides)

    def get_demand(self):
        """Overwrite get_demand and replace with custom functions.

        Returns:
            (dict): A dict of VQIPs, where the keys match with directions
                in Demand/create_demand
        """
        water_output = {}

        water_output["garden"] = self.get_garden_demand()
        water_output["house"] = self.get_house_demand()

        return water_output

    def get_garden_demand(self):
        """Calculate garden water demand in the current timestep by get_connected to all
        attached land nodes. This check should return garden water demand. Applies
        irrigation coefficient. Can function when a single population node is connected
        to multiple land nodes, however, the capacity and preferences of arcs should be
        updated to reflect what is possible based on area.

        Returns:
            vqip (dict): A VQIP of garden water use (including pollutants) to be
                pushed to land
        """
        # Get garden water demand
        excess = self.get_connected(
            direction="push", of_type="Land", tag=("Demand", "Garden")
        )["avail"]

        # Apply garden_efficiency
        excess = self.excess_to_garden_demand(excess)

        # Apply any pollutants
        vqip = self.apply_gardening_pollutants(excess)
        return vqip

    def apply_gardening_pollutants(self, excess):
        """Holder function to apply pollutants (i.e., presumably fertiliser) to the
        garden.

        Args:
            excess (float): A volume of water applied to a garden

        Returns:
            (dict): A VQIP of water that includes pollutants to be sent to land
        """
        # TODO Fertilisers are currently applied in the land node... which is
        # preferable?
        vqip = self.empty_vqip()
        vqip["volume"] = excess
        return vqip

    def excess_to_garden_demand(self, excess):
        """Apply garden_efficiency.

        Args:
            excess (float): Volume of water required to satisfy garden irrigation

        Returns:
            (float): Amount of water actually applied to garden
        """
        # TODO Anything more than this needed? (yes - population presence if eventually
        # included!)

        return excess * self.gardening_efficiency

    def get_house_demand(self):
        """Per capita calculations for household wastewater generation. Applies weighted
        temperature calculation.

        Returns:
            (dict): A VQIP containg foul water
        """
        # TODO water that is consumed but not sent onwards as foul Total water required
        consumption = self.population * self.per_capita
        # Apply pollutants
        foul = self.copy_vqip(self.pollutant_load)
        # Scale to population
        for pol in constants.ADDITIVE_POLLUTANTS:
            foul[pol] *= self.population
        # Update volume and temperature (which is weighted based on air temperature and
        # constant_temp)
        foul["volume"] = consumption
        foul["temperature"] = (
            self.get_data_input("temperature") * (1 - self.constant_weighting)
            + self.constant_temp * self.constant_weighting
        )
        return foul

__init__(population=1, pollutant_load={}, per_capita=0.12, gardening_efficiency=0.6 * 0.7, data_input_dict={}, constant_temp=30, constant_weighting=0.2, **kwargs)

Subclass of demand with functions to handle internal and external water use.

Parameters:

Name Type Description Default
population float

population of node. Defaults to 1. per_capita

 1
(float, optional

Volume per person per timestep of water used. Defaults to 0.12.

 required
pollutant_load dict

Mass per person per timestep of different pollutants generated. Defaults to {}.

 {}
gardening_efficiency float

Value between 0 and 1 that translates irrigation demand from GardenSurface into water requested from the distribution network. Should account for percent of garden that is irrigated and the efficacy of people in meeting their garden water demand. Defaults to 0.6*0.7.

 0.6 * 0.7
data_input_dict dict

Dictionary of data inputs relevant for the node (temperature). Keys are tuples where first value is the name of the variable to read from the dict and the second value is the time. Defaults to {}

 {}
constant_temp float

A constant temperature associated with generated water. Defaults to 30

 30
constant_weighting float

Proportion of temperature that is made up from by constant_temp. Defaults to 0.2.

 0.2
Key assumptions
  • Per capita calculations to generate demand based on population.
  • Pollutant concentration of generated demand uses a fixed mass per person per timestep.
  • Temperature of generated wastewater is based partially on air temperature and partially on a constant.
  • Can interact with land.py/GardenSurface to simulate garden water use.
Input data and parameter requirements
  • population. Units: n
  • per_capita. Units: m3/timestep
  • data_input_dict should contain air temperature at model timestep. Units: C
Source code in wsimod\nodes\demand.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def __init__(
    self,
    population=1,
    pollutant_load={},
    per_capita=0.12,
    gardening_efficiency=0.6 * 0.7,  # Watering efficiency by irrigated area
    data_input_dict={},  # For temperature
    constant_temp=30,
    constant_weighting=0.2,
    **kwargs,
):
    """Subclass of demand with functions to handle internal and external water use.

    Args:
        population (float, optional): population of node. Defaults to 1. per_capita
        (float, optional): Volume per person per timestep of water
            used. Defaults to 0.12.
        pollutant_load (dict, optional): Mass per person per timestep of
            different pollutants generated. Defaults to {}.
        gardening_efficiency (float, optional): Value between 0 and 1 that
            translates irrigation demand from GardenSurface into water requested
            from the distribution network. Should account for percent of garden that
            is irrigated and the efficacy of people in meeting their garden water
            demand. Defaults to 0.6*0.7.
        data_input_dict (dict, optional):  Dictionary of data inputs relevant for
            the node (temperature). Keys are tuples where first value is the name of
            the variable to read from the dict and the second value is the time.
            Defaults to {}
        constant_temp (float, optional): A constant temperature associated with
            generated water. Defaults to 30
        constant_weighting (float, optional): Proportion of temperature that is
            made up from by constant_temp. Defaults to 0.2.

    Key assumptions:
        - Per capita calculations to generate demand based on population.
        - Pollutant concentration of generated demand uses a fixed mass per person
          per timestep.
        - Temperature of generated wastewater is based partially on air temperature
          and partially on a constant.
        - Can interact with `land.py/GardenSurface` to simulate garden water use.

    Input data and parameter requirements:
        - `population`.
            _Units_: n
        - `per_capita`.
            _Units_: m3/timestep
        - `data_input_dict` should contain air temperature at model timestep.
            _Units_: C
    """
    self.gardening_efficiency = gardening_efficiency
    self.population = population
    self.per_capita = per_capita
    self.constant_weighting = constant_weighting
    self.constant_temp = constant_temp
    super().__init__(
        data_input_dict=data_input_dict, pollutant_load=pollutant_load, **kwargs
    )
    # Label as Demand class so that other nodes treat it the same
    self.__class__.__name__ = "Demand"

apply_gardening_pollutants(excess)

Holder function to apply pollutants (i.e., presumably fertiliser) to the garden.

Parameters:

Name Type Description Default
excess float

A volume of water applied to a garden

 required

Returns:

Type Description
dict

A VQIP of water that includes pollutants to be sent to land
Source code in wsimod\nodes\demand.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def apply_gardening_pollutants(self, excess):
    """Holder function to apply pollutants (i.e., presumably fertiliser) to the
    garden.

    Args:
        excess (float): A volume of water applied to a garden

    Returns:
        (dict): A VQIP of water that includes pollutants to be sent to land
    """
    # TODO Fertilisers are currently applied in the land node... which is
    # preferable?
    vqip = self.empty_vqip()
    vqip["volume"] = excess
    return vqip

apply_overrides(overrides={})

Apply overrides to the sewer.

Enables a user to override any of the following parameters: gardening_efficiency, population, per_capita, constant_weighting, constant_temp.

Parameters:

Name Type Description Default
overrides dict

Dictionary of overrides. Defaults to {}.

 {}
Source code in wsimod\nodes\demand.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def apply_overrides(self, overrides: Dict[str, Any] = {}):
    """Apply overrides to the sewer.

    Enables a user to override any of the following parameters:
    gardening_efficiency, population, per_capita, constant_weighting, constant_temp.

    Args:
        overrides (dict, optional): Dictionary of overrides. Defaults to {}.
    """
    self.gardening_efficiency = overrides.pop(
        "gardening_efficiency", self.gardening_efficiency
    )
    self.population = overrides.pop("population", self.population)
    self.per_capita = overrides.pop("per_capita", self.per_capita)
    self.constant_weighting = overrides.pop(
        "constant_weighting", self.constant_weighting
    )
    self.constant_temp = overrides.pop("constant_temp", self.constant_temp)
    super().apply_overrides(overrides)

excess_to_garden_demand(excess)

Apply garden_efficiency.

Parameters:

Name Type Description Default
excess float

Volume of water required to satisfy garden irrigation

 required

Returns:

Type Description
float

Amount of water actually applied to garden
Source code in wsimod\nodes\demand.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
def excess_to_garden_demand(self, excess):
    """Apply garden_efficiency.

    Args:
        excess (float): Volume of water required to satisfy garden irrigation

    Returns:
        (float): Amount of water actually applied to garden
    """
    # TODO Anything more than this needed? (yes - population presence if eventually
    # included!)

    return excess * self.gardening_efficiency

get_demand()

Overwrite get_demand and replace with custom functions.

Returns:

Type Description
dict

A dict of VQIPs, where the keys match with directions in Demand/create_demand
Source code in wsimod\nodes\demand.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
def get_demand(self):
    """Overwrite get_demand and replace with custom functions.

    Returns:
        (dict): A dict of VQIPs, where the keys match with directions
            in Demand/create_demand
    """
    water_output = {}

    water_output["garden"] = self.get_garden_demand()
    water_output["house"] = self.get_house_demand()

    return water_output

get_garden_demand()

Calculate garden water demand in the current timestep by get_connected to all attached land nodes. This check should return garden water demand. Applies irrigation coefficient. Can function when a single population node is connected to multiple land nodes, however, the capacity and preferences of arcs should be updated to reflect what is possible based on area.

Returns:

Name Type Description
vqip dict

A VQIP of garden water use (including pollutants) to be pushed to land
Source code in wsimod\nodes\demand.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def get_garden_demand(self):
    """Calculate garden water demand in the current timestep by get_connected to all
    attached land nodes. This check should return garden water demand. Applies
    irrigation coefficient. Can function when a single population node is connected
    to multiple land nodes, however, the capacity and preferences of arcs should be
    updated to reflect what is possible based on area.

    Returns:
        vqip (dict): A VQIP of garden water use (including pollutants) to be
            pushed to land
    """
    # Get garden water demand
    excess = self.get_connected(
        direction="push", of_type="Land", tag=("Demand", "Garden")
    )["avail"]

    # Apply garden_efficiency
    excess = self.excess_to_garden_demand(excess)

    # Apply any pollutants
    vqip = self.apply_gardening_pollutants(excess)
    return vqip

get_house_demand()

Per capita calculations for household wastewater generation. Applies weighted temperature calculation.

Returns:

Type Description
dict

A VQIP containg foul water
Source code in wsimod\nodes\demand.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def get_house_demand(self):
    """Per capita calculations for household wastewater generation. Applies weighted
    temperature calculation.

    Returns:
        (dict): A VQIP containg foul water
    """
    # TODO water that is consumed but not sent onwards as foul Total water required
    consumption = self.population * self.per_capita
    # Apply pollutants
    foul = self.copy_vqip(self.pollutant_load)
    # Scale to population
    for pol in constants.ADDITIVE_POLLUTANTS:
        foul[pol] *= self.population
    # Update volume and temperature (which is weighted based on air temperature and
    # constant_temp)
    foul["volume"] = consumption
    foul["temperature"] = (
        self.get_data_input("temperature") * (1 - self.constant_weighting)
        + self.constant_temp * self.constant_weighting
    )
    return foul

Created on Sun Aug 14 16:27:14 2022.

@author: bdobson

Distribution

Bases: Node

Source code in wsimod\nodes\distribution.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class Distribution(Node):
    """"""

    def __init__(self, leakage=0, **kwargs):
        """A Node that cannot be pushed to. Intended to pass calls to FWTW - though this
        currently relies on the user to connect it properly.

        Args:
            leakage (float, optional): 1 > float >= 0 to express how much
                water should be leaked to any attached groundwater nodes. This
                number represents the proportion of total flow through the node
                that should be leaked.
                Defaults to 0.

        Functions intended to call in orchestration:
            None

        Key assumptions:
            - No distribution processes yet represented, this class is just
                for conveyance.

        Input data and parameter requirements:
            - None
        """
        self.leakage = leakage
        super().__init__(**kwargs)
        # Update handlers
        self.push_set_handler["default"] = self.push_set_deny
        self.push_check_handler["default"] = self.push_check_deny
        self.decorate_pull_handlers()

    def decorate_pull_handlers(self):
        """Decorate handlers if there is leakage ratio."""
        if self.leakage > 0:
            self.pull_set_handler["default"] = decorate_leakage_set(
                self, self.pull_set_handler["default"]
            )
            self.pull_check_handler["default"] = decorate_leakage_check(
                self, self.pull_check_handler["default"]
            )

    def apply_overrides(self, overrides: Dict[str, Any] = {}):
        """Apply overrides to the sewer.

        Enables a user to override any of the following parameters:
        leakage.

        Args:
            overrides (dict, optional): Dictionary of overrides. Defaults to {}.
        """
        self.leakage = overrides.pop("leakage", self.leakage)
        self.decorate_pull_handlers()
        super().apply_overrides(overrides)

__init__(leakage=0, **kwargs)

A Node that cannot be pushed to. Intended to pass calls to FWTW - though this currently relies on the user to connect it properly.

Parameters:

Name Type Description Default
leakage float

1 > float >= 0 to express how much water should be leaked to any attached groundwater nodes. This number represents the proportion of total flow through the node that should be leaked. Defaults to 0.

 0
Functions intended to call in orchestration

None

Key assumptions
  • No distribution processes yet represented, this class is just for conveyance.
Input data and parameter requirements
  • None
Source code in wsimod\nodes\distribution.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __init__(self, leakage=0, **kwargs):
    """A Node that cannot be pushed to. Intended to pass calls to FWTW - though this
    currently relies on the user to connect it properly.

    Args:
        leakage (float, optional): 1 > float >= 0 to express how much
            water should be leaked to any attached groundwater nodes. This
            number represents the proportion of total flow through the node
            that should be leaked.
            Defaults to 0.

    Functions intended to call in orchestration:
        None

    Key assumptions:
        - No distribution processes yet represented, this class is just
            for conveyance.

    Input data and parameter requirements:
        - None
    """
    self.leakage = leakage
    super().__init__(**kwargs)
    # Update handlers
    self.push_set_handler["default"] = self.push_set_deny
    self.push_check_handler["default"] = self.push_check_deny
    self.decorate_pull_handlers()

apply_overrides(overrides={})

Apply overrides to the sewer.

Enables a user to override any of the following parameters: leakage.

Parameters:

Name Type Description Default
overrides dict

Dictionary of overrides. Defaults to {}.

 {}
Source code in wsimod\nodes\distribution.py 
142
143
144
145
146
147
148
149
150
151
152
153
def apply_overrides(self, overrides: Dict[str, Any] = {}):
    """Apply overrides to the sewer.

    Enables a user to override any of the following parameters:
    leakage.

    Args:
        overrides (dict, optional): Dictionary of overrides. Defaults to {}.
    """
    self.leakage = overrides.pop("leakage", self.leakage)
    self.decorate_pull_handlers()
    super().apply_overrides(overrides)

decorate_pull_handlers()

Decorate handlers if there is leakage ratio.

Source code in wsimod\nodes\distribution.py 
132
133
134
135
136
137
138
139
140
def decorate_pull_handlers(self):
    """Decorate handlers if there is leakage ratio."""
    if self.leakage > 0:
        self.pull_set_handler["default"] = decorate_leakage_set(
            self, self.pull_set_handler["default"]
        )
        self.pull_check_handler["default"] = decorate_leakage_check(
            self, self.pull_check_handler["default"]
        )

UnlimitedDistribution

Bases: Distribution

Source code in wsimod\nodes\distribution.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class UnlimitedDistribution(Distribution):
    """"""

    def __init__(self, **kwargs):
        """A distribution node that provides unlimited water while tracking pass
        balance.

        Functions intended to call in orchestration:
            None

        Key assumptions:
            - Water demand is always satisfied.

        Input data and parameter requirements:
            - None
        """
        super().__init__(**kwargs)
        # Update handlers
        self.pull_set_handler["default"] = self.pull_set_unlimited
        self.pull_check_handler["default"] = lambda x: self.v_change_vqip(
            self.empty_vqip(), constants.UNBOUNDED_CAPACITY
        )

        # States
        self.supplied = self.empty_vqip()

        self.mass_balance_in.append(lambda: self.supplied)

    def pull_set_unlimited(self, vqip):
        """Respond that VQIP was fulfilled and update state variables for mass balance.

        Args:
            vqip (dict): A VQIP amount to request

        Returns:
            vqip (dict): A VQIP amount that was supplied
        """
        # TODO maybe need some pollutant concentrations?
        vqip = self.v_change_vqip(self.empty_vqip(), vqip["volume"])
        self.supplied = self.sum_vqip(self.supplied, vqip)
        return vqip

    def end_timestep(self):
        """Update state variables."""
        self.supplied = self.empty_vqip()

__init__(**kwargs)

A distribution node that provides unlimited water while tracking pass balance.

Functions intended to call in orchestration

None

Key assumptions
  • Water demand is always satisfied.
Input data and parameter requirements
  • None
Source code in wsimod\nodes\distribution.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def __init__(self, **kwargs):
    """A distribution node that provides unlimited water while tracking pass
    balance.

    Functions intended to call in orchestration:
        None

    Key assumptions:
        - Water demand is always satisfied.

    Input data and parameter requirements:
        - None
    """
    super().__init__(**kwargs)
    # Update handlers
    self.pull_set_handler["default"] = self.pull_set_unlimited
    self.pull_check_handler["default"] = lambda x: self.v_change_vqip(
        self.empty_vqip(), constants.UNBOUNDED_CAPACITY
    )

    # States
    self.supplied = self.empty_vqip()

    self.mass_balance_in.append(lambda: self.supplied)

end_timestep()

Update state variables.

Source code in wsimod\nodes\distribution.py 
198
199
200
def end_timestep(self):
    """Update state variables."""
    self.supplied = self.empty_vqip()

pull_set_unlimited(vqip)

Respond that VQIP was fulfilled and update state variables for mass balance.

Parameters:

Name Type Description Default
vqip dict

A VQIP amount to request

 required

Returns:

Name Type Description
vqip dict

A VQIP amount that was supplied
Source code in wsimod\nodes\distribution.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
def pull_set_unlimited(self, vqip):
    """Respond that VQIP was fulfilled and update state variables for mass balance.

    Args:
        vqip (dict): A VQIP amount to request

    Returns:
        vqip (dict): A VQIP amount that was supplied
    """
    # TODO maybe need some pollutant concentrations?
    vqip = self.v_change_vqip(self.empty_vqip(), vqip["volume"])
    self.supplied = self.sum_vqip(self.supplied, vqip)
    return vqip

decorate_leakage_check(self, f)

Decorator to extend the functionality of f by introducing leakage. This is achieved by adjusting the volume of the request (vqip) to include anticipated leakage and then calling the original function f.

Parameters:

Name Type Description Default
self instance of Distribution class

The Distribution object to be extended

 required
f function

The function to be extended. Expected to be the Distribution object's pull_set function.

 required

Returns:

Name Type Description
pull_check function

The decorated function which includes the original functionality of f and additional leakage operations.
Source code in wsimod\nodes\distribution.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def decorate_leakage_check(self, f):
    """Decorator to extend the functionality of `f` by introducing leakage. This is
    achieved by adjusting the volume of the request (vqip) to include anticipated
    leakage and then calling the original function `f`.

    Args:
        self (instance of Distribution class): The Distribution object to be
            extended
        f (function): The function to be extended. Expected to be the
            Distribution object's pull_set function.

    Returns:
        pull_check (function): The decorated function which includes the
            original functionality of `f` and additional leakage operations.
    """

    def pull_check(vqip, **kwargs):
        """

        Args:
            vqip:
            **kwargs:

        Returns:

        """
        if vqip is not None:
            vqip["volume"] /= 1 - self.leakage
        reply = f(vqip, **kwargs)
        amount_leaked = self.v_change_vqip(reply, reply["volume"] * self.leakage)

        reply = self.extract_vqip(reply, amount_leaked)
        return reply

    return pull_check

decorate_leakage_set(self, f)

Decorator to extend the functionality of f by introducing leakage. This is achieved by adjusting the volume of the request (vqip) to include anticipated leakage, calling the original function f, and then distributing the leaked amount to groundwater.

Parameters:

Name Type Description Default
self instance of Distribution class

The Distribution object to be extended

 required
f function

The function to be extended. Expected to be the Distribution object's pull_set function.

 required

Returns:

Name Type Description
pull_set function

The decorated function which includes the original functionality of f and additional leakage operations.
Source code in wsimod\nodes\distribution.py 
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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def decorate_leakage_set(self, f):
    """Decorator to extend the functionality of `f` by introducing leakage. This is
    achieved by adjusting the volume of the request (vqip) to include anticipated
    leakage, calling the original function `f`, and then distributing the leaked amount
    to groundwater.

    Args:
        self (instance of Distribution class): The Distribution object to be
            extended
        f (function): The function to be extended. Expected to be the
            Distribution object's pull_set function.

    Returns:
        pull_set (function): The decorated function which includes the
            original functionality of `f` and additional leakage operations.
    """

    def pull_set(vqip, **kwargs):
        """

        Args:
            vqip:
            **kwargs:

        Returns:

        """
        vqip["volume"] /= 1 - self.leakage

        reply = f(vqip, **kwargs)

        amount_leaked = self.v_change_vqip(reply, reply["volume"] * self.leakage)

        reply = self.extract_vqip(reply, amount_leaked)

        unsuccessful_leakage = self.push_distributed(
            amount_leaked, of_type="Groundwater"
        )
        if unsuccessful_leakage["volume"] > constants.FLOAT_ACCURACY:
            print(
                "warning, distribution leakage not going to GW in {0} at {1}".format(
                    self.name, self.t
                )
            )
            reply = self.sum_vqip(reply, unsuccessful_leakage)

        return reply

    return pull_set

Created on Mon Nov 15 14:20:36 2021.

@author: bdobson

Waste

Bases: Node

Source code in wsimod\nodes\waste.py 
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
42
43
44
45
46
47
48
49
class Waste(Node):
    """"""

    def __init__(self, name):
        """Outlet node that can receive any amount of water by pushes.

        Args:
            name (str): Node name

        Functions intended to call in orchestration:
            None

        Key assumptions:
            - Water 'disappears' (leaves the model) from these nodes.

        Input data and parameter requirements:
            - None
        """
        # Update args
        super().__init__(name)

        # Update handlers
        self.pull_set_handler["default"] = self.pull_set_deny
        self.pull_check_handler["default"] = self.pull_check_deny
        self.push_set_handler["default"] = self.push_set_accept
        self.push_check_handler["default"] = self.push_check_accept

        # Mass balance
        self.mass_balance_out.append(self.total_in)

    def push_set_accept(self, vqip):
        """Push set function that accepts all water.

        Args:
            vqip (dict): A VQIP that has been pushed (ignored)

        Returns:
            (dict): An empty VQIP, indicating all water was received
        """
        return self.empty_vqip()

__init__(name)

Outlet node that can receive any amount of water by pushes.

Parameters:

Name Type Description Default
name str

Node name

 required
Functions intended to call in orchestration

None

Key assumptions
  • Water 'disappears' (leaves the model) from these nodes.
Input data and parameter requirements
  • None
Source code in wsimod\nodes\waste.py 
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
def __init__(self, name):
    """Outlet node that can receive any amount of water by pushes.

    Args:
        name (str): Node name

    Functions intended to call in orchestration:
        None

    Key assumptions:
        - Water 'disappears' (leaves the model) from these nodes.

    Input data and parameter requirements:
        - None
    """
    # Update args
    super().__init__(name)

    # Update handlers
    self.pull_set_handler["default"] = self.pull_set_deny
    self.pull_check_handler["default"] = self.pull_check_deny
    self.push_set_handler["default"] = self.push_set_accept
    self.push_check_handler["default"] = self.push_check_accept

    # Mass balance
    self.mass_balance_out.append(self.total_in)

push_set_accept(vqip)

Push set function that accepts all water.

Parameters:

Name Type Description Default
vqip dict

A VQIP that has been pushed (ignored)

 required

Returns:

Type Description
dict

An empty VQIP, indicating all water was received
Source code in wsimod\nodes\waste.py 
40
41
42
43
44
45
46
47
48
49
def push_set_accept(self, vqip):
    """Push set function that accepts all water.

    Args:
        vqip (dict): A VQIP that has been pushed (ignored)

    Returns:
        (dict): An empty VQIP, indicating all water was received
    """
    return self.empty_vqip()