Coverage for backend/flowsheetInternals/graphicData/models/groupingModel.py: 59%
144 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-11-06 23:27 +0000
1from typing import List
2from django.db import models
3from pydantic import BaseModel
4from django.db.models import Prefetch
6from PinchAnalysis.models.StreamDataProject import StreamDataProject
7from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
8from core.auxiliary.enums.unitOpGraphics import ConType
9from core.auxiliary.enums.unitOpData import SimulationObjectClass
10from core.auxiliary.enums.generalEnums import AbstractionType
11from .graphicObjectModel import GraphicObject
14from core.managers import AccessControlManager
15class Connection(BaseModel):
16 unitOp: int # Id of unitop graphic object
17 stream: int # Id of stream graphic object
18 port: int # Port
19 direction: ConType # Direction of connection
21class Breadcrumbs(BaseModel):
22 groupId: int # Id of group graphic object
23 simulationObjectId: int # Id of simulation object
24 name: str
26class Grouping(models.Model):
27 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="Groupings")
28 simulationObject = models.OneToOneField("flowsheetInternals_unitops.SimulationObject", on_delete=models.CASCADE, related_name="grouping", null=True)
29 propertyInfos = models.ManyToManyField("core_auxiliary.PropertyInfo")
30 abstractionType = models.CharField(choices=AbstractionType.choices, default=AbstractionType.Zone)
32 created_at = models.DateTimeField(auto_now_add=True)
34 objects = AccessControlManager()
37 @classmethod
38 def create(cls, flowsheet: "core_auxiliary.Flowsheet", group: "Grouping", componentName: str = "Module", visible=True, isRoot=False) -> "Grouping":
39 """
40 Creates a new Grouping instance with the provided simulation object.
42 param: simulationObject - The simulation object that this grouping is associated with
43 """
44 from flowsheetInternals.unitops.models import SimulationObject # avoid circular import
45 from core.auxiliary.models.PropertySet import PropertySet
46 from core.auxiliary.models.ObjectTypeCounter import ObjectTypeCounter
47 from flowsheetInternals.unitops.config.objects import group_config
48 group_graphic = group_config.graphicObject
50 if componentName == "Module":
51 idx_for_type = ObjectTypeCounter.next_for(flowsheet, "module")
52 componentName = f"Module {idx_for_type}"
53 simulationObject = SimulationObject.objects.create(flowsheet=flowsheet, componentName=componentName, objectType="group" )
54 PropertySet.objects.create(simulationObject=simulationObject, flowsheet=flowsheet)
55 instance = Grouping(simulationObject=simulationObject, flowsheet=flowsheet)
56 GraphicObject.objects.create(simulationObject=simulationObject, visible=visible, width=group_graphic.width, height=group_graphic.height, group=group, flowsheet=flowsheet)
57 instance.save()
58 return instance
60 def get_parent_group(self):
61 """
62 Returns the parent group of the current group
63 """
64 graphic_obj = self.simulationObject.graphicObject.last() # The simulation object only has one graphic object
65 return graphic_obj.group
67 def get_connections(self) -> List[Connection]:
68 """
69 Returns a list of connections that this object has to the same object in a different grouping
70 """
71 if hasattr(self, '_cached_connections'): 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 return self._cached_connections
74 connections: List[Connection] = []
76 graphicObjects: List[GraphicObject] = self.graphicObjects.select_related("simulationObject").prefetch_related("simulationObject__connectedPorts__unitOp", "simulationObject__connectedPorts__stream").all()
77 simulationObjects: List[SimulationObject] = [gobj.simulationObject for gobj in graphicObjects]
78 sub_groups: List[Grouping] = [obj.grouping for obj in simulationObjects if obj.objectType == SimulationObjectClass.Group]
80 # Iterate through all graphic object in the group and their associated simulationObject
81 for graphicObject in graphicObjects:
82 stream: SimulationObject = graphicObject.simulationObject
83 for port in stream.connectedPorts.all():
84 # if it's not connected to a simulation object in this group level:
85 connectedUnitOp = port.unitOp
86 if connectedUnitOp in simulationObjects: 86 ↛ 88line 86 didn't jump to line 88 because the condition on line 86 was never true
87 # No problems, both the stream and the unit op are visible in this group level
88 connections.append(Connection(
89 unitOp=connectedUnitOp.id,
90 stream=stream.id,
91 port=port.id,
92 direction=port.direction,
93 ))
94 else:
95 # This is connnected to something inside a subgroup.
96 # Find the unit ops' group:
97 group = connectedUnitOp.get_group()
98 # Iterate up until we find a sub group
99 while group not in sub_groups and group is not None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 group = group.get_parent_group()
101 if group is not None: 101 ↛ 83line 101 didn't jump to line 83 because the condition on line 101 was always true
102 # add connection to the groups graphicobject
103 unitOpId = group.simulationObject.id
104 connections.append(Connection(
105 unitOp=unitOpId,
106 stream=stream.id,
107 port=port.id,
108 direction=port.direction,
109 ))
111 self._cached_connections = connections
112 return connections
114 def get_breadcrumbs_trail(self) -> List[Breadcrumbs]:
115 crumbs: List[Breadcrumbs] = []
116 current_group = self
117 while current_group:
118 crumbs.append(Breadcrumbs(
119 groupId=current_group.id,
120 simulationObjectId=current_group.simulationObject.id,
121 name=current_group.simulationObject.componentName,
122 ))
123 graphic_obj = current_group.simulationObject.graphicObject.last()
124 current_group = graphic_obj.group
126 crumbs.reverse()
127 return crumbs
130 def update_internal_simulation_objects(self, simulationObjects) -> None:
131 """
132 Adds any new simulation objects to the grouping,
133 and removes any existing simulation objects from the grouping
134 that are not present in the input list.
135 """
136 graphicObjects = [simObj.graphicObject.first() for simObj in simulationObjects]
137 self.graphicObjects.set(graphicObjects)
139 def get_graphic_object(self) -> GraphicObject:
140 """
141 Returns the graphic object associated with this grouping.
142 """
143 return self.simulationObject.graphicObject.last()
145 def get_simulation_objects(self):
146 """
147 Returns a set of all simulation objects contained within this grouping.
148 """
149 from flowsheetInternals.unitops.models import SimulationObject # avoid circular import
150 return SimulationObject.objects.filter(graphicObject__in=set(self.graphicObjects.all()))
152 def clear_group(self) -> None:
153 """
154 Clears the grouping of all simulation objects and resets graphic object or deletes the group entirely.
155 """
156 from flowsheetInternals.unitops.models.delete_factory import DeleteFactory
157 simulationObjects = list(self.get_simulation_objects())
158 DeleteFactory.delete_multiple_objects(simulationObjects + [self.simulationObject])
160 def set_group_size(self) -> None:
161 """
162 Updates the graphic object to be the size of the contained simulation objects.
163 """
165 containedObjects = self.graphicObjects.all()
166 if len(containedObjects) == 0: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 return
168 gObj = self.get_graphic_object()
169 minX = float("inf")
170 minY = float("inf")
171 maxX = float("-inf")
172 maxY = float("-inf")
173 for graphicObject in self.graphicObjects.all():
174 minX = min(minX, graphicObject.x)
175 minY = min(minY, graphicObject.y)
176 maxX = max(maxX, graphicObject.x + graphicObject.width)
177 maxY = max(maxY, graphicObject.y + graphicObject.height)
178 gObj.x = ((maxX + minX) / 2) - (graphicObject.width * 2)
179 gObj.y = minY
180 gObj.save()
182 def get_recursive_simulation_objects(self) -> set:
183 """
184 Iteratively collects all child group of the current group
185 """
186 queue = [self]
187 all_objects = set()
189 while queue:
190 group = queue.pop(0)
191 group_objects = self.get_simulation_objects()
192 for obj in group_objects:
193 if obj.objectType == obj.objectType == SimulationObjectClass.Group:
194 try:
195 queue.append(obj.grouping)
196 except obj._meta.model.grouping.RelatedObjectDoesNotExist:
197 continue
198 else:
199 all_objects.add(obj)
201 return all_objects
203 def get_unconnected_streams(self):
204 """
205 Returns a set of inlet streams (The inputs for the system)
206 """
207 from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
208 sim_objects: SimulationObject = self.get_recursive_simulation_objects()
209 streams: List[SimulationObject] = [obj for obj in sim_objects if obj.objectType == SimulationObjectClass.Stream or obj.objectType == SimulationObjectClass.HumidAirStream]
211 inlet_streams = []
212 outlet_streams = []
213 for stream in streams:
214 connected_ports = stream.connectedPorts.all()
215 inlet_port = stream.connectedPorts.filter(direction=ConType.Inlet).first()
216 if len(connected_ports) == 1:
217 if connected_ports[0].direction == ConType.Inlet:
218 inlet_streams.append(stream)
219 else:
220 outlet_streams.append(stream)
221 elif inlet_port and inlet_port.unitOp.graphicObject.last().group != stream.graphicObject.last().group:
222 inlet_streams.append(stream)
224 return {'inlets': inlet_streams, 'outlets':outlet_streams}
226 def generate_name_prefix(self) -> str:
227 """
228 Generates the namePrefix for a grouping by traversing the parent hierarchy.
229 :return: A string representing the full hierarchical name
230 """
231 prefix_parts = []
232 current_group = self
234 while current_group:
235 prefix_parts.append(current_group.simulationObject.componentName)
236 graphic_obj = current_group.simulationObject.graphicObject.last()
237 current_group = graphic_obj.group
239 # Reverse the parts to get the hierarchy from root to current group
240 prefix_parts.reverse()
241 return ".".join(prefix_parts)