Coverage for backend/django/Economics/results/services/financial_metrics/baselines.py: 81%
163 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Financial metric baseline resolution."""
3from __future__ import annotations
5from decimal import Decimal
7from core.managers import has_flowsheet_read_access
8from core.validation import get_current_flowsheet
9from Economics.costing.operating.line_calculation import annualized_basis_quantity
10from Economics.results.models import EconomicsResultLine, EconomicsResultRun
11from .calculation import currency_unit
12from .contracts import (
13 ZERO,
14 AssumptionSet,
15 BaselineResolution,
16 FinancialContext,
17 FinancialMetricsError,
18 FinancialWarning,
19 TargetAssumptions,
20)
21from .warnings import target_assumption_warnings
22from .metric_catalog import FinancialMetricKey, required_financial_metric_spec
23from Economics.results.services.lifecycle.runs import ResultRunClassification, classify_result_run
24from Economics.settings_profiles.models import EconomicsSettingsProfile
25from Economics.settings_profiles.services.settings_profiles import get_settings_profile
26from Economics.shared.choices import EconomicsBaselineMode, ResultRunStatus
27from Economics.studies.models import EconomicsStudy
28from Economics.studies.services.baseline_access import (
29 baseline_study_matches_target_study,
30 resolve_baseline_study,
31 stable_flowsheet_id,
32)
33from Economics.costing.capital.electrical_upgrade import derive_peak_demand_basis
36def validate_manual_baseline(baseline: EconomicsSettingsProfile, *, target_assumptions: TargetAssumptions) -> None:
37 """Validate the full manual baseline before presenting calculated metrics."""
38 errors = _manual_baseline_errors(baseline, target_assumptions=target_assumptions)
39 if errors:
40 raise FinancialMetricsError(
41 "manual_baseline_invalid",
42 "Manual economics baseline is incomplete or invalid.",
43 context={"errors": errors, "baseline_id": baseline.pk},
44 )
46def _manual_baseline_errors(baseline: EconomicsSettingsProfile, *, target_assumptions: TargetAssumptions) -> dict[str, str]:
47 """Return field-addressable blockers for calculations that depend on the manual baseline."""
48 errors: dict[str, str] = {}
49 if baseline.manual_capex is None: 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true
50 errors["manual_capex"] = "Manual baseline capex is required."
51 elif baseline.manual_capex < 0: 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 errors["manual_capex"] = "Manual baseline capex cannot be negative."
54 if baseline.manual_annual_opex is None:
55 errors["manual_annual_opex"] = "Manual baseline annual opex is required."
56 elif baseline.manual_annual_opex < 0: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 errors["manual_annual_opex"] = "Manual baseline annual opex cannot be negative."
59 if baseline.annual_heat_basis_mode == "average_power":
60 if baseline.average_power_input is None: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 errors["average_power_input"] = "Hourly heat quantity is required when the baseline heat basis uses hourly heat."
62 elif baseline.average_power_input <= 0: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 errors["average_power_input"] = "Hourly heat quantity must be positive when supplied."
64 if target_assumptions.annual_operating_hours is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 errors["annual_operating_hours"] = "Annual operating hours are required for an hourly-heat baseline heat basis."
66 elif target_assumptions.annual_operating_hours <= 0: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true
67 errors["annual_operating_hours"] = "Annual operating hours must be positive for an hourly-heat baseline heat basis."
68 elif baseline.manual_annual_heat_basis is not None and baseline.manual_annual_heat_basis <= 0: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 errors["manual_annual_heat_basis"] = "Manual annual heat basis must be positive when supplied."
71 if target_assumptions.project_lifetime_years is None: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 errors["project_lifetime_years"] = "Study project lifetime is required for manual baseline metrics."
73 elif target_assumptions.project_lifetime_years <= 0: 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true
74 errors["project_lifetime_years"] = "Study project lifetime must be positive for manual baseline metrics."
76 if target_assumptions.discount_rate_percent is None: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 errors["discount_rate_percent"] = "Study discount rate is required for manual baseline metrics."
78 elif target_assumptions.discount_rate_percent < 0: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 errors["discount_rate_percent"] = "Study discount rate cannot be negative for manual baseline metrics."
81 if baseline.residual_value is not None and baseline.residual_value < 0: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 errors["residual_value"] = "Residual value cannot be negative."
84 return errors
86def resolve_baseline_for_study(
87 *,
88 study: EconomicsStudy,
89 target_capex: Decimal,
90 target_annual_opex: Decimal,
91 target_assumptions: TargetAssumptions,
92) -> tuple[BaselineResolution, list[FinancialWarning]]:
93 """Resolve baseline inputs using explicit v1 precedence rules.
95 V1 resolves manual baselines only. Incomplete manual drafts are preserved
96 as source state, but field-addressable warnings block dependent metrics
97 until required values are present.
98 """
99 warnings = target_assumption_warnings(study=study, target_assumptions=target_assumptions)
100 if study.baseline_mode == EconomicsBaselineMode.STUDY:
101 resolution, baseline_warnings = resolve_study_baseline(
102 target_study=study,
103 baseline_study=resolve_baseline_study(study),
104 target_assumptions=target_assumptions,
105 )
106 warnings.extend(baseline_warnings)
107 return resolution, warnings
109 baseline = _get_baseline(study)
110 if baseline is None:
111 return _resolve_missing_baseline(study=study, target_assumptions=target_assumptions, warnings=warnings)
113 resolution, baseline_warnings = _resolve_manual_baseline(
114 study=study,
115 baseline=baseline,
116 target_assumptions=target_assumptions,
117 )
118 warnings.extend(baseline_warnings)
119 return resolution, warnings
121def resolve_study_baseline(
122 *,
123 target_study: EconomicsStudy,
124 baseline_study: EconomicsStudy | None,
125 target_assumptions: TargetAssumptions,
126) -> tuple[BaselineResolution, list[FinancialWarning]]:
127 """Resolve baseline inputs from another study's current result run."""
128 if baseline_study is None: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 return _study_baseline_unavailable_resolution(
130 target_study=target_study,
131 target_assumptions=target_assumptions,
132 code="study_baseline_missing",
133 message="Select a baseline study before calculating study-baseline metrics.",
134 context={"target_study_id": target_study.pk},
135 )
136 if not baseline_study_matches_target_study(target_study, baseline_study):
137 return _study_baseline_unavailable_resolution(
138 target_study=target_study,
139 target_assumptions=target_assumptions,
140 code="study_baseline_invalid_scope",
141 message="The selected baseline study is no longer available in this project.",
142 context={"baseline_study_id": baseline_study.pk},
143 )
144 request_user = _current_request_user()
145 if request_user is not None and not has_flowsheet_read_access( 145 ↛ 149line 145 didn't jump to line 149 because the condition on line 145 was never true
146 request_user,
147 stable_flowsheet_id(baseline_study),
148 ):
149 return _study_baseline_unavailable_resolution(
150 target_study=target_study,
151 target_assumptions=target_assumptions,
152 code="study_baseline_inaccessible",
153 message="The selected baseline study is no longer accessible.",
154 context={"baseline_study_id": baseline_study.pk},
155 )
156 run = _current_result_run(baseline_study)
157 if run is None: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 return _study_baseline_unavailable_resolution(
159 target_study=target_study,
160 target_assumptions=target_assumptions,
161 code="study_baseline_result_missing",
162 message="The selected baseline study has no current economics result.",
163 context={"baseline_study_id": baseline_study.pk},
164 )
165 if classify_result_run(run) != ResultRunClassification.CURRENT:
166 return _study_baseline_unavailable_resolution(
167 target_study=target_study,
168 target_assumptions=target_assumptions,
169 code="study_baseline_result_stale",
170 message="The selected baseline study result is stale and must be recalculated before it can be used as a study baseline.",
171 context={"baseline_study_id": baseline_study.pk, "baseline_result_run_id": run.pk},
172 )
174 capex_line = _metric_line(
175 run=run,
176 row_key=required_financial_metric_spec(FinancialMetricKey.CAPEX).row_key,
177 )
178 annual_opex_line = _metric_line(
179 run=run,
180 row_key=required_financial_metric_spec(FinancialMetricKey.ANNUAL_OPEX).row_key,
181 )
182 warnings = _study_baseline_line_warnings(
183 target_assumptions=target_assumptions,
184 baseline_study=baseline_study,
185 capex_line=capex_line,
186 annual_opex_line=annual_opex_line,
187 )
188 capex = capex_line.amount if capex_line is not None and capex_line.amount is not None else None
189 annual_opex = annual_opex_line.amount if annual_opex_line is not None and annual_opex_line.amount is not None else None
190 if _line_unit_mismatch(capex_line, currency_unit(target_assumptions.as_assumption_set())):
191 capex = None
192 if _line_unit_mismatch(annual_opex_line, currency_unit(target_assumptions.as_assumption_set(), per_year=True)):
193 annual_opex = None
195 return BaselineResolution(
196 source="study",
197 is_guided_default=capex is None or annual_opex is None or bool(warnings),
198 capex=capex,
199 annual_opex=annual_opex,
200 annual_heat_basis=None,
201 annual_heat_basis_unit=None,
202 residual_value=target_residual_value(target_study),
203 project_lifetime_years=target_assumptions.project_lifetime_years,
204 discount_rate_percent=target_assumptions.discount_rate_percent,
205 assumptions=AssumptionSet.from_mapping(
206 {
207 "target_study_id": target_study.pk,
208 "baseline_source": "study",
209 "baseline_study_id": baseline_study.pk,
210 "baseline_result_run_id": run.pk,
211 "baseline_capex_source": required_financial_metric_spec(FinancialMetricKey.CAPEX).row_key,
212 "baseline_annual_opex_source": required_financial_metric_spec(FinancialMetricKey.ANNUAL_OPEX).row_key,
213 "project_lifetime_source": "target_study",
214 "discount_rate_source": "target_study",
215 "residual_value": target_residual_value(target_study),
216 }
217 ),
218 ), warnings
220def manual_baseline_profile_for_study(study: EconomicsStudy) -> EconomicsSettingsProfile | None:
221 """Return the configured manual baseline profile for a study, if present."""
222 return _get_baseline(study)
224def resolve_manual_baseline_for_target(
225 *,
226 target_study: EconomicsStudy,
227 baseline_profile: EconomicsSettingsProfile,
228 target_assumptions: TargetAssumptions,
229) -> tuple[BaselineResolution, list[FinancialWarning]]:
230 """Resolve manual baseline values against a target study's assumptions."""
231 return _resolve_manual_baseline(
232 study=target_study,
233 baseline=baseline_profile,
234 target_assumptions=target_assumptions,
235 )
237def target_assumptions_for_study(study: EconomicsStudy) -> TargetAssumptions:
238 """Public boundary for callers that need target study financial assumptions."""
239 return _target_assumptions(study)
241def target_only_baseline_resolution(
242 *,
243 study: EconomicsStudy,
244 target_assumptions: TargetAssumptions,
245) -> BaselineResolution:
246 """Return a target-only baseline resolution for unavailable baseline paths."""
248 return _target_only_baseline_resolution(study=study, target_assumptions=target_assumptions)
250def target_residual_value(study: EconomicsStudy) -> Decimal:
251 """Return the target study residual value used by cash-flow formulas."""
252 profile = get_settings_profile(study)
253 if profile is None or profile.residual_value is None:
254 return ZERO
255 return profile.residual_value
257def _study_baseline_unavailable_resolution(
258 *,
259 target_study: EconomicsStudy,
260 target_assumptions: TargetAssumptions,
261 code: str,
262 message: str,
263 context: FinancialContext,
264) -> tuple[BaselineResolution, list[FinancialWarning]]:
265 """Preserve target assumptions while marking study-baseline inputs unavailable."""
266 resolution = _target_only_baseline_resolution(
267 study=target_study,
268 target_assumptions=target_assumptions,
269 ).model_copy(
270 update={
271 "source": "study",
272 "residual_value": target_residual_value(target_study),
273 }
274 )
275 return resolution, [
276 FinancialWarning(
277 code=code,
278 severity="warning",
279 message=message,
280 context=context,
281 )
282 ]
284def _study_baseline_line_warnings(
285 *,
286 target_assumptions: TargetAssumptions,
287 baseline_study: EconomicsStudy,
288 capex_line: EconomicsResultLine | None,
289 annual_opex_line: EconomicsResultLine | None,
290) -> list[FinancialWarning]:
291 """Report missing or incompatible current-result rows from the selected baseline study."""
292 warnings: list[FinancialWarning] = []
293 expected_capex_unit = currency_unit(target_assumptions.as_assumption_set())
294 expected_opex_unit = currency_unit(target_assumptions.as_assumption_set(), per_year=True)
295 if capex_line is None or capex_line.amount is None: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
296 warnings.append(
297 FinancialWarning(
298 code="study_baseline_capex_missing",
299 severity="warning",
300 message="The selected baseline study does not have a total capital cost metric.",
301 context={"baseline_study_id": baseline_study.pk},
302 )
303 )
304 elif _line_unit_mismatch(capex_line, expected_capex_unit):
305 warnings.append(
306 FinancialWarning(
307 code="study_baseline_capex_unit_mismatch",
308 severity="warning",
309 message="The selected baseline study capital cost unit does not match this study currency.",
310 context={"baseline_study_id": baseline_study.pk, "unit": capex_line.unit, "expected_unit": expected_capex_unit},
311 )
312 )
313 if annual_opex_line is None or annual_opex_line.amount is None: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 warnings.append(
315 FinancialWarning(
316 code="study_baseline_annual_opex_missing",
317 severity="warning",
318 message="The selected baseline study does not have an annual operating cost metric.",
319 context={"baseline_study_id": baseline_study.pk},
320 )
321 )
322 elif _line_unit_mismatch(annual_opex_line, expected_opex_unit):
323 warnings.append(
324 FinancialWarning(
325 code="study_baseline_annual_opex_unit_mismatch",
326 severity="warning",
327 message="The selected baseline study annual operating cost unit does not match this study currency.",
328 context={"baseline_study_id": baseline_study.pk, "unit": annual_opex_line.unit, "expected_unit": expected_opex_unit},
329 )
330 )
331 return warnings
333def _current_result_run(study: EconomicsStudy) -> EconomicsResultRun | None:
334 """Load the current run for the selected baseline study, including cross-flowsheet baselines.
336 Project-baseline selection validates read access before storing the baseline
337 study relation. At calculation time the active request context is the target
338 study's flowsheet, so the access-control manager would hide a valid
339 cross-flowsheet baseline run. The explicit flowsheet/study/status filters
340 keep this read pinned to the already-selected baseline study.
341 """
342 return (
343 # Use the unscoped manager because the active request context belongs
344 # to the target study; explicit filters pin this to the stored baseline.
345 EconomicsResultRun._base_manager.filter(
346 flowsheet_state=study.flowsheet_state,
347 study=study,
348 status=ResultRunStatus.CURRENT,
349 )
350 .order_by("-created_at", "-pk")
351 .first()
352 )
355def _current_request_user():
356 """Return the request user from flowsheet context when calculation is request-scoped."""
357 context = get_current_flowsheet() or {}
358 return context.get("user")
360def _metric_line(*, run: EconomicsResultRun, row_key: str) -> EconomicsResultLine | None:
361 """Load one financial metric line from the baseline run in its owning flowsheet."""
362 return (
363 # Use the unscoped manager for cross-flowsheet baseline runs, then pin
364 # the read to the exact result run and owning flowsheet.
365 EconomicsResultLine._base_manager.filter(
366 flowsheet_state=run.flowsheet_state,
367 result_run=run,
368 group="financial_metrics",
369 row_key=row_key,
370 )
371 .order_by("-pk")
372 .first()
373 )
375def _line_unit_mismatch(line: EconomicsResultLine | None, expected_unit: str) -> bool:
376 """Treat a populated result line as unusable when its unit differs from the target study."""
377 return line is not None and line.amount is not None and bool(expected_unit) and line.unit != expected_unit
379def _resolve_missing_baseline(
380 *,
381 study: EconomicsStudy,
382 target_assumptions: TargetAssumptions,
383 warnings: list[FinancialWarning],
384) -> tuple[BaselineResolution, list[FinancialWarning]]:
385 warnings.append(
386 FinancialWarning(
387 code="baseline_missing",
388 severity="warning",
389 message="No baseline has been configured; savings and comparison metrics are not fully calculated.",
390 context={"study_id": study.pk},
391 )
392 )
393 return _target_only_baseline_resolution(study=study, target_assumptions=target_assumptions), warnings
395def _resolve_manual_baseline(
396 *,
397 study: EconomicsStudy,
398 baseline: EconomicsSettingsProfile,
399 target_assumptions: TargetAssumptions,
400) -> tuple[BaselineResolution, list[FinancialWarning]]:
401 errors = _manual_baseline_errors(baseline, target_assumptions=target_assumptions)
402 is_incomplete = bool(errors)
403 warnings = []
404 if errors:
405 warnings.append(
406 FinancialWarning(
407 code="manual_baseline_invalid",
408 severity="error",
409 message="Manual economics baseline is incomplete or invalid.",
410 context={"errors": errors, "baseline_id": baseline.pk},
411 )
412 )
413 return BaselineResolution(
414 source="manual",
415 is_guided_default=is_incomplete,
416 capex=baseline.manual_capex,
417 annual_opex=baseline.manual_annual_opex,
418 annual_heat_basis=_resolve_baseline_annual_heat_basis(
419 study=study,
420 baseline=baseline,
421 target_assumptions=target_assumptions,
422 ),
423 annual_heat_basis_unit=_resolve_baseline_annual_heat_basis_unit(baseline),
424 residual_value=baseline.residual_value or ZERO,
425 project_lifetime_years=target_assumptions.project_lifetime_years,
426 discount_rate_percent=target_assumptions.discount_rate_percent,
427 assumptions=AssumptionSet.from_mapping(
428 {
429 "baseline_id": baseline.pk,
430 "baseline_source": "manual",
431 "baseline_capex_source": "manual_capex",
432 "baseline_annual_opex_source": "manual_annual_opex",
433 "annual_heat_basis_mode": baseline.annual_heat_basis_mode,
434 "annual_heat_basis": _resolve_baseline_annual_heat_basis(
435 study=study,
436 baseline=baseline,
437 target_assumptions=target_assumptions,
438 ),
439 "annual_heat_basis_unit": _resolve_baseline_annual_heat_basis_unit(baseline),
440 "average_power_input": baseline.average_power_input,
441 "average_power_unit": baseline.average_power_unit,
442 "currency_source": "study_assumptions",
443 "basis_date_source": "study_assumptions",
444 "project_lifetime_source": "study_assumptions",
445 "discount_rate_source": "study_assumptions",
446 "index_method_source": "study_assumptions",
447 "residual_value": baseline.residual_value or ZERO,
448 }
449 ),
450 ), warnings
452def _resolve_baseline_annual_heat_basis(
453 *,
454 study: EconomicsStudy,
455 baseline: EconomicsSettingsProfile,
456 target_assumptions: TargetAssumptions,
457) -> Decimal | None:
458 """Resolve the baseline heat basis from explicit annual heat or hourly heat quantity."""
459 if baseline.annual_heat_basis_mode != "average_power":
460 return baseline.manual_annual_heat_basis
461 if baseline.average_power_input is None or target_assumptions.annual_operating_hours is None: 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true
462 return None
463 if baseline.average_power_input <= 0 or target_assumptions.annual_operating_hours <= 0: 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true
464 return None
465 return annualized_basis_quantity(
466 baseline.average_power_input,
467 source_unit=baseline.average_power_unit,
468 target_unit="GJ",
469 study=study,
470 )
472def _resolve_baseline_annual_heat_basis_unit(baseline: EconomicsSettingsProfile) -> str:
473 if baseline.annual_heat_basis_mode == "average_power":
474 return "GJ/year"
475 return baseline.manual_annual_heat_basis_unit
477def _target_only_baseline_resolution(
478 *,
479 study: EconomicsStudy,
480 target_assumptions: TargetAssumptions,
481) -> BaselineResolution:
482 return BaselineResolution(
483 source="target_only",
484 is_guided_default=True,
485 capex=None,
486 annual_opex=None,
487 annual_heat_basis=None,
488 annual_heat_basis_unit=None,
489 residual_value=ZERO,
490 project_lifetime_years=target_assumptions.project_lifetime_years,
491 discount_rate_percent=target_assumptions.discount_rate_percent,
492 assumptions=AssumptionSet.from_mapping(
493 {
494 "target_study_id": study.pk,
495 "baseline_source": "target_only",
496 "project_lifetime_source": "target_study",
497 "discount_rate_source": "target_study",
498 }
499 ),
500 )
502def _target_assumptions(study: EconomicsStudy) -> TargetAssumptions:
503 assumptions = get_settings_profile(study)
504 if assumptions is None:
505 return TargetAssumptions(target_study_id=study.pk, assumptions_source="missing")
506 peak_demand_basis = derive_peak_demand_basis(study)
507 return TargetAssumptions(
508 target_study_id=study.pk,
509 assumptions_id=assumptions.pk,
510 project_lifetime_years=assumptions.project_lifetime_years,
511 discount_rate_percent=assumptions.discount_rate_percent,
512 currency=assumptions.currency,
513 basis_date=assumptions.basis_date.isoformat() if assumptions.basis_date else None,
514 inflation_method=assumptions.inflation_method,
515 capital_index_series_id=assumptions.capital_index_series_id,
516 operating_index_series_id=assumptions.operating_index_series_id,
517 annual_operating_hours=assumptions.annual_operating_hours,
518 tax_rate_percent=assumptions.tax_rate_percent,
519 depreciation_enabled=assumptions.depreciation_enabled,
520 default_depreciation_life_years=assumptions.default_depreciation_life_years,
521 default_depreciation_salvage_percent=assumptions.default_depreciation_salvage_percent,
522 contingency_percent=assumptions.contingency_percent,
523 electrical_upgrade_rate_amount=assumptions.electrical_upgrade_rate_amount,
524 electrical_upgrade_rate_unit=assumptions.electrical_upgrade_rate_unit,
525 peak_demand_kw=peak_demand_basis.quantity_kw,
526 default_lang_factor=assumptions.default_lang_factor,
527 assumptions_source="study",
528 )
530def _get_baseline(study: EconomicsStudy) -> EconomicsSettingsProfile | None:
531 baseline = get_settings_profile(study)
532 if baseline is None or not _has_manual_baseline_values(baseline):
533 return None
534 return baseline
536def _has_manual_baseline_values(profile: EconomicsSettingsProfile) -> bool:
537 return any(
538 value is not None
539 for value in (
540 profile.manual_capex,
541 profile.manual_annual_opex,
542 profile.manual_annual_heat_basis,
543 profile.average_power_input,
544 profile.residual_value,
545 )
546 )