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