Coverage for wsimod\nodes\demand.py: 21%
90 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-10-24 11:16 +0100
1# -*- coding: utf-8 -*-
2"""Created on Mon Nov 15 14:20:36 2021.
4@author: bdobson
6Converted to totals BD 2022-05-03
7"""
8from typing import Any, Dict
10from wsimod.core import constants
11from wsimod.nodes.nodes import Node
14class Demand(Node):
15 """"""
17 def __init__(
18 self,
19 name,
20 constant_demand=0,
21 pollutant_load={},
22 data_input_dict={},
23 ):
24 """Node that generates and moves water. Currently only subclass
25 ResidentialDemand is in use.
27 Args:
28 name (str): node name constant_demand (float, optional): A constant portion
29 of demand if no subclass
30 is used. Defaults to 0.
31 pollutant_load (dict, optional): Pollutant mass per timestep of
32 constant_demand.
33 Defaults to 0.
34 data_input_dict (dict, optional): Dictionary of data inputs relevant for
35 the node (temperature). Keys are tuples where first value is the name of
36 the variable to read from the dict and the second value is the time.
37 Defaults to {}
39 Functions intended to call in orchestration:
40 create_demand
41 """
42 # TODO should temperature be defined in pollutant dict? TODO a lot of this
43 # should be moved to ResidentialDemand Assign parameters
44 self.constant_demand = constant_demand
45 self.pollutant_load = pollutant_load
46 # Update args
47 super().__init__(name, data_input_dict=data_input_dict)
48 # Update handlers
49 self.push_set_handler["default"] = self.push_set_deny
50 self.push_check_handler["default"] = self.push_check_deny
51 self.pull_set_handler["default"] = self.pull_set_deny
52 self.pull_check_handler["default"] = self.pull_check_deny
54 # Initialise states
55 self.total_demand = self.empty_vqip()
56 self.total_backup = self.empty_vqip() # ew
57 self.total_received = self.empty_vqip()
59 # Mass balance Because we assume demand is always satisfied received water
60 # 'disappears' for mass balance and consumed water 'appears' (this makes)
61 # introduction of pollutants easy
62 self.mass_balance_in.append(lambda: self.total_demand)
63 self.mass_balance_out.append(lambda: self.total_backup)
64 self.mass_balance_out.append(lambda: self.total_received)
66 def apply_overrides(self, overrides: Dict[str, Any] = {}):
67 """Apply overrides to the sewer.
69 Enables a user to override any of the following parameters:
70 constant_demand, pollutant_load.
72 Args:
73 overrides (dict, optional): Dictionary of overrides. Defaults to {}.
74 """
75 self.constant_demand = overrides.pop("constant_demand", self.constant_demand)
76 self.pollutant_load.update(overrides.pop("pollutant_load", {}))
77 super().apply_overrides(overrides)
79 def create_demand(self):
80 """Function to call get_demand, which should return a dict with keys that match
81 the keys in directions.
83 A dict that determines how to push_distributed the generated wastewater/garden
84 irrigation. Water is drawn from attached nodes.
85 """
86 demand = self.get_demand()
87 total_requested = 0
88 for dem in demand.values():
89 total_requested += dem["volume"]
91 self.total_received = self.pull_distributed({"volume": total_requested})
93 # TODO Currently just assume all water is received and then pushed onwards
94 if (total_requested - self.total_received["volume"]) > constants.FLOAT_ACCURACY:
95 print(
96 "demand deficit of {2} at {0} on {1}".format(
97 self.name, self.t, total_requested - self.total_received["volume"]
98 )
99 )
101 directions = {
102 "garden": {"tag": ("Demand", "Garden"), "of_type": "Land"},
103 "house": {"tag": "Demand", "of_type": "Sewer"},
104 "default": {"tag": "default", "of_type": None},
105 }
107 # Send water where it needs to go
108 for key, item in demand.items():
109 # Distribute
110 remaining = self.push_distributed(
111 item, of_type=directions[key]["of_type"], tag=directions[key]["tag"]
112 )
113 self.total_backup = self.sum_vqip(self.total_backup, remaining)
114 if remaining["volume"] > constants.FLOAT_ACCURACY:
115 print("Demand not able to push")
117 # Update for mass balance
118 for dem in demand.values():
119 self.total_demand = self.sum_vqip(self.total_demand, dem)
121 def get_demand(self):
122 """Holder function to enable constant demand generation.
124 Returns:
125 (dict): A VQIP that will contain constant demand
126 """
127 # TODO read/gen demand
128 pol = self.v_change_vqip(self.empty_vqip(), self.constant_demand)
129 for key, item in self.pollutant_load.items():
130 pol[key] = item
131 return {"default": pol}
133 def end_timestep(self):
134 """Reset state variable trackers."""
135 self.total_demand = self.empty_vqip()
136 self.total_backup = self.empty_vqip()
137 self.total_received = self.empty_vqip()
140class NonResidentialDemand(Demand):
141 """Holder class to enable non-residential demand generation."""
143 def get_demand(self):
144 """Holder function.
146 Returns:
147 (dict): A dict of VQIPs, where the keys match with directions
148 in Demand/create_demand
149 """
150 return {"house": self.get_demand()}
153class ResidentialDemand(Demand):
154 """"""
156 def __init__(
157 self,
158 population=1,
159 pollutant_load={},
160 per_capita=0.12,
161 gardening_efficiency=0.6 * 0.7, # Watering efficiency by irrigated area
162 data_input_dict={}, # For temperature
163 constant_temp=30,
164 constant_weighting=0.2,
165 **kwargs,
166 ):
167 """Subclass of demand with functions to handle internal and external water use.
169 Args:
170 population (float, optional): population of node. Defaults to 1. per_capita
171 (float, optional): Volume per person per timestep of water
172 used. Defaults to 0.12.
173 pollutant_load (dict, optional): Mass per person per timestep of
174 different pollutants generated. Defaults to {}.
175 gardening_efficiency (float, optional): Value between 0 and 1 that
176 translates irrigation demand from GardenSurface into water requested
177 from the distribution network. Should account for percent of garden that
178 is irrigated and the efficacy of people in meeting their garden water
179 demand. Defaults to 0.6*0.7.
180 data_input_dict (dict, optional): Dictionary of data inputs relevant for
181 the node (temperature). Keys are tuples where first value is the name of
182 the variable to read from the dict and the second value is the time.
183 Defaults to {}
184 constant_temp (float, optional): A constant temperature associated with
185 generated water. Defaults to 30
186 constant_weighting (float, optional): Proportion of temperature that is
187 made up from by constant_temp. Defaults to 0.2.
189 Key assumptions:
190 - Per capita calculations to generate demand based on population.
191 - Pollutant concentration of generated demand uses a fixed mass per person
192 per timestep.
193 - Temperature of generated wastewater is based partially on air temperature
194 and partially on a constant.
195 - Can interact with `land.py/GardenSurface` to simulate garden water use.
197 Input data and parameter requirements:
198 - `population`.
199 _Units_: n
200 - `per_capita`.
201 _Units_: m3/timestep
202 - `data_input_dict` should contain air temperature at model timestep.
203 _Units_: C
204 """
205 self.gardening_efficiency = gardening_efficiency
206 self.population = population
207 self.per_capita = per_capita
208 self.constant_weighting = constant_weighting
209 self.constant_temp = constant_temp
210 super().__init__(
211 data_input_dict=data_input_dict, pollutant_load=pollutant_load, **kwargs
212 )
213 # Label as Demand class so that other nodes treat it the same
214 self.__class__.__name__ = "Demand"
216 def apply_overrides(self, overrides: Dict[str, Any] = {}):
217 """Apply overrides to the sewer.
219 Enables a user to override any of the following parameters:
220 gardening_efficiency, population, per_capita, constant_weighting, constant_temp.
222 Args:
223 overrides (dict, optional): Dictionary of overrides. Defaults to {}.
224 """
225 self.gardening_efficiency = overrides.pop(
226 "gardening_efficiency", self.gardening_efficiency
227 )
228 self.population = overrides.pop("population", self.population)
229 self.per_capita = overrides.pop("per_capita", self.per_capita)
230 self.constant_weighting = overrides.pop(
231 "constant_weighting", self.constant_weighting
232 )
233 self.constant_temp = overrides.pop("constant_temp", self.constant_temp)
234 super().apply_overrides(overrides)
236 def get_demand(self):
237 """Overwrite get_demand and replace with custom functions.
239 Returns:
240 (dict): A dict of VQIPs, where the keys match with directions
241 in Demand/create_demand
242 """
243 water_output = {}
245 water_output["garden"] = self.get_garden_demand()
246 water_output["house"] = self.get_house_demand()
248 return water_output
250 def get_garden_demand(self):
251 """Calculate garden water demand in the current timestep by get_connected to all
252 attached land nodes. This check should return garden water demand. Applies
253 irrigation coefficient. Can function when a single population node is connected
254 to multiple land nodes, however, the capacity and preferences of arcs should be
255 updated to reflect what is possible based on area.
257 Returns:
258 vqip (dict): A VQIP of garden water use (including pollutants) to be
259 pushed to land
260 """
261 # Get garden water demand
262 excess = self.get_connected(
263 direction="push", of_type="Land", tag=("Demand", "Garden")
264 )["avail"]
266 # Apply garden_efficiency
267 excess = self.excess_to_garden_demand(excess)
269 # Apply any pollutants
270 vqip = self.apply_gardening_pollutants(excess)
271 return vqip
273 def apply_gardening_pollutants(self, excess):
274 """Holder function to apply pollutants (i.e., presumably fertiliser) to the
275 garden.
277 Args:
278 excess (float): A volume of water applied to a garden
280 Returns:
281 (dict): A VQIP of water that includes pollutants to be sent to land
282 """
283 # TODO Fertilisers are currently applied in the land node... which is
284 # preferable?
285 vqip = self.empty_vqip()
286 vqip["volume"] = excess
287 return vqip
289 def excess_to_garden_demand(self, excess):
290 """Apply garden_efficiency.
292 Args:
293 excess (float): Volume of water required to satisfy garden irrigation
295 Returns:
296 (float): Amount of water actually applied to garden
297 """
298 # TODO Anything more than this needed? (yes - population presence if eventually
299 # included!)
301 return excess * self.gardening_efficiency
303 def get_house_demand(self):
304 """Per capita calculations for household wastewater generation. Applies weighted
305 temperature calculation.
307 Returns:
308 (dict): A VQIP containg foul water
309 """
310 # TODO water that is consumed but not sent onwards as foul Total water required
311 consumption = self.population * self.per_capita
312 # Apply pollutants
313 foul = self.copy_vqip(self.pollutant_load)
314 # Scale to population
315 for pol in constants.ADDITIVE_POLLUTANTS:
316 foul[pol] *= self.population
317 # Update volume and temperature (which is weighted based on air temperature and
318 # constant_temp)
319 foul["volume"] = consumption
320 foul["temperature"] = (
321 self.get_data_input("temperature") * (1 - self.constant_weighting)
322 + self.constant_temp * self.constant_weighting
323 )
324 return foul