Coverage for backend/django/core/auxiliary/models/PropertyInfo.py: 96%

1from django.db import models 

2from django.db.models import Q, Count 

3 

4from core.auxiliary.models.IndexedItem import IndexedItem 

5from core.auxiliary.enums.uiEnums import DisplayType 

6from core.auxiliary.enums.unitsOfMeasure import UnitOfMeasure 

7from idaes_factory.unit_conversion import convert_value 

8from idaes_factory.queryset_lookup import get_value_object 

9from common.config_types import * 

10from core.auxiliary.models.ControlValue import ControlValue 

11from flowsheetInternals.unitops.config.config_base import configuration 

12 

13from core.managers import AccessControlManager 

14from core.auxiliary.models.PropertyValue import PropertyValue 

15from typing import TYPE_CHECKING 

16if TYPE_CHECKING: 

17 from core.auxiliary.models.PropertyValue import PropertyValue 

18 from core.auxiliary.models.RecycleData import RecycleProperty 

19 from core.auxiliary.models.PropertySet import PropertySet 

20 

21 

22class HistoricalValue(models.Model): 

23 flowsheet = models.ForeignKey("Flowsheet", on_delete=models.CASCADE, related_name="HistoricalValues") 

24 value = models.FloatField() 

25 property = models.ForeignKey("PropertyInfo", on_delete=models.CASCADE, related_name="history", null=True) 

26 

27 created_at = models.DateTimeField(auto_now_add=True) 

28 objects = AccessControlManager() 

29 

30 

31 

32 class Meta: 

33 ordering = ['created_at'] 

34 

35 

36class ProcessPathProperty(models.Model): 

37 flowsheet = models.ForeignKey("Flowsheet", on_delete=models.CASCADE, related_name="ProcessPathProperties") 

38 value = models.FloatField(null=True, blank=True) 

39 property = models.ForeignKey("PropertyInfo", on_delete=models.CASCADE, related_name="ProcessPathProperties", null=True) 

40 path = models.ForeignKey("ProcessPath", on_delete=models.CASCADE, related_name="ProcessPathProperties", null=True) 

41 

42 created_at = models.DateTimeField(auto_now_add=True) 

43 objects = AccessControlManager() 

44 

45 

46 class Meta: 

47 ordering = ['created_at'] 

48 

49 

50class PropertyInfo(models.Model): 

51 flowsheet = models.ForeignKey("Flowsheet", on_delete=models.CASCADE, related_name="propertyInfos") 

52 set = models.ForeignKey("PropertySet", on_delete=models.CASCADE, related_name="ContainedProperties", null=True) 

53 type = models.CharField(choices=DisplayType.choices , default=DisplayType.numeric) 

54 unitType = models.CharField(choices=UnitOfMeasure.choices , default=UnitOfMeasure.none) 

55 unit = models.CharField(max_length=32, blank=True) # selected unit 

56 key = models.CharField(max_length=64) 

57 displayName = models.CharField(max_length=64) 

58 index = models.IntegerField(default=0) 

59 managed = models.BooleanField(default=False) 

60 managed_source = models.CharField(max_length=64, blank=True) 

61 can_edit = models.BooleanField(default=True) 

62 can_edit_formula = models.BooleanField(default=True) 

63 can_delete = models.BooleanField(default=True) 

64 formula_incomplete = models.BooleanField(default=False) 

65 formula_incomplete_reason = models.TextField(blank=True) 

66 created_at = models.DateTimeField(auto_now_add=True) 

67 values: models.QuerySet["PropertyValue"] 

68 recycleConnection: "RecycleProperty" 

69 set: "PropertySet" 

70 

71 objects = AccessControlManager() 

72 

73 class Meta: 

74 ordering = ['created_at'] 

75 

76 @classmethod 

77 def create(cls, indexes:dict[str,list[IndexedItem]]={}, **fields) -> tuple["PropertyInfo", list[PropertyValue]]: 

78 

79 value = fields.pop("value") 

80 property_info = PropertyInfo(**fields) 

81 # Create a property value object with this value 

82 if indexes == {}: 82 ↛ 85line 82 didn't jump to line 85 because the condition on line 82 was always true

83 property_value = PropertyValue(value=value, property=property_info, flowsheet=fields.get("flowsheet")) 

84 

85 return property_info, [property_value] 

86 

87 

88 @classmethod 

89 def create_save(cls, **fields) -> "PropertyInfo": 

90 # Create and save a new property info object 

91 property_info, property_values = cls.create(**fields) 

92 property_info.save() 

93 for property_value in property_values: 

94 property_value.save() 

95 return property_info 

96 

97 

98 def get_value_bulk(self, indexes: list | None = None) -> Any: 

99 """ 

100 Get the value of the property at the specified indexes, 

101 or the first value if no indexes are specified. 

102 

103 This method should be used in bulk operations (where 

104 all the related property values are prefetched). 

105 """ 

106 property_value = get_value_object(self, indexes) 

107 if property_value is None: 

108 raise ValueError(f"No property values found with indexes={indexes}") 

109 return property_value.value 

110 

111 

112 def set_value_bulk(self, value: Any, indexes: list | None = None) -> PropertyValue: 

113 """ 

114 Set the value of the property at the given indexes, 

115 or the first value if no indexes are specified. 

116 

117 This method should be used in bulk operations (where 

118 all the related property values are prefetched). 

119 """ 

120 property_value = get_value_object(self, indexes) 

121 property_value.value = value 

122 return property_value 

123 

124 

125 def get_value_object(self, indexes: list[str] | None = None) -> PropertyValue: 

126 """ 

127 Get the property value object at the specified indexes, 

128 or the first value if no indexes are specified. 

129 

130 This method should not be used in bulk operations. 

131 """ 

132 if indexes is None: 

133 property_value = self.values.first() 

134 else: 

135 annotated_values = self.values.annotate( 

136 total_indexes=Count("indexedItems"), 

137 matching_indexes=Count("indexedItems", filter=Q(indexedItems__key__in=indexes)) 

138 ) 

139 property_value = annotated_values.get(total_indexes=len(indexes), matching_indexes=len(indexes)) 

140 return property_value 

141 

142 

143 def get_value(self, indexes: list | None = None) -> Any: 

144 """ 

145 Get the value of the property at the specified indexes, 

146 or the first value if no indexes are specified. 

147 

148 This method should not be used in bulk operations. 

149 """ 

150 value_object = self.get_value_object(indexes=indexes) 

151 if value_object is None: 

152 return None 

153 return value_object.value 

154 

155 

156 def set_value(self, value: Any, indexes=None) -> None: 

157 """ 

158 Set the value of the property at the given indexes, 

159 or the first value if no indexes are specified. 

160 

161 This method should not be used in bulk operations. 

162 """ 

163 property_value = self.get_value_object(indexes=indexes) 

164 property_value.value = value 

165 property_value.save() 

166 

167 

168 def get_indexes(self, index_set: str) -> list[IndexedItem]: 

169 """ 

170 Get all the indexed items of the specified type for this property. 

171 """ 

172 property_values = self.values.all() 

173 return IndexedItem.objects.filter(propertyValues__in=property_values, type=index_set) 

174 

175 def has_value(self) -> bool: 

176 property_value = self.get_value_object() 

177 if property_value is None: 

178 return False 

179 return property_value.has_value() 

180 

181 def has_value_bulk(self) -> bool: 

182 property_value = get_value_object(self) 

183 if property_value is None: 

184 return False 

185 return property_value.has_value() 

186 

187 def get_cutoff_and_property_values(self, config: ObjectType): 

188 first_index = config.indexSets[0] # get the outermost level e.g: "outlet 1, outlet 2" 

189 first_indexed_items = IndexedItem.objects.filter(owner=self.set.simulationObject, type=first_index) 

190 

191 property_values = [] 

192 for indexed_item in first_indexed_items: 

193 property_values.extend(indexed_item.propertyValues.all()) 

194 

195 count = first_indexed_items.count() 

196 

197 if not count: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true

198 return None, None 

199 

200 cutoff = len(property_values) - 1 - (len(property_values) / count) # get the index where it starts to be disabled 

201 

202 return cutoff, property_values 

203 

204 

205 def isSpecified(self) -> bool: 

206 """ 

207 Checks if all the required property values have been specified. 

208 This includes values that are enabled, control set points, and recycle guesses. 

209 Both guesses and set points are needed. 

210 """ 

211 value: PropertyValue 

212 for value in self.values.all(): 

213 if ( 

214 ( 

215 value.enabled 

216 or value.is_control_set_point() 

217 or self.is_recycle_var() 

218 ) 

219 and not value.has_value() 

220 ): 

221 return False 

222 return True 

223 

224 

225 def enable(self, condition: bool = True) -> list[PropertyValue]: 

226 """Enables or disables property values based on configuration rules.""" 

227 object_type = self.set.simulationObject.objectType 

228 config = configuration.get(object_type) 

229 

230 list_prop_val = [] 

231 

232 if self.key not in config.properties: 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true

233 propertySetGroup_key = None; # default to none, we are assuming the property doesn't exist. 

234 else: 

235 propertySetGroup_key = config.properties[self.key].propertySetGroup 

236 propertySetGroup_config = config.propertySetGroups.get(propertySetGroup_key, None) 

237 

238 if propertySetGroup_config is not None and propertySetGroup_config.type == "exceptLast" and condition == True: 

239 # in ExceptLast properties, you don't enable all of them, you enable  

240 # all of them except the last one 

241 cutoff, property_values = self.get_cutoff_and_property_values(config) 

242 if property_values: 

243 for i in range(len(property_values)): 

244 if i <= cutoff: 

245 # if the property is before the cutoff, it should be enabled 

246 property_values[i].enabled = condition 

247 else: 

248 property_values[i].enabled = False 

249 list_prop_val.append(property_values[i]) 

250 else: 

251 # this must be a custom property or a machine learning property. 

252 values = list(self.values.all()) 

253 for value in values: 

254 value.enabled = condition 

255 list_prop_val.append(value) 

256 return list_prop_val 

257 

258 def is_recycle_var(self) -> bool: 

259 return hasattr(self, "recycleConnection") 

260 

261 def add_control(self, prop: "PropertyValue") -> ControlValue: 

262 return ControlValue.create(setPoint=self.values.first(), manipulated=prop.values.first(), flowsheet=self.flowsheet) 

263 

264 def unit_conversion(self, new_unit: str) -> None: 

265 """ 

266 Perform a unit conversion on the value field of the PropertyInfo instance 

267 """ 

268 

269 # update both the unit and the value 

270 original_unit = self.unit 

271 self.unit = new_unit 

272 # original_value = self.get_value() 

273 original_value = self.values.all() 

274 for value in list(original_value): 

275 # we only want to update the units of properties that can't be manually edited. 

276 # this is a design choice, so users don't enter 100, switch from Pa to kPa, and it gets converted to 0.1 kPa 

277 if not value.is_enabled(): 

278 if value.value != None: 278 ↛ 274line 278 didn't jump to line 274 because the condition on line 278 was always true

279 float_value = float(value.value) 

280 new_value = convert_value(float_value, original_unit, new_unit) 

281 value.value = new_value 

282 value.save() 

283 

284 def get_schema(self) -> PropertyType | None: 

285 #get the type of the unit operation 

286 if self.set.simulationObject == None: 

287 return None 

288 else: 

289 return self.set.simulationObject.schema.properties.get(self.key, None) 

290 

291 def is_custom_property(self) -> bool: 

292 return self.get_schema() is None 

293 

294def check_is_except_last(property_info: PropertyInfo) -> bool: 

295 """ 

296 Check if the property is of type "exceptLast" 

297 """ 

298 object_type = property_info.set.simulationObject.objectType 

299 config = configuration.get(object_type) 

300 for schema in config.propertySetGroups.values(): 

301 if schema.type == "exceptLast": 

302 return True 

303 return False