Coverage for backend/flowsheetInternals/unitops/models/simulation_object_factory.py: 90%
267 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-11-06 23:27 +0000
1import itertools
2from typing import List, Tuple
4from core.auxiliary.models.Flowsheet import Flowsheet
5from core.auxiliary.models.IndexedItem import IndexedItem
6from core.auxiliary.models.PropertyValue import PropertyValue, PropertyValueIntermediate
7from core.auxiliary.models.PropertyInfo import PropertyInfo
8from core.auxiliary.models.RecycleData import RecycleData
9from core.auxiliary.models.PropertySet import PropertySet
10from core.auxiliary.enums import ConType, heat_exchange_ops
12from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
13from flowsheetInternals.unitops.models.Port import Port
14from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
15from flowsheetInternals.graphicData.models.groupingModel import Grouping
16from flowsheetInternals.propertyPackages.models.SimulationObjectPropertyPackages import SimulationObjectPropertyPackages
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.property_packages: list[SimulationObjectPropertyPackages] = []
36 self.ports: list[Port] = []
37 self.index_items: list[IndexedItem] = []
38 self.property_value_indexed_items: List[PropertyValueIntermediate] = []
40 # conditionally created objects
41 self.recycle_data: list[RecycleData] = []
43 self._unitop: SimulationObject = None # the unitop of the object being created
44 self._current_port: Port = None # the current port for the stream being created
46 self._idx_map: dict[int, SimulationObject] = {} # map of id to simulation object
47 self._property_set_map: dict[int, list[PropertySet]] = {} # map of simulation object to property sets for the many-to-many relationship
48 self._index_items_map: dict[int, dict[str, list[IndexedItem]]] = {} # map of simulation object to indexed items for the many-to-many relationship
51 @classmethod
52 def create_simulation_object(cls, coordinates: dict[str, float] | None = None, createPropertySet: bool = True, flowsheet=None, parentGroup=None, **kwargs) -> SimulationObject:
53 """
54 Creates a new unitop with attached ports, streams etc
55 """
56 factory = SimulationObjectFactory()
58 # Create unitop
59 # if parentGroup is provided, fetch it, else fallback to flowsheet.rootGrouping.
60 if parentGroup:
61 group = Grouping.objects.get(pk=parentGroup)
62 else:
63 if flowsheet: 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true
64 group = flowsheet.rootGrouping
65 else:
66 group = None # or raise an error if flowsheet is required
67 factory._flowsheet = flowsheet
68 object_type = kwargs.pop("objectType")
69 if kwargs.get("schema") is not None:
70 object_schema = kwargs.pop("schema")
71 else:
72 object_schema: ObjectType = configuration[object_type]
75 if coordinates is None:
76 coordinates = {"x": 0, "y": 0}
78 unitop = factory.create(
79 object_type=object_type,
80 object_schema=object_schema,
81 coordinates=coordinates,
82 flowsheet=flowsheet,
83 createPropertySet=createPropertySet,
84 parentGroup= group,
85 )
87 factory._unitop = unitop
88 graphic_object = factory.graphic_objects[-1]
90 # count the number of inlet and outlet ports
91 num_inlets = len([port for port in factory.ports if port.direction == ConType.Inlet])
93 num_outlets = len([port for port in factory.ports
94 if ((port.direction == ConType.Outlet) and
95 (factory._unitop.schema.ports[port.key].makeStream==True))])
100 inlet_index = 0
101 outlet_index = 0
103 if kwargs.get("create_attached_streams", True):
104 for port in factory.ports:
105 factory._current_port = port
106 port_schema = factory._unitop.schema.ports[port.key]
107 if port_schema.makeStream is False:
108 continue
109 stream_type = port_schema.streamType
111 stream_schema = configuration[stream_type]
112 # evenly space the ports along the sides of the unitop
113 if port.direction == ConType.Inlet:
114 coordinates = port.default_stream_position(factory._unitop, graphic_object, inlet_index, num_inlets)
115 inlet_index += 1
116 else:
117 coordinates = port.default_stream_position(factory._unitop, graphic_object, outlet_index, num_outlets)
118 outlet_index += 1
120 stream_name = port.default_stream_name(factory._unitop)
121 stream = factory.create(
122 object_type=stream_type,
123 object_schema=stream_schema,
124 coordinates=coordinates,
125 flowsheet=flowsheet,
126 componentName=stream_name,
127 parentGroup=group,
128 )
130 port.stream = stream
132 # save all created objects
133 factory.perform_bulk_create()
135 if unitop.objectType in heat_exchange_ops:
136 create_he_streams(unitop, group)
138 # Propagate streams to other groups for the newly created unitop
139 inlet_streams = []
140 outlet_streams = []
141 for port in unitop.ports.all():
142 if port.stream:
143 if port.direction == ConType.Inlet:
144 inlet_streams.append(port.stream)
145 else:
146 outlet_streams.append(port.stream)
147 from flowsheetInternals.graphicData.logic.make_group import propagate_streams
148 propagate_streams(inlet_streams, ConType.Inlet)
149 propagate_streams(outlet_streams, ConType.Outlet)
151 return unitop
154 @classmethod
155 def create_stream_at_port(cls, port: Port) -> SimulationObject:
156 """
157 Creates a new stream attached to the specified port.
158 """
159 factory = SimulationObjectFactory()
160 factory._current_port = port
161 factory._unitop = port.unitOp
162 factory._flowsheet = factory._unitop.flowsheet
164 # Figure out what type of stream to create based on port config
165 # E.g energy_stream, stream, etc
166 object_type = factory._unitop.schema.ports[port.key].streamType
167 object_schema: ObjectType = configuration[object_type]
169 # get the coordinates for the stream
170 coordinates = cls.default_stream_position(factory._unitop, port)
172 #get the group of the unitop that the stream is attached to
173 group = factory._unitop.graphicObject.last().group
175 # create the stream
176 stream = factory.create(
177 object_type=object_type,
178 object_schema=object_schema,
179 coordinates=coordinates,
180 flowsheet=factory._unitop.flowsheet,
181 parentGroup=group,
182 )
183 factory.perform_bulk_create()
185 # attach the stream to the port
186 port.stream = stream
187 port.save()
189 #populate the stream with compounds
190 update_compounds_on_add_stream(port, stream)
192 return stream
195 @classmethod
196 def default_stream_position(cls, unitop: SimulationObject, port: Port) -> dict[str, float]:
197 """
198 Returns the default position for a stream attached to the specified port.
199 """
200 graphic_object = unitop.graphicObject.last() # the unit op only has one graphic object so this is okay.
201 attached_ports = unitop.ports.filter(direction=port.direction)
202 port_index = list(attached_ports).index(port)
203 return port.default_stream_position(unitop, graphic_object, port_index, attached_ports.count())
206 def perform_bulk_create(self) -> None:
207 """
208 Saves all created objects to the database.
209 """
210 SimulationObject.objects.bulk_create(self.simulation_objects)
211 GraphicObject.objects.bulk_create(self.graphic_objects)
212 PropertySet.objects.bulk_create(self.property_sets)
213 PropertyInfo.objects.bulk_create(self.property_infos)
214 PropertyValue.objects.bulk_create(self.property_values)
215 SimulationObjectPropertyPackages.objects.bulk_create(self.property_packages)
216 Port.objects.bulk_create(self.ports)
217 RecycleData.objects.bulk_create(self.recycle_data)
218 IndexedItem.objects.bulk_create(self.index_items)
219 PropertyValueIntermediate.objects.bulk_create(self.property_value_indexed_items)
221 def create(
222 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
223 ) -> SimulationObject:
224 """
225 Creates a new simulation object
226 does not save the instance to the database.
228 @param object_type: The type of object to create
229 @param object_schema: The schema of the object
230 @param coordinates: The coordinates of the object
231 @param flowsheet: The flowsheet owner of the object
233 @return: The created simulation object, graphic object, property sets, property infos, property packages, and ports.
234 this is so that the caller can create multiple objects at once and do a bulk_create
235 """
236 if flowsheet is not None: 236 ↛ 239line 236 didn't jump to line 239 because the condition on line 236 was always true
237 idx_for_type = ObjectTypeCounter.next_for(flowsheet, object_type)
238 else:
239 last_sim_obj = SimulationObject.objects.last()
240 if last_sim_obj is None:
241 id = 1
242 else:
243 last_sim_obj = SimulationObject.objects.last()
244 idx_for_type = (last_sim_obj.id + 1) if last_sim_obj else 1
246 if componentName is None:
247 componentName = f"{object_schema.displayType}{idx_for_type}"
249 elif "stream" not in object_type and componentName != object_schema.displayName: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 componentName = componentName
251 else:
252 componentName = f"{componentName}{idx_for_type}"
255 # create simulation object
256 fields = {
257 "objectType": object_type,
258 "componentName": componentName,
259 }
260 instance = SimulationObject(**fields)
261 self.simulation_objects.append(instance)
262 idx = len(self._idx_map)
263 self._idx_map[idx] = instance
264 self._property_set_map[idx] = []
266 # create index items
267 self._index_items_map[idx] = {}
268 for index_set in object_schema.indexSets:
269 self.create_indexed_items(instance, index_set, idx)
271 graphic_object_schema = object_schema.graphicObject
273 graphicObject = GraphicObject(
274 simulationObject=instance,
275 x=coordinates["x"] - graphic_object_schema.width / 2, # center the object horizontally
276 y=coordinates["y"] - graphic_object_schema.height / 2, # center the object vertically
277 width = graphic_object_schema.width,
278 height = graphic_object_schema.height,
279 visible=True,
280 group=parentGroup,
281 flowsheet=flowsheet
282 )
283 self.graphic_objects.append(graphicObject)
285 # If flowsheetKey is given, set flowsheet
286 if flowsheet is not None: 286 ↛ 291line 286 didn't jump to line 291 because the condition on line 286 was always true
287 instance.flowsheet = flowsheet
290 # Create property sets
291 if createPropertySet: 291 ↛ 295line 291 didn't jump to line 295 because the condition on line 291 was always true
292 self.create_property_set(object_schema, idx)
294 # Create ports
295 ports_schema = object_schema.ports or {}
296 # replace any many=True ports with multiple ports
297 for key, port_dict in ports_schema.items():
298 if port_dict.many:
299 # replace this key value pair with a list of ports up to the default provided
300 for i in range(port_dict.default):
301 self.ports.append(Port(
302 key=key,
303 index=i,
304 direction=port_dict.type,
305 displayName=port_dict.displayName + f" {i+1}",
306 unitOp=instance,
307 flowsheet=flowsheet
308 ))
309 else:
310 self.ports.append(Port(
311 key=key,
312 direction=port_dict.type,
313 displayName=port_dict.displayName,
314 unitOp=instance,
315 flowsheet=flowsheet
316 ))
318 # Create property package slots
319 property_packages = object_schema.propertyPackagePorts
320 if property_packages is not None:
321 for slot in property_packages.keys():
322 if slot != "__none__": # special case for connecting together streams that don't actually have a property package. 322 ↛ 321line 322 didn't jump to line 321 because the condition on line 322 was always true
323 simulationObjectPropertyPackage = SimulationObjectPropertyPackages(
324 simulationObject=instance,
325 name=slot,
326 propertyPackage="helmholtz",
327 flowsheet=flowsheet
328 )
329 self.property_packages.append(simulationObjectPropertyPackage)
331 # conditionally create attached objects based on object type
332 if object_type == "recycle":
333 self.recycle_data.append(RecycleData(
334 simulationObject=instance,
335 flowsheet=flowsheet
336 ))
338 return instance
340 def replace_the_gut(self, instance: SimulationObject):
341 self.simulation_objects.append(instance)
342 self._unitop = instance
343 self._current_port = instance.connectedPorts.first()
345 object_schema = configuration[instance.objectType]
346 idx = len(self._idx_map)
347 self._idx_map[idx] = instance
348 self._property_set_map[idx] = []
350 # create index items
351 self._index_items_map[idx] = {}
352 for index_set in object_schema.indexSets:
353 self.create_indexed_items(instance, index_set, idx)
355 self.create_property_set(object_schema, idx)
358 def create_property_set(self, object_schema: ObjectType, idx: int) -> PropertySet:
359 """
360 Creates a new SimulationObjectPropertySet instance with the specified schema.
361 """
362 defaults = {
363 'compoundMode': "",
364 "simulationObject": self._idx_map[idx],
365 }
366 for schema in object_schema.propertySetGroups.values():
367 if schema.type == "composition":
368 defaults['compoundMode'] = "MassFraction"
370 # Create SimulationObjectPropertySet
371 instance = PropertySet(**defaults, flowsheet=self._flowsheet)
373 if object_schema.properties != {}: 373 ↛ 384line 373 didn't jump to line 384 because the condition on line 373 was always true
374 # Create PropertyInfo objects
375 property_pairs = self.create_property_infos(instance, object_schema.properties, idx)
377 # set access based on the schema
378 self.set_properties_access(
379 config=object_schema,
380 properties=property_pairs,
381 idx=idx,
382 )
384 self.property_sets.append(instance)
385 self._property_set_map[idx].append(instance)
387 return instance
390 def create_property_infos(self, property_set: PropertySet, schema: PropertiesType, idx) -> List[Tuple[PropertyInfo, List[PropertyValue]]]:
391 """
392 Creates PropertyInfo objects based on the specified schema.
393 """
394 res: List[Tuple[PropertyInfo, List[PropertyValue]]] = []
396 # if schema.type == "composition":
397 # # the composition property set has no property infos, since these are compounds selected by the user
398 # # TODO: Initialise the properties based on the compounds upstream (if any)
399 # flowsheet = self._flowsheet
400 for key, prop in schema.items():
401 fields = get_property_fields(key, prop, property_set)
402 new_property_info, new_property_values = self.create_property_info(idx, prop.indexSets, **fields)
403 res.append((new_property_info, new_property_values))
405 return res
408 def create_property_info(self, idx, index_sets=None, **fields):
409 value = fields.pop("value")
410 property_info = PropertyInfo(**fields, flowsheet=self._flowsheet)
411 self.property_infos.append(property_info)
412 # Create a property value object with this value
413 if index_sets == None:
414 property_value = PropertyValue(value=value, property=property_info, flowsheet=self._flowsheet)
415 self.property_values.append(property_value)
416 return property_info, [property_value]
417 else:
418 combinations = self.get_combinations(self._index_items_map[idx], index_sets)
419 property_values = []
420 for indexes in combinations:
421 property_value = PropertyValue(value=value, property=property_info, flowsheet=self._flowsheet)
422 for indexed_item in indexes:
423 self.property_value_indexed_items.append(
424 PropertyValueIntermediate(propertyvalue=property_value, indexeditem=indexed_item)
425 )
426 property_values.append(property_value)
427 self.property_values.extend(property_values)
428 return property_info, property_values
431 def get_combinations(self, index_items_map: dict[str, list[IndexedItem]], index_sets=[]) -> list[list[IndexedItem]]:
432 """
433 Returns all possible combinations of indexed items for the specified index sets
434 (taking one item from each index set).
435 """
436 if index_items_map == {}: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true
437 return []
438 indexes = [value for key, value in index_items_map.items() if key in index_sets]
439 return list(itertools.product(*indexes))
442 def set_properties_access(
443 self,
444 config: ObjectType,
445 properties: List[Tuple[PropertyInfo, List[PropertyValue]]],
446 idx: int,
447 ) -> None:
449 if (
450 self.simulation_objects[-1].is_stream()
451 and self._unitop.objectType != "recycle"
452 and self._current_port.direction == ConType.Outlet
453 ):
454 # disable outlet/intermediate stream properties
455 for propertyInfo, propVals in properties:
456 for property_value in propVals:
457 property_value.enabled = False
458 else:
459 index_map_items = self._index_items_map[idx]
460 # TODO: Refactor this.
461 # not disabled if it is a state variable
462 for prop, propVals in properties:
463 config_prop = config.properties.get(prop.key)
464 # # figure out the number of items in the first index set
465 if config_prop.indexSets:
466 first_index_type = config_prop.indexSets[0]
467 if first_index_type in index_map_items: 467 ↛ 474line 467 didn't jump to line 474 because the condition on line 467 was always true
468 if len(index_map_items[first_index_type]) == 0:
469 last_item = None
470 else:
471 last_item = index_map_items[first_index_type][-1]
472 # last_item is only needed on index sets.
473 # it should be fine if it's undefined otherwise.
474 for index, property_value in enumerate(propVals):
475 if config_prop: 475 ↛ 478line 475 didn't jump to line 478 because the condition on line 475 was always true
476 group = config_prop.propertySetGroup
477 else:
478 group = "default"
480 config_group = config.propertySetGroups.get(group, None)
481 if config_group is None: 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 property_value.enabled = True
483 elif config_group.type == "exceptLast" or config_prop.sumToOne:
484 # TODO: Deprecate exceptLast in favor of sumToOne. See phase_seperator_config
485 # Instead of using len_last_item, just use the last item in the index set
486 indexed_item_links = [item for item in self.property_value_indexed_items
487 if item.propertyvalue == property_value]
488 indexed_items = [item.indexeditem for item in indexed_item_links]
489 is_last_item = last_item in indexed_items
491 if is_last_item:
492 property_value.enabled = False
493 else:
494 property_value.enabled = True
495 elif config_group.type == "stateVars": 495 ↛ 499line 495 didn't jump to line 499 because the condition on line 495 was always true
496 state_vars = getattr(config_group, "stateVars") or ()
497 property_value.enabled = prop.key in state_vars
498 else: # eg. All
499 property_value.enabled = True
502 def create_indexed_items(self, instance: SimulationObject, index_set: str, idx: int) -> None:
503 """
504 Creates IndexedItem instances for the specified index set.
505 """
506 items = []
508 outlet_name = instance.schema.splitter_fraction_name
509 match index_set:
510 case "splitter_fraction":
511 items = []
512 # create indexed items for the splitter_fraction index set
513 items.append(IndexedItem(
514 owner=instance,
515 key="outlet_1",
516 displayName= outlet_name + " 1",
517 type=index_set,
518 flowsheet=self._flowsheet
519 ))
520 items.append(IndexedItem(
521 owner=instance,
522 key="outlet_2",
523 displayName= outlet_name + " 2",
524 type=index_set,
525 flowsheet=self._flowsheet
526 ))
528 case "phase":
529 # create indexed items for the phase index set
530 items.append(IndexedItem(
531 owner=instance,
532 key="Liq",
533 displayName="Liquid",
534 type=index_set,
535 flowsheet=self._flowsheet
536 ))
537 items.append(IndexedItem(
538 owner=instance,
539 key="Vap",
540 displayName="Vapor",
541 type=index_set,
542 flowsheet=self._flowsheet
543 ))
545 case "compound":
546 pass # have to wait for user to select compounds
547 self._index_items_map[idx][index_set] = items
548 self.index_items.extend(items)