Coverage for backend/django/flowsheetInternals/graphicData/models/groupingModel.py: 83%
189 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from operator import concat
2from typing import Iterable, List, TYPE_CHECKING
3from django.db import models
4from pydantic import BaseModel
5from django.db.models import QuerySet
7from PinchAnalysis.models.StreamDataProject import StreamDataProject
8from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
9from core.auxiliary.enums.unitOpGraphics import ConType
10from core.auxiliary.enums import SimulationObjectClass
11from core.auxiliary.enums.generalEnums import AbstractionType
12from .graphicObjectModel import GraphicObject
14from core.managers import AccessControlManager, AllFlowsheetStatesManager
15from core.auxiliary.models.FlowsheetHistoryModel import FlowsheetHistoryModel
17if TYPE_CHECKING:
18 from core.auxiliary.models.PropertyInfo import PropertyInfo
19 from core.auxiliary.models.FlowsheetState import FlowsheetState
22class Connection(BaseModel):
23 unitOp: int # Id of unitop graphic object
24 stream: int # Id of stream graphic object
25 port: int # Port
26 direction: ConType # Direction of connection
27 anchor_placement: int | None = (
28 None # Visual anchor placement row for this rendered endpoint
29 )
30 anchor_graphic_object: int | None = (
31 None # Graphic object the anchor placement belongs to
32 )
33 anchor_endpoint_kind: str | None = None # Rendered endpoint kind for this placement
34 anchor_side: str | None = None # Logical side selected for the port visual
35 anchor_index: int | None = None # Logical anchor index selected for the port visual
38class Breadcrumbs(BaseModel):
39 groupId: int # Id of group graphic object
40 simulationObjectId: int # Id of simulation object
41 name: str
44class Grouping(FlowsheetHistoryModel, models.Model):
45 flowsheet_state = models.ForeignKey(
46 "core_auxiliary.FlowsheetState",
47 on_delete=models.CASCADE,
48 related_name="Groupings",
49 )
50 simulationObject = models.OneToOneField(
51 "flowsheetInternals_unitops.SimulationObject",
52 on_delete=models.CASCADE,
53 related_name="grouping",
54 null=True,
55 )
56 propertyInfos = models.ManyToManyField("core_auxiliary.PropertyInfo")
57 abstractionType = models.CharField(
58 choices=AbstractionType.choices, default=AbstractionType.Zone
59 )
61 created_at = models.DateTimeField(auto_now_add=True)
63 objects = AccessControlManager()
64 all_states = AllFlowsheetStatesManager()
66 # runtime-accessed relations
67 flowsheet_state: "FlowsheetState"
68 simulationObject: "SimulationObject | None"
69 propertyInfos: "models.Manager[PropertyInfo]"
70 graphicObjects: "models.Manager[GraphicObject]"
72 @classmethod
73 def create(
74 cls,
75 flowsheet_state: "FlowsheetState",
76 group: "Grouping",
77 componentName: str = "Module",
78 visible=True,
79 isRoot=False,
80 ) -> "Grouping":
81 """
82 Creates a new Grouping instance with the provided simulation object.
84 param: simulationObject - The simulation object that this grouping is associated with
85 """
86 from flowsheetInternals.unitops.models import (
87 SimulationObject,
88 ) # avoid circular import
89 from core.auxiliary.models.PropertySet import PropertySet
90 from core.auxiliary.models.ObjectTypeCounter import ObjectTypeCounter
91 from flowsheetInternals.unitops.config.config_base import configuration
93 group_graphic = configuration["group"].graphicObject
95 if componentName == "Module":
96 idx_for_type = ObjectTypeCounter.next_for(flowsheet_state, "module")
97 componentName = f"Module {idx_for_type}"
98 simulationObject = SimulationObject.objects.create(
99 flowsheet_state=flowsheet_state,
100 componentName=componentName,
101 objectType="group",
102 )
103 PropertySet.objects.create(
104 simulationObject=simulationObject,
105 flowsheet_state=flowsheet_state,
106 )
107 instance = Grouping(
108 simulationObject=simulationObject,
109 flowsheet_state=flowsheet_state,
110 )
111 GraphicObject.objects.create(
112 simulationObject=simulationObject,
113 visible=visible,
114 width=group_graphic.width,
115 height=group_graphic.height,
116 group=group,
117 flowsheet_state=flowsheet_state,
118 )
119 instance.save()
120 return instance
122 def get_parent_group(self):
123 """
124 Returns the parent group of the current group
125 """
126 graphic_obj = (
127 self.simulationObject.graphicObject.last()
128 ) # The simulation object only has one graphic object
129 return graphic_obj.group
131 def get_connections(self) -> List[Connection]:
132 """
133 Returns a list of connections that this object has to the same object in a different grouping
134 """
135 if hasattr(self, "_cached_connections"): 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 return self._cached_connections
138 graphicObjects: List[GraphicObject] = (
139 self.graphicObjects.select_related("simulationObject")
140 .prefetch_related(
141 "simulationObject__connectedPorts__unitOp",
142 "simulationObject__connectedPorts__stream",
143 )
144 .all()
145 )
146 connections = self.get_connections_for_graphics(graphicObjects)
148 self._cached_connections = connections
149 return connections
151 @staticmethod
152 def get_anchor_placements_for_graphics(graphicObjects: Iterable[GraphicObject]):
153 """
154 Return anchor placements keyed by rendered endpoint.
156 The key includes the grouping and graphic object because the same port can
157 be rendered on its owning unit operation in one layer and on a group
158 boundary in another.
159 """
160 from .portAnchorPlacementModel import PortAnchorPlacement
162 graphicObjects = list(graphicObjects)
163 graphic_object_ids = [graphic.id for graphic in graphicObjects]
164 group_ids = {
165 graphic.group_id
166 for graphic in graphicObjects
167 if graphic.group_id is not None
168 }
169 port_ids = {
170 port.id
171 for graphic in graphicObjects
172 for port in graphic.simulationObject.connectedPorts.all()
173 }
175 if not graphic_object_ids or not group_ids or not port_ids:
176 return {}
178 placements = PortAnchorPlacement.objects.filter(
179 grouping_id__in=group_ids,
180 graphicObject_id__in=graphic_object_ids,
181 port_id__in=port_ids,
182 )
183 return {
184 (
185 placement.port_id,
186 placement.grouping_id,
187 placement.graphicObject_id,
188 placement.endpointKind,
189 ): placement
190 for placement in placements
191 }
193 def get_connections_for_graphics(
194 self,
195 graphicObjects: Iterable[GraphicObject],
196 placements_by_endpoint=None,
197 ) -> List[Connection]:
198 """
199 Build rendered stream connections for a specific visible group layer.
201 This supports both the live canvas, which uses every graphic in the
202 current group, and preview surfaces, which may pass a limited subset of
203 visible graphics.
204 """
205 from .portAnchorPlacementModel import PortAnchorPlacement
207 connections: List[Connection] = []
208 graphicObjects = list(graphicObjects)
209 placements_by_endpoint = (
210 placements_by_endpoint
211 if placements_by_endpoint is not None
212 else self.get_anchor_placements_for_graphics(graphicObjects)
213 )
214 graphic_object_by_simulation_object = {
215 gobj.simulationObject_id: gobj for gobj in graphicObjects
216 }
217 visible_simulation_object_ids = set(graphic_object_by_simulation_object)
218 simulationObjects: List[SimulationObject] = [
219 gobj.simulationObject for gobj in graphicObjects
220 ]
221 sub_groups: List[Grouping] = [
222 obj.grouping
223 for obj in simulationObjects
224 if obj.objectType == SimulationObjectClass.Group
225 ]
227 def get_placement_data(
228 port, endpoint_graphic_object: GraphicObject | None, endpoint_kind: str
229 ):
230 if endpoint_graphic_object is None: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 return None, None, None, None, None
233 placement = placements_by_endpoint.get(
234 (port.id, self.id, endpoint_graphic_object.id, endpoint_kind),
235 )
236 if placement is None:
237 return None, endpoint_graphic_object.id, endpoint_kind, None, None
239 return (
240 placement.id,
241 endpoint_graphic_object.id,
242 endpoint_kind,
243 placement.anchor_side,
244 placement.anchor_index,
245 )
247 def add_connection(
248 stream: SimulationObject, port, unit_op_id: int, endpoint_kind: str
249 ):
250 (
251 anchor_placement,
252 anchor_graphic_object,
253 anchor_endpoint_kind,
254 anchor_side,
255 anchor_index,
256 ) = get_placement_data(
257 port,
258 graphic_object_by_simulation_object.get(unit_op_id),
259 endpoint_kind,
260 )
261 connections.append(
262 Connection(
263 unitOp=unit_op_id,
264 stream=stream.id,
265 port=port.id,
266 direction=port.direction,
267 anchor_placement=anchor_placement,
268 anchor_graphic_object=anchor_graphic_object,
269 anchor_endpoint_kind=anchor_endpoint_kind,
270 anchor_side=anchor_side,
271 anchor_index=anchor_index,
272 )
273 )
275 for graphicObject in graphicObjects:
276 stream: SimulationObject = graphicObject.simulationObject
277 for port in stream.connectedPorts.all():
278 connectedUnitOp = port.unitOp
279 if connectedUnitOp.id in visible_simulation_object_ids:
280 add_connection(
281 stream,
282 port,
283 connectedUnitOp.id,
284 PortAnchorPlacement.EndpointKind.UnitOperation,
285 )
286 continue
288 group = connectedUnitOp.get_group()
289 while group not in sub_groups and group is not None:
290 group = group.get_parent_group()
291 if group is not None:
292 add_connection(
293 stream,
294 port,
295 group.simulationObject.id,
296 PortAnchorPlacement.EndpointKind.GroupBoundary,
297 )
299 return connections
301 def get_breadcrumbs_trail(self) -> List[Breadcrumbs]:
302 crumbs: List[Breadcrumbs] = []
303 current_group = self
304 while current_group:
305 crumbs.append(
306 Breadcrumbs(
307 groupId=current_group.id,
308 simulationObjectId=current_group.simulationObject.id,
309 name=current_group.simulationObject.componentName,
310 )
311 )
312 graphic_obj = current_group.simulationObject.graphicObject.last()
313 current_group = graphic_obj.group
315 crumbs.reverse()
316 return crumbs
318 def update_internal_graphic_objects(self, graphicObjects) -> None:
319 """
320 Adds any new graphic objects to the grouping,
321 and removes any existing graphic objects from the grouping
322 that are not present in the input list.
323 Also updates port anchor placements for any ports connected to graphic objects within the group
324 """
325 from flowsheetInternals.unitops.services.edit_operations.recorder import (
326 tracked_queryset_update,
327 )
329 for graphicObject in graphicObjects:
330 tracked_queryset_update(
331 GraphicObject.objects.filter(pk=graphicObject.pk),
332 group=self,
333 )
334 for placement in graphicObject.portAnchorPlacements.all():
335 placement.grouping = self
336 placement.save(update_fields=["grouping"])
337 if hasattr(self, "_cached_connections"): 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 delattr(self, "_cached_connections")
340 def get_graphic_object(self) -> GraphicObject:
341 """
342 Returns the graphic object associated with this grouping.
343 """
344 return self.simulationObject.graphicObject.last()
346 def get_simulation_objects(self):
347 """
348 Returns a set of all simulation objects contained within this grouping.
349 """
350 from flowsheetInternals.unitops.models import (
351 SimulationObject,
352 ) # avoid circular import
354 return SimulationObject.objects.filter(
355 graphicObject__in=set(self.graphicObjects.all())
356 )
358 def clear_group(self) -> None:
359 """
360 Clears the grouping of all simulation objects and resets graphic object or deletes the group entirely.
361 """
362 from flowsheetInternals.unitops.models.delete_factory import DeleteFactory
364 simulationObjects = list(self.get_simulation_objects())
365 DeleteFactory.delete_multiple_objects(
366 simulationObjects + [self.simulationObject]
367 )
369 def set_group_size(self) -> None:
370 """
371 Updates the graphic object to be the size of the contained simulation objects.
372 """
374 containedObjects = self.graphicObjects.all()
375 if len(containedObjects) == 0: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 return
377 gObj = self.get_graphic_object()
378 minX = float("inf")
379 minY = float("inf")
380 maxX = float("-inf")
381 maxY = float("-inf")
382 for graphicObject in self.graphicObjects.all():
383 minX = min(minX, graphicObject.x)
384 minY = min(minY, graphicObject.y)
385 maxX = max(maxX, graphicObject.x + graphicObject.width)
386 maxY = max(maxY, graphicObject.y + graphicObject.height)
387 gObj.x = ((maxX + minX) / 2) - (graphicObject.width * 2)
388 gObj.y = minY
389 gObj.save()
391 def get_recursive_simulation_objects(self) -> QuerySet[SimulationObject]:
392 """
393 Iteratively collects all child group of the current group
394 """
395 queue = [self]
396 sim_objs = set() # Use a set to avoid duplicates
398 while queue:
399 current = queue.pop(0)
400 simulation_objects = current.get_simulation_objects()
402 sim_objs = sim_objs.union(set(simulation_objects))
404 # add current obj to sib_objs
405 sim_objs.add(current.simulationObject)
407 group_objects = simulation_objects.filter(
408 objectType=SimulationObjectClass.Group
409 )
410 for group_obj in group_objects:
411 queue.append(group_obj.grouping)
413 queryset = SimulationObject.objects.filter(id__in=[obj.id for obj in sim_objs])
414 return queryset
416 def get_unconnected_streams(self):
417 """
418 Returns a set of inlet streams (The inputs for the system)
419 """
420 sim_objects: QuerySet[SimulationObject] = (
421 self.get_recursive_simulation_objects()
422 )
423 streams: List[SimulationObject] = [
424 obj
425 for obj in sim_objects
426 if obj.objectType == SimulationObjectClass.Stream
427 or obj.objectType == SimulationObjectClass.HumidAirStream
428 ]
430 inlet_streams = []
431 outlet_streams = []
432 for stream in streams:
433 connected_ports = stream.connectedPorts.all()
434 inlet_port = stream.connectedPorts.filter(direction=ConType.Inlet).first()
435 if len(connected_ports) == 1:
436 if connected_ports[0].direction == ConType.Inlet:
437 inlet_streams.append(stream)
438 else:
439 outlet_streams.append(stream)
440 elif (
441 inlet_port
442 and inlet_port.unitOp.graphicObject.last().group
443 != stream.graphicObject.last().group
444 ):
445 inlet_streams.append(stream)
447 return {"inlets": inlet_streams, "outlets": outlet_streams}
449 def generate_name_prefix(self) -> str:
450 """
451 Generates the namePrefix for a grouping by traversing the parent hierarchy.
452 :return: A string representing the full hierarchical name
453 """
454 prefix_parts = []
455 current_group = self
457 while current_group:
458 prefix_parts.append(current_group.simulationObject.componentName)
459 graphic_obj = current_group.simulationObject.graphicObject.last()
460 current_group = graphic_obj.group
462 # Reverse the parts to get the hierarchy from root to current group
463 prefix_parts.reverse()
464 return ".".join(prefix_parts)