Coverage for backend/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 

15 

16 

17class HistoricalValue(models.Model): 

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

19 value = models.FloatField() 

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

21 

22 created_at = models.DateTimeField(auto_now_add=True) 

23 objects = AccessControlManager() 

24 

25 

26 

27 class Meta: 

28 ordering = ['created_at'] 

29 

30 

31class ProcessPathProperty(models.Model): 

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

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

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

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

36 

37 created_at = models.DateTimeField(auto_now_add=True) 

38 objects = AccessControlManager() 

39 

40 

41 class Meta: 

42 ordering = ['created_at'] 

43 

44 

45class PropertyInfo(models.Model): 

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

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

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

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

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

51 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. 

52 key = models.CharField(max_length=64) 

53 displayName = models.CharField(max_length=64) 

54 index = models.IntegerField(default=0) 

55 created_at = models.DateTimeField(auto_now_add=True) 

56 

57 objects = AccessControlManager() 

58 

59 class Meta: 

60 ordering = ['created_at'] 

61 

62 @classmethod 

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

64 

65 value = fields.pop("value") 

66 property_info = PropertyInfo(**fields) 

67 # Create a property value object with this value 

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

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

70 

71 return property_info, [property_value] 

72 

73 

74 @classmethod 

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

76 # Create and save a new property info object 

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

78 property_info.save() 

79 for property_value in property_values: 

80 property_value.save() 

81 return property_info 

82 

83 

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

85 """ 

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

87 or the first value if no indexes are specified. 

88 

89 This method should be used in bulk operations (where 

90 all the related property values are prefetched). 

91 """ 

92 property_value = get_value_object(self, indexes) 

93 if property_value is None: 

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

95 return property_value.value 

96 

97 

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

99 """ 

100 Set the value of the property at the given 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 property_value.value = value 

108 return property_value 

109 

110 

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

112 """ 

113 Get the property value object at the specified indexes, 

114 or the first value if no indexes are specified. 

115 

116 This method should not be used in bulk operations. 

117 """ 

118 if indexes is None: 

119 property_value = self.values.first() 

120 else: 

121 annotated_values = self.values.annotate( 

122 total_indexes=Count("indexedItems"), 

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

124 ) 

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

126 return property_value 

127 

128 

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

130 """ 

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

132 or the first value if no indexes are specified. 

133 

134 This method should not be used in bulk operations. 

135 """ 

136 value_object = self.get_value_object(indexes=indexes) 

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

138 return None 

139 return value_object.value 

140 

141 

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

143 """ 

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

145 or the first value if no indexes are specified. 

146 

147 This method should not be used in bulk operations. 

148 """ 

149 property_value = self.get_value_object(indexes=indexes) 

150 property_value.value = value 

151 property_value.save() 

152 

153 

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

155 """ 

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

157 """ 

158 property_values = self.values.all() 

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

160 

161 def has_value(self) -> bool: 

162 property_value = self.get_value_object() 

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

164 return False 

165 return property_value.has_value() 

166 

167 def has_value_bulk(self) -> bool: 

168 property_value = get_value_object(self) 

169 if property_value is None: 

170 return False 

171 return property_value.has_value() 

172 

173 def get_cutoff_and_property_values(self, config: ObjectType): 

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

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

176 

177 property_values = [] 

178 for indexed_item in first_indexed_items: 

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

180 

181 count = first_indexed_items.count() 

182 

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

184 return None, None 

185 

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

187 

188 return cutoff, property_values 

189 

190 

191 def isSpecified(self) -> bool: 

192 value: PropertyValue 

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

194 if ( 

195 ( 

196 (value.enabled and not value.is_control_manipulated() ) 

197 or value.is_control_set_point() 

198 or self.is_recycle_var() 

199 ) 

200 and not value.has_value() 

201 ): 

202 return False 

203 return True 

204 

205 

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

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

208 object_type = self.set.simulationObject.objectType 

209 config = configuration.get(object_type) 

210 

211 list_prop_val = [] 

212 

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

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

215 else: 

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

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

218 

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

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

221 # all of them except the last one 

222 cutoff, property_values = self.get_cutoff_and_property_values(config) 

223 if property_values: 

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

225 if i <= cutoff: 

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

227 property_values[i].enabled = condition 

228 else: 

229 property_values[i].enabled = False 

230 list_prop_val.append(property_values[i]) 

231 else: 

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

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

234 for value in values: 

235 value.enabled = condition 

236 list_prop_val.append(value) 

237 return list_prop_val 

238 

239 def is_recycle_var(self) -> bool: 

240 return hasattr(self, "recycleConnection") 

241 

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

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

244 

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

246 """ 

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

248 """ 

249 

250 # update both the unit and the value 

251 original_unit = self.unit 

252 self.unit = new_unit 

253 # original_value = self.get_value() 

254 original_value = self.values.all() 

255 for value in list(original_value): 

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

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

258 if not value.is_enabled(): 

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

260 float_value = float(value.value) 

261 new_value = convert_value(float_value, original_unit, new_unit) 

262 value.value = new_value 

263 value.save() 

264 

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

266 #get the type of the unit operation 

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

268 return None 

269 else: 

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

271 

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

273 """ 

274 Check if the property is of type "exceptLast" 

275 """ 

276 object_type = property_info.set.simulationObject.objectType 

277 config = configuration.get(object_type) 

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

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

280 return True 

281 return False 

282