Coverage for backend/django/Economics/costing/capital/generated_lines.py: 87%

1"""Generated capital-line synchronization for capital costing. 

2 

3This module materializes generated ``CapitalCostLine`` rows before 

4fingerprinting and result calculation. Rows are created as soon as an equipment 

5mapping exists so work-capable units can expose editable peak demand even before 

6the user has selected a cost curve. 

7""" 

8 

9from __future__ import annotations 

10 

11from decimal import Decimal, DecimalException 

12from typing import Any, NamedTuple, TypedDict 

13 

14from django.db.models import Prefetch 

15from django.utils import timezone 

16 

17from core.auxiliary.enums import ConType 

18from core.auxiliary.models.PropertyInfo import PropertyInfo 

19from Economics.costing.models import CapitalCostLine, CostCurve, CostDriver, EquipmentMapping 

20 

21from Economics.shared.choices import CostBasis, CostDriverSource 

22 

23from Economics.studies.models import EconomicsStudy 

24from Economics.costing.capital.capital_line_sources import GENERATED_CAPITAL_LINE_SOURCE 

25from Economics.costing.cost_curves.driver_specs import ( 

26 CapitalCostDriverInput, 

27 CapitalCostDriverInputsPayload, 

28 CostCurveDriverSpec, 

29 CostCurveDriverSpecPayload, 

30 default_driver_inputs_payload, 

31 driver_specs_payload, 

32 normalize_capital_cost_driver_inputs, 

33 parse_required_driver_specs, 

34) 

35from Economics.costing.cost_curves.evaluation import ( 

36 CostCurveEvaluationError, 

37 cost_curve_units_compatible, 

38 evaluate_cost_curve, 

39 normalize_economics_unit_notation, 

40) 

41from Economics.costing.capital.electrical_upgrade import PeakDemandScheduleError, unit_work_peak_demand_kw 

42from Economics.costing.capital.schedule_sizing import resolve_schedule_driver_input 

43from Economics.costing.cost_curves.driver_properties import validate_cost_driver_property 

44from Economics.formulas.builders.capital import build_generated_capital_line_formula 

45from Economics.formulas.engine.core import FormulaError 

46from Economics.shared.payloads import json_ready, result_amount, warning_record 

47from Economics.scheduling.series import study_schedule_property_resolutions 

48from idaes_factory.unit_conversion.unit_conversion import convert_value 

49from pint.errors import PintError 

50 

51 

52class ResolvedDriverInput(TypedDict): 

53 value: Decimal | float | int | str 

54 unit: str 

55 

56 

57class ResolvedDriverInputs(NamedTuple): 

58 inputs: dict[str, ResolvedDriverInput] 

59 schedule_inputs: dict[str, dict[str, Any]] 

60 

61 

62def sync_generated_capital_lines(study: EconomicsStudy) -> bool: 

63 """Materialize generated capital lines before result fingerprinting. 

64 

65 Fingerprints include capital line rows, so generated rows have to be updated 

66 first or the dependency comparison would describe stale generated costs. 

67 """ 

68 changed = False 

69 mappings = list( 

70 _generated_line_mapping_queryset().filter( 

71 flowsheet_state=study.flowsheet_state, 

72 costable_item__study=study, 

73 costable_item__simulation_object__is_deleted=False, 

74 ) 

75 .order_by("pk") 

76 ) 

77 active_costable_item_ids = set() 

78 pending_lines: list[CapitalCostLine] = [] 

79 pending_updates: list[CapitalCostLine] = [] 

80 calculation_cache: dict = {} 

81 _prime_schedule_resolution_cache(mappings, calculation_cache) 

82 for mapping in mappings: 

83 active_costable_item_ids.add(mapping.costable_item_id) 

84 changed = _sync_generated_capital_line( 

85 mapping, 

86 pending_lines=pending_lines, 

87 pending_updates=pending_updates, 

88 calculation_cache=calculation_cache, 

89 ) or changed 

90 if pending_lines: 

91 CapitalCostLine.objects.bulk_create(pending_lines) 

92 _bulk_update_generated_lines(pending_updates) 

93 stale_lines = CapitalCostLine.objects.filter( 

94 flowsheet_state=study.flowsheet_state, 

95 study=study, 

96 source=GENERATED_CAPITAL_LINE_SOURCE, 

97 ) 

98 if active_costable_item_ids: 

99 stale_lines = stale_lines.exclude(costable_item_id__in=active_costable_item_ids) 

100 deleted_count, _ = stale_lines.delete() 

101 return changed or deleted_count > 0 

102 

103 

104def sync_generated_capital_lines_for_mappings(mappings: list[EquipmentMapping]) -> bool: 

105 """Refresh generated lines for mappings changed by a targeted mutation.""" 

106 

107 mapping_ids = [mapping.pk for mapping in mappings] 

108 if not mapping_ids: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 return False 

110 hydrated_mappings = list( 

111 _generated_line_mapping_queryset().filter(pk__in=mapping_ids).order_by("pk") 

112 ) 

113 changed = False 

114 pending_lines: list[CapitalCostLine] = [] 

115 pending_updates: list[CapitalCostLine] = [] 

116 calculation_cache: dict = {} 

117 _prime_schedule_resolution_cache(hydrated_mappings, calculation_cache) 

118 for mapping in hydrated_mappings: 

119 changed = _sync_generated_capital_line( 

120 mapping, 

121 pending_lines=pending_lines, 

122 pending_updates=pending_updates, 

123 calculation_cache=calculation_cache, 

124 ) or changed 

125 if pending_lines: 

126 CapitalCostLine.objects.bulk_create(pending_lines) 

127 _bulk_update_generated_lines(pending_updates) 

128 return changed 

129 

130 

131def _generated_line_mapping_queryset(): 

132 """Return mappings with the graph required to evaluate generated lines in bulk.""" 

133 

134 generated_lines = CapitalCostLine.objects.filter( 

135 source=GENERATED_CAPITAL_LINE_SOURCE, 

136 ).select_related("cost_curve").order_by("pk") 

137 properties = ( 

138 PropertyInfo.objects.select_related("set", "set__simulationObject") 

139 .prefetch_related("values") 

140 .order_by("pk") 

141 ) 

142 return EquipmentMapping.objects.select_related( 

143 "costable_item", 

144 "costable_item__study", 

145 "costable_item__study__settings_profile", 

146 "costable_item__study__schedule_scenario", 

147 "costable_item__simulation_object", 

148 "costable_item__simulation_object__properties", 

149 "costable_item__cost_driver", 

150 "costable_item__cost_driver__property_info", 

151 "cost_curve", 

152 ).prefetch_related( 

153 Prefetch( 

154 "costable_item__capital_lines", 

155 queryset=generated_lines, 

156 to_attr="_economics_generated_capital_lines", 

157 ), 

158 Prefetch( 

159 "costable_item__simulation_object__properties__ContainedProperties", 

160 queryset=properties, 

161 to_attr="_economics_contained_properties", 

162 ), 

163 "costable_item__cost_driver__property_info__values", 

164 ) 

165 

166 

167def _sync_generated_capital_line( 

168 mapping: EquipmentMapping, 

169 *, 

170 pending_lines: list[CapitalCostLine] | None = None, 

171 pending_updates: list[CapitalCostLine] | None = None, 

172 calculation_cache: dict | None = None, 

173) -> bool: 

174 """Create or update the single generated capital line for one equipment mapping.""" 

175 driver = getattr(mapping.costable_item, "cost_driver", None) 

176 prefetched_lines = getattr(mapping.costable_item, "_economics_generated_capital_lines", None) 

177 generated_lines = ( 

178 prefetched_lines 

179 if prefetched_lines is not None 

180 else list( 

181 CapitalCostLine.objects.filter( 

182 flowsheet_state=mapping.flowsheet_state, 

183 study=mapping.costable_item.study, 

184 costable_item=mapping.costable_item, 

185 source=GENERATED_CAPITAL_LINE_SOURCE, 

186 ).order_by("pk") 

187 ) 

188 ) 

189 line = generated_lines[0] if generated_lines else None 

190 changed = False 

191 for duplicate in generated_lines[1:]: 191 ↛ 192line 191 didn't jump to line 192 because the loop on line 191 never started

192 duplicate.delete() 

193 changed = True 

194 

195 fields = _generated_capital_line_fields( 

196 mapping=mapping, 

197 driver=driver, 

198 existing_line=line, 

199 calculation_cache=calculation_cache, 

200 ) 

201 if line is None: 

202 fields["peak_demand_kw"] = fields.get("minimum_peak_demand_kw") 

203 new_line = CapitalCostLine( 

204 flowsheet_state=mapping.flowsheet_state, 

205 study=mapping.costable_item.study, 

206 costable_item=mapping.costable_item, 

207 **fields, 

208 ) 

209 if pending_lines is None: 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true

210 new_line.save() 

211 else: 

212 pending_lines.append(new_line) 

213 return True 

214 

215 minimum_peak_demand_kw = fields.get("minimum_peak_demand_kw") 

216 selected_peak_demand_kw = line.peak_demand_kw 

217 previous_minimum_peak_demand_kw = line.minimum_peak_demand_kw 

218 selected_peak_demand_is_auto = selected_peak_demand_kw == previous_minimum_peak_demand_kw 

219 if minimum_peak_demand_kw is None: 

220 fields["peak_demand_kw"] = None 

221 elif ( 

222 selected_peak_demand_kw is None 

223 or selected_peak_demand_is_auto 

224 or selected_peak_demand_kw < minimum_peak_demand_kw 

225 ): 

226 fields["peak_demand_kw"] = minimum_peak_demand_kw 

227 

228 changed_fields = [] 

229 for field_name, value in fields.items(): 

230 if getattr(line, field_name) != value: 

231 setattr(line, field_name, value) 

232 changed_fields.append(field_name) 

233 if changed_fields: 

234 if pending_updates is None: 

235 line.save(update_fields=[*changed_fields, "updated_at"]) 

236 else: 

237 line.updated_at = timezone.now() 

238 pending_updates.append(line) 

239 changed = True 

240 return changed 

241 

242 

243def _bulk_update_generated_lines(lines: list[CapitalCostLine]) -> None: 

244 """Persist generated-line refreshes without per-line model saves.""" 

245 if not lines: 

246 return 

247 CapitalCostLine.objects.bulk_update( 

248 lines, 

249 fields=( 

250 "cost_curve", "label", "line_type", "amount", "currency", "included", 

251 "manual", "source", "confidence", "peak_demand_kw", 

252 "minimum_peak_demand_kw", "driver_inputs", "warning_payload", "updated_at", 

253 ), 

254 ) 

255 

256 

257def sync_generated_capital_lines_for_property(property_info) -> list[EconomicsStudy]: 

258 """Refresh generated economics rows affected by an edited flowsheet property.""" 

259 property_set = getattr(property_info, "set", None) 

260 simulation_object = getattr(property_set, "simulationObject", None) 

261 changed_studies = [] 

262 study_ids: set[int] = set() 

263 flowsheet_studies = EconomicsStudy.objects.filter(flowsheet_state=property_info.flowsheet_state) 

264 if simulation_object is not None: 

265 study_ids.update( 

266 flowsheet_studies.filter( 

267 costable_items__simulation_object=simulation_object, 

268 ).values_list("pk", flat=True) 

269 ) 

270 connected_unit_ids = simulation_object.connectedPorts.filter( 

271 unitOp__isnull=False, 

272 ).values_list("unitOp_id", flat=True) 

273 study_ids.update( 

274 flowsheet_studies.filter( 

275 costable_items__simulation_object_id__in=connected_unit_ids, 

276 ).values_list("pk", flat=True) 

277 ) 

278 referenced_study_ids = set() 

279 for line in CapitalCostLine.objects.filter( 

280 flowsheet_state=property_info.flowsheet_state, 

281 source=GENERATED_CAPITAL_LINE_SOURCE, 

282 ).only("study_id", "driver_inputs"): 

283 if _driver_inputs_reference_property(line.driver_inputs, property_info.pk): 

284 referenced_study_ids.add(line.study_id) 

285 input_studies = EconomicsStudy.objects.filter( 

286 pk__in=referenced_study_ids, 

287 flowsheet_state=property_info.flowsheet_state, 

288 ) 

289 study_ids.update(input_studies.values_list("pk", flat=True)) 

290 for study in flowsheet_studies.filter(pk__in=study_ids).order_by("pk"): 

291 if sync_generated_capital_lines(study): 291 ↛ 290line 291 didn't jump to line 290 because the condition on line 291 was always true

292 changed_studies.append(study) 

293 return changed_studies 

294 

295 

296def _generated_capital_line_fields( 

297 *, 

298 mapping: EquipmentMapping, 

299 driver: CostDriver | None, 

300 existing_line: CapitalCostLine | None = None, 

301 calculation_cache: dict | None = None, 

302) -> dict[str, Any]: 

303 """Return persisted fields for the generated capital line tied to one mapping.""" 

304 curve = mapping.cost_curve 

305 peak_demand_blocked_reason = "" 

306 try: 

307 minimum_peak_demand_kw = unit_work_peak_demand_kw( 

308 mapping.costable_item, 

309 study=mapping.costable_item.study, 

310 bulk_property_values=True, 

311 ) 

312 except PeakDemandScheduleError as exc: 

313 minimum_peak_demand_kw = None 

314 peak_demand_blocked_reason = exc.message 

315 if curve is None: 

316 return _pending_generated_capital_line_fields( 

317 mapping=mapping, 

318 minimum_peak_demand_kw=minimum_peak_demand_kw, 

319 peak_demand_blocked_reason=peak_demand_blocked_reason, 

320 ) 

321 

322 warning_payload: dict[str, Any] = { 

323 "calculation_method": "cost_curve", 

324 "cost_curve_id": curve.pk, 

325 "cost_curve_key": curve.curve_key, 

326 "cost_basis": curve.cost_basis, 

327 "required_driver_specs": _curve_required_driver_specs_payload(curve), 

328 "minimum_peak_demand_kw": None if minimum_peak_demand_kw is None else str(minimum_peak_demand_kw), 

329 "peak_demand_blocked_reason": peak_demand_blocked_reason, 

330 "warnings": [], 

331 } 

332 if peak_demand_blocked_reason: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true

333 warning_payload["warnings"].append( 

334 warning_record( 

335 code="missing_schedule_peak_demand", 

336 severity="warning", 

337 message=peak_demand_blocked_reason, 

338 context={"costable_item_id": mapping.costable_item_id}, 

339 ) 

340 ) 

341 amount = None 

342 confidence = "calculated" 

343 try: 

344 driver_inputs = _reconciled_driver_inputs( 

345 curve=curve, 

346 existing_inputs=( 

347 normalize_capital_cost_driver_inputs(existing_line.driver_inputs) 

348 if existing_line is not None and existing_line.driver_inputs 

349 else {} 

350 ), 

351 ) 

352 resolved_driver_inputs = _resolved_driver_inputs( 

353 curve=curve, 

354 driver_inputs=driver_inputs, 

355 mapping=mapping, 

356 calculation_cache=calculation_cache, 

357 ) 

358 resolved_inputs = resolved_driver_inputs.inputs 

359 generated_formula = build_generated_capital_line_formula( 

360 mapping, 

361 driver=driver, 

362 existing_line=existing_line, 

363 calculation_cache=calculation_cache, 

364 ) 

365 evaluation = evaluate_cost_curve( 

366 curve, 

367 inputs_by_key=resolved_inputs, 

368 apply_installation_factor=generated_formula.applies_lang_factor, 

369 ) 

370 base_amount = evaluation.amount 

371 if base_amount < 0: 

372 raise CostCurveEvaluationError( 

373 "negative_cost_output", 

374 "Cost curve evaluated to a negative cost.", 

375 context={"curve_key": curve.curve_key, "amount": str(base_amount), "output_unit": curve.output_unit}, 

376 ) 

377 factor_rows = [ 

378 _capital_factor_row( 

379 kind="base_curve_cost", 

380 label="Curve base cost", 

381 amount=base_amount, 

382 detail=_base_curve_detail(evaluation), 

383 ) 

384 ] 

385 indexed_amount = base_amount * generated_formula.index_adjustment.factor 

386 purchase_basis_amount = indexed_amount if curve.cost_basis == CostBasis.PURCHASE else Decimal("0") 

387 factor_rows.append( 

388 _capital_factor_row( 

389 kind="index_adjustment", 

390 label="CPI/index adjustment", 

391 amount=indexed_amount, 

392 factor=generated_formula.index_adjustment.factor, 

393 detail=generated_formula.index_adjustment.detail, 

394 ) 

395 ) 

396 installed_basis_amount = indexed_amount if curve.cost_basis == CostBasis.INSTALLED else Decimal("0") 

397 uplift_base_amount = indexed_amount 

398 if generated_formula.applies_lang_factor and generated_formula.lang_factor is not None: 398 ↛ 410line 398 didn't jump to line 410 because the condition on line 398 was always true

399 uplift_base_amount = indexed_amount * generated_formula.lang_factor 

400 installed_basis_amount = uplift_base_amount 

401 factor_rows.append( 

402 _capital_factor_row( 

403 kind="lang_factor", 

404 label="Lang factor", 

405 amount=uplift_base_amount, 

406 factor=generated_formula.lang_factor, 

407 detail=generated_formula.lang_factor_source, 

408 ) 

409 ) 

410 contingency_percent = generated_formula.contingency_percent 

411 contingency_amount = uplift_base_amount * (contingency_percent / Decimal("100")) 

412 amount = uplift_base_amount * generated_formula.contingency_factor 

413 factor_rows.append( 

414 _capital_factor_row( 

415 kind="contingency", 

416 label="Contingency", 

417 amount=amount, 

418 factor=generated_formula.contingency_factor, 

419 percent=contingency_percent, 

420 ) 

421 ) 

422 amount = result_amount(amount) 

423 warning_payload.update( 

424 { 

425 "input_value": None if evaluation.input_value is None else str(evaluation.input_value), 

426 "input_unit": evaluation.input_unit, 

427 "normalized_inputs": evaluation.normalized_inputs, 

428 "schedule_inputs": resolved_driver_inputs.schedule_inputs, 

429 "selected_variant": evaluation.selected_variant, 

430 "selector_diagnostics": list(evaluation.selector_diagnostics), 

431 "base_amount": str(result_amount(evaluation.amount)), 

432 "index_factor": str(generated_formula.index_adjustment.factor), 

433 "lang_factor": ( 

434 None 

435 if generated_formula.lang_factor is None 

436 else str(generated_formula.lang_factor) 

437 ), 

438 "lang_factor_source": generated_formula.lang_factor_source, 

439 "purchase_basis_amount": str(result_amount(purchase_basis_amount)), 

440 "installed_basis_amount": str(result_amount(installed_basis_amount)), 

441 "contingency_percent": str(contingency_percent), 

442 "contingency_amount": str(result_amount(contingency_amount)), 

443 "amount": None if amount is None else str(amount), 

444 "output_unit": evaluation.output_unit, 

445 "capital_factors": factor_rows, 

446 } 

447 ) 

448 warning_payload["warnings"].extend(evaluation.warnings_payload()) 

449 warning_payload["warnings"].extend( 

450 _flow_capacity_warnings( 

451 mapping=mapping, 

452 curve=curve, 

453 driver_inputs=driver_inputs, 

454 resolved_inputs=resolved_inputs, 

455 ) 

456 ) 

457 except (CostCurveEvaluationError, FormulaError, ValueError) as exc: 

458 warning_payload["warnings"].append(_cost_curve_error_warning(exc, mapping=mapping, driver=driver)) 

459 confidence = "blocked" 

460 

461 fields: dict[str, Any] = { 

462 "cost_curve": curve, 

463 "label": f"{mapping.costable_item.name} capital cost", 

464 "line_type": "equipment_capital", 

465 "amount": amount, 

466 "currency": curve.currency or curve.output_unit or "NZD", 

467 "included": mapping.costable_item.included, 

468 "manual": False, 

469 "source": GENERATED_CAPITAL_LINE_SOURCE, 

470 "confidence": confidence, 

471 "minimum_peak_demand_kw": minimum_peak_demand_kw, 

472 "warning_payload": json_ready(warning_payload), 

473 } 

474 # Driver inputs are user-editable on generated capital lines. Populate the 

475 # declarative defaults only for a new or still-empty row so recalculation 

476 # cannot erase the user's property/manual selections. 

477 reconciled_inputs = _reconciled_driver_inputs( 

478 curve=curve, 

479 existing_inputs=( 

480 normalize_capital_cost_driver_inputs(existing_line.driver_inputs) 

481 if existing_line is not None and existing_line.driver_inputs 

482 else {} 

483 ), 

484 ) 

485 if existing_line is None or existing_line.driver_inputs != reconciled_inputs: 

486 fields["driver_inputs"] = reconciled_inputs 

487 return fields 

488 

489 

490def _pending_generated_capital_line_fields( 

491 *, 

492 mapping: EquipmentMapping, 

493 minimum_peak_demand_kw: Decimal | None, 

494 peak_demand_blocked_reason: str = "", 

495) -> dict[str, Any]: 

496 """Return a generated capital line that is configurable but not costed yet.""" 

497 warning_payload = { 

498 "calculation_method": "cost_curve", 

499 "cost_curve_id": None, 

500 "cost_curve_key": "", 

501 "cost_basis": mapping.cost_basis, 

502 "minimum_peak_demand_kw": None if minimum_peak_demand_kw is None else str(minimum_peak_demand_kw), 

503 "peak_demand_blocked_reason": peak_demand_blocked_reason, 

504 "warnings": ( 

505 [ 

506 warning_record( 

507 code="missing_schedule_peak_demand", 

508 severity="warning", 

509 message=peak_demand_blocked_reason, 

510 context={"costable_item_id": mapping.costable_item_id}, 

511 ) 

512 ] 

513 if peak_demand_blocked_reason 

514 else [] 

515 ), 

516 } 

517 return { 

518 "cost_curve": None, 

519 "label": f"{mapping.costable_item.name} capital cost", 

520 "line_type": "equipment_capital", 

521 "amount": None, 

522 "currency": "NZD", 

523 "included": mapping.costable_item.included, 

524 "manual": False, 

525 "source": GENERATED_CAPITAL_LINE_SOURCE, 

526 "driver_inputs": {}, 

527 "confidence": "blocked", 

528 "minimum_peak_demand_kw": minimum_peak_demand_kw, 

529 "warning_payload": json_ready(warning_payload), 

530 } 

531 

532 

533def _curve_required_driver_specs(curve: CostCurve) -> tuple[CostCurveDriverSpec, ...]: 

534 """Return the selected curve's validated driver spec models.""" 

535 return parse_required_driver_specs(curve.required_driver_specs) 

536 

537 

538def _curve_required_driver_specs_payload(curve: CostCurve) -> list[CostCurveDriverSpecPayload]: 

539 """Return the selected curve's JSON-ready driver spec payload.""" 

540 return driver_specs_payload(_curve_required_driver_specs(curve)) 

541 

542 

543def _default_driver_inputs(curve: CostCurve) -> CapitalCostDriverInputsPayload: 

544 """Create JSON-ready per-spec capital-line input rows from typed specs.""" 

545 return default_driver_inputs_payload(_curve_required_driver_specs(curve)) 

546 

547 

548def _reconciled_driver_inputs( 

549 *, 

550 curve: CostCurve, 

551 existing_inputs: CapitalCostDriverInputsPayload, 

552) -> CapitalCostDriverInputsPayload: 

553 """Preserve matching driver inputs and reset keys that no longer match the curve.""" 

554 defaults = _default_driver_inputs(curve) 

555 reconciled: CapitalCostDriverInputsPayload = {} 

556 specs_by_key = {spec.key: spec for spec in _curve_required_driver_specs(curve)} 

557 for key, default_input in defaults.items(): 

558 existing_input = existing_inputs.get(key) 

559 spec = specs_by_key[key] 

560 reconciled[key] = ( 

561 existing_input 

562 if _driver_input_matches_spec(existing_input, spec) 

563 else default_input 

564 ) 

565 return reconciled 

566 

567 

568def _driver_input_matches_spec(driver_input: Any, spec: CostCurveDriverSpec) -> bool: 

569 if not isinstance(driver_input, dict): 

570 return False 

571 try: 

572 parsed_input = CapitalCostDriverInput.model_validate(driver_input) 

573 except ValueError: 

574 return False 

575 if parsed_input.source and parsed_input.source not in spec.source_options: 

576 return False 

577 return cost_curve_units_compatible(parsed_input.unit, spec.unit) 

578 

579 

580def _driver_inputs_reference_property(driver_inputs: Any, property_info_id: int) -> bool: 

581 """Return whether a spec-keyed driver-input JSON payload references a property.""" 

582 if not isinstance(driver_inputs, dict): 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true

583 return False 

584 return any( 

585 isinstance(driver_input, dict) 

586 and driver_input.get("source") == "property" 

587 and driver_input.get("property_info") == property_info_id 

588 for driver_input in driver_inputs.values() 

589 ) 

590 

591 

592def _capital_factor_row( 

593 *, 

594 kind: str, 

595 label: str, 

596 amount: Decimal, 

597 factor: Decimal | None = None, 

598 percent: Decimal | None = None, 

599 detail: str = "", 

600) -> dict[str, Any]: 

601 """Serialize one capital factor step for generated-line audit payloads.""" 

602 return { 

603 "kind": kind, 

604 "label": label, 

605 "factor": None if factor is None else str(factor), 

606 "percent": None if percent is None else str(percent), 

607 "amount": str(result_amount(amount)), 

608 "detail": detail, 

609 } 

610 

611 

612def _resolved_driver_inputs( 

613 *, 

614 curve: CostCurve, 

615 driver_inputs: CapitalCostDriverInputsPayload, 

616 mapping: EquipmentMapping, 

617 calculation_cache: dict | None = None, 

618) -> ResolvedDriverInputs: 

619 """Resolve persisted driver-input selections into evaluator-ready values.""" 

620 resolved: dict[str, ResolvedDriverInput] = {} 

621 schedule_inputs: dict[str, dict[str, Any]] = {} 

622 for spec in _curve_required_driver_specs(curve): 

623 driver_input_payload = driver_inputs.get(spec.key) 

624 if driver_input_payload is None: 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true

625 continue 

626 driver_input = CapitalCostDriverInput.model_validate(driver_input_payload) 

627 if driver_input.source == "property": 

628 property_info_id = driver_input.property_info 

629 if property_info_id is None: 629 ↛ 630line 629 didn't jump to line 630 because the condition on line 629 was never true

630 raise CostCurveEvaluationError( 

631 "missing_cost_curve_input", 

632 "Property-backed cost curve input has no selected property.", 

633 context={"curve_key": curve.curve_key, "input_key": spec.key}, 

634 ) 

635 cached_property_info = ( 

636 calculation_cache.get(("property_info", mapping.flowsheet_state_id, property_info_id)) 

637 if calculation_cache is not None 

638 else None 

639 ) 

640 property_info = cached_property_info or spec_property_info( 

641 mapping=mapping, 

642 property_info_id=property_info_id, 

643 ) 

644 cached_resolution = ( 

645 calculation_cache.get( 

646 ("schedule_resolution", mapping.costable_item.study_id, property_info_id) 

647 ) 

648 if calculation_cache is not None 

649 else None 

650 ) 

651 schedule_result = resolve_schedule_driver_input( 

652 study=mapping.costable_item.study, 

653 spec=spec, 

654 driver_input=driver_input, 

655 property_info=property_info, 

656 resolution=cached_resolution, 

657 ) 

658 if schedule_result is not None: 

659 resolved[spec.key] = { 

660 "value": schedule_result.value, 

661 "unit": schedule_result.unit, 

662 } 

663 schedule_inputs[spec.key] = schedule_result.audit_payload 

664 continue 

665 prefetched_values = getattr(property_info, "_prefetched_objects_cache", {}).get("values") 

666 value = property_info.get_value_bulk() if prefetched_values is not None else property_info.get_value() 

667 if value in (None, ""): 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true

668 raise CostCurveEvaluationError( 

669 "missing_cost_curve_input", 

670 "Selected cost curve input property has no value.", 

671 context={ 

672 "curve_key": curve.curve_key, 

673 "input_key": spec.key, 

674 "property_info_id": property_info_id, 

675 }, 

676 ) 

677 resolved[spec.key] = { 

678 "value": value, 

679 "unit": property_info.unit or driver_input.unit or spec.unit, 

680 } 

681 elif driver_input.source == "manual": 

682 if driver_input.manual_value == "": 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true

683 raise CostCurveEvaluationError( 

684 "missing_cost_curve_input", 

685 "Manual cost curve input has no value.", 

686 context={"curve_key": curve.curve_key, "input_key": spec.key}, 

687 ) 

688 resolved[spec.key] = { 

689 "value": driver_input.manual_value, 

690 "unit": driver_input.unit or spec.unit, 

691 } 

692 elif spec.required: 692 ↛ 622line 692 didn't jump to line 622 because the condition on line 692 was always true

693 raise CostCurveEvaluationError( 

694 "missing_cost_curve_input", 

695 "Cost curve input has no selected source.", 

696 context={"curve_key": curve.curve_key, "input_key": spec.key}, 

697 ) 

698 return ResolvedDriverInputs(inputs=resolved, schedule_inputs=schedule_inputs) 

699 

700 

701def _prime_schedule_resolution_cache( 

702 mappings: list[EquipmentMapping], 

703 calculation_cache: dict, 

704) -> None: 

705 """Load schedule-backed driver properties once for a generated-line batch.""" 

706 

707 property_ids_by_study: dict[int, set[int]] = {} 

708 studies_by_id: dict[int, EconomicsStudy] = {} 

709 flowsheet_ids: set[int] = set() 

710 for mapping in mappings: 

711 study = mapping.costable_item.study 

712 studies_by_id.setdefault(study.pk, study) 

713 flowsheet_ids.add(mapping.flowsheet_state_id) 

714 property_ids = property_ids_by_study.setdefault(study.pk, set()) 

715 driver = getattr(mapping.costable_item, "cost_driver", None) 

716 if driver is not None and driver.property_info_id is not None: 

717 property_ids.add(driver.property_info_id) 

718 for line in getattr(mapping.costable_item, "_economics_generated_capital_lines", ()): 

719 if not isinstance(line.driver_inputs, dict): 719 ↛ 720line 719 didn't jump to line 720 because the condition on line 719 was never true

720 continue 

721 for driver_input in line.driver_inputs.values(): 

722 if not isinstance(driver_input, dict) or driver_input.get("source") != "property": 

723 continue 

724 property_info_id = driver_input.get("property_info") 

725 if isinstance(property_info_id, int): 725 ↛ 721line 725 didn't jump to line 721 because the condition on line 725 was always true

726 property_ids.add(property_info_id) 

727 

728 all_property_ids = set().union(*property_ids_by_study.values()) if property_ids_by_study else set() 

729 if not all_property_ids: 

730 return 

731 properties = list( 

732 PropertyInfo.objects.filter( 

733 pk__in=all_property_ids, 

734 flowsheet_state_id__in=flowsheet_ids, 

735 ) 

736 .select_related("set", "set__simulationObject") 

737 .prefetch_related("values") 

738 .order_by("pk") 

739 ) 

740 properties_by_id = {property_info.pk: property_info for property_info in properties} 

741 for property_info in properties: 

742 calculation_cache[("property_info", property_info.flowsheet_state_id, property_info.pk)] = property_info 

743 for study_id, property_ids in property_ids_by_study.items(): 

744 study_properties = tuple( 

745 properties_by_id[property_info_id] 

746 for property_info_id in property_ids 

747 if property_info_id in properties_by_id 

748 ) 

749 resolutions = study_schedule_property_resolutions( 

750 study=studies_by_id[study_id], 

751 property_infos=study_properties, 

752 ) 

753 for property_info_id, resolution in resolutions.items(): 

754 calculation_cache[("schedule_resolution", study_id, property_info_id)] = resolution 

755 

756 

757def spec_property_info(*, mapping: EquipmentMapping, property_info_id: int): 

758 from core.auxiliary.models import PropertyInfo 

759 

760 driver = getattr(mapping.costable_item, "cost_driver", None) 

761 if driver is not None and driver.property_info_id == property_info_id: 

762 # The selected primary driver has already passed CostDriver serializer 

763 # or bulk-setup validation; reuse the hydrated relation and values. 

764 return driver.property_info 

765 try: 

766 property_info = PropertyInfo.objects.get(pk=property_info_id, flowsheet_state=mapping.flowsheet_state) 

767 except PropertyInfo.DoesNotExist as exc: 

768 raise CostCurveEvaluationError( 

769 "invalid_cost_curve_input_property", 

770 "Selected cost curve input property does not belong to this flowsheet.", 

771 context={"property_info_id": property_info_id}, 

772 ) from exc 

773 if driver is not None: 773 ↛ 774line 773 didn't jump to line 774 because the condition on line 773 was never true

774 try: 

775 validate_cost_driver_property(driver, property_info) 

776 except ValueError as exc: 

777 raise CostCurveEvaluationError( 

778 "invalid_cost_curve_input_property", 

779 str(exc), 

780 context={"property_info_id": property_info_id, "costable_item_id": mapping.costable_item_id}, 

781 ) from exc 

782 return property_info 

783 

784 

785def _flow_capacity_warnings( 

786 *, 

787 mapping: EquipmentMapping, 

788 curve: CostCurve, 

789 driver_inputs: CapitalCostDriverInputsPayload, 

790 resolved_inputs: dict[str, ResolvedDriverInput], 

791) -> list[dict[str, Any]]: 

792 """Warn when manual HX flow capacity is smaller than inlet stream flow.""" 

793 warnings: list[dict[str, Any]] = [] 

794 for spec in _curve_required_driver_specs(curve): 

795 flow_property_key = _flow_capacity_property_key(spec) 

796 if flow_property_key is None: 

797 continue 

798 driver_input_payload = driver_inputs.get(spec.key) 

799 if not isinstance(driver_input_payload, dict): 799 ↛ 800line 799 didn't jump to line 800 because the condition on line 799 was never true

800 continue 

801 driver_input = CapitalCostDriverInput.model_validate(driver_input_payload)