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

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 

19class HistoricalValue(models.Model): 

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

21 value = models.FloatField() 

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

23 

24 created_at = models.DateTimeField(auto_now_add=True) 

25 objects = AccessControlManager() 

26 

27 

28 

29 class Meta: 

30 ordering = ['created_at'] 

31 

32 

33class ProcessPathProperty(models.Model): 

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

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

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

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

38 

39 created_at = models.DateTimeField(auto_now_add=True) 

40 objects = AccessControlManager() 

41 

42 

43 class Meta: 

44 ordering = ['created_at'] 

45 

46 

47class PropertyInfo(models.Model): 

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

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

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

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

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

53 expression = models.OneToOneField("Expression", on_delete=models.SET_NULL, related_name="property", null=True) # this should probably be refactored to a one-to-one relationship, and could be moved to the expression model. 

54 key = models.CharField(max_length=64) 

55 displayName = models.CharField(max_length=64) 

56 index = models.IntegerField(default=0) 

57 created_at = models.DateTimeField(auto_now_add=True) 

58 values: models.QuerySet["PropertyValue"] 

59 

60 objects = AccessControlManager() 

61 

62 class Meta: 

63 ordering = ['created_at'] 

64 

65 @classmethod 

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

67 

68 value = fields.pop("value") 

69 property_info = PropertyInfo(**fields) 

70 # Create a property value object with this value 

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

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

73 

74 return property_info, [property_value] 

75 

76 

77 @classmethod 

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

79 # Create and save a new property info object 

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

81 property_info.save() 

82 for property_value in property_values: 

83 property_value.save() 

84 return property_info 

85 

86 

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

88 """ 

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

90 or the first value if no indexes are specified. 

91 

92 This method should be used in bulk operations (where 

93 all the related property values are prefetched). 

94 """ 

95 property_value = get_value_object(self, indexes) 

96 if property_value is None: 

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

98 return property_value.value 

99 

100 

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

102 """ 

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

104 or the first value if no indexes are specified. 

105 

106 This method should be used in bulk operations (where 

107 all the related property values are prefetched). 

108 """ 

109 property_value = get_value_object(self, indexes) 

110 property_value.value = value 

111 return property_value 

112 

113 

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

115 """ 

116 Get the property value object at the specified indexes, 

117 or the first value if no indexes are specified. 

118 

119 This method should not be used in bulk operations. 

120 """ 

121 if indexes is None: 

122 property_value = self.values.first() 

123 else: 

124 annotated_values = self.values.annotate( 

125 total_indexes=Count("indexedItems"), 

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

127 ) 

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

129 return property_value 

130 

131 

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

133 """ 

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

135 or the first value if no indexes are specified. 

136 

137 This method should not be used in bulk operations. 

138 """ 

139 value_object = self.get_value_object(indexes=indexes) 

140 if value_object is None: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 return None 

142 return value_object.value 

143 

144 

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

146 """ 

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

148 or the first value if no indexes are specified. 

149 

150 This method should not be used in bulk operations. 

151 """ 

152 property_value = self.get_value_object(indexes=indexes) 

153 property_value.value = value 

154 property_value.save() 

155 

156 

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

158 """ 

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

160 """ 

161 property_values = self.values.all() 

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

163 

164 def has_value(self) -> bool: 

165 property_value = self.get_value_object() 

166 if property_value is None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 return False 

168 return property_value.has_value() 

169 

170 def has_value_bulk(self) -> bool: 

171 property_value = get_value_object(self) 

172 if property_value is None: 

173 return False 

174 return property_value.has_value() 

175 

176 def get_cutoff_and_property_values(self, config: ObjectType): 

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

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

179 

180 property_values = [] 

181 for indexed_item in first_indexed_items: 

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

183 

184 count = first_indexed_items.count() 

185 

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

187 return None, None 

188 

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

190 

191 return cutoff, property_values 

192 

193 

194 def isSpecified(self) -> bool: 

195 """ 

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

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

198 Both guesses and set points are needed. 

199 """ 

200 value: PropertyValue 

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

202 if ( 

203 ( 

204 value.enabled 

205 or value.is_control_set_point() 

206 or self.is_recycle_var() 

207 ) 

208 and not value.has_value() 

209 ): 

210 return False 

211 return True 

212 

213 

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

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

216 object_type = self.set.simulationObject.objectType 

217 config = configuration.get(object_type) 

218 

219 list_prop_val = [] 

220 

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

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

223 else: 

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

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

226 

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

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

229 # all of them except the last one 

230 cutoff, property_values = self.get_cutoff_and_property_values(config) 

231 if property_values: 

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

233 if i <= cutoff: 

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

235 property_values[i].enabled = condition 

236 else: 

237 property_values[i].enabled = False 

238 list_prop_val.append(property_values[i]) 

239 else: 

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

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

242 for value in values: 

243 value.enabled = condition 

244 list_prop_val.append(value) 

245 return list_prop_val 

246 

247 def is_recycle_var(self) -> bool: 

248 return hasattr(self, "recycleConnection") 

249 

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

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

252 

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

254 """ 

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

256 """ 

257 

258 # update both the unit and the value 

259 original_unit = self.unit 

260 self.unit = new_unit 

261 # original_value = self.get_value() 

262 original_value = self.values.all() 

263 for value in list(original_value): 

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

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

266 if not value.is_enabled(): 

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

268 float_value = float(value.value) 

269 new_value = convert_value(float_value, original_unit, new_unit) 

270 value.value = new_value 

271 value.save() 

272 

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

274 #get the type of the unit operation 

275 if self.set.simulationObject == None: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true

276 return None 

277 else: 

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

279 

280 def is_custom_property(self) -> bool: 

281 return self.get_schema() is None 

282 

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

284 """ 

285 Check if the property is of type "exceptLast" 

286 """ 

287 object_type = property_info.set.simulationObject.objectType 

288 config = configuration.get(object_type) 

289 for schema in config.propertySetGroups.values(): 289 ↛ 292line 289 didn't jump to line 292 because the loop on line 289 didn't complete

290 if schema.type == "exceptLast": 290 ↛ 289line 290 didn't jump to line 289 because the condition on line 290 was always true

291 return True 

292 return False 

293