Coverage for backend/django/core/auxiliary/methods/CopyFlowsheet.py: 89%
73 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-05-13 02:47 +0000
1from django.db import transaction
2from authentication.user.models import User
3from core.auxiliary.enums import ConType
4from core.auxiliary.models.Flowsheet import Flowsheet
5from flowsheetInternals.graphicData.models.groupingModel import Grouping
6from flowsheetInternals.graphicData.logic.make_group import propagate_streams, propagate_intermediate_streams
7from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
8from django.utils import timezone
9from authentication.user.AccessTable import AccessTable
10from core.validation import flowsheet_ctx
11from .copy_flowsheet.copy_caching import ModelLookup
12from .copy_flowsheet.copy_primary_keys import update_primary_keys
13from .copy_flowsheet.copy_foreign_keys import create_foreign_key_lookups, create_foreign_key_lookups_for_modules, update_foreign_keys
14from .copy_flowsheet.copy_many_to_many import create_many_to_many_lookups, update_many_to_many_relationships
15from .copy_flowsheet.copy_formulas import update_formulas
18def copy_flowsheet_data(flowsheet: Flowsheet, user: User):
19 try:
20 with transaction.atomic():
22 model_lookups = create_foreign_key_lookups(flowsheet)
23 # Manually add the flowsheet to the model lookups.
24 new_flowsheet = Flowsheet(
25 pk=flowsheet.pk, # We need the old primary key for now, so the ModelLookup can use it. update_primary_keys will set it later.
26 name=f"{flowsheet.name} (copy)"[:32], # Ensure name is not too long,
27 owner=user,
28 created_at=flowsheet.created_at,
29 rootGrouping=flowsheet.rootGrouping, # This will be updated later to be the cloned version, because it doesn't exist yet.
30 savedDate= timezone.now(),
31 )
32 model_lookups[Flowsheet] = ModelLookup([new_flowsheet])
33 many_to_many_lookup = create_many_to_many_lookups(model_lookups)
35 # Normally, bulk_create will automatically set the flowsheet ID, because of
36 # our access control Manager.
37 # So we need to escape that behavior to allow us to set the flowsheet ID manually.
39 # Update everything to point to the new flowsheet.
40 update_primary_keys(model_lookups)
42 new_flowsheet.save()
44 AccessTable.objects.create(
45 user=user,
46 flowsheet=new_flowsheet,
47 read_only=False,
48 )
49 update_foreign_keys(model_lookups)
50 update_many_to_many_relationships(many_to_many_lookup, model_lookups)
51 update_formulas(model_lookups)
53 root_group_pk = new_flowsheet.rootGrouping.pk
54 new_flowsheet.rootGrouping = model_lookups[Grouping].get_model(root_group_pk)
55 new_flowsheet.created_at = timezone.now()
56 new_flowsheet.save()
57 return new_flowsheet
58 # Bulk update the models with the new foreign keys.
59 # for Model_Type, models in model_lookups.items():
60 except Exception as e:
61 import traceback
62 print(f"Error copying flowsheet: {e}")
63 print(traceback.format_exc())
64 raise Exception(
65 f"Error copying flowsheet: {e} " + traceback.format_exc()
66 ) # Re-raise the exception to be handled by the caller.
70def create_module_from_template_logic(template_flowsheet: Flowsheet, target_flowsheet: Flowsheet,
71 user, current_group: Grouping, x=None, y=None) -> Grouping:
72 """
73 Create a module (grouping) from a template flowsheet within an existing flowsheet
74 """
75 with transaction.atomic():
77 # This maps old PKs to new model instances
78 model_lookups = create_foreign_key_lookups_for_modules(template_flowsheet)
79 many_to_many_lookup = create_many_to_many_lookups(model_lookups)
81 # Generate new primary keys for all copied models
82 # Sets pk=None so Django creates new PKs when saving
83 update_primary_keys(model_lookups)
85 # Map template flowsheet to target flowsheet
86 # All copied models will now reference the target flowsheet instead of template
87 model_lookups[Flowsheet] = ModelLookup([])
88 model_lookups[Flowsheet].model_map[template_flowsheet.pk] = target_flowsheet
90 update_foreign_keys(model_lookups)
91 update_many_to_many_relationships(many_to_many_lookup, model_lookups)
92 update_formulas(model_lookups)
94 # Find the copied root group from the template
95 template_root_group_pk = template_flowsheet.rootGrouping.pk
96 module_root_group: Grouping = model_lookups[Grouping].get_model(template_root_group_pk)
98 # Set the copied root group's parent to current group
99 # This "inserts" the template module into the target flowsheet hierarchy
100 module_graphic_object = module_root_group.get_graphic_object()
101 module_graphic_object.group = current_group
102 if x is not None: 102 ↛ 104line 102 didn't jump to line 104 because the condition on line 102 was always true
103 module_graphic_object.x = x
104 if y is not None: 104 ↛ 106line 104 didn't jump to line 106 because the condition on line 104 was always true
105 module_graphic_object.y = y
106 module_graphic_object.save()
108 # Get all copied simulation objects
109 copied_simulation_objects = list(model_lookups[SimulationObject])
110 contained_objects_id = [sim_obj.pk for sim_obj in copied_simulation_objects]
112 # Handle streams for all copied simulation objects
113 inlet_streams = []
114 outlet_streams = []
115 for simulation_object in copied_simulation_objects:
116 ports_mgr = simulation_object.connectedPorts # RelatedManager
118 # Propagate streams connected to one port
119 if simulation_object.is_stream() and ports_mgr.count() == 1:
120 port = ports_mgr.first()
121 if port and port.direction == ConType.Inlet:
122 inlet_streams.append(simulation_object)
123 else:
124 outlet_streams.append(simulation_object)
126 # Handle intermediate streams connected to the module
127 if simulation_object.is_stream() and ports_mgr.count() == 2: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 propagate_intermediate_streams(simulation_object, contained_objects_id)
130 # Return the new module group for further use
131 propagate_streams(inlet_streams, ConType.Inlet)
132 propagate_streams(outlet_streams, ConType.Outlet)
133 return module_root_group