Coverage for backend/django/flowsheetInternals/graphicData/services/auto_sort.py: 89%
117 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from dataclasses import dataclass
2import shlex
3from typing import Iterable
4from django.db import transaction
5from django.db.models import Q
6import pydot
8from core.auxiliary.enums import ConType
9from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
10from flowsheetInternals.unitops.models.Port import Port
11from flowsheetInternals.unitops.services.edit_operations.recorder import (
12 tracked_bulk_update,
13)
16GRAPHVIZ_SCALE = 40
17RECYCLE_VERTICAL_OFFSET = 80
18RECYCLE_STACK_OFFSET = 40
21@dataclass(frozen=True)
22class Position:
23 x: float
24 y: float
27def _graphic_sort_key(graphic_object: GraphicObject) -> tuple[float, float, int]:
28 return (
29 float(graphic_object.y or 0),
30 float(graphic_object.x or 0),
31 graphic_object.simulationObject_id or 0,
32 )
35def _get_recycle_tear_object_id(graphic_object: GraphicObject) -> int | None:
36 simulation_object = getattr(graphic_object, "simulationObject", None)
37 if getattr(simulation_object, "objectType", None) != "recycle":
38 return None
40 try:
41 recycle_data = simulation_object.recycleData
42 except AttributeError:
43 return None
45 return getattr(recycle_data, "tearObject_id", None)
48def _split_recycles(
49 graphic_objects: list[GraphicObject],
50) -> tuple[list[GraphicObject], dict[int, list[int]]]:
51 """
52 Connected recycle blocks are annotations on a tear stream, so place them
53 relative to that stream instead of letting them influence the graph layout.
54 """
55 visible_object_ids = {graphic.simulationObject_id for graphic in graphic_objects}
56 graphviz_graphics = []
57 recycle_ids_by_tear_object_id = {}
59 for graphic in graphic_objects:
60 tear_object_id = _get_recycle_tear_object_id(graphic)
61 if tear_object_id is None or tear_object_id not in visible_object_ids:
62 graphviz_graphics.append(graphic)
63 continue
65 recycle_ids_by_tear_object_id.setdefault(tear_object_id, []).append(
66 graphic.simulationObject_id
67 )
69 return graphviz_graphics, recycle_ids_by_tear_object_id
72def _iter_connection_edges(
73 ports: Iterable[Port], visible_object_ids: set[int]
74) -> Iterable[tuple[int, int]]:
75 for port in ports:
76 if (
77 port.unitOp_id not in visible_object_ids
78 or port.stream_id not in visible_object_ids
79 ):
80 continue
82 if port.direction == ConType.Outlet:
83 yield port.unitOp_id, port.stream_id
84 elif port.direction == ConType.Inlet: 84 ↛ 75line 84 didn't jump to line 75 because the condition on line 84 was always true
85 yield port.stream_id, port.unitOp_id
88def _build_graph(
89 graphic_objects: list[GraphicObject],
90 ports: Iterable[Port],
91) -> pydot.Dot:
92 visible_object_ids = {graphic.simulationObject_id for graphic in graphic_objects}
93 graph = pydot.Dot(
94 graph_type="digraph",
95 rankdir="LR",
96 ranksep="2.0 equally",
97 nodesep="2.0",
98 )
99 graph.set_node_defaults(
100 shape="box",
101 fixedsize="true",
102 width="1.5",
103 height="1.0",
104 )
105 graph.set_edge_defaults(weight="2")
107 for sort_index, graphic in enumerate(graphic_objects):
108 object_id = str(graphic.simulationObject_id)
109 graph.add_node(
110 pydot.Node(
111 object_id,
112 label=object_id,
113 sortv=str(sort_index),
114 )
115 )
117 for source, target in _iter_connection_edges(ports, visible_object_ids):
118 graph.add_edge(pydot.Edge(str(source), str(target)))
120 return graph
123def _parse_plain_positions(plain_output: str) -> dict[int, Position]:
124 positions = {}
125 for line in plain_output.splitlines():
126 parts = shlex.split(line)
127 if not parts or parts[0] != "node":
128 continue
130 object_id = int(parts[1])
131 positions[object_id] = Position(
132 x=float(parts[2]) * GRAPHVIZ_SCALE,
133 y=float(parts[3]) * GRAPHVIZ_SCALE,
134 )
136 return positions
139def _layout_graph(graph: pydot.Dot) -> dict[int, Position]:
140 plain_output = graph.create(format="plain", prog="dot").decode("utf-8")
141 return _parse_plain_positions(plain_output)
144def compute_auto_sort_positions(
145 graphic_objects: Iterable[GraphicObject],
146 ports: Iterable[Port],
147) -> dict[int, Position]:
148 sorted_graphics = sorted(
149 [
150 graphic
151 for graphic in graphic_objects
152 if graphic.simulationObject_id is not None
153 ],
154 key=_graphic_sort_key,
155 )
156 if not sorted_graphics: 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true
157 return {}
159 graphviz_graphics, recycle_ids_by_tear_object_id = _split_recycles(sorted_graphics)
160 if not graphviz_graphics: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 return {}
163 graphviz_positions = _layout_graph(_build_graph(graphviz_graphics, ports))
164 if not graphviz_positions: 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 return {}
167 min_graphviz_x = min(position.x for position in graphviz_positions.values())
168 max_graphviz_y = max(position.y for position in graphviz_positions.values())
169 anchor_x = min(float(graphic.x or 0) for graphic in sorted_graphics)
170 anchor_y = min(float(graphic.y or 0) for graphic in sorted_graphics)
172 positions = {
173 object_id: Position(
174 x=round(anchor_x + position.x - min_graphviz_x, 2),
175 y=round(anchor_y + max_graphviz_y - position.y, 2),
176 )
177 for object_id, position in graphviz_positions.items()
178 }
180 for tear_object_id, recycle_object_ids in recycle_ids_by_tear_object_id.items():
181 tear_position = positions.get(tear_object_id)
182 if tear_position is None: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 continue
185 for stack_index, recycle_object_id in enumerate(sorted(recycle_object_ids)):
186 positions[recycle_object_id] = Position(
187 x=tear_position.x,
188 y=tear_position.y - RECYCLE_VERTICAL_OFFSET - stack_index * RECYCLE_STACK_OFFSET,
189 )
191 return positions
194def auto_sort(flowsheet_id: int, group_id: int):
195 graphic_objects = list(
196 GraphicObject.objects.filter(
197 group_id=group_id,
198 visible=True,
199 simulationObject__is_deleted=False,
200 )
201 .select_related("simulationObject", "simulationObject__recycleData")
202 .only(
203 "id",
204 "x",
205 "y",
206 "group_id",
207 "simulationObject_id",
208 "simulationObject__id",
209 "simulationObject__objectType",
210 "simulationObject__recycleData__id",
211 "simulationObject__recycleData__tearObject_id",
212 )
213 )
214 object_ids = {
215 graphic_object.simulationObject_id
216 for graphic_object in graphic_objects
217 if graphic_object.simulationObject_id is not None
218 }
220 ports = Port.objects.none()
221 if object_ids: 221 ↛ 227line 221 didn't jump to line 227 because the condition on line 221 was always true
222 ports = Port.objects.filter(
223 Q(unitOp_id__in=object_ids) | Q(stream_id__in=object_ids),
224 flowsheet_state__flowsheet_id=flowsheet_id,
225 ).only("id", "direction", "unitOp_id", "stream_id")
227 next_positions = compute_auto_sort_positions(graphic_objects, ports)
228 changed_graphics = []
229 moved_objects = []
231 for graphic_object in graphic_objects:
232 object_id = graphic_object.simulationObject_id
233 if object_id is None or object_id not in next_positions: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 continue
236 old_position = {
237 "x": float(graphic_object.x or 0),
238 "y": float(graphic_object.y or 0),
239 }
240 new_position = {
241 "x": next_positions[object_id].x,
242 "y": next_positions[object_id].y,
243 }
244 if old_position == new_position: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 continue
247 graphic_object.x = new_position["x"]
248 graphic_object.y = new_position["y"]
249 changed_graphics.append(graphic_object)
250 moved_objects.append(
251 {
252 "objectId": object_id,
253 "graphicObjectId": graphic_object.id,
254 "oldPosition": old_position,
255 "newPosition": new_position,
256 }
257 )
259 with transaction.atomic():
260 if changed_graphics: 260 ↛ 267line 260 didn't jump to line 267
261 tracked_bulk_update(
262 GraphicObject.objects,
263 changed_graphics,
264 ["x", "y"],
265 )
267 return moved_objects