Coverage for wsimod/nodes/demand.py: 22%
79 statements
« prev ^ index » next coverage.py v7.3.2, created at 2024-01-11 16:39 +0000
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 wsimod.core import constants
9from wsimod.nodes.nodes import Node
12class Demand(Node):
13 """"""
15 def __init__(
16 self,
17 name,
18 constant_demand=0,
19 pollutant_load={},
20 data_input_dict={},
21 ):
22 """Node that generates and moves water. Currently only subclass
23 ResidentialDemand is in use.
25 Args:
26 name (str): node name constant_demand (float, optional): A constant portion
27 of demand if no subclass
28 is used. Defaults to 0.
29 pollutant_load (dict, optional): Pollutant mass per timestep of
30 constant_demand.
31 Defaults to {}.
32 data_input_dict (dict, optional): Dictionary of data inputs relevant for
33 the node (temperature). Keys are tuples where first value is the name of
34 the variable to read from the dict and the second value is the time.
35 Defaults to {}
37 Functions intended to call in orchestration:
38 create_demand
39 """
40 # TODO should temperature be defined in pollutant dict? TODO a lot of this
41 # should be moved to ResidentialDemand Assign parameters
42 self.constant_demand = constant_demand
43 self.pollutant_load = pollutant_load
44 # Update args
45 super().__init__(name, data_input_dict=data_input_dict)
46 # Update handlers
47 self.push_set_handler["default"] = self.push_set_deny
48 self.push_check_handler["default"] = self.push_check_deny
49 self.pull_set_handler["default"] = self.pull_set_deny
50 self.pull_check_handler["default"] = self.pull_check_deny
52 # Initialise states
53 self.total_demand = self.empty_vqip()
54 self.total_backup = self.empty_vqip() # ew
55 self.total_received = self.empty_vqip()
57 # Mass balance Because we assume demand is always satisfied received water
58 # 'disappears' for mass balance and consumed water 'appears' (this makes)
59 # introduction of pollutants easy
60 self.mass_balance_in.append(lambda: self.total_demand)
61 self.mass_balance_out.append(lambda: self.total_backup)
62 self.mass_balance_out.append(lambda: self.total_received)
64 def create_demand(self):
65 """Function to call get_demand, which should return a dict with keys that match
66 the keys in directions.
68 A dict that determines how to push_distributed the generated wastewater/garden
69 irrigation. Water is drawn from attached nodes.
70 """
71 demand = self.get_demand()
72 total_requested = 0
73 for dem in demand.values():
74 total_requested += dem["volume"]
76 self.total_received = self.pull_distributed({"volume": total_requested})
78 # TODO Currently just assume all water is received and then pushed onwards
79 if (total_requested - self.total_received["volume"]) > constants.FLOAT_ACCURACY:
80 print(
81 "demand deficit of {2} at {0} on {1}".format(
82 self.name, self.t, total_requested - self.total_received["volume"]
83 )
84 )
86 directions = {
87 "garden": {"tag": ("Demand", "Garden"), "of_type": "Land"},
88 "house": {"tag": "Demand", "of_type": "Sewer"},
89 "default": {"tag": "default", "of_type": None},
90 }
92 # Send water where it needs to go
93 for key, item in demand.items():
94 # Distribute
95 remaining = self.push_distributed(
96 item, of_type=directions[key]["of_type"], tag=directions[key]["tag"]
97 )
98 self.total_backup = self.sum_vqip(self.total_backup, remaining)
99 if remaining["volume"] > constants.FLOAT_ACCURACY:
100 print("Demand not able to push")
102 # Update for mass balance
103 for dem in demand.values():
104 self.total_demand = self.sum_vqip(self.total_demand, dem)
106 def get_demand(self):
107 """Holder function to enable constant demand generation.
109 Returns:
110 (dict): A VQIP that will contain constant demand
111 """
112 # TODO read/gen demand
113 pol = self.v_change_vqip(self.empty_vqip(), self.constant_demand)
114 for key, item in self.pollutant_load.items():
115 pol[key] = item
116 return {"default": pol}
118 def end_timestep(self):
119 """Reset state variable trackers."""
120 self.total_demand = self.empty_vqip()
121 self.total_backup = self.empty_vqip()
122 self.total_received = self.empty_vqip()
125class NonResidentialDemand(Demand):
126 """Holder class to enable non-residential demand generation."""
128 def get_demand(self):
129 """Holder function.
131 Returns:
132 (dict): A dict of VQIPs, where the keys match with directions
133 in Demand/create_demand
134 """
135 return {"house": self.get_demand()}
138class ResidentialDemand(Demand):
139 """"""
141 def __init__(
142 self,
143 population=1,
144 pollutant_load={},
145 per_capita=0.12,
146 gardening_efficiency=0.6 * 0.7, # Watering efficiency by irrigated area
147 data_input_dict={}, # For temperature
148 constant_temp=30,
149 constant_weighting=0.2,
150 **kwargs,
151 ):
152 """Subclass of demand with functions to handle internal and external water use.
154 Args:
155 population (float, optional): population of node. Defaults to 1. per_capita
156 (float, optional): Volume per person per timestep of water
157 used. Defaults to 0.12.
158 pollutant_load (dict, optional): Mass per person per timestep of
159 different pollutants generated. Defaults to {}.
160 gardening_efficiency (float, optional): Value between 0 and 1 that
161 translates irrigation demand from GardenSurface into water requested
162 from the distribution network. Should account for percent of garden that
163 is irrigated and the efficacy of people in meeting their garden water
164 demand. Defaults to 0.6*0.7.
165 data_input_dict (dict, optional): Dictionary of data inputs relevant for
166 the node (temperature). Keys are tuples where first value is the name of
167 the variable to read from the dict and the second value is the time.
168 Defaults to {}
169 constant_temp (float, optional): A constant temperature associated with
170 generated water. Defaults to 30
171 constant_weighting (float, optional): Proportion of temperature that is
172 made up from by constant_temp. Defaults to 0.2.
174 Key assumptions:
175 - Per capita calculations to generate demand based on population.
176 - Pollutant concentration of generated demand uses a fixed mass per person
177 per timestep.
178 - Temperature of generated wastewater is based partially on air temperature
179 and partially on a constant.
180 - Can interact with `land.py/GardenSurface` to simulate garden water use.
182 Input data and parameter requirements:
183 - `population`.
184 _Units_: n
185 - `per_capita`.
186 _Units_: m3/timestep
187 - `data_input_dict` should contain air temperature at model timestep.
188 _Units_: C
189 """
190 self.gardening_efficiency = gardening_efficiency
191 self.population = population
192 self.per_capita = per_capita
193 self.constant_weighting = constant_weighting
194 self.constant_temp = constant_temp
195 super().__init__(
196 data_input_dict=data_input_dict, pollutant_load=pollutant_load, **kwargs
197 )
198 # Label as Demand class so that other nodes treat it the same
199 self.__class__.__name__ = "Demand"
201 def get_demand(self):
202 """Overwrite get_demand and replace with custom functions.
204 Returns:
205 (dict): A dict of VQIPs, where the keys match with directions
206 in Demand/create_demand
207 """
208 water_output = {}
210 water_output["garden"] = self.get_garden_demand()
211 water_output["house"] = self.get_house_demand()
213 return water_output
215 def get_garden_demand(self):
216 """Calculate garden water demand in the current timestep by get_connected to all
217 attached land nodes. This check should return garden water demand. Applies
218 irrigation coefficient. Can function when a single population node is connected
219 to multiple land nodes, however, the capacity and preferences of arcs should be
220 updated to reflect what is possible based on area.
222 Returns:
223 vqip (dict): A VQIP of garden water use (including pollutants) to be
224 pushed to land
225 """
226 # Get garden water demand
227 excess = self.get_connected(
228 direction="push", of_type="Land", tag=("Demand", "Garden")
229 )["avail"]
231 # Apply garden_efficiency
232 excess = self.excess_to_garden_demand(excess)
234 # Apply any pollutants
235 vqip = self.apply_gardening_pollutants(excess)
236 return vqip
238 def apply_gardening_pollutants(self, excess):
239 """Holder function to apply pollutants (i.e., presumably fertiliser) to the
240 garden.
242 Args:
243 excess (float): A volume of water applied to a garden
245 Returns:
246 (dict): A VQIP of water that includes pollutants to be sent to land
247 """
248 # TODO Fertilisers are currently applied in the land node... which is
249 # preferable?
250 vqip = self.empty_vqip()
251 vqip["volume"] = excess
252 return vqip
254 def excess_to_garden_demand(self, excess):
255 """Apply garden_efficiency.
257 Args:
258 excess (float): Volume of water required to satisfy garden irrigation
260 Returns:
261 (float): Amount of water actually applied to garden
262 """
263 # TODO Anything more than this needed? (yes - population presence if eventually
264 # included!)
266 return excess * self.gardening_efficiency
268 def get_house_demand(self):
269 """Per capita calculations for household wastewater generation. Applies weighted
270 temperature calculation.
272 Returns:
273 (dict): A VQIP containg foul water
274 """
275 # TODO water that is consumed but not sent onwards as foul Total water required
276 consumption = self.population * self.per_capita
277 # Apply pollutants
278 foul = self.copy_vqip(self.pollutant_load)
279 # Scale to population
280 for pol in constants.ADDITIVE_POLLUTANTS:
281 foul[pol] *= self.population
282 # Update volume and temperature (which is weighted based on air temperature and
283 # constant_temp)
284 foul["volume"] = consumption
285 foul["temperature"] = (
286 self.get_data_input("temperature") * (1 - self.constant_weighting)
287 + self.constant_temp * self.constant_weighting
288 )
289 return foul