Coverage for backend/ahuora-builder/src/ahuora_builder/custom/custom_valve.py: 61%

114 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-05-13 02:47 +0000

1################################################################################# 

2# The Institute for the Design of Advanced Energy Systems Integrated Platform 

3# Framework (IDAES IP) was produced under the DOE Institute for the 

4# Design of Advanced Energy Systems (IDAES). 

5# 

6# Copyright (c) 2018-2024 by the software owners: The Regents of the 

7# University of California, through Lawrence Berkeley National Laboratory, 

8# National Technology & Engineering Solutions of Sandia, LLC, Carnegie Mellon 

9# University, West Virginia University Research Corporation, et al. 

10# All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md 

11# for full copyright and license information. 

12################################################################################# 

13""" 

14This provides standard valve models for adiabatic control valves. Beyond the 

15most common valve models, and adiabatic valve model can be added by supplying 

16custom callbacks for the pressure-flow relation or valve function. 

17""" 

18# Changing existing config block attributes 

19# pylint: disable=protected-access 

20 

21__Author__ = "John Eslick" 

22 

23from enum import Enum 

24 

25from ahuora_builder.custom.valve_pressure_changer import ValvePressureChangerData 

26import pyomo.environ as pyo 

27from pyomo.common.config import ConfigValue, In 

28 

29from idaes.core import declare_process_block_class 

30from .updated_pressure_changer import ( 

31 ThermodynamicAssumption, 

32 MaterialBalanceType, 

33) 

34 

35from .updated_pressure_changer import PressureChangerData 

36 

37from idaes.core.util.exceptions import ConfigurationError 

38import idaes.logger as idaeslog 

39import idaes.core.util.scaling as iscale 

40from .inverted import add_inverted, initialise_inverted 

41 

42_log = idaeslog.getLogger(__name__) 

43 

44 

45class ValveFunctionType(Enum): 

46 """ 

47 Enum of supported valve types. 

48 """ 

49 

50 linear = 1 

51 quick_opening = 2 

52 equal_percentage = 3 

53 

54 

55def linear_cb(valve): 

56 """ 

57 Linear opening valve function callback. 

58 """ 

59 

60 @valve.Expression(valve.flowsheet().time) 

61 def valve_function(b, t): 

62 return b.valve_opening[t] 

63 

64 

65def quick_cb(valve): 

66 """ 

67 Quick opening valve function callback. 

68 """ 

69 

70 @valve.Expression(valve.flowsheet().time) 

71 def valve_function(b, t): 

72 return pyo.sqrt(b.valve_opening[t]) 

73 

74 

75def equal_percentage_cb(valve): 

76 """ 

77 Equal percentage valve function callback. 

78 """ 

79 valve.alpha = pyo.Var(initialize=100, doc="Valve function parameter") 

80 valve.alpha.fix() 

81 

82 @valve.Expression(valve.flowsheet().time) 

83 def valve_function(b, t): 

84 return b.alpha ** (b.valve_opening[t] - 1) 

85 

86 

87def pressure_flow_default_callback(valve): 

88 """ 

89 Add the default pressure flow relation constraint. This will be used in the 

90 valve model, a custom callback is provided. 

91 """ 

92 umeta = ( 

93 valve.control_volume.config.property_package.get_metadata().get_derived_units 

94 ) 

95 

96 valve.Cv = pyo.Var( 

97 initialize=0.1, 

98 doc="Valve flow coefficient", 

99 units=umeta("amount") / umeta("time") / umeta("pressure") ** 0.5, 

100 ) 

101 # valve.Cv.fix() 

102 

103 valve.flow_var = pyo.Reference(valve.control_volume.properties_in[:].flow_mol) 

104 valve.pressure_flow_equation_scale = lambda x: x**2 

105 

106 @valve.Constraint(valve.flowsheet().time) 

107 def pressure_flow_equation(b, t): 

108 Po = b.control_volume.properties_out[t].pressure 

109 Pi = b.control_volume.properties_in[t].pressure 

110 F = b.control_volume.properties_in[t].flow_mol 

111 Cv = b.Cv 

112 fun = b.valve_function[t] 

113 return F**2 == Cv**2 * (Pi - Po) * fun**2 

114 

115 

116@declare_process_block_class("Valve", doc="Adiabatic valves") 

117class ValveData(PressureChangerData): 

118 """ 

119 Basic valve model class. 

120 """ 

121 

122 # Same settings as the default pressure changer, but force to expander with 

123 # isentropic efficiency 

124 CONFIG = PressureChangerData.CONFIG() 

125 CONFIG.compressor = False 

126 CONFIG.get("compressor")._default = False 

127 CONFIG.get("compressor")._domain = In([False]) 

128 CONFIG.material_balance_type = MaterialBalanceType.componentTotal 

129 CONFIG.get("material_balance_type")._default = MaterialBalanceType.componentTotal 

130 CONFIG.thermodynamic_assumption = ThermodynamicAssumption.adiabatic 

131 CONFIG.get("thermodynamic_assumption")._default = ThermodynamicAssumption.adiabatic 

132 CONFIG.get("thermodynamic_assumption")._domain = In( 

133 [ThermodynamicAssumption.adiabatic] 

134 ) 

135 CONFIG.declare( 

136 "valve_function_callback", 

137 ConfigValue( 

138 default=ValveFunctionType.linear, 

139 description="Valve function type or callback for custom", 

140 doc="""This takes either an enumerated valve function type in: { 

141ValveFunctionType.linear, ValveFunctionType.quick_opening, 

142ValveFunctionType.equal_percentage, ValveFunctionType.custom} or a callback 

143function that takes a valve model object as an argument and adds a time-indexed 

144valve_function expression to it. Any additional required variables, expressions, 

145or constraints required can also be added by the callback.""", 

146 ), 

147 ) 

148 CONFIG.declare( 

149 "pressure_flow_callback", 

150 ConfigValue( 

151 default=pressure_flow_default_callback, 

152 description="Callback function providing the valve_function expression", 

153 doc="""This callback function takes a valve model object as an argument 

154and adds a time-indexed valve_function expression to it. Any additional required 

155variables, expressions, or constraints required can also be added by the callback.""", 

156 ), 

157 ) 

158 

159 def build(self): 

160 super().build() 

161 

162 self.valve_opening = pyo.Var( 

163 self.flowsheet().time, 

164 initialize=1, 

165 bounds=(0, 1), 

166 doc="Fraction open for valve from 0 to 1", 

167 ) 

168 

169 

170 #commented out to allow for valve opening to be set by the user 

171 #self.valve_opening.fix() 

172 

173 

174 

175 # If the valve function callback is set to one of the known enumerated 

176 # types, set the callback appropriately. If not callable and not a known 

177 # type raise ConfigurationError. 

178 vfcb = self.config.valve_function_callback 

179 if not callable(vfcb): 179 ↛ 188line 179 didn't jump to line 188 because the condition on line 179 was always true

180 if vfcb == ValveFunctionType.linear: 

181 self.config.valve_function_callback = linear_cb 

182 elif vfcb == ValveFunctionType.quick_opening: 

183 self.config.valve_function_callback = quick_cb 

184 elif vfcb == ValveFunctionType.equal_percentage: 184 ↛ 187line 184 didn't jump to line 187 because the condition on line 184 was always true

185 self.config.valve_function_callback = equal_percentage_cb 

186 else: 

187 raise ConfigurationError("Invalid valve function callback.") 

188 self.config.valve_function_callback(self) 

189 self.config.pressure_flow_callback(self) 

190 

191 # add deltaP_inverted as a property 

192 add_inverted(self,"deltaP") 

193 

194 def initialize_build( 

195 self, 

196 state_args=None, 

197 outlvl=idaeslog.NOTSET, 

198 solver=None, 

199 optarg=None, 

200 ): 

201 """ 

202 Initialize the valve based on a deltaP guess. 

203 

204 Args: 

205 state_args (dict): Initial state for property initialization 

206 outlvl : sets output level of initialization routine 

207 solver (str): Solver to use for initialization 

208 optarg (dict): Solver arguments dictionary 

209 """ 

210 initialise_inverted(self,"deltaP") 

211 

212 for t in self.flowsheet().time: 

213 if ( 213 ↛ 221line 213 didn't jump to line 221 because the condition on line 213 was always true

214 self.deltaP[t].fixed or self.deltaP_inverted[t].fixed 

215 or self.ratioP[t].fixed 

216 or self.outlet.pressure[t].fixed 

217 ): 

218 continue 

219 # Generally for the valve initialization pressure drop won't be 

220 # fixed, so if there is no good guess on deltaP try to out one in 

221 Pout = self.outlet.pressure[t] 

222 Pin = self.inlet.pressure[t] 

223 if self.deltaP[t].value is not None: 

224 prdp = pyo.value((self.deltaP[t] - Pin) / Pin) 

225 else: 

226 prdp = -100 # crazy number to say don't use deltaP as guess 

227 if pyo.value(Pout / Pin) > 1 or pyo.value(Pout / Pin) < 0.0: 

228 if pyo.value(self.ratioP[t]) <= 1 and pyo.value(self.ratioP[t]) >= 0: 

229 Pout.value = pyo.value(Pin * self.ratioP[t]) 

230 elif prdp <= 1 and prdp >= 0: 

231 Pout.value = pyo.value(prdp * Pin) 

232 else: 

233 Pout.value = pyo.value(Pin * 0.95) 

234 self.deltaP[t] = pyo.value(Pout - Pin) 

235 self.ratioP[t] = pyo.value(Pout / Pin) 

236 

237 # one bad thing about reusing this is that the log messages aren't 

238 # really compatible with being nested inside another initialization 

239 super().initialize_build( 

240 state_args=state_args, outlvl=outlvl, solver=solver, optarg=optarg 

241 ) 

242 

243 def calculate_scaling_factors(self): 

244 """ 

245 Calculate pressure flow constraint scaling from flow variable scale. 

246 """ 

247 # The value of the valve opening and the output of the valve function 

248 # expression are between 0 and 1, so the only thing that needs to be 

249 # scaled here is the pressure-flow constraint, which can be scaled by 

250 # using the flow variable scale. The flow variable could be defined 

251 # in different ways, so the flow variable is determined here from a 

252 # "flow_var[t]" reference set in the pressure-flow callback. The flow 

253 # term could be in various forms, so an optional 

254 # "pressure_flow_equation_scale" function can be defined in the callback 

255 # as well. The pressure-flow function could be flow = f(Pin, Pout), but 

256 # it could also be flow**2 = f(Pin, Pout), ... The so 

257 # "pressure_flow_equation_scale" provides the form of the LHS side as 

258 # a function of the flow variable. 

259 

260 super().calculate_scaling_factors() 

261 

262 # Do some error trapping. 

263 if not hasattr(self, "pressure_flow_equation"): 

264 raise AttributeError( 

265 "Pressure-flow callback must define pressure_flow_equation[t]" 

266 ) 

267 # Check for flow term form if none assume flow = f(Pin, Pout) 

268 if hasattr(self, "pressure_flow_equation_scale"): 

269 ff = self.pressure_flow_equation_scale 

270 else: 

271 # pylint: disable-next=unnecessary-lambda-assignment 

272 ff = lambda x: x 

273 # if the "flow_var" is not set raise an exception 

274 if not hasattr(self, "flow_var"): 

275 raise AttributeError( 

276 "Pressure-flow callback must define flow_var[t] reference" 

277 ) 

278 

279 # Calculate and set the pressure-flow relation scale. 

280 if hasattr(self, "pressure_flow_equation"): 

281 for t, c in self.pressure_flow_equation.items(): 

282 iscale.constraint_scaling_transform( 

283 c, 

284 ff( 

285 iscale.get_scaling_factor( 

286 self.flow_var[t], default=1, warning=True 

287 ) 

288 ), 

289 ) 

290 

291 def _get_performance_contents(self, time_point=0): 

292 pc = super()._get_performance_contents(time_point=time_point) 

293 

294 pc["vars"]["Opening"] = self.valve_opening[time_point] 

295 try: 

296 pc["vars"]["Valve Coefficient"] = self.Cv 

297 except AttributeError: 

298 pass 

299 if self.config.valve_function_callback == ValveFunctionType.equal_percentage: 

300 pc["vars"]["alpha"] = self.alpha 

301 return pc 

302 

303 # Use the same diagnostic method as the ValvePressureChanger. 

304 diagnose = ValvePressureChangerData.diagnose 

305 

306