Coverage for backend/django/core/auxiliary/methods/CopyFlowsheet.py: 90%
75 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +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
16from Economics.studies.services.flowsheet_copy import copy_economics_configuration_for_flowsheet
19def copy_flowsheet_data(flowsheet: Flowsheet, user: User):
20 try:
21 with transaction.atomic():
23 model_lookups = create_foreign_key_lookups(flowsheet)
24 # Manually add the flowsheet to the model lookups.
25 new_flowsheet = Flowsheet(
26 pk=flowsheet.pk, # We need the old primary key for now, so the ModelLookup can use it. update_primary_keys will set it later.
27 name=f"{flowsheet.name} (copy)"[:32], # Ensure name is not too long,
28 owner=user,
29 created_at=flowsheet.created_at,
30 rootGrouping=flowsheet.rootGrouping, # This will be updated later to be the cloned version, because it doesn't exist yet.
31 savedDate= timezone.now(),
32 )
33 model_lookups[Flowsheet] = ModelLookup([new_flowsheet])
34 many_to_many_lookup = create_many_to_many_lookups(model_lookups)
36 # Normally, bulk_create will automatically set the flowsheet ID, because of
37 # our access control Manager.
38 # So we need to escape that behavior to allow us to set the flowsheet ID manually.
40 # Update everything to point to the new flowsheet.
41 update_primary_keys(model_lookups)
43 new_flowsheet.save()
45 AccessTable.objects.create(
46 user=user,
47 flowsheet=new_flowsheet,
48 read_only=False,
49 )
50 update_foreign_keys(model_lookups)
51 update_many_to_many_relationships(many_to_many_lookup, model_lookups)
52 update_formulas(model_lookups)
53 copy_economics_configuration_for_flowsheet(
54 source_flowsheet=flowsheet,
55 target_flowsheet=new_flowsheet,
56 model_lookups=model_lookups,
57 )
59 root_group_pk = new_flowsheet.rootGrouping.pk
60 new_flowsheet.rootGrouping = model_lookups[Grouping].get_model(root_group_pk)
61 new_flowsheet.created_at = timezone.now()
62 new_flowsheet.save()
63 return new_flowsheet
64 # Bulk update the models with the new foreign keys.
65 # for Model_Type, models in model_lookups.items():
66 except Exception as e:
67 import traceback
68 print(f"Error copying flowsheet: {e}")
69 print(traceback.format_exc())
70 raise Exception(
71 f"Error copying flowsheet: {e} " + traceback.format_exc()
72 ) # Re-raise the exception to be handled by the caller.
76def create_module_from_template_logic(template_flowsheet: Flowsheet, target_flowsheet: Flowsheet,
77 user, current_group: Grouping, x=None, y=None) -> Grouping:
78 """
79 Create a module (grouping) from a template flowsheet within an existing flowsheet
80 """
81 with transaction.atomic():
83 # This maps old PKs to new model instances
84 model_lookups = create_foreign_key_lookups_for_modules(template_flowsheet)
85 many_to_many_lookup = create_many_to_many_lookups(model_lookups)
87 # Generate new primary keys for all copied models
88 # Sets pk=None so Django creates new PKs when saving
89 update_primary_keys(model_lookups)
91 # Map template flowsheet to target flowsheet
92 # All copied models will now reference the target flowsheet instead of template
93 model_lookups[Flowsheet] = ModelLookup([])
94 model_lookups[Flowsheet].model_map[template_flowsheet.pk] = target_flowsheet
96 update_foreign_keys(model_lookups)
97 update_many_to_many_relationships(many_to_many_lookup, model_lookups)
98 update_formulas(model_lookups)
100 # Find the copied root group from the template
101 template_root_group_pk = template_flowsheet.rootGrouping.pk
102 module_root_group: Grouping = model_lookups[Grouping].get_model(template_root_group_pk)
104 # Set the copied root group's parent to current group
105 # This "inserts" the template module into the target flowsheet hierarchy
106 module_graphic_object = module_root_group.get_graphic_object()
107 module_graphic_object.group = current_group
108 if x is not None: 108 ↛ 110line 108 didn't jump to line 110 because the condition on line 108 was always true
109 module_graphic_object.x = x
110 if y is not None: 110 ↛ 112line 110 didn't jump to line 112 because the condition on line 110 was always true
111 module_graphic_object.y = y
112 module_graphic_object.save()
114 # Get all copied simulation objects
115 copied_simulation_objects = list(model_lookups[SimulationObject])
116 contained_objects_id = [sim_obj.pk for sim_obj in copied_simulation_objects]
118 # Handle streams for all copied simulation objects
119 inlet_streams = []
120 outlet_streams = []
121 for simulation_object in copied_simulation_objects:
122 ports_mgr = simulation_object.connectedPorts # RelatedManager
124 # Propagate streams connected to one port
125 if simulation_object.is_stream() and ports_mgr.count() == 1:
126 port = ports_mgr.first()
127 if port and port.direction == ConType.Inlet:
128 inlet_streams.append(simulation_object)
129 else:
130 outlet_streams.append(simulation_object)
132 # Handle intermediate streams connected to the module
133 if simulation_object.is_stream() and ports_mgr.count() == 2: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 propagate_intermediate_streams(simulation_object, contained_objects_id)
136 # Return the new module group for further use
137 propagate_streams(inlet_streams, ConType.Inlet)
138 propagate_streams(outlet_streams, ConType.Outlet)
139 return module_root_group