Coverage for backend/django/core/auxiliary/models/PropertyValue.py: 93%
89 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.db import models
2from core.auxiliary.models.IndexedItem import IndexedItem
3from core.managers import AccessControlManager, AllFlowsheetStatesManager
4from core.auxiliary.models.ControlValue import ControlValue
5from core.auxiliary.models.FlowsheetHistoryModel import FlowsheetHistoryModel
6from typing import TYPE_CHECKING
8if TYPE_CHECKING:
9 from core.auxiliary.models.PropertyInfo import PropertyInfo
10 from core.auxiliary.models.PropertySet import PropertySet
11 from core.auxiliary.models.ControlValue import ControlValue
14class PropertyValue(FlowsheetHistoryModel, models.Model):
15 flowsheet_state = models.ForeignKey(
16 "FlowsheetState", on_delete=models.CASCADE, related_name="propertyValues"
17 )
18 value = models.JSONField(null=True, blank=True)
19 displayValue = models.JSONField(null=True, blank=True)
20 enabled = models.BooleanField(default=True)
21 formula = models.CharField(max_length=2048, null=True, blank=True, default=None)
22 property: models.ForeignKey["PropertyInfo"] = models.ForeignKey(
23 "PropertyInfo", on_delete=models.CASCADE, related_name="values", null=True
24 )
25 indexedItems = models.ManyToManyField(IndexedItem, related_name="propertyValues")
26 created_at = models.DateTimeField(auto_now_add=True)
27 tag = models.CharField(
28 max_length=255, null=True, blank=True, default=None
29 ) # tag can be used to sync the digital model with the rest of the sytem.
30 objects = AccessControlManager()
31 all_states = AllFlowsheetStatesManager()
32 controlManipulated: ControlValue | None
33 controlSetPoint: ControlValue | None
34 # Bulk update doesn't work with access control manager so we
35 # sometimes need to just use the normal manager.
36 _unsafe_objects = models.Manager()
38 class Meta:
39 ordering = ["created_at"]
41 def enable(self, state=True) -> None:
42 self.enabled = state
43 self.save()
45 def get_index(self, index_set: str) -> list[IndexedItem]:
46 """
47 Get the indexed item of the specified type for this property.
48 """
49 return self.indexedItems.get(type=index_set)
51 def has_value(self) -> bool:
52 return self.value not in [None, ""]
54 def get_indexed_items(self) -> list[IndexedItem]:
55 """
56 Gets the list of indexed items for this property value, in order
57 Note that this is kinda expensive if you're doing it for everything,
58 as it looks up the unit op type every time.
59 But it's helpful for testing.
60 """
61 if self.property.is_custom_property():
62 return [] # no special indexes on custom properties as they don't exist in the schema
63 indexes = list(self.indexedItems.all())
64 property_key = self.property.key
65 index_set_order = self.property.set.simulationObject.schema.properties[
66 property_key
67 ].indexSets
68 return sort_indexes(index_set_order, indexes)
70 def get_indexes(self) -> list[str]:
71 """
72 Gets the list of indexes for this property value, in order. e.g "outlet_1
73 , Note that this is kinda expensive if you're doing it for everything,
74 as it looks up the unit op type every time.
75 But it's helpful for testing.
76 """
77 return [index.key for index in self.get_indexed_items()]
79 def get_simulation_object(self):
80 return self.property.set.simulationObject
82 def get_index_names(self) -> list[str]:
83 return [index.displayName for index in self.get_indexed_items()]
85 def is_control_set_point(self) -> bool:
86 return hasattr(self, "controlSetPoint")
88 def is_control_manipulated(self) -> bool:
89 return hasattr(self, "controlManipulated")
91 def is_externally_controlled(self) -> bool:
92 """
93 Returns true if the property is controlled or controlling an external property
94 """
95 if not self.is_control_manipulated() and not self.is_control_set_point():
96 return False
97 if self.is_control_manipulated(): 97 ↛ 99line 97 didn't jump to line 99 because the condition on line 97 was never true
98 # set point is from a different property set
99 return self.controlManipulated.setPoint.property.set != self.property.set
100 if self.is_control_set_point(): 100 ↛ 103line 100 didn't jump to line 103 because the condition on line 100 was always true
101 # manipulated property is from a different property set
102 return self.controlSetPoint.manipulated.property.set != self.property.set
103 return False
105 def is_enabled(self) -> bool:
106 """
107 Returns true if the property is to be enabled in simulation: either a statevar or a controlled property
108 """
109 return (
110 self.enabled and not self.is_control_manipulated()
111 ) or self.is_control_set_point()
113 def add_control(self, prop: "PropertyValue") -> ControlValue:
114 # This property is the set point controlling the manipulated property
115 return ControlValue.create(
116 setPoint=self,
117 manipulated=prop,
118 flowsheet_state=prop.flowsheet_state,
119 )
121 def auto_replace(self):
122 """
123 Automatically create control relationships as autocomplete option:
124 e.g.
125 - Mass Flow -> Molar Flow
126 - Vapour Fraction -> Temperature
128 Logic:
129 - Non-enabled property (self) becomes controlSetPoint
130 - Enabled property becomes manipulated
131 - No properties are enabled/disabled here
132 """
133 property_info: PropertyInfo = self.property
134 property_set: PropertySet = property_info.set
135 simulation_object = property_set.simulationObject
137 control_mapping = {}
139 if simulation_object.objectType == "stream": 139 ↛ 158line 139 didn't jump to line 158 because the condition on line 139 was always true
140 control_mapping.update(
141 {
142 "flow_mass": "flow_mol",
143 "vapor_frac": "temperature",
144 "enth_mol": "temperature",
145 "enth_mass": "temperature",
146 "entr_mol": "pressure",
147 "entr_mass": "pressure",
148 # "total_energy_flow": "",
149 # "flow_vol": "",
150 # "mole_frac_phase_comp": "",
151 # "efficiency_isentropic": "", // are unitops still streams??
152 # "deltaP": "work_mechanical",
153 # "ratioP": "",
154 # "work_mechanical": "",
155 }
156 )
158 setpoint_key = property_info.key
159 manipulated_key = control_mapping.get(setpoint_key)
160 if not manipulated_key:
161 return
163 # find enabled property to manipulate
164 manipulated_pvs = [
165 pv
166 for prop_info in property_set.containedProperties.all()
167 if prop_info.key == manipulated_key
168 for pv in prop_info.values.all()
169 if pv.enabled # only check the actual enabled flag
170 ]
172 if not manipulated_pvs: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 return
175 manipulated_pv = manipulated_pvs[0]
177 if ControlValue.objects.filter(manipulated=manipulated_pv).exists():
178 return
180 # create the control relationship
181 ControlValue.create(
182 manipulated=manipulated_pv,
183 setPoint=self,
184 flowsheet_state=self.flowsheet_state,
185 )
188PropertyValueIntermediate = PropertyValue.indexedItems.through
191def sort_indexes(
192 index_set_order: list[str], indexes: list[IndexedItem]
193) -> list[IndexedItem]:
194 """
195 Sorts the list of indexed items based on the order defined in the config file.
196 index_set_order: list[str], a list of index set types, the order in which the indexes should be sorted. e.g ["splitter_fraction","compound","phase"]
197 indexes: list[IndexedItem], a list of indexed items to be sorted. e.g a list of indexed items attached to a PropertyValue
198 """
199 item = [index.type for index in indexes]
200 reordered_indexes = [None] * len(indexes)
202 for i in range(len(item)):
203 if item[i] != index_set_order[i]:
204 # Find the correct position for the current item in object_properties
205 correct_position = index_set_order.index(item[i])
206 # Place the current index in the correct position
207 reordered_indexes[correct_position] = indexes[i]
208 else:
209 # If the item is already in the correct position, keep it as is
210 reordered_indexes[i] = indexes[i]
212 # Update the original indexes list with the reordered indexes
213 for i in range(len(indexes)):
214 indexes[i] = reordered_indexes[i]
215 return indexes