Coverage for backend/django/PinchAnalysis/models/InputModels.py: 91%

132 statements  

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

1from django.db import models 

2from core.auxiliary.enums.unitOpGraphics import ConType 

3from core.auxiliary.enums import pinchEnums 

4from core.managers import AccessControlManager, AllFlowsheetStatesManager 

5 

6from flowsheetInternals.unitops.models.SimulationObject import SimulationObject 

7from core.auxiliary.enums import SimulationObjectClass 

8from core.auxiliary.models.PropertyInfo import PropertyInfo 

9from core.auxiliary.models.FlowsheetHistoryModel import ( 

10 DependentFlowsheetHistoryModel, 

11) 

12 

13 

14class PinchInputs(models.Model): 

15 flowsheet_state = models.ForeignKey( 

16 "core_auxiliary.FlowsheetState", 

17 on_delete=models.CASCADE, 

18 related_name="PinchInputs", 

19 ) 

20 project_owner = models.OneToOneField( 

21 "PinchAnalysis.StreamDataProject", 

22 on_delete=models.CASCADE, 

23 related_name="Inputs", 

24 null=True, 

25 ) 

26 name = models.CharField(max_length=64, default="Inputs") 

27 

28 created_at = models.DateTimeField(auto_now_add=True) 

29 objects = AccessControlManager() 

30 all_states = AllFlowsheetStatesManager() 

31 

32 

33class StreamDataEntry(DependentFlowsheetHistoryModel, models.Model): 

34 flowsheet_state = models.ForeignKey( 

35 "core_auxiliary.FlowsheetState", 

36 on_delete=models.CASCADE, 

37 related_name="StreamDataEntries", 

38 ) 

39 custom = models.BooleanField(default=False) 

40 group = models.ForeignKey( 

41 "flowsheetInternals_graphicData.Grouping", 

42 on_delete=models.CASCADE, 

43 related_name="StreamDataEntries", 

44 ) 

45 streamDataProject = models.ForeignKey( 

46 "StreamDataProject", on_delete=models.CASCADE, related_name="StreamDataEntries" 

47 ) 

48 unitop = models.ForeignKey( 

49 "flowsheetInternals_unitops.SimulationObject", 

50 on_delete=models.CASCADE, 

51 related_name="StreamDataEntries", 

52 null=True, 

53 ) 

54 property_package_mapping = models.CharField( 

55 max_length=32, null=True 

56 ) # eg: "Cold Side" 

57 # nextStream = models.ManyToManyField("StreamDataEntry", blank=True) 

58 t_h_data = models.JSONField(null=True, blank=True) 

59 states = models.JSONField(null=True, blank=True) 

60 expanded = models.BooleanField(default=False) 

61 

62 Segments: models.QuerySet["Segment"] 

63 

64 created_at = models.DateTimeField(auto_now_add=True) 

65 objects = AccessControlManager() 

66 all_states = AllFlowsheetStatesManager() 

67 

68 @property 

69 def zone(self) -> str: 

70 return self.group.simulationObject.componentName 

71 

72 # TODO: Null checks are really bad on these, if we have a custom StreamDataEntry, these would break. 

73 # We can use the first and last segment to get the inlet and outlet streams, but thats for the future 

74 @property 

75 def inlet_outlet_stream(self) -> tuple[SimulationObject, SimulationObject]: 

76 unitop_schema = self.unitop.schema 

77 property_package_ports = unitop_schema.propertyPackagePorts[ 

78 self.property_package_mapping 

79 ] 

80 # We can use get here because Pinch only worries about heaters and heat exchangers, which only 

81 # have one inlet and one outlet for each property package. 

82 inletPort = self.unitop.ports.get( 

83 key__in=property_package_ports, direction=ConType.Inlet 

84 ) 

85 inletStream = inletPort.stream 

86 

87 outletPort = self.unitop.ports.get( 

88 key__in=property_package_ports, direction=ConType.Outlet 

89 ) 

90 outletStream = outletPort.stream 

91 return (inletStream, outletStream) 

92 

93 @property 

94 def temperatures(self) -> tuple[PropertyInfo, PropertyInfo]: 

95 inlet, outlet = self.inlet_outlet_stream 

96 inlet_temp = inlet.properties.get_property("temperature") 

97 outlet_temp = outlet.properties.get_property("temperature") 

98 return (inlet_temp, outlet_temp) 

99 

100 @property 

101 def pressures(self) -> tuple[PropertyInfo, PropertyInfo]: 

102 inlet, outlet = self.inlet_outlet_stream 

103 inlet_pressure = inlet.properties.get_property("pressure") 

104 outlet_pressure = outlet.properties.get_property("pressure") 

105 return (inlet_pressure, outlet_pressure) 

106 

107 @property 

108 def enthalpies(self) -> tuple[PropertyInfo, PropertyInfo]: 

109 inlet, outlet = self.inlet_outlet_stream 

110 inlet_enthalpy = inlet.properties.get_property("enth_mol") 

111 outlet_enthalpy = outlet.properties.get_property("enth_mol") 

112 return (inlet_enthalpy, outlet_enthalpy) 

113 

114 @property 

115 def heat_flow(self) -> PropertyInfo: 

116 match self.unitop.objectType: 

117 case SimulationObjectClass.HeatExchanger: 

118 heat_flow = self.unitop.properties.get_property("heat_duty") 

119 case SimulationObjectClass.Heater: 

120 heat_flow = self.unitop.properties.get_property("heat_duty") 

121 case SimulationObjectClass.Cooler: 121 ↛ 123line 121 didn't jump to line 123 because the pattern on line 121 always matched

122 heat_flow = self.unitop.properties.get_property("heat_duty_inverted") 

123 case _: 

124 raise ValueError( 

125 "Invalid simulation object type for heat flow extraction." 

126 ) 

127 return heat_flow 

128 

129 

130class Segment(DependentFlowsheetHistoryModel, models.Model): 

131 flowsheet_state = models.ForeignKey( 

132 "core_auxiliary.FlowsheetState", 

133 on_delete=models.CASCADE, 

134 related_name="Segments", 

135 ) 

136 stream_data_entry = models.ForeignKey( 

137 StreamDataEntry, on_delete=models.CASCADE, related_name="Segments", null=True 

138 ) 

139 hen_node = models.ForeignKey( 

140 "HenNode", on_delete=models.CASCADE, related_name="Segments", null=True 

141 ) 

142 name = models.CharField(max_length=128) 

143 t_supply = models.FloatField(default=0) 

144 t_target = models.FloatField(default=0) 

145 p_supply = models.FloatField(default=0) 

146 p_target = models.FloatField(default=0) 

147 h_supply = models.FloatField(default=0) 

148 h_target = models.FloatField(default=0) 

149 heat_flow = models.FloatField(default=0) 

150 dt_cont = models.FloatField(default=5.0) 

151 htc = models.FloatField(default=1.0) 

152 area = models.FloatField(default=0) 

153 

154 created_at = models.DateTimeField(auto_now_add=True) 

155 objects = AccessControlManager() 

156 all_states = AllFlowsheetStatesManager() 

157 

158 @property 

159 def zone(self) -> str: 

160 return self.zone_object.simulationObject.componentName 

161 

162 @property 

163 def zone_object(self): 

164 return self.stream_data_entry.group 

165 

166 class Meta: 

167 ordering = ["created_at"] 

168 

169 @classmethod 

170 def create(cls, **kwargs): 

171 instance = cls.objects.create(**kwargs) 

172 instance.save() 

173 return instance 

174 

175 def _calc_area(self) -> float: 

176 """ 

177 A = Q / (htc * deltaT) 

178 Q:kW 

179 htc: kw/m^2/degC 

180 deltaT: t_supply - t_target in degC 

181 THIS IS WRONG AND NEEDS REVISION (dT is wrong). 

182 Needs ln((dT_HOT)/(dT_COLD)) 

183 """ 

184 htc = float(self.htc or 1) 

185 Q = float(self.heat_flow or 0) 

186 dT = abs(float(self.t_supply) - float(self.t_target)) 

187 if htc <= 0 or dT <= 0: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 return 0.0 

189 return Q / (htc * dT) 

190 

191 

192class PinchUtility(models.Model): 

193 flowsheet_state = models.ForeignKey( 

194 "core_auxiliary.FlowsheetState", 

195 on_delete=models.CASCADE, 

196 related_name="PinchUtilities", 

197 ) 

198 input_owner = models.ForeignKey( 

199 PinchInputs, on_delete=models.CASCADE, related_name="PinchUtilities", null=False 

200 ) 

201 # zone = models.CharField(max_length=32, null=True) ############ ADD ################### 

202 name = models.CharField(max_length=64) 

203 type = models.CharField( 

204 choices=pinchEnums.StreamType.choices, default=pinchEnums.StreamType.Both 

205 ) 

206 t_supply = models.FloatField(null=True) 

207 t_target = models.FloatField(null=True) 

208 heat_flow = models.FloatField( 

209 default=0, null=True 

210 ) ############ CHANGE TO MAX HEAT FLOW ################### 

211 dt_cont = models.FloatField(default=5.0, null=True) 

212 price = models.FloatField(default=30.0, null=True) 

213 htc = models.FloatField(default=1.0, null=True) 

214 

215 created_at = models.DateTimeField(auto_now_add=True) 

216 objects = AccessControlManager() 

217 all_states = AllFlowsheetStatesManager() 

218 

219 class Meta: 

220 ordering = ["created_at"] 

221 

222 @classmethod 

223 def create(cls, **kwargs): 

224 # Default stream values 

225 defaults = { 

226 "input_owner": kwargs.get("input_owner", None), 

227 # 'zone': kwargs.get('zone', "Site"), 

228 "name": kwargs.get("name", "Utility"), 

229 "type": kwargs.get("type", pinchEnums.StreamType.Both), 

230 "t_supply": kwargs.get("t_supply", None), 

231 "t_target": kwargs.get("t_target", None), 

232 "heat_flow": kwargs.get("heat_flow", 0.0), 

233 "dt_cont": kwargs.get("dt_cont", 5.0), 

234 "htc": kwargs.get("htc", 1.0), 

235 "price": kwargs.get("price", 1.0), 

236 } 

237 

238 kwargs.update(defaults) 

239 instance = cls.objects.create(**kwargs) 

240 instance.save() 

241 return instance