Coverage for backend/django/core/auxiliary/services/flowsheet_states/new_flowsheet.py: 91%
89 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Create a new stable flowsheet through the NEW_FLOWSHEET clone profile."""
3from dataclasses import dataclass
5from django.db import transaction
7from authentication.user.models import User
8from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType
9from core.auxiliary.models.DataRow import DataRow
10from core.auxiliary.models.Flowsheet import Flowsheet
11from core.auxiliary.models.Project import Project
12from core.auxiliary.models.Solution import Solution
14from .cloning import (
15 ProjectCloneContext,
16 clone_state,
17 finalize_project_clone,
18)
19from .errors import FlowsheetStateCloneError, merge_clone_validation_errors
20from .registry import CloneProfile
21from .validation import (
22 repair_legacy_formula_unit_references,
23 validate_state_for_clone,
24)
27@dataclass(frozen=True)
28class ScenarioResultCopySummary:
29 """User-facing size of the scenario results available to duplicate."""
31 has_results: bool
32 row_count: int
33 property_count: int
36def summarize_scenario_results(*, flowsheet_state_ids) -> ScenarioResultCopySummary:
37 """Count result-table rows and distinct result properties in the scope."""
39 state_ids = tuple(flowsheet_state_ids)
40 if not state_ids:
41 return ScenarioResultCopySummary(False, 0, 0)
42 solutions = Solution.all_states.filter(flowsheet_state_id__in=state_ids)
43 if not solutions.exists():
44 return ScenarioResultCopySummary(False, 0, 0)
45 result_scenario_ids = solutions.values("scenario_id")
46 return ScenarioResultCopySummary(
47 has_results=True,
48 row_count=DataRow.all_states.filter(
49 flowsheet_state_id__in=state_ids,
50 scenario_id__in=result_scenario_ids,
51 ).count(),
52 property_count=solutions.exclude(property_id=None)
53 .values("property_id")
54 .distinct()
55 .count(),
56 )
59@transaction.atomic
60def clone_to_new_flowsheet(
61 *,
62 source_flowsheet: Flowsheet,
63 user,
64 target_project=None,
65 project_context: ProjectCloneContext | None = None,
66 name: str | None = None,
67 repair_legacy_formula_units: bool = False,
68 include_scenario_results: bool = False,
69) -> Flowsheet:
70 """Clone current durable configuration under a fresh stable identity."""
72 source_queryset = Flowsheet.objects.select_related(
73 "current_state",
74 "project",
75 )
76 if repair_legacy_formula_units: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 source_queryset = source_queryset.select_for_update()
78 source_flowsheet = source_queryset.get(pk=source_flowsheet.pk)
79 if repair_legacy_formula_units: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 repair_legacy_formula_unit_references(
81 source_state=source_flowsheet.current_state,
82 profile=CloneProfile.NEW_FLOWSHEET,
83 )
84 project = target_project or source_flowsheet.project
85 owner = project.owner if project is not None else user
86 owns_context = False
87 if (
88 project_context is None
89 and source_flowsheet.project_id is not None
90 and project is not None
91 and project.pk != source_flowsheet.project_id
92 ):
93 project_context = ProjectCloneContext(source_flowsheet.project, project)
94 owns_context = True
96 copied = Flowsheet.create(
97 name=(name or f"{source_flowsheet.name} (copy)")[:64],
98 project=project,
99 owner=owner,
100 flowsheet_template_type=FlowsheetTemplateType.NotTemplate,
101 auto_snapshot_after_single_solve=(
102 source_flowsheet.auto_snapshot_after_single_solve
103 ),
104 buildVersion=source_flowsheet.current_state.build_version,
105 buildDate=source_flowsheet.current_state.build_date,
106 initialize_state=False,
107 )
108 if project_context is not None:
109 project_context.register_flowsheet(source_flowsheet, copied)
110 try:
111 clone_state(
112 source_state=source_flowsheet.current_state,
113 target_state=copied.current_state,
114 profile=CloneProfile.NEW_FLOWSHEET,
115 project_context=project_context,
116 include_scenario_results=include_scenario_results,
117 )
118 except FlowsheetStateCloneError as exc:
119 raise exc.for_source_flowsheet(source_flowsheet) from exc
120 if owns_context:
121 finalize_project_clone(project_context)
122 return copied
125@transaction.atomic
126def create_project_from_template(
127 *,
128 source_template: Flowsheet,
129 user: User,
130) -> tuple[Project, Flowsheet]:
131 """Create a project whose first working state comes from a template."""
133 project = Project.create_empty(
134 name=f"{source_template.name} (copy)"[:64],
135 owner=user,
136 active_flowsheet=None,
137 is_starred=False,
138 is_binned=False,
139 binned_at=None,
140 )
141 copied = clone_to_new_flowsheet(
142 source_flowsheet=source_template,
143 user=user,
144 target_project=project,
145 )
146 copied.flowsheet_template_type = FlowsheetTemplateType.NotTemplate
147 copied.save(update_fields=["flowsheet_template_type"])
148 project.set_active_flowsheet(copied)
149 return project, copied
152@transaction.atomic
153def copy_project_data(
154 *,
155 source_project: Project,
156 user: User,
157 repair_legacy_formula_units: bool = False,
158 include_scenario_results: bool = False,
159) -> Project:
160 """Duplicate a project and all of its non-template flowsheet states."""
162 source_queryset = (
163 source_project.flowsheets.filter(
164 flowsheet_template_type=FlowsheetTemplateType.NotTemplate,
165 )
166 .select_related("current_state")
167 .order_by("created_at", "id")
168 )
169 if repair_legacy_formula_units: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 source_queryset = source_queryset.select_for_update()
171 source_flowsheets = list(source_queryset)
172 validation_errors = []
173 for source_flowsheet in source_flowsheets:
174 if repair_legacy_formula_units: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 repair_legacy_formula_unit_references(
176 source_state=source_flowsheet.current_state,
177 profile=CloneProfile.NEW_FLOWSHEET,
178 )
179 try:
180 validate_state_for_clone(
181 source_state=source_flowsheet.current_state,
182 profile=CloneProfile.NEW_FLOWSHEET,
183 )
184 except FlowsheetStateCloneError as exc:
185 validation_errors.append(exc.for_source_flowsheet(source_flowsheet))
186 if validation_errors:
187 raise merge_clone_validation_errors(validation_errors)
189 copied_project = Project.create_empty(
190 name=f"{source_project.name} (copy)"[:32],
191 owner=user,
192 active_flowsheet=None,
193 is_starred=False,
194 is_binned=False,
195 binned_at=None,
196 )
197 project_context = ProjectCloneContext(
198 source_project=source_project,
199 target_project=copied_project,
200 )
201 project_context.validated_source_state_ids.update(
202 flowsheet.current_state_id for flowsheet in source_flowsheets
203 )
205 active_source_id = source_project.active_flowsheet_id
206 copied_active_flowsheet = None
207 for source_flowsheet in source_flowsheets:
208 copied_flowsheet = clone_to_new_flowsheet(
209 source_flowsheet=source_flowsheet,
210 user=user,
211 target_project=copied_project,
212 project_context=project_context,
213 include_scenario_results=include_scenario_results,
214 )
215 if source_flowsheet.pk == active_source_id:
216 copied_active_flowsheet = copied_flowsheet
218 if copied_active_flowsheet is None: 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true
219 copied_active_flowsheet = copied_project.flowsheets.order_by(
220 "created_at",
221 "id",
222 ).first()
223 if copied_active_flowsheet is not None: 223 ↛ 226line 223 didn't jump to line 226 because the condition on line 223 was always true
224 copied_project.set_active_flowsheet(copied_active_flowsheet)
226 finalize_project_clone(project_context)
227 return copied_project