Coverage for backend/django/flowsheetInternals/unitops/models/simulation_object_factory.py: 93%
304 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +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.FlowsheetState import FlowsheetState
7from core.auxiliary.models.IndexedItem import IndexedItem
8from core.auxiliary.models.PropertyValue import PropertyValue, PropertyValueIntermediate
9from core.auxiliary.models.PropertyInfo import PropertyInfo
10from core.auxiliary.models.RecycleData import RecycleData
11from core.auxiliary.models.PropertySet import PropertySet
12from core.auxiliary.enums import ConType, heat_exchange_ops
14from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
15from flowsheetInternals.unitops.models.Port import Port
16from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
17from flowsheetInternals.graphicData.models.groupingModel import Grouping
18from flowsheetInternals.graphicData.models.portAnchorPlacementModel import (
19 PortAnchorPlacement,
20)
21from flowsheetInternals.unitops.models.compound_propogation import (
22 update_compounds_on_add_stream,
23)
24from flowsheetInternals.unitops.services.edit_operations.recorder import (
25 tracked_bulk_create,
26)
28from flowsheetInternals.unitops.config.config_methods import *
29from flowsheetInternals.unitops.config.config_base import configuration
30from common.config_types import *
31from core.auxiliary.models.ObjectTypeCounter import ObjectTypeCounter
32from core.auxiliary.views.ExtractSegmentDataFromFS import create_he_streams
34ANCHORS_PER_SIDE = 7
36PORT_ANCHOR_EXCLUDED_OBJECT_TYPES = {
37 "header",
38 "simple_header",
39 "boiler",
40 "mixer",
41 "energy_mixer",
42 "splitter",
43 "energy_splitter",
44}
47class SimulationObjectFactory:
48 def __init__(self) -> None:
49 self._flowsheet_state: FlowsheetState | None = None
51 self.simulation_objects: list[SimulationObject] = []
52 self.graphic_objects: list[GraphicObject] = []
53 self.property_sets: list[PropertySet] = []
54 self.property_infos: list[PropertyInfo] = []
55 self.property_values: list[PropertyValue] = []
56 self.ports: list[Port] = []
57 self.index_items: list[IndexedItem] = []
58 self.property_value_indexed_items: List[PropertyValueIntermediate] = []
60 # conditionally created objects
61 self.recycle_data: list[RecycleData] = []
63 self._unitop: SimulationObject = None # the unitop of the object being created
64 self._current_port: Port = None # the current port for the stream being created
66 self._idx_map: dict[
67 int, SimulationObject
68 ] = {} # map of id to simulation object
69 self._property_set_map: dict[
70 int, list[PropertySet]
71 ] = {} # map of simulation object to property sets for the many-to-many relationship
72 self._index_items_map: dict[
73 int, dict[str, list[IndexedItem]]
74 ] = {} # map of simulation object to indexed items for the many-to-many relationship
76 # For storing old property data when replacing the gut
77 self.old_properties: dict[str, any] = {}
79 @classmethod
80 def create_simulation_object(
81 cls,
82 coordinates: dict[str, float] | None = None,
83 createPropertySet: bool = True,
84 flowsheet: Flowsheet | None = None,
85 parentGroup=None,
86 **kwargs,
87 ) -> SimulationObject:
88 """
89 Creates a new unitop with attached ports, streams etc
90 """
91 factory = SimulationObjectFactory()
93 flowsheet_state = flowsheet.current_state if flowsheet is not None else None
95 # Create unitop
96 # If parentGroup is omitted, use the active state's root grouping.
97 if parentGroup:
98 group = Grouping.objects.get(pk=parentGroup)
99 else:
100 if flowsheet_state: 100 ↛ 103line 100 didn't jump to line 103 because the condition on line 100 was always true
101 group = flowsheet_state.root_grouping
102 else:
103 group = None
104 factory._flowsheet_state = flowsheet_state
105 object_type = kwargs.pop("objectType")
106 if kwargs.get("schema") is not None:
107 object_schema = kwargs.pop("schema")
108 else:
109 object_schema: ObjectType = configuration[object_type]
111 if coordinates is None:
112 coordinates = {"x": 0, "y": 0}
114 unitop = factory.create(
115 object_type=object_type,
116 object_schema=object_schema,
117 coordinates=coordinates,
118 flowsheet_state=flowsheet_state,
119 createPropertySet=createPropertySet,
120 parentGroup=group,
121 )
123 factory._unitop = unitop
124 graphic_object = factory.graphic_objects[-1]
126 # count the number of inlet and outlet ports
127 num_inlets = len(
128 [port for port in factory.ports if port.direction == ConType.Inlet]
129 )
131 num_outlets = len(
132 [
133 port
134 for port in factory.ports
135 if (
136 (port.direction == ConType.Outlet)
137 and (factory._unitop.schema.ports[port.key].makeStream == True)
138 )
139 ]
140 )
142 inlet_index = 0
143 outlet_index = 0
145 if kwargs.get("create_attached_streams", True):
146 for port in factory.ports:
147 factory._current_port = port
148 port_schema = factory._unitop.schema.ports[port.key]
149 if port_schema.makeStream is False:
150 continue
151 stream_type = port_schema.streamType
153 stream_schema = configuration[stream_type]
154 # evenly space the ports along the sides of the unitop
155 if port.direction == ConType.Inlet:
156 coordinates = port.default_stream_position(
157 factory._unitop, graphic_object, inlet_index, num_inlets
158 )
159 inlet_index += 1
160 else:
161 coordinates = port.default_stream_position(
162 factory._unitop, graphic_object, outlet_index, num_outlets
163 )
164 outlet_index += 1
166 stream_name = port.default_stream_name(factory._unitop)
167 stream = factory.create(
168 object_type=stream_type,
169 object_schema=stream_schema,
170 coordinates=coordinates,
171 flowsheet_state=flowsheet_state,
172 componentName=stream_name,
173 parentGroup=group,
174 )
176 port.stream = stream
178 # save all created objects
179 factory.perform_bulk_create()
181 if unitop.objectType in heat_exchange_ops:
182 create_he_streams(unitop, group)
183 factory.create_default_port_anchor_placements(unitop, group)
185 # Propagate streams to other groups for the newly created unitop
186 inlet_streams = []
187 outlet_streams = []
188 for port in unitop.ports.all():
189 if port.stream:
190 if port.direction == ConType.Inlet:
191 inlet_streams.append(port.stream)
192 else:
193 outlet_streams.append(port.stream)
194 from flowsheetInternals.graphicData.logic.make_group import propagate_streams
196 propagate_streams(inlet_streams, ConType.Inlet)
197 propagate_streams(outlet_streams, ConType.Outlet)
199 # create the attached ML model object if it's a machine learning block
200 if unitop.objectType == "machineLearningBlock":
201 MLModel.objects.create(
202 simulationObject=unitop,
203 flowsheet_state=flowsheet_state,
204 )
205 return unitop
207 @classmethod
208 def create_stream_at_port(cls, port: Port) -> SimulationObject:
209 """
210 Creates a new stream attached to the specified port.
211 """
212 factory = SimulationObjectFactory()
213 factory._current_port = port
214 factory._unitop = port.unitOp
215 factory._flowsheet_state = factory._unitop.flowsheet_state
217 # Figure out what type of stream to create based on port config
218 # E.g energy_stream, stream, etc
219 object_type = factory._unitop.schema.ports[port.key].streamType
220 object_schema: ObjectType = configuration[object_type]
222 # get the coordinates for the stream
223 coordinates = cls.default_stream_position(factory._unitop, port)
225 # get the group of the unitop that the stream is attached to
226 group = factory._unitop.graphicObject.last().group
228 # create the stream
229 stream = factory.create(
230 object_type=object_type,
231 object_schema=object_schema,
232 coordinates=coordinates,
233 flowsheet_state=factory._unitop.flowsheet_state,
234 parentGroup=group,
235 )
236 factory.perform_bulk_create()
238 # attach the stream to the port
239 port.stream = stream
240 port.save()
241 factory.create_default_port_anchor_placements(factory._unitop, group)
243 # populate the stream with compounds
244 update_compounds_on_add_stream(port, stream)
246 return stream
248 @classmethod
249 def default_stream_position(
250 cls, unitop: SimulationObject, port: Port
251 ) -> dict[str, float]:
252 """
253 Returns the default position for a stream attached to the specified port.
254 """
255 graphic_object = (
256 unitop.graphicObject.last()
257 ) # the unit op only has one graphic object so this is okay.
258 attached_ports = unitop.ports.filter(direction=port.direction)
259 port_index = list(attached_ports).index(port)
260 return port.default_stream_position(
261 unitop, graphic_object, port_index, attached_ports.count()
262 )
264 @staticmethod
265 def default_anchor_index(port_index: int, num_ports: int) -> int:
266 """
267 Maps the previous side-relative port ordering to the fixed seven-anchor grid.
268 """
269 if num_ports <= 1:
270 return ANCHORS_PER_SIDE // 2
272 proportional_index = ((port_index + 0.5) / num_ports) * (
273 ANCHORS_PER_SIDE + 1
274 ) - 1
275 return max(0, min(ANCHORS_PER_SIDE - 1, round(proportional_index)))
277 @classmethod
278 def create_default_port_anchor_placements(
279 cls,
280 unitop: SimulationObject,
281 group: Grouping | None,
282 ) -> None:
283 """
284 Creates explicit unit-operation anchor placements for newly attached streams.
286 The frontend now treats anchor placement rows as the source of truth for
287 connection endpoint occupancy, so default streams need placements before
288 the user moves them manually.
289 """
290 if group is None or unitop.objectType in PORT_ANCHOR_EXCLUDED_OBJECT_TYPES:
291 return
293 graphic_object = unitop.graphicObject.last()
294 if graphic_object is None: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true
295 return
297 for direction, anchor_side in (
298 (ConType.Inlet, "left"),
299 (ConType.Outlet, "right"),
300 ):
301 connected_ports = list(
302 unitop.ports.filter(direction=direction, stream__isnull=False)
303 )
304 num_ports = len(connected_ports)
305 for port_index, port in enumerate(connected_ports):
306 defaults = {
307 "flowsheet_state": unitop.flowsheet_state,
308 "anchor_side": anchor_side,
309 "anchor_index": cls.default_anchor_index(port_index, num_ports),
310 }
311 placement, created = PortAnchorPlacement.objects.get_or_create(
312 port=port,
313 grouping=group,
314 graphicObject=graphic_object,
315 endpointKind=PortAnchorPlacement.EndpointKind.UnitOperation,
316 defaults=defaults,
317 )
318 if not created and ( 318 ↛ 321line 318 didn't jump to line 321 because the condition on line 318 was never true
319 placement.anchor_side is None or placement.anchor_index is None
320 ):
321 placement.anchor_side = defaults["anchor_side"]
322 placement.anchor_index = defaults["anchor_index"]
323 placement.save(update_fields=["anchor_side", "anchor_index"])
325 def perform_bulk_create(self) -> None:
326 """
327 Saves all created objects to the database.
328 """
329 tracked_bulk_create(SimulationObject.objects, self.simulation_objects)
330 tracked_bulk_create(GraphicObject.objects, self.graphic_objects)
331 tracked_bulk_create(PropertySet.objects, self.property_sets)
332 tracked_bulk_create(PropertyInfo.objects, self.property_infos)
333 tracked_bulk_create(PropertyValue.objects, self.property_values)
334 tracked_bulk_create(Port.objects, self.ports)
335 tracked_bulk_create(RecycleData.objects, self.recycle_data)
336 tracked_bulk_create(IndexedItem.objects, self.index_items)
337 tracked_bulk_create(
338 PropertyValueIntermediate.objects,
339 self.property_value_indexed_items,
340 )
342 def create(
343 self,
344 object_type: str,
345 object_schema: ObjectType,
346 coordinates: dict[str, float],
347 flowsheet_state: FlowsheetState | None = None,
348 parentGroup: Grouping | None = None,
349 componentName: str | None = None,
350 createPropertySet: bool = True,
351 ) -> SimulationObject:
352 """
353 Creates a new simulation object
354 does not save the instance to the database.
356 @param object_type: The type of object to create
357 @param object_schema: The schema of the object
358 @param coordinates: The coordinates of the object
359 @param flowsheet_state: The state that owns the object.
361 @return: The created simulation object, graphic object, property sets, property infos, property packages, and ports.
362 this is so that the caller can create multiple objects at once and do a bulk_create
363 """
364 if flowsheet_state is not None: 364 ↛ 367line 364 didn't jump to line 367 because the condition on line 364 was always true
365 idx_for_type = ObjectTypeCounter.next_for(flowsheet_state, object_type)
366 else:
367 last_sim_obj = SimulationObject.objects.last()
368 if last_sim_obj is None:
369 idx_for_type = 1
370 else:
371 last_sim_obj = SimulationObject.objects.last()
372 idx_for_type = (last_sim_obj.id + 1) if last_sim_obj else 1
374 if componentName is None:
375 componentName = f"{object_schema.displayType}{idx_for_type}"
377 elif "stream" not in object_type and componentName != object_schema.displayName:
378 componentName = componentName
379 else:
380 componentName = f"{componentName}{idx_for_type}"
382 # create simulation object
383 fields = {
384 "objectType": object_type,
385 "componentName": componentName,
386 }
387 instance = SimulationObject(**fields)
388 self.simulation_objects.append(instance)
389 idx = len(self._idx_map)
390 self._idx_map[idx] = instance
391 self._property_set_map[idx] = []
393 # create index items
394 self._index_items_map[idx] = {}
395 for index_set in object_schema.indexSets:
396 self.create_indexed_items(instance, index_set, idx)
398 graphic_object_schema = object_schema.graphicObject
400 graphicObject = GraphicObject(
401 simulationObject=instance,
402 x=coordinates["x"]
403 - graphic_object_schema.width / 2, # center the object horizontally
404 y=coordinates["y"]
405 - graphic_object_schema.height / 2, # center the object vertically
406 width=graphic_object_schema.width,
407 height=graphic_object_schema.height,
408 visible=True,
409 group=parentGroup,
410 flowsheet_state=flowsheet_state,
411 )
412 self.graphic_objects.append(graphicObject)
414 # Assign state ownership after constructing the unsaved object.
415 if flowsheet_state is not None: 415 ↛ 419line 415 didn't jump to line 419 because the condition on line 415 was always true
416 instance.flowsheet_state = flowsheet_state
418 # Create property sets
419 if createPropertySet: 419 ↛ 423line 419 didn't jump to line 423 because the condition on line 419 was always true
420 self.create_property_set(object_schema, idx)
422 # Create ports
423 ports_schema = object_schema.ports or {}
424 # replace any many=True ports with multiple ports
425 for key, port_dict in ports_schema.items():
426 if port_dict.many:
427 # replace this key value pair with a list of ports up to the default provided
428 for i in range(port_dict.default):
429 self.ports.append(
430 Port(
431 key=key,
432 index=i,
433 direction=port_dict.type,
434 displayName=port_dict.displayName + f" {i + 1}",
435 unitOp=instance,
436 flowsheet_state=flowsheet_state,
437 )
438 )
439 else:
440 self.ports.append(
441 Port(
442 key=key,
443 direction=port_dict.type,
444 displayName=port_dict.displayName,
445 unitOp=instance,
446 flowsheet_state=flowsheet_state,
447 )
448 )
450 # conditionally create attached objects based on object type
451 if object_type == "recycle":
452 self.recycle_data.append(
453 RecycleData(simulationObject=instance, flowsheet_state=flowsheet_state)
454 )
456 return instance
458 def store_old_properties(self, instance: SimulationObject):
459 """
460 Stores the old properties of the stream.
461 Used when switching stream types to keep properties that exist in both schema.
462 Prevents losing user input in those properties.
463 e.g. If Molar Flow has a value, switching to Humid Air won't remove that value.
464 """
465 self.old_properties = {}
466 if instance.properties: 466 ↛ exitline 466 didn't return from function 'store_old_properties' because the condition on line 466 was always true
467 for prop_info in instance.properties.containedProperties.all():
468 try:
469 self.old_properties[prop_info.key] = prop_info.get_value_bulk()
470 except ValueError:
471 self.old_properties[prop_info.key] = None
473 def replace_the_gut(self, instance: SimulationObject):
474 self.simulation_objects.append(instance)
475 self._unitop = instance
476 self._current_port = instance.connectedPorts.first()
478 object_schema = configuration[instance.objectType]
479 idx = len(self._idx_map)
480 self._idx_map[idx] = instance
481 self._property_set_map[idx] = []
483 # create index items
484 self._index_items_map[idx] = {}
485 for index_set in object_schema.indexSets:
486 self.create_indexed_items(instance, index_set, idx)
488 self.create_property_set(object_schema, idx)
490 def create_property_set(self, object_schema: ObjectType, idx: int) -> PropertySet:
491 """
492 Creates a new SimulationObjectPropertySet instance with the specified schema.
493 """
494 defaults = {
495 "compoundMode": "",
496 "simulationObject": self._idx_map[idx],
497 }
498 for schema in object_schema.propertySetGroups.values():
499 if schema.type == "composition":
500 defaults["compoundMode"] = "MassFraction"
502 # Create SimulationObjectPropertySet
503 instance = PropertySet(**defaults, flowsheet_state=self._flowsheet_state)
505 if object_schema.properties != {}: 505 ↛ 518line 505 didn't jump to line 518 because the condition on line 505 was always true
506 # Create PropertyInfo objects
507 property_pairs = self.create_property_infos(
508 instance, object_schema.properties, idx
509 )
511 # set access based on the schema
512 self.set_properties_access(
513 config=object_schema,
514 properties=property_pairs,
515 idx=idx,
516 )
518 self.property_sets.append(instance)
519 self._property_set_map[idx].append(instance)
521 return instance
523 def create_property_infos(
524 self, property_set: PropertySet, schema: PropertiesType, idx
525 ) -> List[Tuple[PropertyInfo, List[PropertyValue]]]:
526 """
527 Creates PropertyInfo objects based on the specified schema.
528 """
529 res: List[Tuple[PropertyInfo, List[PropertyValue]]] = []
531 # if schema.type == "composition":
532 # # the composition property set has no property infos, since these are compounds selected by the user
533 # # TODO: Initialise the properties based on the compounds upstream (if any)
534 # flowsheet_state = self._flowsheet_state
535 for key, prop in schema.items():
536 fields = get_property_fields(key, prop, property_set)
538 if self.old_properties:
539 # If property in the schema is also in the old properties, keep the value
540 if key in self.old_properties.keys():
541 fields["value"] = self.old_properties[key]
543 new_property_info, new_property_values = self.create_property_info(
544 idx, prop.indexSets, **fields
545 )
546 res.append((new_property_info, new_property_values))
548 return res
550 def create_property_info(self, idx, index_sets=None, **fields):
551 value = fields.pop("value")
552 property_info = PropertyInfo(**fields, flowsheet_state=self._flowsheet_state)
553 self.property_infos.append(property_info)
554 # Create a property value object with this value
555 if index_sets == None:
556 property_value = PropertyValue(
557 value=value,
558 property=property_info,
559 flowsheet_state=self._flowsheet_state,
560 )
561 self.property_values.append(property_value)
562 return property_info, [property_value]
563 else:
564 combinations = self.get_combinations(self._index_items_map[idx], index_sets)
565 property_values = []
566 for indexes in combinations:
567 property_value = PropertyValue(
568 value=value,
569 property=property_info,
570 flowsheet_state=self._flowsheet_state,
571 )
572 for indexed_item in indexes:
573 self.property_value_indexed_items.append(
574 PropertyValueIntermediate(
575 propertyvalue=property_value, indexeditem=indexed_item
576 )
577 )
578 property_values.append(property_value)
579 self.property_values.extend(property_values)
580 return property_info, property_values
582 def get_combinations(
583 self, index_items_map: dict[str, list[IndexedItem]], index_sets=[]
584 ) -> list[list[IndexedItem]]:
585 """
586 Returns all possible combinations of indexed items for the specified index sets
587 (taking one item from each index set).
588 """
589 if index_items_map == {}: 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true
590 return []
591 indexes = [value for key, value in index_items_map.items() if key in index_sets]
592 return list(itertools.product(*indexes))
594 def set_properties_access(
595 self,
596 config: ObjectType,
597 properties: List[Tuple[PropertyInfo, List[PropertyValue]]],
598 idx: int,
599 ) -> None:
600 if (
601 self.simulation_objects[-1].is_stream()
602 and self._unitop.objectType != "recycle"
603 and self._current_port
604 and self._current_port.direction == ConType.Outlet
605 ):
606 # disable outlet/intermediate stream properties
607 for propertyInfo, propVals in properties:
608 for property_value in propVals:
609 property_value.enabled = False
610 else:
611 index_map_items = self._index_items_map[idx]
612 # TODO: Refactor this.
613 # not disabled if it is a state variable
614 for prop, propVals in properties:
615 config_prop = config.properties.get(prop.key)
616 # # figure out the number of items in the first index set
617 if config_prop.indexSets:
618 first_index_type = config_prop.indexSets[0]
619 if first_index_type in index_map_items: 619 ↛ 626line 619 didn't jump to line 626 because the condition on line 619 was always true
620 if len(index_map_items[first_index_type]) == 0:
621 last_item = None
622 else:
623 last_item = index_map_items[first_index_type][-1]
624 # last_item is only needed on index sets.
625 # it should be fine if it's undefined otherwise.
626 for index, property_value in enumerate(propVals):
627 if config_prop: 627 ↛ 630line 627 didn't jump to line 630 because the condition on line 627 was always true
628 group = config_prop.propertySetGroup
629 else:
630 group = "default"
632 config_group = config.propertySetGroups.get(group, None)
633 if config_group is None: 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 property_value.enabled = True
635 elif config_group.type == "exceptLast" or config_prop.sumToOne:
636 # TODO: Deprecate exceptLast in favor of sumToOne. See phase_seperator_config
637 # Instead of using len_last_item, just use the last item in the index set
638 indexed_item_links = [
639 item
640 for item in self.property_value_indexed_items
641 if item.propertyvalue == property_value
642 ]
643 indexed_items = [
644 item.indexeditem for item in indexed_item_links
645 ]
646 is_last_item = last_item in indexed_items
648 if is_last_item:
649 property_value.enabled = False
650 else:
651 property_value.enabled = True
652 elif config_group.type == "stateVars": 652 ↛ 656line 652 didn't jump to line 656 because the condition on line 652 was always true
653 state_vars = getattr(config_group, "stateVars") or ()
654 property_value.enabled = prop.key in state_vars
655 else: # eg. All
656 property_value.enabled = True
658 def create_indexed_items(
659 self, instance: SimulationObject, index_set: str, idx: int
660 ) -> None:
661 """
662 Creates IndexedItem instances for the specified index set.
663 """
664 items = []
666 outlet_name = instance.schema.splitter_fraction_name
667 match index_set:
668 case "splitter_fraction":
669 items = []
670 # create indexed items for the splitter_fraction index set
671 items.append(
672 IndexedItem(
673 owner=instance,
674 key="outlet_1",
675 displayName=outlet_name + " 1",
676 type=index_set,
677 flowsheet_state=self._flowsheet_state,
678 )
679 )
680 items.append(
681 IndexedItem(
682 owner=instance,
683 key="outlet_2",
684 displayName=outlet_name + " 2",
685 type=index_set,
686 flowsheet_state=self._flowsheet_state,
687 )
688 )
690 case "phase":
691 # create indexed items for the phase index set
692 items.append(
693 IndexedItem(
694 owner=instance,
695 key="Liq",
696 displayName="Liquid",
697 type=index_set,
698 flowsheet_state=self._flowsheet_state,
699 )
700 )
701 items.append(
702 IndexedItem(
703 owner=instance,
704 key="Vap",
705 displayName="Vapor",
706 type=index_set,
707 flowsheet_state=self._flowsheet_state,
708 )
709 )
711 case "compound":
712 pass # have to wait for user to select compounds
713 self._index_items_map[idx][index_set] = items
714 self.index_items.extend(items)