Coverage for wsimod\nodes\distribution.py: 25%

1# -*- coding: utf-8 -*- 

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

3 

4@author: bdobson 

5""" 

6 

7from typing import Any, Dict 

8 

9from wsimod.core import constants 

10from wsimod.nodes.nodes import Node 

11 

12 

13def decorate_leakage_set(self, f): 

14 """Decorator to extend the functionality of `f` by introducing leakage. This is 

15 achieved by adjusting the volume of the request (vqip) to include anticipated 

16 leakage, calling the original function `f`, and then distributing the leaked amount 

17 to groundwater. 

18 

19 Args: 

20 self (instance of Distribution class): The Distribution object to be 

21 extended 

22 f (function): The function to be extended. Expected to be the 

23 Distribution object's pull_set function. 

24 

25 Returns: 

26 pull_set (function): The decorated function which includes the 

27 original functionality of `f` and additional leakage operations. 

28 """ 

29 

30 def pull_set(vqip, **kwargs): 

31 """ 

32 

33 Args: 

34 vqip: 

35 **kwargs: 

36 

37 Returns: 

38 

39 """ 

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

41 

42 reply = f(vqip, **kwargs) 

43 

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

45 

46 reply = self.extract_vqip(reply, amount_leaked) 

47 

48 unsuccessful_leakage = self.push_distributed( 

49 amount_leaked, of_type="Groundwater" 

50 ) 

51 if unsuccessful_leakage["volume"] > constants.FLOAT_ACCURACY: 

52 print( 

53 "warning, distribution leakage not going to GW in {0} at {1}".format( 

54 self.name, self.t 

55 ) 

56 ) 

57 reply = self.sum_vqip(reply, unsuccessful_leakage) 

58 

59 return reply 

60 

61 return pull_set 

62 

63 

64def decorate_leakage_check(self, f): 

65 """Decorator to extend the functionality of `f` by introducing leakage. This is 

66 achieved by adjusting the volume of the request (vqip) to include anticipated 

67 leakage and then calling the original function `f`. 

68 

69 Args: 

70 self (instance of Distribution class): The Distribution object to be 

71 extended 

72 f (function): The function to be extended. Expected to be the 

73 Distribution object's pull_set function. 

74 

75 Returns: 

76 pull_check (function): The decorated function which includes the 

77 original functionality of `f` and additional leakage operations. 

78 """ 

79 

80 def pull_check(vqip, **kwargs): 

81 """ 

82 

83 Args: 

84 vqip: 

85 **kwargs: 

86 

87 Returns: 

88 

89 """ 

90 if vqip is not None: 

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

92 reply = f(vqip, **kwargs) 

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

94 

95 reply = self.extract_vqip(reply, amount_leaked) 

96 return reply 

97 

98 return pull_check 

99 

100 

101class Distribution(Node): 

102 """""" 

103 

104 def __init__(self, leakage=0, **kwargs): 

105 """A Node that cannot be pushed to. Intended to pass calls to FWTW - though this 

106 currently relies on the user to connect it properly. 

107 

108 Args: 

109 leakage (float, optional): 1 > float >= 0 to express how much 

110 water should be leaked to any attached groundwater nodes. This 

111 number represents the proportion of total flow through the node 

112 that should be leaked. 

113 Defaults to 0. 

114 

115 Functions intended to call in orchestration: 

116 None 

117 

118 Key assumptions: 

119 - No distribution processes yet represented, this class is just 

120 for conveyance. 

121 

122 Input data and parameter requirements: 

123 - None 

124 """ 

125 self.leakage = leakage 

126 super().__init__(**kwargs) 

127 # Update handlers 

128 self.push_set_handler["default"] = self.push_set_deny 

129 self.push_check_handler["default"] = self.push_check_deny 

130 self.decorate_pull_handlers() 

131 

132 def decorate_pull_handlers(self): 

133 """Decorate handlers if there is leakage ratio.""" 

134 if self.leakage > 0: 

135 self.pull_set_handler["default"] = decorate_leakage_set( 

136 self, self.pull_set_handler["default"] 

137 ) 

138 self.pull_check_handler["default"] = decorate_leakage_check( 

139 self, self.pull_check_handler["default"] 

140 ) 

141 

142 def apply_overrides(self, overrides: Dict[str, Any] = {}): 

143 """Apply overrides to the sewer. 

144 

145 Enables a user to override any of the following parameters: 

146 leakage. 

147 

148 Args: 

149 overrides (dict, optional): Dictionary of overrides. Defaults to {}. 

150 """ 

151 self.leakage = overrides.pop("leakage", self.leakage) 

152 self.decorate_pull_handlers() 

153 super().apply_overrides(overrides) 

154 

155 

156class UnlimitedDistribution(Distribution): 

157 """""" 

158 

159 def __init__(self, **kwargs): 

160 """A distribution node that provides unlimited water while tracking pass 

161 balance. 

162 

163 Functions intended to call in orchestration: 

164 None 

165 

166 Key assumptions: 

167 - Water demand is always satisfied. 

168 

169 Input data and parameter requirements: 

170 - None 

171 """ 

172 super().__init__(**kwargs) 

173 # Update handlers 

174 self.pull_set_handler["default"] = self.pull_set_unlimited 

175 self.pull_check_handler["default"] = lambda x: self.v_change_vqip( 

176 self.empty_vqip(), constants.UNBOUNDED_CAPACITY 

177 ) 

178 

179 # States 

180 self.supplied = self.empty_vqip() 

181 

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

183 

184 def pull_set_unlimited(self, vqip): 

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

186 

187 Args: 

188 vqip (dict): A VQIP amount to request 

189 

190 Returns: 

191 vqip (dict): A VQIP amount that was supplied 

192 """ 

193 # TODO maybe need some pollutant concentrations? 

194 vqip = self.v_change_vqip(self.empty_vqip(), vqip["volume"]) 

195 self.supplied = self.sum_vqip(self.supplied, vqip) 

196 return vqip 

197 

198 def end_timestep(self): 

199 """Update state variables.""" 

200 self.supplied = self.empty_vqip()