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

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 key = models.CharField(max_length=64) 

54 displayName = models.CharField(max_length=64) 

55 index = models.IntegerField(default=0) 

56 created_at = models.DateTimeField(auto_now_add=True) 

57 values: models.QuerySet["PropertyValue"] 

58 

59 objects = AccessControlManager() 

60 

61 class Meta: 

62 ordering = ['created_at'] 

63 

64 @classmethod 

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

66 

67 value = fields.pop("value") 

68 property_info = PropertyInfo(**fields) 

69 # Create a property value object with this value 

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

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

72 

73 return property_info, [property_value] 

74 

75 

76 @classmethod 

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

78 # Create and save a new property info object 

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

80 property_info.save() 

81 for property_value in property_values: 

82 property_value.save() 

83 return property_info 

84 

85 

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

87 """ 

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

89 or the first value if no indexes are specified. 

90 

91 This method should be used in bulk operations (where 

92 all the related property values are prefetched). 

93 """ 

94 property_value = get_value_object(self, indexes) 

95 if property_value is None: 

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

97 return property_value.value 

98 

99 

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

101 """ 

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

103 or the first value if no indexes are specified. 

104 

105 This method should be used in bulk operations (where 

106 all the related property values are prefetched). 

107 """ 

108 property_value = get_value_object(self, indexes) 

109 property_value.value = value 

110 return property_value 

111 

112 

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

114 """ 

115 Get the property value object at the specified indexes, 

116 or the first value if no indexes are specified. 

117 

118 This method should not be used in bulk operations. 

119 """ 

120 if indexes is None: 

121 property_value = self.values.first() 

122 else: 

123 annotated_values = self.values.annotate( 

124 total_indexes=Count("indexedItems"), 

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

126 ) 

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

128 return property_value 

129 

130 

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

132 """ 

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

134 or the first value if no indexes are specified. 

135 

136 This method should not be used in bulk operations. 

137 """ 

138 value_object = self.get_value_object(indexes=indexes) 

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

140 return None 

141 return value_object.value 

142 

143 

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

145 """ 

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

147 or the first value if no indexes are specified. 

148 

149 This method should not be used in bulk operations. 

150 """ 

151 property_value = self.get_value_object(indexes=indexes) 

152 property_value.value = value 

153 property_value.save() 

154 

155 

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

157 """ 

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

159 """ 

160 property_values = self.values.all() 

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

162 

163 def has_value(self) -> bool: 

164 property_value = self.get_value_object() 

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

166 return False 

167 return property_value.has_value() 

168 

169 def has_value_bulk(self) -> bool: 

170 property_value = get_value_object(self) 

171 if property_value is None: 

172 return False 

173 return property_value.has_value() 

174 

175 def get_cutoff_and_property_values(self, config: ObjectType): 

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

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

178 

179 property_values = [] 

180 for indexed_item in first_indexed_items: 

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

182 

183 count = first_indexed_items.count() 

184 

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

186 return None, None 

187 

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

189 

190 return cutoff, property_values 

191 

192 

193 def isSpecified(self) -> bool: 

194 """ 

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

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

197 Both guesses and set points are needed. 

198 """ 

199 value: PropertyValue 

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

201 if ( 

202 ( 

203 value.enabled 

204 or value.is_control_set_point() 

205 or self.is_recycle_var() 

206 ) 

207 and not value.has_value() 

208 ): 

209 return False 

210 return True 

211 

212 

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

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

215 object_type = self.set.simulationObject.objectType 

216 config = configuration.get(object_type) 

217 

218 list_prop_val = [] 

219 

220 if self.key not in config.properties: 

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

222 else: 

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

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

225 

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

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

228 # all of them except the last one 

229 cutoff, property_values = self.get_cutoff_and_property_values(config) 

230 if property_values: 

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

232 if i <= cutoff: 

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

234 property_values[i].enabled = condition 

235 else: 

236 property_values[i].enabled = False 

237 list_prop_val.append(property_values[i]) 

238 else: 

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

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

241 for value in values: 

242 value.enabled = condition 

243 list_prop_val.append(value) 

244 return list_prop_val 

245 

246 def is_recycle_var(self) -> bool: 

247 return hasattr(self, "recycleConnection") 

248 

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

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

251 

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

253 """ 

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

255 """ 

256 

257 # update both the unit and the value 

258 original_unit = self.unit 

259 self.unit = new_unit 

260 # original_value = self.get_value() 

261 original_value = self.values.all() 

262 for value in list(original_value): 

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

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

265 if not value.is_enabled(): 

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

267 float_value = float(value.value) 

268 new_value = convert_value(float_value, original_unit, new_unit) 

269 value.value = new_value 

270 value.save() 

271 

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

273 #get the type of the unit operation 

274 if self.set.simulationObject == None: 

275 return None 

276 else: 

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

278 

279 def is_custom_property(self) -> bool: 

280 return self.get_schema() is None 

281 

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

283 """ 

284 Check if the property is of type "exceptLast" 

285 """ 

286 object_type = property_info.set.simulationObject.objectType 

287 config = configuration.get(object_type) 

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

289 if schema.type == "exceptLast": 

290 return True 

291 return False 

292