Coverage for backend/django/core/auxiliary/models/PropertyInfo.py: 96%
173 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.db import models
2from django.db.models import Q, Count
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
13from core.managers import AccessControlManager, AllFlowsheetStatesManager
14from core.auxiliary.models.PropertyValue import PropertyValue
15from core.auxiliary.models.FlowsheetHistoryModel import FlowsheetHistoryModel
16from typing import TYPE_CHECKING
18if TYPE_CHECKING:
19 from core.auxiliary.models.PropertyValue import PropertyValue
20 from core.auxiliary.models.RecycleData import RecycleProperty
21 from core.auxiliary.models.PropertySet import PropertySet
24class HistoricalValue(models.Model):
25 flowsheet_state = models.ForeignKey(
26 "FlowsheetState", on_delete=models.CASCADE, related_name="HistoricalValues"
27 )
28 value = models.FloatField()
29 property = models.ForeignKey(
30 "PropertyInfo", on_delete=models.CASCADE, related_name="history", null=True
31 )
33 created_at = models.DateTimeField(auto_now_add=True)
34 objects = AccessControlManager()
35 all_states = AllFlowsheetStatesManager()
37 class Meta:
38 ordering = ["created_at"]
41class ProcessPathProperty(models.Model):
42 flowsheet_state = models.ForeignKey(
43 "FlowsheetState", on_delete=models.CASCADE, related_name="ProcessPathProperties"
44 )
45 value = models.FloatField(null=True, blank=True)
46 property = models.ForeignKey(
47 "PropertyInfo",
48 on_delete=models.CASCADE,
49 related_name="ProcessPathProperties",
50 null=True,
51 )
52 path = models.ForeignKey(
53 "ProcessPath",
54 on_delete=models.CASCADE,
55 related_name="ProcessPathProperties",
56 null=True,
57 )
59 created_at = models.DateTimeField(auto_now_add=True)
60 objects = AccessControlManager()
61 all_states = AllFlowsheetStatesManager()
63 class Meta:
64 ordering = ["created_at"]
67class PropertyInfo(FlowsheetHistoryModel, models.Model):
68 flowsheet_state = models.ForeignKey(
69 "FlowsheetState", on_delete=models.CASCADE, related_name="propertyInfos"
70 )
71 set = models.ForeignKey(
72 "PropertySet",
73 on_delete=models.CASCADE,
74 related_name="ContainedProperties",
75 null=True,
76 )
77 type = models.CharField(choices=DisplayType.choices, default=DisplayType.numeric)
78 unitType = models.CharField(
79 choices=UnitOfMeasure.choices, default=UnitOfMeasure.none
80 )
81 unit = models.CharField(max_length=32, blank=True) # selected unit
82 key = models.CharField(max_length=64)
83 displayName = models.CharField(max_length=64)
84 index = models.IntegerField(default=0)
85 managed = models.BooleanField(default=False)
86 managed_source = models.CharField(max_length=64, blank=True)
87 can_edit = models.BooleanField(default=True)
88 can_edit_formula = models.BooleanField(default=True)
89 can_delete = models.BooleanField(default=True)
90 formula_incomplete = models.BooleanField(default=False)
91 formula_incomplete_reason = models.TextField(blank=True)
92 created_at = models.DateTimeField(auto_now_add=True)
93 values: models.QuerySet["PropertyValue"]
94 recycleConnection: "RecycleProperty"
95 set: "PropertySet"
97 objects = AccessControlManager()
98 all_states = AllFlowsheetStatesManager()
100 class Meta:
101 ordering = ["created_at"]
103 @classmethod
104 def create(
105 cls, indexes: dict[str, list[IndexedItem]] = {}, **fields
106 ) -> tuple["PropertyInfo", list[PropertyValue]]:
108 value = fields.pop("value")
109 property_info = PropertyInfo(**fields)
110 # Create a property value object with this value
111 if indexes == {}: 111 ↛ 118line 111 didn't jump to line 118 because the condition on line 111 was always true
112 property_value = PropertyValue(
113 value=value,
114 property=property_info,
115 flowsheet_state=fields.get("flowsheet_state"),
116 )
118 return property_info, [property_value]
120 @classmethod
121 def create_save(cls, **fields) -> "PropertyInfo":
122 # Create and save a new property info object
123 property_info, property_values = cls.create(**fields)
124 property_info.save()
125 for property_value in property_values:
126 property_value.save()
127 return property_info
129 def get_value_bulk(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.
134 This method should be used in bulk operations (where
135 all the related property values are prefetched).
136 """
137 property_value = get_value_object(self, indexes)
138 if property_value is None:
139 raise ValueError(f"No property values found with indexes={indexes}")
140 return property_value.value
142 def set_value_bulk(self, value: Any, indexes: list | None = None) -> PropertyValue:
143 """
144 Set the value of the property at the given indexes,
145 or the first value if no indexes are specified.
147 This method should be used in bulk operations (where
148 all the related property values are prefetched).
149 """
150 property_value = get_value_object(self, indexes)
151 property_value.value = value
152 return property_value
154 def get_value_object(self, indexes: list[str] | None = None) -> PropertyValue:
155 """
156 Get the property value object at the specified indexes,
157 or the first value if no indexes are specified.
159 This method should not be used in bulk operations.
160 """
161 if indexes is None:
162 property_value = self.values.first()
163 else:
164 annotated_values = self.values.annotate(
165 total_indexes=Count("indexedItems"),
166 matching_indexes=Count(
167 "indexedItems", filter=Q(indexedItems__key__in=indexes)
168 ),
169 )
170 property_value = annotated_values.get(
171 total_indexes=len(indexes), matching_indexes=len(indexes)
172 )
173 return property_value
175 def get_value(self, indexes: list | None = None) -> Any:
176 """
177 Get the value of the property at the specified indexes,
178 or the first value if no indexes are specified.
180 This method should not be used in bulk operations.
181 """
182 value_object = self.get_value_object(indexes=indexes)
183 if value_object is None:
184 return None
185 return value_object.value
187 def set_value(self, value: Any, indexes=None) -> None:
188 """
189 Set the value of the property at the given indexes,
190 or the first value if no indexes are specified.
192 This method should not be used in bulk operations.
193 """
194 property_value = self.get_value_object(indexes=indexes)
195 property_value.value = value
196 property_value.save()
198 def get_indexes(self, index_set: str) -> list[IndexedItem]:
199 """
200 Get all the indexed items of the specified type for this property.
201 """
202 property_values = self.values.all()
203 return IndexedItem.objects.filter(
204 propertyValues__in=property_values, type=index_set
205 )
207 def has_value(self) -> bool:
208 property_value = self.get_value_object()
209 if property_value is None:
210 return False
211 return property_value.has_value()
213 def has_value_bulk(self) -> bool:
214 property_value = get_value_object(self)
215 if property_value is None:
216 return False
217 return property_value.has_value()
219 def get_cutoff_and_property_values(self, config: ObjectType):
220 first_index = config.indexSets[
221 0
222 ] # get the outermost level e.g: "outlet 1, outlet 2"
223 first_indexed_items = IndexedItem.objects.filter(
224 owner=self.set.simulationObject, type=first_index
225 )
227 property_values = []
228 for indexed_item in first_indexed_items:
229 property_values.extend(indexed_item.propertyValues.all())
231 count = first_indexed_items.count()
233 if not count: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 return None, None
236 cutoff = (
237 len(property_values) - 1 - (len(property_values) / count)
238 ) # get the index where it starts to be disabled
240 return cutoff, property_values
242 def isSpecified(self) -> bool:
243 """
244 Checks if all the required property values have been specified.
245 This includes values that are enabled, control set points, and recycle guesses.
246 Both guesses and set points are needed.
247 """
248 value: PropertyValue
249 for value in self.values.all():
250 if (
251 value.enabled or value.is_control_set_point() or self.is_recycle_var()
252 ) and not value.has_value():
253 return False
254 return True
256 def enable(self, condition: bool = True) -> list[PropertyValue]:
257 """Enables or disables property values based on configuration rules."""
258 object_type = self.set.simulationObject.objectType
259 config = configuration.get(object_type)
261 list_prop_val = []
263 if self.key not in config.properties: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 propertySetGroup_key = (
265 None # default to none, we are assuming the property doesn't exist.
266 )
267 else:
268 propertySetGroup_key = config.properties[self.key].propertySetGroup
269 propertySetGroup_config = config.propertySetGroups.get(
270 propertySetGroup_key, None
271 )
273 if (
274 propertySetGroup_config is not None
275 and propertySetGroup_config.type == "exceptLast"
276 and condition == True
277 ):
278 # in ExceptLast properties, you don't enable all of them, you enable
279 # all of them except the last one
280 cutoff, property_values = self.get_cutoff_and_property_values(config)
281 if property_values:
282 for i in range(len(property_values)):
283 if i <= cutoff:
284 # if the property is before the cutoff, it should be enabled
285 property_values[i].enabled = condition
286 else:
287 property_values[i].enabled = False
288 list_prop_val.append(property_values[i])
289 else:
290 # this must be a custom property or a machine learning property.
291 values = list(self.values.all())
292 for value in values:
293 value.enabled = condition
294 list_prop_val.append(value)
295 return list_prop_val
297 def is_recycle_var(self) -> bool:
298 return hasattr(self, "recycleConnection")
300 def add_control(self, prop: "PropertyValue") -> ControlValue:
301 return ControlValue.create(
302 setPoint=self.values.first(),
303 manipulated=prop.values.first(),
304 flowsheet_state=self.flowsheet_state,
305 )
307 def unit_conversion(self, new_unit: str) -> None:
308 """
309 Perform a unit conversion on the value field of the PropertyInfo instance
310 """
312 # update both the unit and the value
313 original_unit = self.unit
314 self.unit = new_unit
315 # original_value = self.get_value()
316 original_value = self.values.all()
317 for value in list(original_value):
318 # we only want to update the units of properties that can't be manually edited.
319 # this is a design choice, so users don't enter 100, switch from Pa to kPa, and it gets converted to 0.1 kPa
320 if not value.is_enabled():
321 if value.value != None:
322 float_value = float(value.value)
323 new_value = convert_value(float_value, original_unit, new_unit)
324 value.value = new_value
325 value.save()
327 def get_schema(self) -> PropertyType | None:
328 # get the type of the unit operation
329 if self.set.simulationObject == None:
330 return None
331 else:
332 return self.set.simulationObject.schema.properties.get(self.key, None)
334 def is_custom_property(self) -> bool:
335 return self.get_schema() is None
338def check_is_except_last(property_info: PropertyInfo) -> bool:
339 """
340 Check if the property is of type "exceptLast"
341 """
342 object_type = property_info.set.simulationObject.objectType
343 config = configuration.get(object_type)
344 for schema in config.propertySetGroups.values():
345 if schema.type == "exceptLast":
346 return True
347 return False