Coverage for wsimod/nodes/sewer.py: 19%
72 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
5Converted to totals on 2022-05-03
6"""
7from wsimod.core import constants
8from wsimod.nodes.nodes import Node, QueueTank
11class Sewer(Node):
12 """"""
14 def __init__(
15 self,
16 name,
17 capacity=0,
18 pipe_time=0, # Sewer to sewer travel time
19 pipe_timearea={0: 1},
20 chamber_area=1,
21 chamber_floor=10,
22 data_input_dict={},
23 ):
24 """Sewer node that has a QueueTank and storage capacity. Think carefully about
25 parameterising this tank, because of course the amount of water that can flow
26 through a sewer in a timestep is different in reality than in a.
28 steady state (e.g., a sewer that can handle a peak of 6m3/s in practice could
29 not handle 6 * 86400 m3 of water in a day because that water does not flow
30 uniformly over the day).
32 Args:
33 name (str): node name
34 capacity (float, optional): Sewer tank capacity. Defaults to 0.
35 pipe_time (float, optional): Number of timesteps to spend in the queue of
36 the sewer tank. Defaults to 0.
37 pipe_timearea (dict, optional): Time area diagram that enables flows to
38 take a range of different durations to 'traverse' the tank. The keys
39 of the dict are the number of timesteps while the values are the
40 proportion of flow. E.g., {0 : 0.7, 1 : 0.3} means 70% of flow takes
41 0 timesteps and 30% takes 1 timesteps.
42 chamber_area (float, optional): Sewer tank area. Defaults to 1.
43 chamber_floor (float, optional): Sewer tank datum. Defaults to 10.
44 data_input_dict (dict, optional): Dictionary of data inputs relevant for
45 the node (though I don't think it is used). Defaults to {}.
47 NOTE that currently the queuetank either applies the pipe_timearea
48 (push_set_land) OR the pipe_time (push_set_sewer). Though this behaviour
49 could be changed by setting the number_of_timesteps property to pipe_time of
50 the sewer_tank and removing the pipe_time setting in push_set_sewer.
52 Functions intended to call in orchestration:
53 make_discharge
55 Key assumptions:
56 - Sewer networks can be represented in an aggregated manner, where
57 the behaviour of collections of manholes/pipes can be captured
58 in a single component.
59 - Travel time of water received from either `land.py/Land` objects
60 or `demand.py/Demand` objects is assumed to be received as a
61 non-point source and thus can be represented with the time-area
62 method.
63 - Travel time of water from an upstream `Sewer` object has a fixed
64 travel time through the node.
65 - The flow capacity of sewer network can be represented as with a
66 `Tank`.
67 - The `Sewer` object is not currently biochemically active.
69 Input data and parameter requirements:
70 - `pipe_timearea` is a dictionary containing the timearea diagram.
71 _Units_: duration of flow (in timesteps) and proportion of flow
72 - `pipe_time` describes the travel time of water received from upstream
73 `Sewer`
74 objects.
75 _Units_: number of timesteps
76 - `capacity`, `chamber_area`, `chamber_datum` describe the dimensions of the
77 `Tank` that controls flow.
78 _Units_: cubic metres, squared metres, metres
79 """
80 # Set parameters
81 self.capacity = capacity
82 self.pipe_time = pipe_time
83 self.pipe_timearea = pipe_timearea
84 self.chamber_area = chamber_area
85 self.chamber_floor = chamber_floor
86 # TODO I don't think this is used..
87 self.data_input_dict = data_input_dict
89 # Update args
90 super().__init__(name)
92 # Update handlers
93 self.push_set_handler["Sewer"] = self.push_set_sewer
94 self.push_set_handler["default"] = self.push_set_sewer
95 self.push_set_handler["Land"] = self.push_set_land
96 self.push_set_handler["Demand"] = self.push_set_land
98 self.push_check_handler["default"] = self.push_check_sewer
99 self.push_check_handler["Sewer"] = self.push_check_sewer
100 self.push_check_handler["Demand"] = self.push_check_sewer
101 self.push_check_handler["Land"] = self.push_check_sewer
103 # Create sewer tank
104 # TODO this might work better as a ResidenceTank (maybe also decay?)
105 self.sewer_tank = QueueTank(
106 capacity=self.capacity,
107 number_of_timesteps=0,
108 datum=self.chamber_floor,
109 area=self.chamber_area,
110 )
112 # Mass balance
113 self.mass_balance_ds.append(lambda: self.sewer_tank.ds())
115 def push_check_sewer(self, vqip=None):
116 """Generic push check, simply looks at excess.
118 Args:
119 vqip (dict, optional): A VQIP that can be used to limit the volume in
120 the return value (only volume key is used). Defaults to None.
122 Returns:
123 excess (dict): Sewer tank excess
124 """
125 # Get excess
126 excess = self.sewer_tank.get_excess()
127 if vqip is None:
128 return excess
129 # Limit respone to vqip volume
130 excess = self.v_change_vqip(excess, min(excess["volume"], vqip["volume"]))
131 return excess
133 def push_set_sewer(self, vqip):
134 """Generic push request setting that implements basic queue travel time (it does
135 NOT implement timearea travel time). Updates the sewer tank storage. Assumes
136 that the inflow arc has accurately calculated capacity with push_check_sewer,
137 thus the water is forced.
139 Args:
140 vqip (dict): A VQIP amount of water to push
142 Returns:
143 (dict): A VQIP amount of water that was not received
144 """
145 # Sewer to sewer push, update queued tank
146 return self.sewer_tank.push_storage(vqip, time=self.pipe_time)
148 def push_set_land(self, vqip):
149 """Push request that applies pipe_timearea (see __init__ for description). As
150 with push_set_sewer, push is also forced. Used to receive flow from land or
151 demand that is assumed to occur widely across some kind of sewer catchment.
153 Args:
154 vqip (dict): A VQIP amount to be pushed
156 Returns:
157 (dict): A VQIP amount that was not received
158 """
159 # Land/demand to sewer push, update queued tank
161 reply = self.empty_vqip()
163 # Iterate over timearea diagram
164 for time, normalised in self.pipe_timearea.items():
165 vqip_ = self.v_change_vqip(vqip, vqip["volume"] * normalised)
166 reply_ = self.sewer_tank.push_storage(vqip_, time=time)
167 reply = self.sum_vqip(reply, reply_)
169 return reply
171 def make_discharge(self):
172 """Function to trigger downstream sewer flow.
174 Updates sewer tank travel time, pushes to WWTW, then sewer, then CSO. May flood
175 land if, after these attempts, the sewer tank storage is above capacity.
176 """
177 self.sewer_tank.internal_arc.update_queue(direction="push")
178 # TODO... do I need to do anything with this backflow... does it ever happen?
179 # Discharge to Sewer if possible
180 # remaining = self.push_distributed(self.sewer_tank.active_storage,
181 # of_type = 'Sewer',
182 # tag = 'Sewer')
184 # #Discharge to WWTW if possible
185 # remaining = self.push_distributed(remaining,
186 # of_type = 'WWTW',
187 # tag = 'Sewer')
189 # #CSO discharge
190 # remaining = self.push_distributed(remaining,
191 # of_type = ['Node', 'River'])
193 remaining = self.push_distributed(self.sewer_tank.active_storage)
195 # TODO backflow can cause mass balance errors here
197 # Update tank
198 sent = self.extract_vqip(self.sewer_tank.active_storage, remaining)
199 reply = self.sewer_tank.pull_storage_exact(sent)
200 if (reply["volume"] - sent["volume"]) > constants.FLOAT_ACCURACY:
201 print("Miscalculated tank storage in discharge")
203 # Flood excess
204 ponded = self.sewer_tank.pull_ponded()
205 if ponded["volume"] > constants.FLOAT_ACCURACY:
206 reply_ = self.push_distributed(ponded, of_type=["Land"], tag="Sewer")
207 reply_ = self.sewer_tank.push_storage(reply_, time=0, force=True)
208 if reply_["volume"]:
209 print("ponded water cant reenter")
211 def end_timestep(self):
212 """Overwrite end_timestep behaviour to update tank variables."""
213 self.sewer_tank.end_timestep()
215 def reinit(self):
216 """Call Tank reinit."""
217 self.sewer_tank.reinit()
220class EnfieldFoulSewer(Sewer):
221 """"""
223 # TODO: combine with sewer
224 def __init__(
225 self,
226 name,
227 capacity=0,
228 pipe_time=0, # Sewer to sewer travel time
229 pipe_timearea={0: 1},
230 chamber_area=1,
231 chamber_floor=10,
232 data_input_dict={},
233 ):
234 """Alternate legacy sewer class...
236 I dont think this is needed any more.
237 """
238 # TODO above
240 super().__init__(
241 name,
242 capacity=capacity,
243 pipe_time=pipe_time,
244 pipe_timearea=pipe_timearea,
245 chamber_area=chamber_area,
246 chamber_floor=chamber_floor,
247 data_input_dict=data_input_dict,
248 )
249 self.__class__.__name__ = "Sewer"
251 def make_discharge(self):
252 """"""
253 _ = self.sewer_tank.internal_arc.update_queue(direction="push")
255 # Discharge downstream
256 if (
257 self.sewer_tank.storage["volume"]
258 > self.storm_exchange * self.sewer_tank.capacity
259 ):
260 exchange_v = min(
261 (1 - self.storm_exchange) * self.sewer_tank.capacity,
262 self.sewer_tank.active_storage["volume"],
263 )
264 exchange = self.v_change_vqip(self.sewer_tank.active_storage, exchange_v)
265 remaining = self.push_distributed(exchange)
266 sent_to_exchange = self.v_change_vqip(
267 self.sewer_tank.active_storage, exchange_v - remaining["volume"]
268 )
269 self.sewer_tank.pull_storage(sent_to_exchange)
271 remaining = self.push_distributed(
272 self.sewer_tank.active_storage, of_type=["Waste"]
273 )
275 # Update tank
276 sent = self.sewer_tank.active_storage["volume"] - remaining["volume"]
277 sent = self.v_change_vqip(self.sewer_tank.active_storage, sent)
278 reply = self.sewer_tank.pull_storage(sent)
279 if (reply["volume"] - sent["volume"]) > constants.FLOAT_ACCURACY:
280 print("Miscalculated tank storage in discharge")