Coverage for backend/django/flowsheetInternals/graphicData/logic/make_group.py: 87%
181 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from typing import List
2from core.auxiliary.enums import SimulationObjectClass
3from core.auxiliary.enums import ConType
4from flowsheetInternals.graphicData.models.groupingModel import Breadcrumbs, Grouping, GraphicObject
5from flowsheetInternals.graphicData.models.portAnchorPlacementModel import PortAnchorPlacement
6from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
7import random
8from core.auxiliary.models.ObjectTypeCounter import ObjectTypeCounter
9from django.db.models import Count, QuerySet, Prefetch
10from flowsheetInternals.unitops.models.Port import Port
13def make_group(contained_objects_id: List[int]) -> Grouping:
14 """
15 Creates a group
17 Args:
18 containedObjects_id (List[SimulationObject id]): Simulation objects to include in the group.
19 is_abstract: (bool): This hasn't really been implemented yet, but the idea was to have
20 groups shown fully in the parent, with modules hidden inside something.
21 """
23 # Ensure all unit operations share the same group
24 simulation_objects = SimulationObject.objects.filter(pk__in=contained_objects_id).prefetch_related(
25 Prefetch(
26 "connectedPorts",
27 queryset=Port.objects.select_related("unitOp").prefetch_related()
28 )
29 )
30 unit_operations = (
31 simulation_objects.exclude(objectType=SimulationObjectClass.Stream)
32 .exclude(objectType=SimulationObjectClass.Recycle) # exclude recycle block because we want to propagate them like streams
33 .all()
34 )
35 streams = simulation_objects.filter(objectType=SimulationObjectClass.Stream)
36 group_ids = [g.id for g in simulation_objects.filter(objectType=SimulationObjectClass.Group)]
37 unitop_ids = [uo.id for uo in unit_operations]
39 #Remove orphaned streams
40 for stream in streams:
41 stream_group_ids = [g.simulationObject.pk for g in stream.get_groups()]
42 connected_unitop_ids = [p.unitOp.id for p in stream.connectedPorts.all()]
43 #check for orphan and catch case where stream is between two packed groups
44 if not bool(set(connected_unitop_ids).intersection(unitop_ids)) and not bool(set(stream_group_ids).intersection(group_ids)): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 contained_objects_id.remove(stream.id)
47 if not unit_operations.exists(): 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true
48 raise ValueError("No unit operations provided to determine group.")
50 # Get the common group from the unit operations
51 parent_group: Grouping = unit_operations.first().graphicObject.last().group
52 if any(unitop.graphicObject.last().group != parent_group for unitop in unit_operations): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise ValueError("All unit operations must belong to the same group.")
55 # Get all connections in current group within the contained objects
56 simulation_object_ids = [obj.pk for obj in simulation_objects]
57 selected_connections = [connection for connection in parent_group.get_connections() if
58 connection.unitOp in simulation_object_ids]
60 # Find intermediate streams which were not part of the initial selection (hanging streams)
61 no_duplicate_list = set()
62 hanging_streams = [connection.stream for connection in selected_connections if
63 connection.stream in no_duplicate_list or no_duplicate_list.add(connection.stream)]
64 # Add hanging streams to contained objects if not already included
65 for stream in hanging_streams:
66 if stream not in contained_objects_id:
67 contained_objects_id.append(stream)
69 # Include streams only if they belong to the same group
70 for unitop in unit_operations:
71 for port in unitop.ports.all():
72 if port.stream and port.stream.id not in contained_objects_id:
73 group_ids = [g.pk for g in port.stream.get_groups()]
74 if parent_group.pk in group_ids: 74 ↛ 71line 74 didn't jump to line 71 because the condition on line 74 was always true
75 contained_objects_id.append(port.stream.id)
77 # Get inlet and outlet streams connected to groups
78 in_out_streams : list[int] = SimulationObject.objects.annotate(port_count=Count('connectedPorts')).filter(pk__in=no_duplicate_list, port_count=1).values_list('pk', flat=True)
80 # Add inlet and outlet streams to contained objects if not already included
81 for stream in in_out_streams:
82 if stream not in contained_objects_id:
83 contained_objects_id.append(stream)
85 # Re-filter simulation objects after appending streams
86 simulation_objects = SimulationObject.objects.filter(pk__in=contained_objects_id)
88 flowsheet_state = simulation_objects.first().flowsheet_state
90 idx_for_type = ObjectTypeCounter.next_for(flowsheet_state, "module")
91 componentName = f"Module {idx_for_type}"
93 # Create the new group
94 simulation_object = simulation_objects.first()
95 new_group = Grouping.create(
96 flowsheet_state,
97 parent_group,
98 componentName=componentName,
99 )
100 new_group.update_internal_graphic_objects([simObj.graphicObject.first() for simObj in simulation_objects])
101 new_group.set_group_size()
103 inlet_streams = []
104 outlet_streams = []
106 # handle streams for all simulation objects
107 for simulation_object in simulation_objects:
109 # propagate streams connected to one port
110 if simulation_object.is_stream() and simulation_object.connectedPorts.count() == 1:
111 if simulation_object.connectedPorts.all()[0].direction == ConType.Inlet:
112 inlet_streams.append(simulation_object)
113 else:
114 outlet_streams.append(simulation_object)
116 # handle intermediate streams connected to the module
117 if simulation_object.is_stream() and simulation_object.connectedPorts.count() == 2:
118 propagate_intermediate_streams(simulation_object, contained_objects_id)
120 propagate_streams(inlet_streams, ConType.Inlet)
121 propagate_streams(outlet_streams, ConType.Outlet)
123 # update zone of stream data entries
124 for unitop in unit_operations:
125 for stream_data_entry in unitop.StreamDataEntries.all():
126 stream_data_entry.group = new_group
127 stream_data_entry.save()
129 return new_group
132def create_group_boundary_port_anchor_placements(group: Grouping) -> None:
133 """
134 Creates default anchor placements for ports rendered on a packed group's boundary.
136 A port can be shown on its real unit operation inside the group, or on the
137 group object in the parent layer. Those are separate rendered endpoints, so
138 each needs its own placement row.
139 """
140 from flowsheetInternals.unitops.models.simulation_object_factory import (
141 SimulationObjectFactory,
142 )
144 parent_group = group.get_parent_group()
145 if parent_group is None: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 return
148 group_graphic = group.get_graphic_object()
149 if group_graphic is None: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return
152 parent_graphics = parent_group.graphicObjects.select_related(
153 "simulationObject",
154 "simulationObject__grouping",
155 ).prefetch_related(
156 Prefetch(
157 "simulationObject__connectedPorts",
158 queryset=Port.objects.select_related("unitOp", "stream"),
159 )
160 )
161 boundary_connections = [
162 connection
163 for connection in parent_group.get_connections_for_graphics(parent_graphics)
164 if (
165 connection.unitOp == group.simulationObject.id
166 and connection.anchor_endpoint_kind
167 == PortAnchorPlacement.EndpointKind.GroupBoundary
168 )
169 ]
171 changed = False
172 for direction, anchor_side in (
173 (ConType.Inlet, "left"),
174 (ConType.Outlet, "right"),
175 ):
176 connections = sorted(
177 [
178 connection
179 for connection in boundary_connections
180 if connection.direction == direction
181 ],
182 key=lambda connection: connection.port,
183 )
184 num_ports = len(connections)
185 for port_index, connection in enumerate(connections):
186 defaults = {
187 "flowsheet_state": group.flowsheet_state,
188 "anchor_side": anchor_side,
189 "anchor_index": SimulationObjectFactory.default_anchor_index(
190 port_index,
191 num_ports,
192 ),
193 }
194 placement, created = PortAnchorPlacement.objects.get_or_create(
195 port_id=connection.port,
196 grouping=parent_group,
197 graphicObject=group_graphic,
198 endpointKind=PortAnchorPlacement.EndpointKind.GroupBoundary,
199 defaults=defaults,
200 )
201 changed = changed or created
202 if not created and ( 202 ↛ 205line 202 didn't jump to line 205 because the condition on line 202 was never true
203 placement.anchor_side is None or placement.anchor_index is None
204 ):
205 placement.anchor_side = defaults["anchor_side"]
206 placement.anchor_index = defaults["anchor_index"]
207 placement.save(update_fields=["anchor_side", "anchor_index"])
208 changed = True
210 if changed and hasattr(parent_group, "_cached_connections"): 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 delattr(parent_group, "_cached_connections")
214def get_group_visible_from_parent(
215 unit_op: SimulationObject,
216 parent_group: Grouping,
217) -> Grouping | None:
218 """
219 Return the packed group that represents a unit op in an ancestor layer.
221 A unit operation is rendered directly in its own group, but in ancestor
222 groups it is rendered through the nearest child group boundary.
223 """
224 group = unit_op.get_group()
225 while group is not None:
226 if group.get_parent_group() == parent_group:
227 return group
228 group = group.get_parent_group()
230 return None
233def propagate_streams(streams: List[SimulationObject], direction: ConType):
234 """
235 Propagates a stream's graphic objects to all parent groups of its connected unit operations.
237 Args:
238 simulation_object (SimulationObject): The stream to propagate.
239 """
241 if direction == ConType.Inlet:
242 x_offset = -1
243 else:
244 x_offset = 2
245 groups_needing_boundary_anchors = {}
246 for simulation_object in streams:
247 if(len(streams) <= 2):
248 y_offset = (0.5 + streams.index(simulation_object)) / len(streams)
249 y_gap = -12.5
250 # space out the streams if there are too many of them
251 else:
252 y_offset = 0.5 * streams.index(simulation_object)
253 y_gap = -12.5 * (2 * len(streams) - 5)
255 if not simulation_object.is_stream(): 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 pass
258 # make sure the stream is connected to one port (0 and 2 or more should not propagate)
259 connected_ports: QuerySet[Port]= simulation_object.connectedPorts.all()
260 if len(connected_ports) != 1: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 pass
263 # get the original graphic object for reference
264 original_stream_graphic_object : GraphicObject = simulation_object.graphicObject.first()
265 if not original_stream_graphic_object: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 pass
268 # propagate to all parent groups of the connected unit operation
269 unit_op : SimulationObject = connected_ports[0].unitOp
270 for parent_group in unit_op.get_parent_groups():
271 visible_group = get_group_visible_from_parent(unit_op, parent_group)
272 if visible_group is not None:
273 groups_needing_boundary_anchors[visible_group.pk] = visible_group
275 if not simulation_object.graphicObject.filter(group=parent_group).exists():
277 module_graphic_object = original_stream_graphic_object.group.get_graphic_object()
279 GraphicObject.objects.create(
280 flowsheet_state=simulation_object.flowsheet_state,
281 simulationObject=simulation_object,
282 width=original_stream_graphic_object.width,
283 height=original_stream_graphic_object.height,
284 x=module_graphic_object.x -12.5 + x_offset*module_graphic_object.width,
285 y=module_graphic_object.y + y_gap + y_offset*module_graphic_object.height,
286 group=parent_group,
287 )
289 for group in groups_needing_boundary_anchors.values():
290 create_group_boundary_port_anchor_placements(group)
293def propagate_intermediate_streams(simulation_object: SimulationObject, contained_objects_id: List[int]):
294 """
295 Handles the special case where an intermediate stream is connected to a module.
297 Args:
298 simulation_object (SimulationObject): The intermediate stream to propagate.
299 contained_objects_id (List[int]): The IDs of objects being abstracted.
300 """
301 if not simulation_object.is_stream(): 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 return
304 # make sure the stream is an intermediate stream (connected to exactly two ports)
305 connected_ports = simulation_object.connectedPorts.all()
306 if len(connected_ports) != 2: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 return
309 # Count the number of connections to contained objects (unitops or modules)
310 connections_to_contained_objects = 0
311 for port in connected_ports:
312 # Check if the connected unitop is in the contained objects
313 if port.unitOp.id in contained_objects_id:
314 connections_to_contained_objects += 1
316 # get the original graphic object for reference
317 original_graphic_object = simulation_object.graphicObject.last()
318 if not original_graphic_object: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 return
322 def groups_to_add_intermediate():
323 """
324 Check if both ends of the intermediate stream are visible in the given group.
325 If they are, return the groups to which the intermediate stream should be added.
326 """
327 port1, port2 = connected_ports
329 def get_group_path(group):
330 path = []
331 while group:
332 path.append(group)
333 group = group.get_parent_group()
334 return path # [leaf, ..., root]
336 path1 = get_group_path(port1.unitOp.get_group())
337 path2 = get_group_path(port2.unitOp.get_group())
339 # The LCA is the first group in path1 that is also in path2
340 lca = next((group for group in path1 if group in path2), None)
341 if lca is None: 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true
342 return []
344 # Collect all groups from path1 up to and including the LCA
345 groups = []
346 for group in path1: 346 ↛ 351line 346 didn't jump to line 351 because the loop on line 346 didn't complete
347 groups.append(group)
348 if group == lca:
349 break
350 # Collect all groups from path2 up to but not including the LCA (avoid duplicates)
351 for group in path2: 351 ↛ 356line 351 didn't jump to line 356 because the loop on line 351 didn't complete
352 if group == lca:
353 break
354 if group not in groups: 354 ↛ 351line 354 didn't jump to line 351 because the condition on line 354 was always true
355 groups.append(group)
356 return groups
359 # we can use the groups_to_add_intermediate function to determine which groups to add the stream to
360 # we get the groups to put the intermediate streams so just add it to those groups
361 simulation_object.graphicObject.all().delete()
362 groups_to_add = groups_to_add_intermediate()
363 for group in groups_to_add:
364 if not simulation_object.graphicObject.filter(group=group).exists(): 364 ↛ 363line 364 didn't jump to line 363 because the condition on line 364 was always true
365 GraphicObject.objects.create(
366 flowsheet_state=simulation_object.flowsheet_state,
367 simulationObject=simulation_object,
368 width=original_graphic_object.width,
369 height=original_graphic_object.height,
370 x=original_graphic_object.x,
371 y=original_graphic_object.y,
372 group=group
373 )
375 # Creates recycle graphic objects in this group iteration if stream has a recycle connection
376 if simulation_object.has_recycle_connection:
377 # Get the recycle associated with this stream
378 recycle = simulation_object.recycleConnection
380 # Prevent duplicating the recycle graphic object in the same group
381 if recycle.simulationObject.graphicObject.filter(group=group).exists():
382 continue
384 # Set graphic objects for the recycle block in the same groups as the stream
385 default_graphic = recycle.simulationObject.graphicObject.last()
386 default_width = default_graphic.width if default_graphic else 32
387 default_height = default_graphic.height if default_graphic else 32\
389 # Make sure a stream graphic exists in this group
390 stream_graphic = simulation_object.graphicObject.filter(group=group).last()
391 if stream_graphic is None: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 continue
394 # Recreate the recycle graphic object
395 GraphicObject.objects.create(
396 flowsheet_state=simulation_object.flowsheet_state,
397 simulationObject=recycle.simulationObject,
398 width=default_width,
399 height=default_height,
400 x=stream_graphic.x + (stream_graphic.width - default_width) / 2,
401 y=stream_graphic.y + stream_graphic.height + 30,
402 group=group,
403 )