Coverage for backend/django/Economics/studies/services/study_copy.py: 68%
76 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import ObjectDoesNotExist
2from django.db import transaction
3from rest_framework.exceptions import ValidationError
5from Economics.costing.models import CapitalCostLine, CostDriver, CostableItem, EquipmentMapping, OperatingCostLine
6from Economics.settings_profiles.models import EconomicsAssumptions, EconomicsBaseline
7from Economics.shared.choices import EconomicsScheduleMode
8from Economics.studies.models import EconomicsSchedulePlan, EconomicsScheduleRule, EconomicsStudy
11def duplicate_study(source: EconomicsStudy, *, requested_name: str) -> EconomicsStudy:
12 """Copy editable study configuration without copying result runs or active-study state."""
13 with transaction.atomic():
14 study = EconomicsStudy.objects.create(
15 flowsheet_state=source.flowsheet_state,
16 settings_profile=source.settings_profile,
17 schedule_mode=(
18 EconomicsScheduleMode.STEADY_STATE
19 if source.schedule_mode == EconomicsScheduleMode.COMPOSITE
20 else source.schedule_mode
21 ),
22 schedule_scenario=(
23 source.schedule_scenario
24 if source.schedule_mode != EconomicsScheduleMode.COMPOSITE
25 else None
26 ),
27 baseline_mode=source.baseline_mode,
28 baseline_flowsheet_id=source.baseline_flowsheet_id,
29 baseline_study_lineage_id=source.baseline_study_lineage_id,
30 name=_duplicate_name(source, requested_name=requested_name),
31 description=source.description,
32 )
33 costable_item_map = _copy_costable_items(source=source, target=study)
34 _copy_capital_lines(source=source, target=study, costable_item_map=costable_item_map)
35 _copy_operating_lines(source=source, target=study, costable_item_map=costable_item_map)
36 _copy_schedule_plan(source=source, target=study)
37 if source.schedule_mode == EconomicsScheduleMode.COMPOSITE: 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true
38 study.schedule_mode = EconomicsScheduleMode.COMPOSITE
39 study.save(update_fields=["schedule_mode", "updated_at"])
40 return study
43def _duplicate_name(source: EconomicsStudy, *, requested_name: str) -> str:
44 if requested_name:
45 if EconomicsStudy.objects.filter(flowsheet_state=source.flowsheet_state, name=requested_name).exists(): 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true
46 raise ValidationError({"name": "An economics study with this name already exists in the flowsheet."})
47 return requested_name
49 base_name = f"{source.name} Copy"
50 candidate = base_name
51 suffix = 2
52 while EconomicsStudy.objects.filter(flowsheet_state=source.flowsheet_state, name=candidate).exists(): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 candidate = f"{base_name} {suffix}"
54 suffix += 1
55 return candidate
58def _copy_assumptions(*, source: EconomicsStudy, target: EconomicsStudy) -> None:
59 try:
60 assumptions = source.assumptions
61 except ObjectDoesNotExist:
62 return
63 EconomicsAssumptions.objects.create(
64 flowsheet_state=target.flowsheet_state,
65 study=target,
66 currency=assumptions.currency,
67 location=assumptions.location,
68 basis_date=assumptions.basis_date,
69 discount_rate_percent=assumptions.discount_rate_percent,
70 project_lifetime_years=assumptions.project_lifetime_years,
71 inflation_method=assumptions.inflation_method,
72 annual_operating_hours=assumptions.annual_operating_hours,
73 tax_rate_percent=assumptions.tax_rate_percent,
74 depreciation_enabled=assumptions.depreciation_enabled,
75 default_depreciation_life_years=assumptions.default_depreciation_life_years,
76 default_depreciation_salvage_percent=assumptions.default_depreciation_salvage_percent,
77 contingency_percent=assumptions.contingency_percent,
78 electrical_upgrade_rate_amount=assumptions.electrical_upgrade_rate_amount,
79 default_lang_factor=assumptions.default_lang_factor,
80 capital_index_series=assumptions.capital_index_series,
81 operating_index_series=assumptions.operating_index_series,
82 default_rate_overrides=assumptions.default_rate_overrides,
83 notes=assumptions.notes,
84 )
87def _copy_baseline(*, source: EconomicsStudy, target: EconomicsStudy) -> None:
88 try:
89 baseline = source.baseline
90 except ObjectDoesNotExist:
91 return
92 EconomicsBaseline.objects.create(
93 flowsheet_state=target.flowsheet_state,
94 study=target,
95 manual_capex=baseline.manual_capex,
96 manual_annual_opex=baseline.manual_annual_opex,
97 annual_heat_basis_mode=baseline.annual_heat_basis_mode,
98 manual_annual_heat_basis=baseline.manual_annual_heat_basis,
99 manual_annual_heat_basis_unit=baseline.manual_annual_heat_basis_unit,
100 average_power_input=baseline.average_power_input,
101 average_power_unit=baseline.average_power_unit,
102 residual_value=baseline.residual_value,
103 notes=baseline.notes,
104 )
107def _copy_costable_items(*, source: EconomicsStudy, target: EconomicsStudy) -> dict[int, CostableItem]:
108 costable_item_map = {}
109 for item in source.costable_items.select_related("simulation_object").order_by("created_at", "pk"):
110 copied = CostableItem.objects.create(
111 flowsheet_state=target.flowsheet_state,
112 study=target,
113 item_type=item.item_type,
114 simulation_object=item.simulation_object,
115 name=item.name,
116 included=item.included,
117 manual=item.manual,
118 notes=item.notes,
119 )
120 costable_item_map[item.pk] = copied
121 _copy_cost_driver(source_item=item, target_item=copied)
122 _copy_equipment_mapping(source_item=item, target_item=copied)
123 return costable_item_map
126def _copy_cost_driver(*, source_item: CostableItem, target_item: CostableItem) -> None:
127 try:
128 driver = source_item.cost_driver
129 except ObjectDoesNotExist:
130 return
131 CostDriver.objects.create(
132 flowsheet_state=target_item.flowsheet_state,
133 costable_item=target_item,
134 source=driver.source,
135 property_info=driver.property_info,
136 manual_property_info=driver.manual_property_info,
137 sizing_mode=driver.sizing_mode,
138 canonical_unit=driver.canonical_unit,
139 design_value=driver.design_value,
140 unresolved_reason_code=driver.unresolved_reason_code,
141 warning_payload=driver.warning_payload,
142 )
145def _copy_equipment_mapping(*, source_item: CostableItem, target_item: CostableItem) -> None:
146 try:
147 mapping = source_item.equipment_mapping
148 except ObjectDoesNotExist:
149 return
150 EquipmentMapping.objects.create(
151 flowsheet_state=target_item.flowsheet_state,
152 costable_item=target_item,
153 cost_curve=mapping.cost_curve,
154 equipment_category=mapping.equipment_category,
155 equipment_subtype=mapping.equipment_subtype,
156 cost_basis=mapping.cost_basis,
157 install_factor_profile=mapping.install_factor_profile,
158 install_factor=mapping.install_factor,
159 use_study_lang_factor=mapping.use_study_lang_factor,
160 applicability_notes=mapping.applicability_notes,
161 )
164def _copy_capital_lines(*, source: EconomicsStudy, target: EconomicsStudy, costable_item_map: dict[int, CostableItem]) -> None:
165 for line in source.capital_lines.order_by("created_at", "pk"):
166 CapitalCostLine.objects.create(
167 flowsheet_state=target.flowsheet_state,
168 study=target,
169 costable_item=costable_item_map.get(line.costable_item_id),
170 cost_curve=line.cost_curve,
171 label=line.label,
172 line_type=line.line_type,
173 calculation_basis=line.calculation_basis,
174 amount=line.amount,
175 basis_percent=line.basis_percent,
176 depreciation_mode=line.depreciation_mode,
177 depreciation_life_years=line.depreciation_life_years,
178 depreciation_salvage_percent=line.depreciation_salvage_percent,
179 peak_demand_kw=line.peak_demand_kw,
180 minimum_peak_demand_kw=line.minimum_peak_demand_kw,
181 currency=line.currency,
182 included=line.included,
183 manual=line.manual,
184 source=line.source,
185 confidence=line.confidence,
186 warning_payload=line.warning_payload,
187 driver_inputs=line.driver_inputs,
188 )
191def _copy_operating_lines(*, source: EconomicsStudy, target: EconomicsStudy, costable_item_map: dict[int, CostableItem]) -> None:
192 for line in source.operating_lines.order_by("created_at", "pk"): 192 ↛ 193line 192 didn't jump to line 193 because the loop on line 192 never started
193 OperatingCostLine.objects.create(
194 flowsheet_state=target.flowsheet_state,
195 study=target,
196 costable_item=costable_item_map.get(line.costable_item_id),
197 label=line.label,
198 line_type=line.line_type,
199 category=line.category,
200 economic_effect=line.economic_effect,
201 currency=line.currency,
202 basis_quantity=line.basis_quantity,
203 basis_unit=line.basis_unit,
204 basis_quantity_source=line.basis_quantity_source,
205 rate_amount=line.rate_amount,
206 rate_unit=line.rate_unit,
207 rate_type=line.rate_type,
208 rate_source_mode=line.rate_source_mode,
209 calculation_method=line.calculation_method,
210 source_property_info=line.source_property_info,
211 source_default_rate=line.source_default_rate,
212 outlet_stream_disposition=line.outlet_stream_disposition,
213 included=line.included,
214 manual=line.manual,
215 source=line.source,
216 warning_payload=line.warning_payload,
217 )
220def _copy_schedule_plan(*, source: EconomicsStudy, target: EconomicsStudy) -> None:
221 """Copy composite schedule rules for same-flowsheet study duplication."""
223 try:
224 plan = source.schedule_plan
225 except ObjectDoesNotExist:
226 return
227 copied_plan = EconomicsSchedulePlan.objects.create(
228 flowsheet_state=target.flowsheet_state,
229 study=target,
230 )
231 for rule in plan.rules.order_by("sort_order", "pk"):
232 EconomicsScheduleRule.objects.create(
233 flowsheet_state=target.flowsheet_state,
234 plan=copied_plan,
235 source_scenario=rule.source_scenario,
236 label=rule.label,
237 days_mask=rule.days_mask,
238 start_time=rule.start_time,
239 sort_order=rule.sort_order,
240 )