Coverage for backend/django/Economics/costing/cost_curves/driver_specs.py: 85%
239 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Typed driver-input and variant contracts for economics cost curves."""
3from __future__ import annotations
5from decimal import Decimal
6import re
7from typing import Any, Iterable, Literal, TypeAlias
9from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator, model_validator
11from Economics.costing.cost_curves.evaluation import normalize_economics_unit_notation
12from Economics.costing.cost_curves.unit_options import cost_curve_input_unit_options
14UnitOptionPayload: TypeAlias = dict[str, str]
15CostCurveDriverSpecPayload: TypeAlias = dict[str, str | bool | list[str] | None]
16CostCurveDriverSpecReadPayload: TypeAlias = dict[
17 str, str | bool | list[str] | list[UnitOptionPayload] | None
18]
19CapitalCostDriverInputPayload: TypeAlias = dict[str, str | int | None]
20CapitalCostDriverInputsPayload: TypeAlias = dict[str, CapitalCostDriverInputPayload]
21CostCurveDiscreteVariantPayload: TypeAlias = dict[str, str | dict[str, str]]
24CostCurveDriverRole = Literal["formula_input", "discrete_selector"]
25CostCurveDriverSource = Literal["property", "manual"]
26CapitalScheduleAggregateFunction = Literal["", "min", "max", "mean", "percentile"]
29class CostCurveDriverSpec(BaseModel):
30 """One declared input needed to size a cost curve.
32 Cost curves still store this in JSON so templates and user-authored curves
33 can remain declarative, but all reads/writes are normalized through this
34 model before the JSON reaches the API or generated capital-line payloads.
35 """
37 model_config = ConfigDict(frozen=True, extra="forbid")
39 key: str
40 label: str
41 unit: str
42 role: CostCurveDriverRole
43 variable_symbol: str = ""
44 required: bool = True
45 primary: bool = False
46 valid_min: str = ""
47 valid_max: str = ""
48 valid_range_note: str = ""
49 soft_maximum_adjustment_percent: str = ""
50 default_manual_value: str = ""
51 source_options: tuple[CostCurveDriverSource, ...] = Field(
52 default=("property", "manual"),
53 min_length=1,
54 )
56 @field_validator("key", "label", "unit")
57 @classmethod
58 def _required_text(cls, value: str) -> str:
59 value = str(value or "").strip()
60 if not value: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 raise ValueError("Required driver spec fields cannot be blank.")
62 return value
64 @field_validator(
65 "variable_symbol",
66 "valid_min",
67 "valid_max",
68 "valid_range_note",
69 "soft_maximum_adjustment_percent",
70 "default_manual_value",
71 mode="before",
72 )
73 @classmethod
74 def _optional_text(cls, value: Any) -> str:
75 return "" if value is None else str(value).strip()
77 @field_validator("unit")
78 @classmethod
79 def _normalize_unit(cls, value: str) -> str:
80 return normalize_economics_unit_notation(value)
82 @model_validator(mode="after")
83 def _validate_role_fields(self) -> "CostCurveDriverSpec":
84 if self.role == "formula_input" and not self.variable_symbol: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 raise ValueError("Formula input driver specs require variable_symbol.")
86 if self.role == "discrete_selector" and self.variable_symbol: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 raise ValueError("Discrete selector driver specs cannot define variable_symbol.")
88 for field_name in ("valid_min", "valid_max", "soft_maximum_adjustment_percent"):
89 value = getattr(self, field_name)
90 if value:
91 _parse_decimal_text(value, field_name=field_name)
92 return self
95class CostCurveDriverSpecRead(CostCurveDriverSpec):
96 """Read contract for driver specs with backend-provided unit choices."""
98 model_config = ConfigDict(frozen=True, extra="forbid", title="CostCurveDriverSpec")
100 unit_options: list[UnitOptionPayload] = Field(default_factory=list)
103class CostCurveDiscreteVariant(BaseModel):
104 """One published curve candidate inside a discrete-family cost curve."""
106 model_config = ConfigDict(frozen=True, extra="forbid")
108 key: str
109 label: str
110 selector_values: dict[str, str]
111 expression_text: str
112 valid_min: str = ""
113 valid_max: str = ""
114 valid_range_note: str = ""
115 source_reference: str = ""
116 notes: str = ""
118 @field_validator("key", "label", "expression_text")
119 @classmethod
120 def _required_text(cls, value: str) -> str:
121 value = str(value or "").strip()
122 if not value: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 raise ValueError("Required discrete variant fields cannot be blank.")
124 return value
126 @field_validator("valid_min", "valid_max", "valid_range_note", "source_reference", "notes", mode="before")
127 @classmethod
128 def _optional_text(cls, value: Any) -> str:
129 return "" if value is None else str(value).strip()
131 @field_validator("selector_values")
132 @classmethod
133 def _selector_values(cls, value: dict[str, Any]) -> dict[str, str]:
134 normalized = {}
135 for key, raw_value in value.items():
136 selector_key = str(key or "").strip()
137 selector_value = str(raw_value or "").strip()
138 if not selector_key or not selector_value: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 raise ValueError("Discrete variant selector values cannot be blank.")
140 _parse_decimal_text(selector_value, field_name=f"selector_values.{selector_key}")
141 normalized[selector_key] = selector_value
142 return normalized
144 @model_validator(mode="after")
145 def _validate_optional_numbers(self) -> "CostCurveDiscreteVariant":
146 for field_name in ("valid_min", "valid_max"):
147 value = getattr(self, field_name)
148 if value:
149 _parse_decimal_text(value, field_name=field_name)
150 return self
153class CapitalCostDriverInput(BaseModel):
154 """One persisted property/manual value selection for a capital-line driver."""
156 model_config = ConfigDict(frozen=True, extra="forbid")
158 source: Literal["", "property", "manual"] = ""
159 property_info: int | None = None
160 manual_value: str = ""
161 unit: str
162 aggregate_function: CapitalScheduleAggregateFunction = ""
163 aggregate_percentile: str = "50"
164 aggregate_adjustment_percent: str = "0"
166 @classmethod
167 def from_spec_default(cls, spec: CostCurveDriverSpec) -> "CapitalCostDriverInput":
168 """Create the blank/default capital-line input for a declared spec."""
169 if "manual" in spec.source_options and spec.default_manual_value not in (None, ""):
170 return cls(
171 source="manual",
172 property_info=None,
173 manual_value=spec.default_manual_value,
174 unit=spec.unit,
175 )
176 if spec.source_options == ("manual",):
177 return cls(source="", property_info=None, manual_value="", unit=spec.unit)
178 return cls(
179 source="",
180 property_info=None,
181 manual_value="",
182 unit=spec.unit,
183 )
185 @field_validator("manual_value", mode="before")
186 @classmethod
187 def _manual_value_text(cls, value: Any) -> str:
188 return "" if value is None else str(value).strip()
190 @field_validator("aggregate_percentile", "aggregate_adjustment_percent", mode="before")
191 @classmethod
192 def _aggregate_decimal_text(cls, value: Any) -> str:
193 return "" if value is None else str(value).strip()
195 @field_validator("unit")
196 @classmethod
197 def _input_unit(cls, value: str) -> str:
198 value = str(value or "").strip()
199 if not value: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 raise ValueError("Capital cost driver input unit cannot be blank.")
201 return normalize_economics_unit_notation(value)
203 @model_validator(mode="after")
204 def _validate_schedule_aggregate_fields(self) -> "CapitalCostDriverInput":
205 percentile = _parse_decimal_text(
206 self.aggregate_percentile or "50",
207 field_name="aggregate_percentile",
208 )
209 if percentile < 0 or percentile > 100: 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true
210 raise ValueError("aggregate_percentile must be between 0 and 100.")
211 adjustment_percent = _parse_decimal_text(
212 self.aggregate_adjustment_percent or "0",
213 field_name="aggregate_adjustment_percent",
214 )
215 if adjustment_percent < 0:
216 raise ValueError("aggregate_adjustment_percent must be greater than or equal to 0.")
217 if self.aggregate_function and self.source != "property": 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 raise ValueError("Schedule aggregate fields require a property-backed driver input.")
219 return self
222_DRIVER_SPECS_ADAPTER = TypeAdapter(list[CostCurveDriverSpec])
223_DRIVER_INPUTS_ADAPTER = TypeAdapter(dict[str, CapitalCostDriverInput])
224_DISCRETE_VARIANTS_ADAPTER = TypeAdapter(list[CostCurveDiscreteVariant])
227def parse_required_driver_specs(specs: Any) -> tuple[CostCurveDriverSpec, ...]:
228 """Validate and return the typed required driver spec list.
230 The uniqueness check is outside the single-spec Pydantic model because it
231 depends on the whole list and keeps ``CapitalCostLine.driver_inputs`` keyed
232 deterministically by spec key.
233 """
234 if specs in (None, ""): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 return ()
236 parsed_specs = tuple(
237 _DRIVER_SPECS_ADAPTER.validate_python(_specs_with_generated_keys(specs))
238 )
239 seen_keys: set[str] = set()
240 seen_symbols: set[str] = set()
241 formula_input_count = 0
242 primary_count = 0
243 for spec in parsed_specs:
244 if spec.key in seen_keys: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise ValueError(f"Driver spec key `{spec.key}` is duplicated.")
246 seen_keys.add(spec.key)
247 if spec.role == "formula_input":
248 formula_input_count += 1
249 if spec.variable_symbol in seen_symbols: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 raise ValueError(f"Formula variable `{spec.variable_symbol}` is duplicated.")
251 seen_symbols.add(spec.variable_symbol)
252 primary_count += 1 if spec.primary else 0
253 elif spec.primary: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 raise ValueError("Only formula input driver specs can be primary.")
255 if formula_input_count == 0: 255 ↛ 256line 255 didn't jump to line 256 because the condition on line 255 was never true
256 raise ValueError("Cost curves require at least one formula input driver spec.")
257 if primary_count != 1: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 raise ValueError("Cost curves require exactly one primary formula input driver spec.")
259 return parsed_specs
262def normalize_required_driver_specs(specs: Any) -> list[CostCurveDriverSpecPayload]:
263 """Validate and serialize the JSON-ready driver spec list."""
264 return [driver_spec_payload(spec) for spec in parse_required_driver_specs(specs)]
267def _specs_with_generated_keys(specs: Any) -> Any:
268 """Fill missing spec keys at the API boundary.
270 Key values are internal identifiers for stored capital-line driver inputs.
271 User-authored payloads do not need to provide them, but existing template
272 and stored keys are preserved exactly so saved selections stay stable.
273 """
274 if not isinstance(specs, list): 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true
275 return specs
276 seen_keys: set[str] = set()
277 normalized_specs: list[Any] = []
278 for index, spec in enumerate(specs):
279 if not isinstance(spec, dict): 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true
280 normalized_specs.append(spec)
281 continue
282 next_spec = {**spec}
283 next_spec.pop("unit_options", None)
284 key = str(next_spec.get("key") or "").strip()
285 if not key:
286 key = _generated_driver_spec_key(next_spec, index=index, seen_keys=seen_keys)
287 next_spec["key"] = key
288 seen_keys.add(key)
289 normalized_specs.append(next_spec)
290 return normalized_specs
293def _generated_driver_spec_key(
294 spec: dict[str, Any], *, index: int, seen_keys: set[str]
295) -> str:
296 base = (
297 _slug_value(str(spec.get("label") or ""))
298 or _slug_value(str(spec.get("variable_symbol") or ""))
299 or f"input_{index + 1}"
300 )
301 candidate = base
302 suffix = 2
303 while candidate in seen_keys:
304 candidate = f"{base}_{suffix}"
305 suffix += 1
306 return candidate
309def _slug_value(value: str) -> str:
310 return re.sub(
311 r"(^_+|_+$)",
312 "",
313 re.sub(r"[^a-z0-9]+", "_", value.strip().lower()),
314 )
317def parse_discrete_variants(variants: Any) -> tuple[CostCurveDiscreteVariant, ...]:
318 """Validate and return typed discrete-family variant rows."""
319 if variants in (None, ""): 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 return ()
321 parsed_variants = tuple(_DISCRETE_VARIANTS_ADAPTER.validate_python(variants))
322 seen_keys: set[str] = set()
323 for variant in parsed_variants:
324 if variant.key in seen_keys: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
325 raise ValueError(f"Discrete variant key `{variant.key}` is duplicated.")
326 seen_keys.add(variant.key)
327 return parsed_variants
330def normalize_discrete_variants(variants: Any) -> list[CostCurveDiscreteVariantPayload]:
331 """Validate and serialize discrete-family variant rows for JSON storage."""
332 return [discrete_variant_payload(variant) for variant in parse_discrete_variants(variants)]
335def normalize_capital_cost_driver_inputs(
336 inputs: Any,
337) -> CapitalCostDriverInputsPayload:
338 """Validate and serialize keyed capital-line driver inputs.
340 `CapitalCostLine.driver_inputs` is a JSON object keyed by required driver
341 spec key. Keeping this path Pydantic-backed gives the API a precise schema
342 without weakening writes to arbitrary JSON.
343 """
344 if inputs in (None, ""): 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 return {}
346 return {
347 key: capital_cost_driver_input_payload(driver_input)
348 for key, driver_input in _DRIVER_INPUTS_ADAPTER.validate_python(inputs).items()
349 }
352def driver_specs_payload(
353 specs: Iterable[CostCurveDriverSpec],
354) -> list[CostCurveDriverSpecPayload]:
355 """Return JSON-ready driver specs from already-validated spec models."""
356 return [driver_spec_payload(spec) for spec in specs]
359def driver_specs_read_payload(
360 specs: Iterable[CostCurveDriverSpec],
361) -> list[CostCurveDriverSpecReadPayload]:
362 """Return API driver specs with read-only unit options attached."""
363 return [driver_spec_read_payload(spec) for spec in specs]
366def default_driver_inputs_payload(
367 specs: Iterable[CostCurveDriverSpec],
368) -> CapitalCostDriverInputsPayload:
369 """Build JSON-ready default capital-line driver inputs from typed specs."""
370 return {
371 spec.key: capital_cost_driver_input_payload(
372 CapitalCostDriverInput.from_spec_default(spec)
373 )
374 for spec in specs
375 }
378def driver_spec_payload(spec: CostCurveDriverSpec) -> CostCurveDriverSpecPayload:
379 """Serialize a typed driver spec without widening the payload type to Any."""
380 return {
381 "key": spec.key,
382 "label": spec.label,
383 "role": spec.role,
384 "variable_symbol": spec.variable_symbol,
385 "unit": spec.unit,
386 "required": spec.required,
387 "primary": spec.primary,
388 "valid_min": spec.valid_min,
389 "valid_max": spec.valid_max,
390 "valid_range_note": spec.valid_range_note,
391 "soft_maximum_adjustment_percent": spec.soft_maximum_adjustment_percent,
392 "default_manual_value": spec.default_manual_value,
393 "source_options": list(spec.source_options),
394 }
397def driver_spec_read_payload(spec: CostCurveDriverSpec) -> CostCurveDriverSpecReadPayload:
398 """Serialize one driver spec for API reads without changing stored JSON."""
399 return CostCurveDriverSpecRead(
400 **driver_spec_payload(spec),
401 unit_options=cost_curve_input_unit_options(spec.unit),
402 ).model_dump(mode="json")
405def discrete_variant_payload(variant: CostCurveDiscreteVariant) -> CostCurveDiscreteVariantPayload:
406 """Serialize one discrete variant without widening the payload type to Any."""
407 return {
408 "key": variant.key,
409 "label": variant.label,
410 "selector_values": variant.selector_values,
411 "expression_text": variant.expression_text,
412 "valid_min": variant.valid_min,
413 "valid_max": variant.valid_max,
414 "valid_range_note": variant.valid_range_note,
415 "source_reference": variant.source_reference,
416 "notes": variant.notes,
417 }
420def capital_cost_driver_input_payload(
421 driver_input: CapitalCostDriverInput,
422) -> CapitalCostDriverInputPayload:
423 """Serialize one typed capital-line driver input without untyped dict access."""
424 return {
425 "source": driver_input.source,
426 "property_info": driver_input.property_info,
427 "manual_value": driver_input.manual_value,
428 "unit": driver_input.unit,
429 "aggregate_function": driver_input.aggregate_function,
430 "aggregate_percentile": driver_input.aggregate_percentile or "50",
431 "aggregate_adjustment_percent": driver_input.aggregate_adjustment_percent or "0",
432 }
435def _parse_decimal_text(value: str, *, field_name: str) -> Decimal:
436 try:
437 decimal_value = Decimal(str(value))
438 except Exception as exc:
439 raise ValueError(f"{field_name} must be a number.") from exc
440 if not decimal_value.is_finite(): 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 raise ValueError(f"{field_name} must be finite.")
442 return decimal_value