Coverage for backend/django/flowsheetInternals/unitops/viewsets/DuplicateSimulationObject.py: 87%
162 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.db import transaction
2from django.db.models import Prefetch
3from core.auxiliary.enums import SimulationObjectClass
4from core.auxiliary.methods.copy_object.formula_remapping import update_formulas
5from core.auxiliary.methods.copy_object.model_lookup import ModelLookup, ModelLookupDict
6from core.auxiliary.methods.copy_object import CopyObject, RemapOrRetainMode
7from core.auxiliary.models.PropertySet import PropertySet
8from core.auxiliary.models.PropertyInfo import PropertyInfo
9from core.auxiliary.models.PropertyValue import PropertyValue, PropertyValueIntermediate
10from core.auxiliary.models.ControlValue import ControlValue
11from core.auxiliary.models.IndexedItem import IndexedItem
12from core.auxiliary.models.RecycleData import RecycleData
13from flowsheetInternals.unitops.models.Port import Port
14from flowsheetInternals.graphicData.models.groupingModel import Grouping
15from flowsheetInternals.graphicData.models.graphicObjectModel import GraphicObject
16from flowsheetInternals.unitops.models import SimulationObject
17from flowsheetInternals.unitops.services.edit_operations.recorder import (
18 tracked_bulk_create,
19 tracked_bulk_update,
20)
22class Coords:
23 def __init__(self, x, y):
24 self.x = x
25 self.y = y
28def calc_centre_simulation_objects(simulation_object_collection):
29 from django.db.models import Min, Max, F, ExpressionWrapper, FloatField
30 aggregated = GraphicObject.objects.filter(
31 simulationObject__in=simulation_object_collection
32 ).aggregate(
33 min_x=Min('x'),
34 max_edge_x=Max(ExpressionWrapper(F('x') + F('width'), output_field=FloatField())),
35 min_y=Min('y'),
36 max_edge_y=Max(ExpressionWrapper(F('y') + F('height'), output_field=FloatField()))
37 )
38 centre_x = (aggregated['min_x'] + aggregated['max_edge_x']) / 2
39 centre_y = (aggregated['min_y'] + aggregated['max_edge_y']) / 2
40 return Coords(centre_x, centre_y)
43class SimulationObjectDuplicator:
44 """
45 `CopyObject` handles the plain row duplication for the selected subset.
47 The remaining methods here handle behaviours that are more specific to
48 simulation-object copy/paste and therefore do not belong in the generic
49 object copier:
50 - rename copied simulation objects
51 - offset copied graphics to the requested drop point
52 - rebuild M2M and through-model relationships
53 - copy controls only when both ends are part of the copied subset
54 """
56 def build_copy_spec(self, original_simulation_objects):
57 return CopyObject(
58 SimulationObject,
59 queryset=original_simulation_objects,
60 copy_relation_to={
61 "properties": CopyObject(
62 PropertySet,
63 copy_relation_to={
64 "ContainedProperties": CopyObject(
65 PropertyInfo,
66 copy_relation_to={
67 "values": CopyObject(
68 PropertyValue,
69 )
70 },
71 )
72 },
73 ),
74 "graphicObject": CopyObject(
75 GraphicObject,
76 forward_relation_policy={"group": RemapOrRetainMode.COPY_IF_SELECTED_ELSE_KEEP},
77 ),
78 "ports": CopyObject(
79 Port,
80 forward_relation_policy={"stream": RemapOrRetainMode.COPY_IF_SELECTED_ELSE_KEEP},
81 ),
82 "grouping": CopyObject(
83 Grouping,
84 ),
85 "recycleData": CopyObject(
86 RecycleData,
87 forward_relation_policy={"tearObject": RemapOrRetainMode.COPY_IF_SELECTED_ELSE_NULL},
88 ),
89 },
90 )
92 def duplicate(self, original_simulation_objects) -> ModelLookupDict:
93 model_lookups: ModelLookupDict = {}
94 self.build_copy_spec(original_simulation_objects).copy(model_lookups)
95 return model_lookups
97 def update_simulation_object_names(self, model_lookups: ModelLookupDict) -> None:
98 simulation_objects = list(model_lookups.get(SimulationObject, ModelLookup([])))
99 for simulation_object in simulation_objects:
100 simulation_object.componentName = simulation_object.componentName + " copy"
101 if simulation_objects: 101 ↛ exitline 101 didn't return from function 'update_simulation_object_names' because the condition on line 101 was always true
102 tracked_bulk_update(
103 SimulationObject.objects,
104 simulation_objects,
105 ["componentName"],
106 )
108 def update_graphic_positions(self, model_lookups: ModelLookupDict, delta: Coords) -> None:
109 graphic_objects = list(model_lookups.get(GraphicObject, ModelLookup([])))
110 for graphic_object in graphic_objects:
111 graphic_object.x += delta.x
112 graphic_object.y += delta.y
113 if graphic_objects: 113 ↛ exitline 113 didn't return from function 'update_graphic_positions' because the condition on line 113 was always true
114 tracked_bulk_update(
115 GraphicObject.objects,
116 graphic_objects,
117 ["x", "y"],
118 )
120 def update_grouping_property_infos(self, model_lookups: ModelLookupDict) -> None:
121 grouping_lookup = model_lookups.get(Grouping)
122 property_info_lookup = model_lookups.get(PropertyInfo)
123 if grouping_lookup is None:
124 return
126 for original_grouping_pk, copied_grouping in grouping_lookup.model_map.items():
127 original_grouping = Grouping.objects.get(pk=original_grouping_pk)
128 copied_property_infos = []
129 for original_property_info in original_grouping.propertyInfos.all(): 129 ↛ 130line 129 didn't jump to line 130 because the loop on line 129 never started
130 if property_info_lookup is None:
131 copied_property_infos.append(original_property_info)
132 continue
133 copied_property_info = property_info_lookup.get_model(original_property_info.pk)
134 copied_property_infos.append(copied_property_info or original_property_info)
136 if copied_property_infos: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 copied_grouping.propertyInfos.set(copied_property_infos)
139 def duplicate_control_values(self, model_lookups: ModelLookupDict) -> None:
140 property_value_lookup = model_lookups.get(PropertyValue)
141 if property_value_lookup is None:
142 return
144 original_value_ids = list(property_value_lookup.model_map.keys())
145 original_control_values = ControlValue.objects.filter(
146 setPoint_id__in=original_value_ids
147 ).select_related("manipulated", "setPoint")
149 new_control_values = []
150 for original_control_value in original_control_values:
151 new_setpoint = property_value_lookup.get_model(original_control_value.setPoint_id)
152 new_manipulated = property_value_lookup.get_model(original_control_value.manipulated_id)
153 if new_setpoint is None or new_manipulated is None: 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 continue
155 new_control_values.append(
156 ControlValue(
157 setPoint=new_setpoint,
158 manipulated=new_manipulated,
159 flowsheet_state=new_setpoint.flowsheet_state,
160 )
161 )
163 if new_control_values:
164 tracked_bulk_create(ControlValue.objects, new_control_values)
166 def duplicate_indexed_items(self, model_lookups: ModelLookupDict) -> None:
167 property_value_lookup = model_lookups.get(PropertyValue)
168 simulation_object_lookup = model_lookups.get(SimulationObject)
169 if property_value_lookup is None:
170 return
172 original_property_values = list(
173 PropertyValue.objects
174 .filter(pk__in=property_value_lookup.model_map.keys())
175 .prefetch_related("indexedItems")
176 )
178 original_indexed_item_map = {}
179 for original_property_value in original_property_values:
180 original_indexed_item_map[original_property_value.pk] = [
181 indexed_item.pk for indexed_item in original_property_value.indexedItems.all()
182 ]
184 original_indexed_ids = {
185 pk for id_list in original_indexed_item_map.values() for pk in id_list
186 }
187 if not original_indexed_ids:
188 return
190 indexed_item_lookup = ModelLookup([])
191 new_indexed_items = []
192 original_indexed_items = IndexedItem.objects.filter(pk__in=original_indexed_ids).select_related("owner")
193 for original_indexed_item in original_indexed_items:
194 new_owner = None
195 if original_indexed_item.owner_id is not None and simulation_object_lookup is not None: 195 ↛ 197line 195 didn't jump to line 197 because the condition on line 195 was always true
196 new_owner = simulation_object_lookup.get_model(original_indexed_item.owner_id)
197 if new_owner is None: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 new_owner = original_indexed_item.owner
200 new_indexed_item = IndexedItem(
201 owner=new_owner,
202 key=original_indexed_item.key,
203 displayName=original_indexed_item.displayName,
204 type=original_indexed_item.type,
205 flowsheet_state=original_indexed_item.flowsheet_state,
206 )
207 new_indexed_item.save()
208 indexed_item_lookup.model_map[original_indexed_item.pk] = new_indexed_item
209 new_indexed_items.append(new_indexed_item)
211 all_intermediates = []
212 for original_property_value in original_property_values:
213 copied_property_value = property_value_lookup.get_model(original_property_value.pk)
214 for indexed_item_pk in original_indexed_item_map[original_property_value.pk]:
215 all_intermediates.append(
216 PropertyValueIntermediate(
217 propertyvalue_id=copied_property_value.pk,
218 indexeditem_id=indexed_item_lookup.get_model(indexed_item_pk).pk,
219 )
220 )
221 if all_intermediates: 221 ↛ exitline 221 didn't return from function 'duplicate_indexed_items' because the condition on line 221 was always true
222 tracked_bulk_create(
223 PropertyValueIntermediate.objects,
224 all_intermediates,
225 )
227 def apply_recycle_updates(self, model_lookups: ModelLookupDict) -> None:
228 recycle_lookup = model_lookups.get(RecycleData)
229 if recycle_lookup is None:
230 return
232 for copied_recycle_data in recycle_lookup:
233 if copied_recycle_data.tearObject_id is not None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 copied_recycle_data.update(copied_recycle_data.tearObject)
237class DuplicateSimulationObject:
238 def handle_duplication_request(self, flowsheet: int, validated_data):
239 object_ids = validated_data.get('objectIDs') or []
240 if not object_ids: 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true
241 return
243 with transaction.atomic():
244 expanded_ids = self._expand_object_ids(object_ids)
245 if not expanded_ids: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 return
248 original_simulation_objects = list(
249 SimulationObject.objects
250 .filter(pk__in=expanded_ids)
251 .select_related('flowsheet_state', 'grouping', 'recycleData')
252 .prefetch_related(
253 'graphicObject',
254 'ports',
255 'properties__ContainedProperties__values__indexedItems',
256 'graphicObject',
257 Prefetch('grouping__graphicObjects', queryset=GraphicObject.objects.select_related('simulationObject')),
258 'grouping__propertyInfos',
259 )
260 )
262 if not original_simulation_objects: 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true
263 return
265 # calculate the centre of the original simulation objects
266 old_centre = calc_centre_simulation_objects(original_simulation_objects)
267 # calculate the new centre of the duplicated simulation objects
268 new_centre = Coords(validated_data.get('x'), validated_data.get('y'))
269 delta = Coords(new_centre.x - old_centre.x, new_centre.y - old_centre.y)
271 duplicator = SimulationObjectDuplicator()
272 model_lookups = duplicator.duplicate(original_simulation_objects)
273 duplicator.update_simulation_object_names(model_lookups)
274 duplicator.update_graphic_positions(model_lookups, delta)
275 duplicator.update_grouping_property_infos(model_lookups)
276 duplicator.duplicate_control_values(model_lookups)
277 duplicator.duplicate_indexed_items(model_lookups)
278 update_formulas(model_lookups, allow_not_found=True)
279 duplicator.apply_recycle_updates(model_lookups)
280 return model_lookups
282 def _expand_object_ids(self, object_ids):
283 """Recursively collect all simulation objects contained within selected groups."""
284 if not object_ids: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 return set()
287 discovered = set()
288 queue = set(object_ids)
290 while queue:
291 batch_ids = list(queue)
292 queue.clear()
293 queryset = (
294 SimulationObject.objects
295 .filter(pk__in=batch_ids)
296 .select_related('grouping')
297 .prefetch_related(
298 Prefetch(
299 'grouping__graphicObjects',
300 queryset=GraphicObject.objects.select_related('simulationObject')
301 )
302 )
303 )
305 for simulation_object in queryset:
306 if simulation_object.pk in discovered: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 continue
309 discovered.add(simulation_object.pk)
311 grouping = getattr(simulation_object, 'grouping', None)
312 if simulation_object.objectType == SimulationObjectClass.Group and grouping is not None:
313 for graphic_object in grouping.graphicObjects.all():
314 child = graphic_object.simulationObject
315 if child and child.pk not in discovered:
316 queue.add(child.pk)
318 return discovered