Coverage for backend/django/flowsheetInternals/graphicData/models/groupingModel.py: 78%
146 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-03-26 20:57 +0000
1from typing import List, TYPE_CHECKING
2from django.db import models
3from pydantic import BaseModel
4from django.db.models import QuerySet
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) -> QuerySet[SimulationObject]:
200 """
201 Iteratively collects all child group of the current group
202 """
203 queue = [self]
204 sim_objs = set() # Use a set to avoid duplicates
206 while queue:
207 current = queue.pop(0)
208 simulation_objects = current.get_simulation_objects()
210 sim_objs = sim_objs.union(set(simulation_objects))
212 # add current obj to sib_objs
213 sim_objs.add(current.simulationObject)
215 group_objects = simulation_objects.filter(objectType=SimulationObjectClass.Group)
216 for group_obj in group_objects:
217 queue.append(group_obj.grouping)
219 queryset = SimulationObject.objects.filter(id__in=[obj.id for obj in sim_objs])
220 return queryset
222 def get_unconnected_streams(self):
223 """
224 Returns a set of inlet streams (The inputs for the system)
225 """
226 sim_objects: QuerySet[SimulationObject] = self.get_recursive_simulation_objects()
227 streams: List[SimulationObject] = [obj for obj in sim_objects if
228 obj.objectType == SimulationObjectClass.Stream or obj.objectType == SimulationObjectClass.HumidAirStream]
230 inlet_streams = []
231 outlet_streams = []
232 for stream in streams:
233 connected_ports = stream.connectedPorts.all()
234 inlet_port = stream.connectedPorts.filter(direction=ConType.Inlet).first()
235 if len(connected_ports) == 1:
236 if connected_ports[0].direction == ConType.Inlet:
237 inlet_streams.append(stream)
238 else:
239 outlet_streams.append(stream)
240 elif inlet_port and inlet_port.unitOp.graphicObject.last().group != stream.graphicObject.last().group:
241 inlet_streams.append(stream)
243 return {'inlets': inlet_streams, 'outlets': outlet_streams}
245 def generate_name_prefix(self) -> str:
246 """
247 Generates the namePrefix for a grouping by traversing the parent hierarchy.
248 :return: A string representing the full hierarchical name
249 """
250 prefix_parts = []
251 current_group = self
253 while current_group:
254 prefix_parts.append(current_group.simulationObject.componentName)
255 graphic_obj = current_group.simulationObject.graphicObject.last()
256 current_group = graphic_obj.group
258 # Reverse the parts to get the hierarchy from root to current group
259 prefix_parts.reverse()
260 return ".".join(prefix_parts)