Coverage for backend/django/flowsheetInternals/unitops/models/SimulationObject.py: 87%
550 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 QuerySet
3from core.managers import AllFlowsheetStatesManager, SoftDeleteManager
4from core.auxiliary.enums import SimulationObjectClass
5from core.auxiliary.models.FlowsheetState import FlowsheetState
6from core.auxiliary.models.FlowsheetHistoryModel import FlowsheetHistoryModel
7from core.auxiliary.models.PropertySet import PropertySet
8from core.auxiliary.models.PropertyInfo import PropertyInfo
9from core.auxiliary.enums.generalEnums import PropertyType as PropertyTypeChoices
11from typing import TYPE_CHECKING
12from core.auxiliary.models.PropertyValue import PropertyValue, PropertyValueIntermediate
13from core.auxiliary.models.IndexedItem import IndexedItem, IndexChoices
14from typing import Set
16from flowsheetInternals.unitops.models.Port import Port
17from typing import Iterable
18from flowsheetInternals.unitops.config.config_methods import *
19from common.config_types import *
20import itertools
22from .compound_propogation import (
23 update_compounds_on_set,
24 update_compounds_on_merge,
25 _get_compound_keys,
26 update_decision_node_and_propagate,
27 update_compounds_on_add_stream,
28)
29from typing import Optional, List
30from ..methods.add_expression import add_expression as _add_expression
31from flowsheetInternals.unitops.services.edit_operations.recorder import (
32 tracked_bulk_create,
33 tracked_bulk_update,
34 tracked_queryset_update,
35)
37if TYPE_CHECKING:
38 from core.auxiliary.models.PropertySet import PropertySet
39 from core.auxiliary.models.PropertyInfo import PropertyInfo
40 from core.auxiliary.models.RecycleData import RecycleData
41 from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
42 from flowsheetInternals.graphicData.models.groupingModel import Grouping
43 from flowsheetInternals.unitops.models.Port import Port
44 from core.auxiliary.models.CustomPropertyPackage import CustomPropertyPackage
47class SimulationObject(FlowsheetHistoryModel, models.Model):
48 flowsheet_history_soft_delete_field = "is_deleted"
50 flowsheet_state = models.ForeignKey(
51 FlowsheetState, on_delete=models.CASCADE, related_name="flowsheetObjects"
52 )
53 componentName = models.CharField(max_length=64)
54 objectType = models.CharField(choices=SimulationObjectClass.choices)
56 created_at = models.DateTimeField(auto_now_add=True)
57 is_deleted = models.BooleanField(default=False)
58 initial_values = models.JSONField(null=True, blank=True)
59 propertyPackageType = models.CharField(max_length=64, default="helmholtz")
60 customPackage = models.ForeignKey["CustomPropertyPackage"](
61 "core_auxiliary.CustomPropertyPackage",
62 on_delete=models.SET_NULL,
63 null=True,
64 blank=True,
65 related_name="simulationObjects",
66 )
68 # add a soft delete manager
69 objects = SoftDeleteManager()
70 all_states = AllFlowsheetStatesManager()
71 add_expression = _add_expression
72 # runtime-accessed attributes
73 properties: "PropertySet"
74 graphicObject: "GraphicObject"
75 ports: QuerySet["Port"]
76 connectedPorts: QuerySet["Port"]
77 recycleConnection: "Optional[RecycleData]"
78 recycleData: "Optional[RecycleData]"
80 @property
81 def schema(self) -> ObjectType:
82 return get_object_schema(self)
84 @property
85 def has_recycle_connection(self) -> bool:
86 return hasattr(self, "recycleConnection")
88 def is_stream(self) -> bool:
89 return self.schema.is_stream
91 def get_stream(self, key: str, index: int = 0) -> "SimulationObject":
92 """
93 Returns the stream attached to the port with the given key
94 """
95 port = self.get_port(key, index)
96 stream = port.stream
97 return stream
99 def get_group(self) -> "Grouping":
100 """
101 Returns the group that this object belongs to
102 """
103 if not self.is_stream(): 103 ↛ 108line 103 didn't jump to line 108 because the condition on line 103 was always true
104 # There is only one graphic object, so we can just return the group
105 return self.graphicObject.last().group
106 else:
107 # Streams arent' really in a group, throw an error
108 raise ValueError("Streams do not belong to a group")
110 def get_groups(self) -> Iterable["Grouping"]:
111 """
112 Returns an iterable of groups that this object belongs to. Mostly needed for streams that have many groups.
113 """
114 return (graphic.group for graphic in self.graphicObject.all())
116 def get_parent_groups(self) -> List["Grouping"]:
117 """
118 Returns an iterable of parent groups that this object belongs to.
119 """
120 parent_groups = []
121 current_group = self.get_group()
122 while current_group:
123 parent_groups.append(current_group)
124 simulationObject = current_group.simulationObject
125 if simulationObject is None: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 break
127 current_group = simulationObject.get_group()
128 return parent_groups
130 def get_property_package(self, name: str | None = None):
131 """
132 Returns the property package slot with the given name
133 """
134 return self.propertyPackageType
136 def set_property_package(self, property_package, name: str | None = None) -> None:
137 """
138 Sets the property package for this object
139 """
140 self.propertyPackageType = property_package
141 self.save()
143 def get_port(self, key: str, index: int = 0) -> Port:
144 """
145 Returns the port with the given key
146 """
147 try:
148 port: Port = self.ports.get(key=key, index=index)
149 return port
150 except Port.DoesNotExist:
151 raise ValueError(
152 f"Port with key {key} does not exist on object {self.componentName}"
153 )
155 def reorder_object_ports(self):
156 """
157 Reorders port mappings by connected unit operation y position
158 :return: None
159 """
160 inlet_connections = [
161 port for port in self.ports.filter(direction=ConType.Inlet).all()
162 ]
163 outlet_connections = [
164 port for port in self.ports.filter(direction=ConType.Outlet).all()
165 ]
166 inlet_connections.sort(
167 key=lambda port: (
168 port.stream.connectedPorts.get(direction=ConType.Outlet)
169 .unitOp.graphicObject.last()
170 .y
171 )
172 )
173 outlet_connections.sort(
174 key=lambda port: (
175 port.stream.connectedPorts.get(direction=ConType.Inlet)
176 .unitOp.graphicObject.last()
177 .y
178 )
179 )
180 for i, port in enumerate(inlet_connections):
181 port.index = i
182 port.save()
183 for i, port in enumerate(outlet_connections):
184 port.index = i
185 port.save()
186 self.save()
188 def horizontally_center_graphic(self) -> None:
189 """
190 Horizontally centers GraphicObject for a stream or unit operation
191 Currently uni-directional - left to right connections
192 Centers based on inlet and outlet graphics.
193 :return: None
194 """
195 # TODO: Make this work as appropriate for streams with multiple graphic objects
196 # Is an Intermediate Stream
197 if ( 197 ↛ 228line 197 didn't jump to line 228 because the condition on line 197 was always true
198 self.objectType == SimulationObjectClass.Stream
199 or SimulationObjectClass.EnergyStream
200 or SimulationObjectClass.acStream
201 and self.connectedPorts.count() > 1
202 ):
203 inlet, outlet = (
204 self.connectedPorts.filter(direction=ConType.Outlet).first(),
205 self.connectedPorts.filter(direction=ConType.Inlet).first(),
206 )
207 inlet_x, outlet_x = (
208 inlet.unitOp.graphicObject.last().x,
209 outlet.unitOp.graphicObject.last().x,
210 )
211 if inlet_x <= outlet_x:
212 self.graphicObject.last().x = (
213 abs(
214 (inlet_x + (inlet.unitOp.graphicObject.last().width))
215 + (outlet_x)
216 )
217 / 2
218 )
219 else:
220 self.graphicObject.last().x = (
221 abs(inlet_x + (outlet_x + outlet.unitOp.graphicObject.last().width))
222 / 2
223 )
224 # Flip Graphic Object horizontally
225 self.graphicObject.last().save()
226 self.save()
227 else:
228 inlet, outlet = (
229 self.ports.filter(direction=ConType.Inlet).first(),
230 self.ports.filter(direction=ConType.Outlet).first(),
231 )
232 self.graphicObject.last().x = abs(
233 (
234 inlet.stream.graphicObject.last().x
235 + outlet.stream.graphicObject.last().x
236 )
237 / 2
238 )
239 self.graphicObject.last().save()
240 self.save()
241 return
243 def vertically_center_graphic(self) -> None:
244 """
245 Vertically centers GraphicObject for a stream or unit operation
246 :return: None
247 """
248 # TODO: Make this work as appropriate for streams with multiple graphic objects
249 # Is an intermediate stream
250 if (
251 self.objectType == SimulationObjectClass.Stream
252 or SimulationObjectClass.EnergyStream
253 or SimulationObjectClass.acStream
254 and self.connectedPorts.count() > 1
255 ):
256 inlet, outlet = (
257 self.connectedPorts.filter(direction=ConType.Inlet).first(),
258 self.connectedPorts.filter(direction=ConType.Outlet).first(),
259 )
260 self.graphicObject.last().y = abs(
261 (
262 inlet.unitOp.graphicObject.last().y
263 + outlet.unitOp.graphicObject.last().y
264 )
265 / 2
266 )
267 self.graphicObject.last().save()
268 self.save()
269 else:
270 inlet, outlet = (
271 self.ports.filter(direction=ConType.Inlet).first(),
272 self.ports.filter(direction=ConType.Outlet).all()[1],
273 )
274 self.graphicObject.last().y = abs(
275 (
276 inlet.stream.graphicObject.last().y
277 + outlet.stream.graphicObject.last().y
278 )
279 / 2
280 )
281 self.graphicObject.last().save()
282 self.save()
283 return
285 def split_stream(self) -> "SimulationObject":
286 """
287 Splits a stream into two separate streams (one inlet and one outlet - disconnected).
288 :return: New Stream Object (outlet stream) if object is a stream.
289 """
290 from flowsheetInternals.unitops.models.simulation_object_factory import (
291 SimulationObjectFactory,
292 )
293 from flowsheetInternals.graphicData.models.graphicObjectModel import (
294 GraphicObject,
295 )
297 new_stream = None
298 if ( 298 ↛ 386line 298 didn't jump to line 386 because the condition on line 298 was always true
299 self.objectType == SimulationObjectClass.Stream
300 or SimulationObjectClass.EnergyStream
301 or SimulationObjectClass.acStream
302 ):
303 connectedPorts = self.connectedPorts.all()
304 inlet_port: Port = connectedPorts.get(direction=ConType.Inlet)
305 outlet_port: Port = connectedPorts.get(direction=ConType.Outlet)
306 new_stream = SimulationObjectFactory.create_stream_at_port(inlet_port)
307 new_stream.save()
308 # reset the stream to default position
309 coordinates = SimulationObjectFactory.default_stream_position(
310 outlet_port.unitOp, outlet_port
311 )
312 stream_graphic_object = self.graphicObject.last()
313 stream_graphic_object.x = coordinates["x"] - stream_graphic_object.width / 2
314 stream_graphic_object.y = (
315 coordinates["y"] - stream_graphic_object.height / 2
316 )
317 stream_graphic_object.save()
319 """
320 If a stream connects two groups together, it will have graphic objects in each of those groups.
321 This next section figures out which graphic object to keep, and which to move to the new stream.
322 """
323 all_graphic_objects = self.graphicObject.all()
324 default_graphic_object = (
325 new_stream.graphicObject.last()
326 ) # a default graphic object is created by create_stream_at_port
327 # The graphic objects in these groups should be kept
328 groups_to_keep: List[int] = [
329 g.id for g in outlet_port.unitOp.get_parent_groups()
330 ]
331 groups_to_move: List[int] = [
332 g.id for g in inlet_port.unitOp.get_parent_groups()
333 ]
335 for gobj in all_graphic_objects:
336 if gobj.group.id in groups_to_keep:
337 # keep it here
338 if gobj.group.id in groups_to_move:
339 # move the default graphic object into this group, so both the inlet and outlet show in this group
340 default_graphic_object.group = gobj.group
341 default_graphic_object.save()
342 else:
343 if gobj.group.id not in groups_to_move: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 print(
345 "Error: this should be in either ggroups to move or the other",
346 gobj.group,
347 )
348 # must be in groups_to_move
349 # move graphic object to this stream
350 gobj.simulationObject = new_stream
351 gobj.save()
353 # make sure both the original and new streams are shown in all relevant groups
354 for graphic_object in self.graphicObject.all():
355 current_group = graphic_object.group
356 while current_group:
357 parent_group = current_group.get_parent_group()
358 if parent_group:
359 # make sure the original stream is shown in the parent group
360 if not self.graphicObject.filter(group=parent_group).exists(): 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 GraphicObject.objects.create(
362 flowsheet_state=self.flowsheet_state,
363 simulationObject=self,
364 width=graphic_object.width,
365 height=graphic_object.height,
366 x=graphic_object.x,
367 y=graphic_object.y,
368 group=parent_group,
369 )
371 # make sure the new stream is shown in the parent group
372 if not new_stream.graphicObject.filter( 372 ↛ 375line 372 didn't jump to line 375 because the condition on line 372 was never true
373 group=parent_group
374 ).exists():
375 GraphicObject.objects.create(
376 flowsheet_state=self.flowsheet_state,
377 simulationObject=new_stream,
378 width=graphic_object.width,
379 height=graphic_object.height,
380 x=graphic_object.x,
381 y=graphic_object.y,
382 group=parent_group,
383 )
384 current_group = parent_group
386 return new_stream
388 def merge_parallel_streams(
389 self, connected_stream: "SimulationObject", decision_node=None, coordinates=None
390 ) -> "SimulationObject":
391 """
392 Merges two streams that have the same direction (Both Inlet or Outlet)
393 :param connected_stream: Stream to connect
394 :param decision_node: Decision Node object to merges streams into (In the case of n-inlet / n-outlet)
395 :return: Decision Node Object
396 """
397 from flowsheetInternals.unitops.models.simulation_object_factory import (
398 SimulationObjectFactory,
399 )
401 is_inlet = self.connectedPorts.first().direction == ConType.Inlet
402 direction = ConType.Outlet if is_inlet else ConType.Inlet
403 if decision_node is None:
404 modified_schema: ObjectType = configuration["decisionNode"].model_copy(
405 deep=True
406 )
407 if is_inlet: 407 ↛ 411line 407 didn't jump to line 411 because the condition on line 407 was always true
408 modified_schema.ports["outlet"].default = 2
409 modified_schema.ports["inlet"].default = 2
410 else:
411 modified_schema.ports["inlet"].default = 2
412 modified_schema.ports["outlet"].default = 2
413 decision_node = SimulationObjectFactory.create_simulation_object(
414 coordinates={
415 "x": self.graphicObject.last().x,
416 "y": self.graphicObject.last().y,
417 }
418 if coordinates is None
419 else coordinates,
420 objectType="decisionNode",
421 schema=modified_schema,
422 flowsheet=self.flowsheet_state.flowsheet,
423 create_attached_streams=False,
424 )
425 decision_node.graphicObject.last().rotation = (
426 self.connectedPorts.first().unitOp.graphicObject.last().rotation
427 )
428 decision_node.graphicObject.last().save()
429 ports = decision_node.ports.filter(direction=direction).all()
430 port1 = ports[0]
431 port2 = ports[1]
432 port1.stream = self
433 port2.stream = connected_stream
434 port1.save()
435 port2.save()
437 decision_node.save()
439 # Center GraphicObjects
440 self.horizontally_center_graphic()
441 connected_stream.horizontally_center_graphic()
442 self.save()
443 connected_stream.save()
445 return decision_node
447 def merge_stream(self, connectStream: "SimulationObject") -> "SimulationObject":
448 """
449 Connects this object (a stream) to another stream
450 :param connectedStream: Stream to connect to this object
451 :return: Decision Node Object
452 """
453 material_stream_1 = self
454 material_stream_2 = connectStream
456 material_stream_1_port = material_stream_1.connectedPorts.first()
457 material_stream_2_port = material_stream_2.connectedPorts.first()
459 # Case 1: Intermediate to Intermediate
460 if (
461 material_stream_1.connectedPorts.count() > 1
462 and material_stream_2.connectedPorts.count() > 1
463 ):
464 coordinates = {
465 "x": material_stream_2.graphicObject.last().x,
466 "y": material_stream_2.graphicObject.last().y,
467 }
468 outlet_stream_1 = material_stream_1.split_stream()
469 outlet_stream_2 = material_stream_2.split_stream()
471 decision_node = outlet_stream_2.merge_parallel_streams(
472 outlet_stream_1, coordinates=coordinates
473 )
474 decision_node = material_stream_2.merge_parallel_streams(
475 material_stream_1, decision_node
476 )
478 update_decision_node_and_propagate(decision_node)
479 decision_node.save()
480 return decision_node
482 # Case 2: Connecting intermediate stream with product or feed
483 if (
484 material_stream_1.connectedPorts.count() > 1
485 or material_stream_2.connectedPorts.count() > 1
486 ):
487 inter_stream, connected_stream = (
488 (material_stream_1, material_stream_2)
489 if material_stream_1.connectedPorts.count()
490 > material_stream_2.connectedPorts.count()
491 else (material_stream_2, material_stream_1)
492 )
493 connected_stream_port = connected_stream.connectedPorts.first()
494 is_inlet = connected_stream_port.direction == ConType.Inlet
496 if is_inlet:
497 # Feed -> Intermediate
498 decision_node = inter_stream.make_decision_node(
499 num_inlets=1, num_outlets=2
500 )
501 empty_port = decision_node.ports.filter(
502 direction=ConType.Outlet, stream=None
503 ).first()
504 empty_port.stream = connected_stream
505 empty_port.save()
507 self.delete_empty_node(connected_stream, connected_stream_port)
509 # Center graphics
510 self.horizontally_center_graphic()
511 connected_stream.horizontally_center_graphic()
512 self.save()
513 connected_stream.save()
515 # Update compounds - First update decision node, then propagate
516 update_decision_node_and_propagate(decision_node)
517 return decision_node
518 else:
519 # Product -> Intermediate
520 decision_node = inter_stream.make_decision_node(
521 num_inlets=2, num_outlets=1
522 )
523 empty_port = decision_node.ports.filter(
524 direction=ConType.Inlet, stream=None
525 ).first()
526 empty_port.stream = connected_stream
527 empty_port.save()
529 self.delete_empty_node(connected_stream, connected_stream_port)
531 # Center graphics
532 self.horizontally_center_graphic()
533 connected_stream.horizontally_center_graphic()
534 self.save()
535 connected_stream.save()
537 # Update compounds - First update decision node, then propagate
538 update_decision_node_and_propagate(decision_node)
539 return decision_node
541 # Case 3: Feed to feed or product to product
542 if material_stream_1_port.direction == material_stream_2_port.direction: 542 ↛ 543line 542 didn't jump to line 543 because the condition on line 542 was never true
543 result = material_stream_2.merge_parallel_streams(material_stream_1)
544 update_decision_node_and_propagate(result)
545 return result
547 # Case 4: Feed to product or product to feed
548 if material_stream_1_port.direction == "inlet":
549 inlet_stream, outlet_stream = material_stream_1, material_stream_2
550 inlet_port, outlet_port = material_stream_1_port, material_stream_2_port
551 else:
552 inlet_stream, outlet_stream = material_stream_2, material_stream_1
553 inlet_port, outlet_port = material_stream_2_port, material_stream_1_port
555 # Get all the graphic objects to preserve
557 # Always preserve the graphic of the stream that is being connected to
558 preserve_graphic = material_stream_2.graphicObject.last()
559 # Update compounds
560 if inlet_stream.objectType == SimulationObjectClass.EnergyStream: 560 ↛ 561line 560 didn't jump to line 561 because the condition on line 560 was never true
561 pass
562 elif inlet_stream.objectType != SimulationObjectClass.acStream: 562 ↛ 565line 562 didn't jump to line 565 because the condition on line 562 was always true
563 update_compounds_on_merge(inlet_stream, outlet_stream)
565 def update_graphic_object_on_merge(preserve_stream, delete_stream) -> None:
566 # Preserve one graphic object for each group either stream is connected to
567 # If the stream is connected to multiple groups, it will be shown in both groups
568 preserve_stream_groups = list(preserve_stream.get_groups())
569 preserve_stream_gobjs = preserve_stream.graphicObject.all()
570 merged_gobjs = []
571 for gobj in delete_stream.graphicObject.all():
572 # Check if there's a preserved stream in this group
573 preserved_gobj = next(
574 (g for g in preserve_stream_gobjs if g.group == gobj.group), None
575 )
576 if preserved_gobj is not None:
577 # Keep this if
578 if preserve_stream == material_stream_2:
579 # Keep this graphic object's position
580 gobj.delete()
581 else:
582 # Keep the position of the stream to be deleted
583 preserved_gobj.copy_position_from(gobj)
584 gobj.delete()
585 # If both streams are shown in multiple parent groups, we only want to show the
586 # stream at the lowest level. make a list of these merged graphicsObjects so we can remove
587 # the extras later.
588 merged_gobjs.append(preserved_gobj)
589 else:
590 # connect this graphic object to the preserved stream
591 gobj.simulationObject = preserve_stream
592 gobj.save()
594 # Of the merged graphics objects, only keep the one in the lowest group
595 parent_groups: list[int] = []
597 for gobj in merged_gobjs:
598 parent_group = gobj.group.get_parent_group()
599 if parent_group is not None:
600 parent_groups.append(parent_group.pk)
602 for gobj in merged_gobjs:
603 if gobj.group.pk in parent_groups:
604 gobj.delete()
606 def _collect_unique_groups(*group_iters) -> set["Grouping"]:
607 """Return a set of unique Grouping objects from one or more iterables.
609 This preserves iteration order only in the sense that groups seen earlier
610 are added first, but the return value is a set because callers expect
611 membership semantics rather than ordering.
612 """
613 unique = set()
614 for iterable in group_iters:
615 for g in iterable:
616 if g not in unique:
617 unique.add(g)
618 return unique
620 # get all groups that are groups of either stream
621 inlet_stream_groups = list(inlet_stream.get_groups())
622 outlet_stream_groups = list(outlet_stream.get_groups())
623 intermediate_groups: set["Grouping"] = _collect_unique_groups(
624 inlet_stream_groups, outlet_stream_groups
625 )
627 from flowsheetInternals.unitops.models.delete_factory import DeleteFactory
629 if not outlet_stream.has_recycle_connection and inlet_stream.has_path_to(
630 outlet_stream
631 ):
632 # preserve the outlet stream
633 preserve_stream = outlet_stream
634 inlet_port.stream = preserve_stream
635 inlet_port.save()
636 DeleteFactory.delete_object(inlet_stream)
638 update_graphic_object_on_merge(preserve_stream, inlet_stream)
640 # attach recycle block to the inlet stream
641 # this also handles updating property access
642 outlet_stream.attach_recycle(intermediate_groups)
643 inlet_stream.delete_control_values()
645 outlet_stream.reevaluate_properties_enabled()
646 else:
647 # Preserve the outlet stream
648 preserve_stream = outlet_stream
649 inlet_port.stream = preserve_stream
650 inlet_port.save()
651 DeleteFactory.delete_object(inlet_stream)
652 # Update graphic object
653 update_graphic_object_on_merge(preserve_stream, inlet_stream)
655 return preserve_stream
657 def delete_control_values(self) -> None:
658 """
659 Deletes all control values connected to properties in this object
660 """
661 property_set: PropertySet = self.properties
662 for prop in property_set.ContainedProperties.all():
663 for value in prop.values.all():
664 if value.is_control_set_point():
665 value.controlSetPoint.delete()
667 def attach_recycle(
668 self, intermediate_groups: set["Grouping"] | None = None
669 ) -> None:
670 """
671 Attaches a new recycle block to the stream
672 """
673 if self.has_recycle_connection: 673 ↛ 674line 673 didn't jump to line 674 because the condition on line 673 was never true
674 return # already has a recycle connection
675 from flowsheetInternals.unitops.models.simulation_object_factory import (
676 SimulationObjectFactory,
677 )
678 from flowsheetInternals.graphicData.models.groupingModel import Grouping
679 from flowsheetInternals.graphicData.models.graphicObjectModel import (
680 GraphicObject,
681 )
683 recycle = SimulationObjectFactory.create_simulation_object(
684 objectType="recycle",
685 flowsheet=self.flowsheet_state.flowsheet,
686 coordinates={
687 "x": self.graphicObject.last().x + self.graphicObject.last().width / 2,
688 "y": self.graphicObject.last().y
689 + self.graphicObject.last().height / 2
690 + 100,
691 },
692 )
694 # Set graphic objects for the recycle block in the same groups as the stream
695 default_graphic = recycle.graphicObject.last()
696 default_width = default_graphic.width if default_graphic else 32
697 default_height = default_graphic.height if default_graphic else 32
699 # Remove all graphic objects associated with the recycle so we can re-add them to proper groups
700 recycle.graphicObject.all().delete()
702 # If caller didn't provide groups, use this stream's groups
703 if intermediate_groups is None:
704 intermediate_groups = set(self.get_groups())
705 else:
706 # ensure we have a set (in case a generator or queryset was passed)
707 intermediate_groups = set(intermediate_groups)
709 normalized_groups: set[Grouping] = set()
710 seen_group_ids: set[int] = set()
712 # Normalise groups to Grouping objects and remove duplicates
713 for group in intermediate_groups:
714 group_obj = None
715 if isinstance(group, Grouping): 715 ↛ 717line 715 didn't jump to line 717 because the condition on line 715 was always true
716 group_obj = group
717 elif group is not None:
718 group_obj = Grouping.objects.filter(pk=group).first()
720 if group_obj and group_obj.pk not in seen_group_ids: 720 ↛ 713line 720 didn't jump to line 713 because the condition on line 720 was always true
721 normalized_groups.add(group_obj)
722 seen_group_ids.add(group_obj.pk)
724 # Create graphic objects for each group
725 for group in normalized_groups:
726 stream_graphic = self.graphicObject.filter(group=group).last()
727 if stream_graphic is None:
728 continue
730 GraphicObject.objects.create(
731 flowsheet_state=self.flowsheet_state,
732 simulationObject=recycle,
733 width=default_width,
734 height=default_height,
735 x=stream_graphic.x + (stream_graphic.width - default_width) / 2,
736 y=stream_graphic.y + stream_graphic.height + 30,
737 group=group,
738 )
740 recycle.recycleData.update(self)
741 recycle.save()
743 def has_path_to(self, end_stream: "SimulationObject", check_recycles=True) -> bool:
744 """
745 Checks if there is a path in the flowsheet from the start stream (self) to the end stream
746 Can be used to check for loops in the flowsheet if these two streams are being merged
748 - param end_stream: The stream to check if there is a path to (from self)
749 - param check_recycles: If True, will skip the path if a stream has a recycle connection
750 """
751 remaining_streams = [self]
752 while remaining_streams:
753 current_stream = remaining_streams.pop()
754 if current_stream == end_stream:
755 # loop detected
756 return True
757 unit_op_port = current_stream.connectedPorts.filter(
758 direction=ConType.Inlet
759 ).first()
760 if not unit_op_port:
761 continue
762 unit_op = unit_op_port.unitOp
763 # get the outlet ports of the unitop
764 connected_port_keys = get_connected_port_keys(
765 unit_op_port.key, unit_op.schema
766 )
767 outlet_ports = unit_op.ports.filter(
768 direction=ConType.Outlet, key__in=connected_port_keys
769 )
770 for outlet_port in outlet_ports:
771 outlet_stream = outlet_port.stream
772 if outlet_stream is not None: 772 ↛ 770line 772 didn't jump to line 770 because the condition on line 772 was always true
773 if check_recycles and outlet_stream.has_recycle_connection: 773 ↛ 774line 773 didn't jump to line 774 because the condition on line 773 was never true
774 continue
775 remaining_streams.append(outlet_stream)
776 return False
778 def make_decision_node(self, num_inlets, num_outlets):
779 """
780 Turns a stream into a decision node with n inlet and m outlet ports
781 :param num_inlets: number of inlet ports to create
782 :param num_outlets: number of outlet ports to create
783 :return: Decision Node object
784 """
785 from flowsheetInternals.unitops.models.simulation_object_factory import (
786 SimulationObjectFactory,
787 )
789 if self.objectType == SimulationObjectClass.Stream: 789 ↛ exitline 789 didn't return from function 'make_decision_node' because the condition on line 789 was always true
790 # Create Decision Node
791 modified_schema: ObjectType = configuration["decisionNode"].model_copy(
792 deep=True
793 )
794 modified_schema.ports["inlet"].default = num_inlets
795 modified_schema.ports["outlet"].default = num_outlets
796 graphicObject = (
797 self.graphicObject.last()
798 ) # For now, we are assuming this only has one graphicObject.
799 parentGroup = self.graphicObject.last().group.id
800 decision_node = SimulationObjectFactory.create_simulation_object(
801 coordinates={"x": graphicObject.x, "y": graphicObject.y},
802 objectType="decisionNode",
803 schema=modified_schema,
804 flowsheet=self.flowsheet_state.flowsheet,
805 create_attached_streams=False,
806 parentGroup=parentGroup,
807 )
809 if self.connectedPorts.count() > 1:
810 # Connect both streams to Decision Node
811 ms_outlet_port = self.connectedPorts.filter(
812 direction=ConType.Inlet
813 ).first()
814 dn_outlet_port = decision_node.ports.filter(
815 direction=ConType.Outlet
816 ).first()
817 dn_inlet_port = decision_node.ports.filter(
818 direction=ConType.Inlet
819 ).first()
820 new_stream = SimulationObjectFactory.create_stream_at_port(
821 port=dn_outlet_port
822 )
823 ms_outlet_port.stream = new_stream
824 dn_inlet_port.stream = self
826 # Save Objects
827 ms_outlet_port.save()
828 dn_outlet_port.save()
829 dn_inlet_port.save()
830 decision_node.save()
832 # Center GraphicObjects
833 self.horizontally_center_graphic()
834 new_stream.horizontally_center_graphic()
835 self.save()
836 new_stream.save()
837 else:
838 # For now, only dealing with the case that a regular stream is initialized with one port
839 port = decision_node.ports.first()
840 port.stream = self
841 port.save()
842 self.horizontally_center_graphic()
843 self.save()
844 self.save()
845 decision_node.save()
846 # handle compounds for decision node
847 update_decision_node_and_propagate(
848 decision_node, updated_via_right_click=True
849 )
850 return decision_node
852 def update_compounds(self, compounds: list[str]) -> None:
853 """
854 Updates the compounds for this stream
855 """
856 update_compounds_on_set(self, compounds)
858 def add_port(
859 self, key: str, existing_stream: "SimulationObject | None" = None
860 ) -> Port:
861 """
862 Adds a port to this object and adds a a new stream if none is provided
863 """
864 from flowsheetInternals.graphicData.logic.make_group import propagate_streams
865 from flowsheetInternals.unitops.models.simulation_object_factory import (
866 SimulationObjectFactory,
867 )
869 # Create ports
870 ports_schema = self.schema.ports
872 # replace any many=True ports with multiple ports
873 port_dict = ports_schema[key]
874 # get the next index for the port (looks at all ports with the same key and gets the length of the l
875 next_index = self.ports.filter(key=key).count()
876 new_port = Port(
877 key=key,
878 index=next_index,
879 direction=port_dict.type,
880 displayName=port_dict.displayName + f" {next_index + 1}",
881 unitOp=self,
882 flowsheet_state=self.flowsheet_state,
883 )
884 new_port.save()
886 self.update_height()
888 if existing_stream:
889 # connect existing stream to the new port
890 new_port.stream = existing_stream
891 new_port.save() # might have to remove this trying to figure out why other streams are being removed
893 if self.objectType == "decisionNode": 893 ↛ 988line 893 didn't jump to line 988 because the condition on line 893 was always true
894 # Use existing function to update decision node compounds
895 update_decision_node_and_propagate(self)
896 else:
897 # create a new strea at the port using the factory
898 new_stream = SimulationObjectFactory.create_stream_at_port(new_port)
899 update_compounds_on_add_stream(new_port, new_stream)
900 outlet_name = self.schema.splitter_fraction_name
901 for property_key, property in self.schema.properties.items():
902 if (
903 property.indexSets is not None
904 and ("splitter_fraction" in property.indexSets)
905 and new_port.direction == ConType.Outlet
906 ):
907 property_infos = self.properties.ContainedProperties.all()
908 property_info = property_infos.first()
909 new_indexed_item = IndexedItem.objects.create(
910 owner=self,
911 key="outlet_" + f"{next_index + 1}",
912 displayName=outlet_name + f" {next_index + 1}",
913 type="splitter_fraction",
914 flowsheet_state=self.flowsheet_state,
915 )
917 # get indexed set for split_fraction or priorities:
918 index_sets = property.indexSets
920 # figure out which other indexed items this should link to
921 indexed_item_sets: List[List[IndexedItem]] = []
922 for index in index_sets:
923 if index == IndexChoices.SplitterFraction:
924 continue # We don't need this, it's the one we're editing
925 indexed_items: List[IndexedItem] = self.get_indexed_items(
926 index
927 ) # e.g get_indexed_items("phase"), "compound", etc
928 indexed_item_sets.append(indexed_items)
930 # create a property value for each combination of indexed items, for the new outlet
931 combinations = list(itertools.product(*indexed_item_sets))
933 # Bulk update existing property values to be enabled
934 property_status = False
935 if (
936 self.objectType == "header"
937 or self.objectType == "simple_header"
938 ):
939 property_status = True
940 else:
941 tracked_queryset_update(
942 property_info.values.all(),
943 enabled=True,
944 )
946 # bulk create indexed items (all disabled as it's the last outlet)
947 index_links_to_create: List[PropertyValueIntermediate] = []
948 for combination in combinations:
949 # Create a propertyValue for this combination
950 combo_property_value = PropertyValue.objects.create(
951 property=property_info,
952 enabled=property_status,
953 flowsheet_state=self.flowsheet_state,
954 )
955 # link it up to the new outlet
956 index_links_to_create.append(
957 PropertyValueIntermediate(
958 propertyvalue=combo_property_value,
959 indexeditem=new_indexed_item,
960 )
961 )
963 # Also link it up to all the other sets
964 for indexed_item in combination:
965 index_links_to_create.append(
966 PropertyValueIntermediate(
967 propertyvalue=combo_property_value,
968 indexeditem=indexed_item,
969 )
970 )
972 # Bulk create the links
973 tracked_bulk_create(
974 PropertyValueIntermediate.objects,
975 index_links_to_create,
976 )
978 new_stream.save()
979 # Rotate the new stream to match the rotation of the unitop it is attached to
980 self.update_stream_rotation(new_stream, self)
982 # Ensure the new stream appears in all parent groups
983 if new_port.direction == ConType.Inlet:
984 propagate_streams([new_stream], ConType.Inlet)
985 elif new_port.direction == ConType.Outlet: 985 ↛ 988line 985 didn't jump to line 988 because the condition on line 985 was always true
986 propagate_streams([new_stream], ConType.Outlet)
988 return new_port
990 def update_height(self):
991 """
992 Updates the height of the graphic object based on the number of ports
993 """
994 if (
995 self.schema.graphicObject.autoHeight
996 ): # e.g header has auto height calculation
997 OFFSET_VALUE = 200
999 max_ports = max(
1000 self.ports.filter(direction=ConType.Inlet).count(),
1001 self.ports.filter(direction=ConType.Outlet).count(),
1002 )
1003 rotation = self.graphicObject.last().rotation
1005 graphic_object: GraphicObject = self.graphicObject.last()
1006 new_length = max_ports * OFFSET_VALUE
1007 obj_width = graphic_object.width
1008 obj_height = graphic_object.height
1010 # Update graphic object position to keep the other streams centered
1011 if rotation == 90:
1012 if new_length > obj_width:
1013 tracked_queryset_update(
1014 self.graphicObject.all(),
1015 x=graphic_object.x - OFFSET_VALUE,
1016 )
1017 elif new_length < obj_width:
1018 tracked_queryset_update(
1019 self.graphicObject.all(),
1020 x=graphic_object.x + OFFSET_VALUE,
1021 )
1022 elif rotation == 180:
1023 if new_length > obj_height:
1024 tracked_queryset_update(
1025 self.graphicObject.all(),
1026 y=graphic_object.y - OFFSET_VALUE,
1027 )
1028 elif new_length < obj_height:
1029 tracked_queryset_update(
1030 self.graphicObject.all(),
1031 y=graphic_object.y + OFFSET_VALUE,
1032 )
1034 # Update the correct dimension based on rotation
1035 if rotation in [90, 270]:
1036 tracked_queryset_update(
1037 self.graphicObject.all(),
1038 width=new_length,
1039 )
1040 else:
1041 tracked_queryset_update(
1042 self.graphicObject.all(),
1043 height=new_length,
1044 )
1046 def update_stream_rotation(
1047 self, stream: "SimulationObject", unitop: "SimulationObject"
1048 ) -> None:
1049 """
1050 Updates the rotation of the stream based on the rotation of the unitop it is attached to.
1051 """
1052 graphic_object = stream.graphicObject.last()
1054 if graphic_object is None: 1054 ↛ 1055line 1054 didn't jump to line 1055 because the condition on line 1054 was never true
1055 return
1057 # Update the rotation of the stream to match the rotation of the unitop
1058 graphic_object.rotation = unitop.graphicObject.last().rotation
1059 graphic_object.flipped = unitop.graphicObject.last().flipped
1060 graphic_object.save()
1062 def get_indexed_items(self, index_set_type: IndexChoices) -> List[IndexedItem]:
1063 match index_set_type:
1064 case IndexChoices.Phase:
1065 # get all the indexedItems that are type=phase
1066 items = IndexedItem.objects.filter(
1067 owner=self, type=IndexChoices.Phase
1068 ).all()
1069 return items
1070 case IndexChoices.Compound: 1070 ↛ 1075line 1070 didn't jump to line 1075 because the pattern on line 1070 always matched
1071 items = IndexedItem.objects.filter(
1072 owner=self, type=IndexChoices.Compound
1073 ).all()
1074 return items
1075 case _:
1076 raise ValueError("Get_indexed_items didn't expect this index set type")
1078 def merge_decision_nodes(
1079 self,
1080 decision_node_active: "SimulationObject",
1081 decision_node_over: "SimulationObject",
1082 ) -> Optional["SimulationObject"]:
1083 """
1084 Merges this decision node with the over decision node and handles graphics positioning
1085 """
1086 import flowsheetInternals.unitops.models.delete_factory as DeleteFactory
1088 # Get all streams from active node
1089 inlet_streams = decision_node_active.ports.filter(direction="inlet").all()
1090 outlet_streams = decision_node_active.ports.filter(direction="outlet").all()
1092 # Transfer and reposition inlet streams
1093 for inlet_port in inlet_streams:
1094 stream = inlet_port.stream
1095 # Add stream to new decision node
1096 decision_node_over.add_port("inlet", stream)
1098 # Update compounds and propagate
1099 update_decision_node_and_propagate(decision_node_over)
1101 # Transfer and reposition outlet streams
1102 for outlet_port in outlet_streams:
1103 stream = outlet_port.stream
1104 # Add stream to new decision node
1105 decision_node_over.add_port("outlet", stream)
1107 # Delete the active node and its graphic object
1108 DeleteFactory.delete_object(decision_node_active)
1109 decision_node_over.save()
1110 return decision_node_over
1112 def reevaluate_properties_enabled(self) -> None:
1113 """
1114 Reevaluates property access for all properties in this object
1116 This should only really be called when connections are changed,
1117 otherwise the property enabling should be handled by adding
1118 or removing control values.
1119 """
1120 if self.objectType == SimulationObjectClass.MachineLearningBlock:
1121 return # We don't want to change from the defaults
1122 properties: list[PropertyInfo] = self.properties.ContainedProperties.all()
1123 config = self.schema
1124 config_properties = config.properties
1125 config_groups = config.propertySetGroups
1127 def _eval_enabled(prop: PropertyInfo, config_group) -> bool:
1128 if self.is_stream():
1129 # disable outlet/intermediate stream properties
1130 ports = self.connectedPorts.all()
1131 if len(ports) == 2 or (
1132 len(ports) == 1 and ports[0].direction == "outlet"
1133 ):
1134 return False
1135 # inlet stream properties should be enabled
1136 if config_group.type == "stateVars":
1137 state_vars = getattr(config_group, "stateVars") or ()
1138 return prop.key in state_vars
1139 return True # eg. All, default to enabled
1141 list_prop_val = []
1143 for prop in properties:
1144 config_prop = config_properties.get(prop.key)
1145 # ignore custom properties
1146 if config_prop:
1147 group = config_prop.propertySetGroup
1148 config_group = config_groups.get(group, None)
1149 res = _eval_enabled(prop, config_group)
1150 list_prop_val.extend(prop.enable(res))
1152 tracked_bulk_update(PropertyValue.objects, list_prop_val, ["enabled"])
1154 def get_unspecified_properties(self) -> list:
1155 if not hasattr(self, "properties") or not self.properties: 1155 ↛ 1156line 1155 didn't jump to line 1156 because the condition on line 1155 was never true
1156 return []
1158 contained_properties: models.QuerySet[PropertyInfo] = (
1159 self.properties.ContainedProperties.all()
1160 )
1161 is_splitter = getattr(self.schema, "displayType", "").lower() == "splitter"
1163 # use schema stateVars to find required properties
1164 required_properties: Set[str] = set()
1165 for group in self.schema.propertySetGroups.values():
1166 if (
1167 group.toggle
1168 and not self.properties.get_property(group.toggle).get_value()
1169 ):
1170 # The group is toggled off, so we don't care that the properties are not specified
1171 continue
1172 if (
1173 group.type == "stateVars"
1174 or group.type == "composition"
1175 or group.type == "exceptLast"
1176 ):
1177 required_properties.update(group.stateVars)
1179 # use pre fetched data to avoid additional queries
1180 unspecified_properties = []
1181 property_info: PropertyInfo
1182 for property_info in contained_properties:
1183 # check if the property has no values or if all values are invalid
1184 has_valid_value = property_info.isSpecified()
1185 if not has_valid_value and property_info.key in required_properties:
1186 unspecified_properties.append(property_info.key)
1188 # check if mole_frac_comp sums to 1
1189 mole_frac_props = [x for x in contained_properties if x.key == "mole_frac_comp"]
1191 for prop in mole_frac_props:
1192 if prop.has_value_bulk():
1193 total = 0.0
1194 for value in prop.values.all():
1195 try:
1196 val = float(value.value)
1197 total += val
1198 except (ValueError, TypeError):
1199 continue
1200 if (
1201 abs(total - 1.0) > 0.001
1202 and "mole_frac_comp" not in unspecified_properties
1203 ):
1204 unspecified_properties.append("mole_frac_comp")
1205 else:
1206 if "mole_frac_comp" not in unspecified_properties:
1207 unspecified_properties.append("mole_frac_comp")
1209 return unspecified_properties
1211 def delete(self, *args, **kwargs):
1212 raise NotImplementedError("Use delete_object method from DeleteFactory")
1214 def permanently_delete(self, *args, **kwargs):
1215 """
1216 Permanently deletes the object from the database.
1217 This should only be used in tests or when you are sure you want to delete the object.
1218 """
1219 super().delete(*args, **kwargs)
1221 def delete_empty_node(
1222 self, connected_stream: "SimulationObject", connected_stream_port: "Port"
1223 ):
1224 """
1225 Deletes the empty_port stream from parent groups when creating a decision node inside a group.
1226 :param connected_stream: Stream connected to the empty port
1227 :param connected_stream_port: Port of the connected stream
1228 """
1229 simulation_object_id = connected_stream.id
1230 current_group = connected_stream_port.unitOp.get_group()
1231 parent_groups = connected_stream_port.unitOp.get_parent_groups()
1233 for parent_group in parent_groups:
1234 if parent_group != current_group:
1235 gobjs_in_group = connected_stream.graphicObject.filter(
1236 group=parent_group
1237 )
1238 for graphic_object in gobjs_in_group:
1239 if graphic_object.simulationObject.id == simulation_object_id: 1239 ↛ 1238line 1239 didn't jump to line 1238 because the condition on line 1239 was always true
1240 graphic_object.delete()