Coverage for backend/django/flowsheetInternals/unitops/models/simulation_object_factory.py: 94%
273 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1import itertools
2from typing import List, Tuple
4from core.auxiliary.models.MLModel import MLModel
5from core.auxiliary.models.Flowsheet import Flowsheet
6from core.auxiliary.models.IndexedItem import IndexedItem
7from core.auxiliary.models.PropertyValue import PropertyValue, PropertyValueIntermediate
8from core.auxiliary.models.PropertyInfo import PropertyInfo
9from core.auxiliary.models.RecycleData import RecycleData
10from core.auxiliary.models.PropertySet import PropertySet
11from core.auxiliary.enums import ConType, heat_exchange_ops
13from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
14from flowsheetInternals.unitops.models.Port import Port
15from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
16from flowsheetInternals.graphicData.models.groupingModel import Grouping
17from flowsheetInternals.unitops.models.compound_propogation import update_compounds_on_add_stream
19from flowsheetInternals.unitops.config.config_methods import *
20from flowsheetInternals.unitops.config.config_base import configuration
21from common.config_types import *
22from core.auxiliary.models.ObjectTypeCounter import ObjectTypeCounter
23from core.auxiliary.views.ExtractSegmentDataFromFS import create_he_streams
26class SimulationObjectFactory:
27 def __init__(self) -> None:
28 self._flowsheet: Flowsheet | None = None
30 self.simulation_objects: list[SimulationObject] = []
31 self.graphic_objects: list[GraphicObject] = []
32 self.property_sets: list[PropertySet] = []
33 self.property_infos: list[PropertyInfo] = []
34 self.property_values: list[PropertyValue] = []
35 self.ports: list[Port] = []
36 self.index_items: list[IndexedItem] = []
37 self.property_value_indexed_items: List[PropertyValueIntermediate] = []
39 # conditionally created objects
40 self.recycle_data: list[RecycleData] = []
42 self._unitop: SimulationObject = None # the unitop of the object being created
43 self._current_port: Port = None # the current port for the stream being created
45 self._idx_map: dict[int, SimulationObject] = {} # map of id to simulation object
46 self._property_set_map: dict[int, list[PropertySet]] = {} # map of simulation object to property sets for the many-to-many relationship
47 self._index_items_map: dict[int, dict[str, list[IndexedItem]]] = {} # map of simulation object to indexed items for the many-to-many relationship
49 # For storing old property data when replacing the gut
50 self.old_properties: dict[str, any] = {}
52 @classmethod
53 def create_simulation_object(cls, coordinates: dict[str, float] | None = None, createPropertySet: bool = True, flowsheet=None, parentGroup=None, **kwargs) -> SimulationObject:
54 """
55 Creates a new unitop with attached ports, streams etc
56 """
57 factory = SimulationObjectFactory()
59 # Create unitop
60 # if parentGroup is provided, fetch it, else fallback to flowsheet.rootGrouping.
61 if parentGroup:
62 group = Grouping.objects.get(pk=parentGroup)
63 else:
64 if flowsheet: 64 ↛ 67line 64 didn't jump to line 67 because the condition on line 64 was always true
65 group = flowsheet.rootGrouping
66 else:
67 group = None # or raise an error if flowsheet is required
68 factory._flowsheet = flowsheet
69 object_type = kwargs.pop("objectType")
70 if kwargs.get("schema") is not None:
71 object_schema = kwargs.pop("schema")
72 else:
73 object_schema: ObjectType = configuration[object_type]
76 if coordinates is None:
77 coordinates = {"x": 0, "y": 0}
79 unitop = factory.create(
80 object_type=object_type,
81 object_schema=object_schema,
82 coordinates=coordinates,
83 flowsheet=flowsheet,
84 createPropertySet=createPropertySet,
85 parentGroup= group,
86 )
88 factory._unitop = unitop
89 graphic_object = factory.graphic_objects[-1]
91 # count the number of inlet and outlet ports
92 num_inlets = len([port for port in factory.ports if port.direction == ConType.Inlet])
94 num_outlets = len([port for port in factory.ports
95 if ((port.direction == ConType.Outlet) and
96 (factory._unitop.schema.ports[port.key].makeStream==True))])
101 inlet_index = 0
102 outlet_index = 0
104 if kwargs.get("create_attached_streams", True):
105 for port in factory.ports:
106 factory._current_port = port
107 port_schema = factory._unitop.schema.ports[port.key]
108 if port_schema.makeStream is False:
109 continue
110 stream_type = port_schema.streamType
112 stream_schema = configuration[stream_type]
113 # evenly space the ports along the sides of the unitop
114 if port.direction == ConType.Inlet:
115 coordinates = port.default_stream_position(factory._unitop, graphic_object, inlet_index, num_inlets)
116 inlet_index += 1
117 else:
118 coordinates = port.default_stream_position(factory._unitop, graphic_object, outlet_index, num_outlets)
119 outlet_index += 1
121 stream_name = port.default_stream_name(factory._unitop)
122 stream = factory.create(
123 object_type=stream_type,
124 object_schema=stream_schema,
125 coordinates=coordinates,
126 flowsheet=flowsheet,
127 componentName=stream_name,
128 parentGroup=group,
129 )
131 port.stream = stream
133 # save all created objects
134 factory.perform_bulk_create()
136 if unitop.objectType in heat_exchange_ops:
137 create_he_streams(unitop, group)
139 # Propagate streams to other groups for the newly created unitop
140 inlet_streams = []
141 outlet_streams = []
142 for port in unitop.ports.all():
143 if port.stream:
144 if port.direction == ConType.Inlet:
145 inlet_streams.append(port.stream)
146 else:
147 outlet_streams.append(port.stream)
148 from flowsheetInternals.graphicData.logic.make_group import propagate_streams
149 propagate_streams(inlet_streams, ConType.Inlet)
150 propagate_streams(outlet_streams, ConType.Outlet)
152 # create the attached ML model object if it's a machine learning block
153 if unitop.objectType == "machineLearningBlock":
154 MLModel.objects.create(
155 simulationObject=unitop,
156 flowsheet=flowsheet,
157 )
159 return unitop
162 @classmethod
163 def create_stream_at_port(cls, port: Port) -> SimulationObject:
164 """
165 Creates a new stream attached to the specified port.
166 """
167 factory = SimulationObjectFactory()
168 factory._current_port = port
169 factory._unitop = port.unitOp
170 factory._flowsheet = factory._unitop.flowsheet
172 # Figure out what type of stream to create based on port config
173 # E.g energy_stream, stream, etc
174 object_type = factory._unitop.schema.ports[port.key].streamType
175 object_schema: ObjectType = configuration[object_type]
177 # get the coordinates for the stream
178 coordinates = cls.default_stream_position(factory._unitop, port)
180 #get the group of the unitop that the stream is attached to
181 group = factory._unitop.graphicObject.last().group
183 # create the stream
184 stream = factory.create(
185 object_type=object_type,
186 object_schema=object_schema,
187 coordinates=coordinates,
188 flowsheet=factory._unitop.flowsheet,
189 parentGroup=group,
190 )
191 factory.perform_bulk_create()
193 # attach the stream to the port
194 port.stream = stream
195 port.save()
197 #populate the stream with compounds
198 update_compounds_on_add_stream(port, stream)
200 return stream
203 @classmethod
204 def default_stream_position(cls, unitop: SimulationObject, port: Port) -> dict[str, float]:
205 """
206 Returns the default position for a stream attached to the specified port.
207 """
208 graphic_object = unitop.graphicObject.last() # the unit op only has one graphic object so this is okay.
209 attached_ports = unitop.ports.filter(direction=port.direction)
210 port_index = list(attached_ports).index(port)
211 return port.default_stream_position(unitop, graphic_object, port_index, attached_ports.count())
214 def perform_bulk_create(self) -> None:
215 """
216 Saves all created objects to the database.
217 """
218 SimulationObject.objects.bulk_create(self.simulation_objects)
219 GraphicObject.objects.bulk_create(self.graphic_objects)
220 PropertySet.objects.bulk_create(self.property_sets)
221 PropertyInfo.objects.bulk_create(self.property_infos)
222 PropertyValue.objects.bulk_create(self.property_values)
223 Port.objects.bulk_create(self.ports)
224 RecycleData.objects.bulk_create(self.recycle_data)
225 IndexedItem.objects.bulk_create(self.index_items)
226 PropertyValueIntermediate.objects.bulk_create(self.property_value_indexed_items)
228 def create(
229 self, object_type: str, object_schema: ObjectType, coordinates: dict[str, float], flowsheet: Flowsheet | None = None, parentGroup: Grouping | None = None, componentName: str | None = None, createPropertySet: bool = True
230 ) -> SimulationObject:
231 """
232 Creates a new simulation object
233 does not save the instance to the database.
235 @param object_type: The type of object to create
236 @param object_schema: The schema of the object
237 @param coordinates: The coordinates of the object
238 @param flowsheet: The flowsheet owner of the object
240 @return: The created simulation object, graphic object, property sets, property infos, property packages, and ports.
241 this is so that the caller can create multiple objects at once and do a bulk_create
242 """
243 if flowsheet is not None: 243 ↛ 246line 243 didn't jump to line 246 because the condition on line 243 was always true
244 idx_for_type = ObjectTypeCounter.next_for(flowsheet, object_type)
245 else:
246 last_sim_obj = SimulationObject.objects.last()
247 if last_sim_obj is None:
248 id = 1
249 else:
250 last_sim_obj = SimulationObject.objects.last()
251 idx_for_type = (last_sim_obj.id + 1) if last_sim_obj else 1
253 if componentName is None:
254 componentName = f"{object_schema.displayType}{idx_for_type}"
256 elif "stream" not in object_type and componentName != object_schema.displayName:
257 componentName = componentName
258 else:
259 componentName = f"{componentName}{idx_for_type}"
262 # create simulation object
263 fields = {
264 "objectType": object_type,
265 "componentName": componentName,
266 }
267 instance = SimulationObject(**fields)
268 self.simulation_objects.append(instance)
269 idx = len(self._idx_map)
270 self._idx_map[idx] = instance
271 self._property_set_map[idx] = []
273 # create index items
274 self._index_items_map[idx] = {}
275 for index_set in object_schema.indexSets:
276 self.create_indexed_items(instance, index_set, idx)
278 graphic_object_schema = object_schema.graphicObject
280 graphicObject = GraphicObject(
281 simulationObject=instance,
282 x=coordinates["x"] - graphic_object_schema.width / 2, # center the object horizontally
283 y=coordinates["y"] - graphic_object_schema.height / 2, # center the object vertically
284 width = graphic_object_schema.width,
285 height = graphic_object_schema.height,
286 visible=True,
287 group=parentGroup,
288 flowsheet=flowsheet
289 )
290 self.graphic_objects.append(graphicObject)
292 # If flowsheetKey is given, set flowsheet
293 if flowsheet is not None: 293 ↛ 298line 293 didn't jump to line 298 because the condition on line 293 was always true
294 instance.flowsheet = flowsheet
297 # Create property sets
298 if createPropertySet: 298 ↛ 302line 298 didn't jump to line 302 because the condition on line 298 was always true
299 self.create_property_set(object_schema, idx)
301 # Create ports
302 ports_schema = object_schema.ports or {}
303 # replace any many=True ports with multiple ports
304 for key, port_dict in ports_schema.items():
305 if port_dict.many:
306 # replace this key value pair with a list of ports up to the default provided
307 for i in range(port_dict.default):
308 self.ports.append(Port(
309 key=key,
310 index=i,
311 direction=port_dict.type,
312 displayName=port_dict.displayName + f" {i+1}",
313 unitOp=instance,
314 flowsheet=flowsheet
315 ))
316 else:
317 self.ports.append(Port(
318 key=key,
319 direction=port_dict.type,
320 displayName=port_dict.displayName,
321 unitOp=instance,
322 flowsheet=flowsheet
323 ))
325 # conditionally create attached objects based on object type
326 if object_type == "recycle":
327 self.recycle_data.append(RecycleData(
328 simulationObject=instance,
329 flowsheet=flowsheet
330 ))
332 return instance
334 def store_old_properties(self, instance: SimulationObject):
335 """
336 Stores the old properties of the stream.
337 Used when switching stream types to keep properties that exist in both schema.
338 Prevents losing user input in those properties.
339 e.g. If Molar Flow has a value, switching to Humid Air won't remove that value.
340 """
341 self.old_properties = {}
342 if instance.properties: 342 ↛ exitline 342 didn't return from function 'store_old_properties' because the condition on line 342 was always true
343 for prop_info in instance.properties.containedProperties.all():
344 try:
345 self.old_properties[prop_info.key] = prop_info.get_value_bulk()
346 except ValueError:
347 self.old_properties[prop_info.key] = None
349 def replace_the_gut(self, instance: SimulationObject):
350 self.simulation_objects.append(instance)
351 self._unitop = instance
352 self._current_port = instance.connectedPorts.first()
354 object_schema = configuration[instance.objectType]
355 idx = len(self._idx_map)
356 self._idx_map[idx] = instance
357 self._property_set_map[idx] = []
359 # create index items
360 self._index_items_map[idx] = {}
361 for index_set in object_schema.indexSets:
362 self.create_indexed_items(instance, index_set, idx)
364 self.create_property_set(object_schema, idx)
367 def create_property_set(self, object_schema: ObjectType, idx: int) -> PropertySet:
368 """
369 Creates a new SimulationObjectPropertySet instance with the specified schema.
370 """
371 defaults = {
372 'compoundMode': "",
373 "simulationObject": self._idx_map[idx],
374 }
375 for schema in object_schema.propertySetGroups.values():
376 if schema.type == "composition":
377 defaults['compoundMode'] = "MassFraction"
379 # Create SimulationObjectPropertySet
380 instance = PropertySet(**defaults, flowsheet=self._flowsheet)
382 if object_schema.properties != {}: 382 ↛ 393line 382 didn't jump to line 393 because the condition on line 382 was always true
383 # Create PropertyInfo objects
384 property_pairs = self.create_property_infos(instance, object_schema.properties, idx)
386 # set access based on the schema
387 self.set_properties_access(
388 config=object_schema,
389 properties=property_pairs,
390 idx=idx,
391 )
393 self.property_sets.append(instance)
394 self._property_set_map[idx].append(instance)
396 return instance
399 def create_property_infos(self, property_set: PropertySet, schema: PropertiesType, idx) -> List[Tuple[PropertyInfo, List[PropertyValue]]]:
400 """
401 Creates PropertyInfo objects based on the specified schema.
402 """
403 res: List[Tuple[PropertyInfo, List[PropertyValue]]] = []
405 # if schema.type == "composition":
406 # # the composition property set has no property infos, since these are compounds selected by the user
407 # # TODO: Initialise the properties based on the compounds upstream (if any)
408 # flowsheet = self._flowsheet
409 for key, prop in schema.items():
410 fields = get_property_fields(key, prop, property_set)
412 if self.old_properties:
413 # If property in the schema is also in the old properties, keep the value
414 if key in self.old_properties.keys():
415 fields["value"] = self.old_properties[key]
417 new_property_info, new_property_values = self.create_property_info(idx, prop.indexSets, **fields)
418 res.append((new_property_info, new_property_values))
420 return res
423 def create_property_info(self, idx, index_sets=None, **fields):
424 value = fields.pop("value")
425 property_info = PropertyInfo(**fields, flowsheet=self._flowsheet)
426 self.property_infos.append(property_info)
427 # Create a property value object with this value
428 if index_sets == None:
429 property_value = PropertyValue(value=value, property=property_info, flowsheet=self._flowsheet)
430 self.property_values.append(property_value)
431 return property_info, [property_value]
432 else:
433 combinations = self.get_combinations(self._index_items_map[idx], index_sets)
434 property_values = []
435 for indexes in combinations:
436 property_value = PropertyValue(value=value, property=property_info, flowsheet=self._flowsheet)
437 for indexed_item in indexes:
438 self.property_value_indexed_items.append(
439 PropertyValueIntermediate(propertyvalue=property_value, indexeditem=indexed_item)
440 )
441 property_values.append(property_value)
442 self.property_values.extend(property_values)
443 return property_info, property_values
446 def get_combinations(self, index_items_map: dict[str, list[IndexedItem]], index_sets=[]) -> list[list[IndexedItem]]:
447 """
448 Returns all possible combinations of indexed items for the specified index sets
449 (taking one item from each index set).
450 """
451 if index_items_map == {}: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 return []
453 indexes = [value for key, value in index_items_map.items() if key in index_sets]
454 return list(itertools.product(*indexes))
457 def set_properties_access(
458 self,
459 config: ObjectType,
460 properties: List[Tuple[PropertyInfo, List[PropertyValue]]],
461 idx: int,
462 ) -> None:
463 if (
464 self.simulation_objects[-1].is_stream()
465 and self._unitop.objectType != "recycle"
466 and self._current_port and self._current_port.direction == ConType.Outlet
467 ):
468 # disable outlet/intermediate stream properties
469 for propertyInfo, propVals in properties:
470 for property_value in propVals:
471 property_value.enabled = False
472 else:
473 index_map_items = self._index_items_map[idx]
474 # TODO: Refactor this.
475 # not disabled if it is a state variable
476 for prop, propVals in properties:
477 config_prop = config.properties.get(prop.key)
478 # # figure out the number of items in the first index set
479 if config_prop.indexSets:
480 first_index_type = config_prop.indexSets[0]
481 if first_index_type in index_map_items: 481 ↛ 488line 481 didn't jump to line 488 because the condition on line 481 was always true
482 if len(index_map_items[first_index_type]) == 0:
483 last_item = None
484 else:
485 last_item = index_map_items[first_index_type][-1]
486 # last_item is only needed on index sets.
487 # it should be fine if it's undefined otherwise.
488 for index, property_value in enumerate(propVals):
489 if config_prop: 489 ↛ 492line 489 didn't jump to line 492 because the condition on line 489 was always true
490 group = config_prop.propertySetGroup
491 else:
492 group = "default"
494 config_group = config.propertySetGroups.get(group, None)
495 if config_group is None: 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 property_value.enabled = True
497 elif config_group.type == "exceptLast" or config_prop.sumToOne:
498 # TODO: Deprecate exceptLast in favor of sumToOne. See phase_seperator_config
499 # Instead of using len_last_item, just use the last item in the index set
500 indexed_item_links = [item for item in self.property_value_indexed_items
501 if item.propertyvalue == property_value]
502 indexed_items = [item.indexeditem for item in indexed_item_links]
503 is_last_item = last_item in indexed_items
505 if is_last_item:
506 property_value.enabled = False
507 else:
508 property_value.enabled = True
509 elif config_group.type == "stateVars": 509 ↛ 513line 509 didn't jump to line 513 because the condition on line 509 was always true
510 state_vars = getattr(config_group, "stateVars") or ()
511 property_value.enabled = prop.key in state_vars
512 else: # eg. All
513 property_value.enabled = True
516 def create_indexed_items(self, instance: SimulationObject, index_set: str, idx: int) -> None:
517 """
518 Creates IndexedItem instances for the specified index set.
519 """
520 items = []
522 outlet_name = instance.schema.splitter_fraction_name
523 match index_set:
524 case "splitter_fraction":
525 items = []
526 # create indexed items for the splitter_fraction index set
527 items.append(IndexedItem(
528 owner=instance,
529 key="outlet_1",
530 displayName= outlet_name + " 1",
531 type=index_set,
532 flowsheet=self._flowsheet
533 ))
534 items.append(IndexedItem(
535 owner=instance,
536 key="outlet_2",
537 displayName= outlet_name + " 2",
538 type=index_set,
539 flowsheet=self._flowsheet
540 ))
542 case "phase":
543 # create indexed items for the phase index set
544 items.append(IndexedItem(
545 owner=instance,
546 key="Liq",
547 displayName="Liquid",
548 type=index_set,
549 flowsheet=self._flowsheet
550 ))
551 items.append(IndexedItem(
552 owner=instance,
553 key="Vap",
554 displayName="Vapor",
555 type=index_set,
556 flowsheet=self._flowsheet
557 ))
559 case "compound":
560 pass # have to wait for user to select compounds
561 self._index_items_map[idx][index_set] = items
562 self.index_items.extend(items)