Coverage for backend/django/Economics/costing/capital/schedule_sizing.py: 67%
78 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Production-schedule sizing helpers for generated capital lines."""
3from __future__ import annotations
5from dataclasses import dataclass
6from decimal import Decimal, DecimalException
7from typing import Any
9from core.auxiliary.models.PropertyInfo import PropertyInfo
10from core.auxiliary.services.result_summary.aggregation import NumericAggregateFunction
11from core.auxiliary.services.result_summary.aggregation import aggregate_values
12from Economics.costing.cost_curves.evaluation import CostCurveEvaluationError
13from Economics.costing.cost_curves.driver_specs import CapitalCostDriverInput, CostCurveDriverSpec
14from Economics.scheduling.series import ScheduleSeriesResolution, study_schedule_property_resolution
15from Economics.shared.choices import EconomicsScheduleMode
16from Economics.studies.models import EconomicsStudy
17from idaes_factory.unit_conversion.unit_conversion import convert_value
18from pint.errors import PintError
21DEFAULT_CAPITAL_AGGREGATE = "max"
24@dataclass(frozen=True)
25class ScheduleSizingResult:
26 """Schedule-derived sizing value and audit metadata for one driver input."""
28 value: Decimal
29 unit: str
30 audit_payload: dict[str, Any]
33def resolve_schedule_driver_input(
34 *,
35 study: EconomicsStudy,
36 spec: CostCurveDriverSpec,
37 driver_input: CapitalCostDriverInput,
38 property_info: PropertyInfo,
39 resolution: ScheduleSeriesResolution | None = None,
40) -> ScheduleSizingResult | None:
41 """Return schedule aggregate sizing when the selected property varies by schedule."""
43 schedule_selected = (
44 (study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id)
45 or study.schedule_mode == EconomicsScheduleMode.COMPOSITE
46 )
47 if not schedule_selected or driver_input.source != "property":
48 return None
50 if resolution is None:
51 resolution = study_schedule_property_resolution(study=study, property_info=property_info)
52 if not resolution.schedule_varying: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 return None
54 if not resolution.points or any(point.value is None for point in resolution.points):
55 raise CostCurveEvaluationError(
56 "missing_schedule_capital_input",
57 resolution.message
58 or "This sizing property is not available in the selected production schedule.",
59 context={
60 "property_info_id": property_info.pk,
61 "input_key": spec.key,
62 "scenario_id": study.schedule_scenario_id,
63 "schedule_mode": study.schedule_mode,
64 },
65 )
67 converted_values = [
68 _convert_schedule_value(
69 value=point.value,
70 source_unit=resolution.unit or property_info.unit,
71 target_unit=driver_input.unit or spec.unit,
72 input_key=spec.key,
73 property_info=property_info,
74 )
75 for point in resolution.points
76 ]
77 aggregate_function = driver_input.aggregate_function or DEFAULT_CAPITAL_AGGREGATE
78 aggregate_percentile = _decimal_text(driver_input.aggregate_percentile or "50")
79 adjustment_percent = _decimal_text(driver_input.aggregate_adjustment_percent or "0")
80 raw_value = _aggregate_schedule_values(
81 resolution=resolution,
82 values=converted_values,
83 aggregate_function=aggregate_function,
84 aggregate_percentile=aggregate_percentile,
85 )
86 adjusted_value = raw_value * (Decimal("1") + (adjustment_percent / Decimal("100")))
87 unit = driver_input.unit or spec.unit
88 return ScheduleSizingResult(
89 value=adjusted_value,
90 unit=unit,
91 audit_payload={
92 "aggregate": aggregate_function,
93 "percentile": str(aggregate_percentile) if aggregate_function == "percentile" else None,
94 "adjustment_percent": str(adjustment_percent),
95 "raw_value": str(raw_value),
96 "adjusted_value": str(adjusted_value),
97 "unit": unit,
98 "scenario": resolution.scenario_id,
99 "schedule_mode": study.schedule_mode,
100 "property_info": resolution.property_info_id,
101 "property": resolution.property_name,
102 "source": resolution.source,
103 "row_count": resolution.row_count,
104 "values": [str(value) for value in converted_values],
105 "durations": [str(point.interval_hours) for point in resolution.points],
106 },
107 )
110def _aggregate_schedule_values(
111 *,
112 resolution: ScheduleSeriesResolution,
113 values: list[Decimal],
114 aggregate_function: str,
115 aggregate_percentile: Decimal,
116) -> Decimal:
117 """Aggregate schedule values, duration-weighting composite mean/percentile inputs."""
119 function = NumericAggregateFunction(str(aggregate_function or "").strip().lower())
120 if resolution.source != "composite" or function in {NumericAggregateFunction.MIN, NumericAggregateFunction.MAX}:
121 return aggregate_values(
122 values,
123 aggregate_function,
124 percentile=aggregate_percentile if function == NumericAggregateFunction.PERCENTILE else None,
125 )
126 weights = [point.interval_hours for point in resolution.points]
127 if function == NumericAggregateFunction.MEAN: 127 ↛ 129line 127 didn't jump to line 129 because the condition on line 127 was always true
128 return _weighted_mean(values=values, weights=weights)
129 if function == NumericAggregateFunction.PERCENTILE:
130 return _weighted_percentile(values=values, weights=weights, percentile=aggregate_percentile)
131 return aggregate_values(values, aggregate_function)
134def _weighted_mean(*, values: list[Decimal], weights: list[Decimal]) -> Decimal:
135 total_weight = sum(weights, Decimal("0"))
136 if total_weight <= 0: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 return aggregate_values(values, NumericAggregateFunction.MEAN)
138 weighted_sum = sum((value * weight for value, weight in zip(values, weights, strict=True)), Decimal("0"))
139 return weighted_sum / total_weight
142def _weighted_percentile(*, values: list[Decimal], weights: list[Decimal], percentile: Decimal) -> Decimal:
143 total_weight = sum(weights, Decimal("0"))
144 if total_weight <= 0:
145 return aggregate_values(values, NumericAggregateFunction.PERCENTILE, percentile=percentile)
146 target = total_weight * (percentile / Decimal("100"))
147 cumulative = Decimal("0")
148 ordered_values = sorted(zip(values, weights, strict=True), key=lambda item: item[0])
149 for value, weight in ordered_values:
150 cumulative += weight
151 if cumulative >= target:
152 return value
153 return ordered_values[-1][0]
156def _convert_schedule_value(
157 *,
158 value: Decimal,
159 source_unit: str,
160 target_unit: str,
161 input_key: str,
162 property_info: PropertyInfo,
163) -> Decimal:
164 if not source_unit or not target_unit or source_unit == target_unit: 164 ↛ 166line 164 didn't jump to line 166 because the condition on line 164 was always true
165 return value
166 try:
167 return Decimal(str(convert_value(value, from_unit=source_unit, to_unit=target_unit)))
168 except (ValueError, DecimalException, PintError) as exc:
169 raise CostCurveEvaluationError(
170 "unsupported_schedule_capital_input_unit",
171 "This sizing property cannot be converted for production-schedule capital costing.",
172 context={
173 "input_key": input_key,
174 "property_info_id": property_info.pk,
175 "source_unit": source_unit,
176 "target_unit": target_unit,
177 },
178 ) from exc
181def _decimal_text(value: str) -> Decimal:
182 decimal_value = Decimal(str(value))
183 if not decimal_value.is_finite(): 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 raise ValueError("Schedule aggregate values must be finite.")
185 return decimal_value