Coverage for backend/django/Economics/costing/cost_curves/driver_properties.py: 84%

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from typing import Any 

5 

6from core.auxiliary.enums import ConType 

7from core.auxiliary.models.PropertyInfo import PropertyInfo 

8from flowsheetInternals.unitops.models.Port import Port 

9from django.core.exceptions import ObjectDoesNotExist 

10from django.utils import timezone 

11from Economics.costing.capital.capital_line_sources import GENERATED_CAPITAL_LINE_SOURCE 

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

13from Economics.shared.choices import CostDriverSource, EconomicsScheduleMode 

14from Economics.scheduling.series import ( 

15 study_property_is_schedule_varying, 

16 study_schedule_varying_property_ids, 

17) 

18from Economics.costing.cost_curves.driver_specs import ( 

19 CapitalCostDriverInput, 

20 CapitalCostDriverInputsPayload, 

21 CostCurveDriverSpec, 

22 capital_cost_driver_input_payload, 

23 normalize_capital_cost_driver_inputs, 

24 parse_required_driver_specs, 

25) 

26from Economics.costing.cost_curves.registry import COST_DRIVER_RULES, CostDriverRule, PreferredProperty 

27from Economics.costing.cost_curves.evaluation import cost_curve_units_compatible 

28 

29 

30BulkDriverInputSetup = dict[str, dict[str, str]] 

31_GENERATED_LINE_NOT_LOADED = object() 

32 

33 

34@dataclass(frozen=True) 

35class CostDriverPropertyOption: 

36 property_info: int 

37 scope: str 

38 object_id: int 

39 object_name: str 

40 object_type: str 

41 property_key: str 

42 display_name: str 

43 unit: str 

44 unit_type: str 

45 value_preview: str 

46 has_value: bool 

47 recommended: bool = False 

48 recommendation_label: str = "" 

49 schedule_varying: bool = False 

50 

51 

52@dataclass(frozen=True) 

53class BulkEquipmentSetupResult: 

54 mapping: int 

55 costable_item: int 

56 unit_name: str 

57 cost_curve: int | None 

58 cost_curve_name: str 

59 curve_status: str 

60 curve_overwrite: bool 

61 cost_driver: int | None 

62 sizing_status: str 

63 sizing_property: int | None = None 

64 sizing_property_label: str = "" 

65 sizing_property_unit: str = "" 

66 sizing_property_scope: str = "" 

67 sizing_property_object_name: str = "" 

68 sizing_overwrite: bool = False 

69 driver_input_results: dict[str, "BulkEquipmentDriverInputResult"] | None = None 

70 message: str = "" 

71 

72 

73@dataclass(frozen=True) 

74class BulkEquipmentDriverInputResult: 

75 key: str 

76 label: str 

77 unit: str 

78 mode: str 

79 status: str 

80 property: int | None = None 

81 property_label: str = "" 

82 property_unit: str = "" 

83 property_scope: str = "" 

84 property_object_name: str = "" 

85 manual_value: str = "" 

86 overwrite: bool = False 

87 message: str = "" 

88 

89 

90def cost_driver_property_options(driver: CostDriver) -> list[CostDriverPropertyOption]: 

91 """Return scalar numeric properties that can size a unit cost curve.""" 

92 return _cost_driver_property_options(driver) 

93 

94 

95def _cost_driver_property_options( 

96 driver: CostDriver, 

97 *, 

98 equipment_category: str | None = None, 

99) -> list[CostDriverPropertyOption]: 

100 """Return scalar numeric sizing properties, optionally for a pending category.""" 

101 simulation_object = driver.costable_item.simulation_object 

102 if simulation_object is None: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 return [] 

104 

105 equipment_category = ( 

106 equipment_category 

107 if equipment_category is not None 

108 else _driver_equipment_category(driver) 

109 ) 

110 options: list[CostDriverPropertyOption] = [] 

111 options.extend( 

112 _property_options_for_object( 

113 driver, 

114 simulation_object, 

115 scope="unit", 

116 equipment_category=equipment_category, 

117 ) 

118 ) 

119 for stream in _connected_streams(simulation_object, direction=ConType.Inlet): 

120 options.extend( 

121 _property_options_for_object( 

122 driver, 

123 stream, 

124 scope="input_stream", 

125 equipment_category=equipment_category, 

126 ) 

127 ) 

128 for stream in _connected_streams(simulation_object, direction=ConType.Outlet): 

129 options.extend( 

130 _property_options_for_object( 

131 driver, 

132 stream, 

133 scope="output_stream", 

134 equipment_category=equipment_category, 

135 ) 

136 ) 

137 

138 seen: set[int] = set() 

139 unique_options: list[CostDriverPropertyOption] = [] 

140 for option in options: 

141 if option.property_info in seen: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true

142 continue 

143 seen.add(option.property_info) 

144 unique_options.append(option) 

145 return unique_options 

146 

147 

148def preview_or_apply_bulk_equipment_setup( 

149 *, 

150 mappings: list[EquipmentMapping], 

151 cost_curve: CostCurve | None, 

152 dry_run: bool, 

153 apply_cost_curve: bool, 

154 apply_recommended_sizing_property: bool, 

155 overwrite_sizing_property: bool, 

156 driver_inputs: BulkDriverInputSetup | None = None, 

157) -> list[BulkEquipmentSetupResult]: 

158 """Preview or apply one bulk capital-line setup decision across mappings.""" 

159 results: list[BulkEquipmentSetupResult] = [] 

160 explicit_driver_inputs = bool(driver_inputs) 

161 equipment_categories = { 

162 mapping.pk: _bulk_equipment_category( 

163 mapping=mapping, 

164 cost_curve=cost_curve, 

165 apply_cost_curve=apply_cost_curve, 

166 ) 

167 for mapping in mappings 

168 } 

169 options_by_driver = _bulk_cost_driver_property_options( 

170 mappings=mappings, 

171 equipment_categories=equipment_categories, 

172 ) 

173 property_infos_by_id = PropertyInfo.objects.in_bulk( 

174 { 

175 option.property_info 

176 for options in options_by_driver.values() 

177 for option in options 

178 } 

179 ) 

180 mapping_updates: list[EquipmentMapping] = [] 

181 driver_updates: list[CostDriver] = [] 

182 generated_lines = {} 

183 for line in CapitalCostLine.objects.filter( 

184 costable_item_id__in=[mapping.costable_item_id for mapping in mappings], 

185 source=GENERATED_CAPITAL_LINE_SOURCE, 

186 ).order_by("pk"): 

187 generated_lines.setdefault(line.costable_item_id, line) 

188 for mapping in mappings: 

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

190 curve_status = _curve_status(mapping, cost_curve, apply_cost_curve) 

191 previous_curve_id = mapping.cost_curve_id 

192 effective_curve = cost_curve if apply_cost_curve and cost_curve is not None else mapping.cost_curve 

193 effective_equipment_category = equipment_categories[mapping.pk] 

194 curve_overwrite = bool( 

195 apply_cost_curve 

196 and cost_curve is not None 

197 and previous_curve_id 

198 and previous_curve_id != cost_curve.pk 

199 ) 

200 needs_options = apply_recommended_sizing_property or any( 

201 setup.get("mode") == "auto_property" 

202 for setup in (driver_inputs or {}).values() 

203 ) 

204 options = ( 

205 options_by_driver.get(driver.pk, []) 

206 if driver is not None and needs_options 

207 else [] 

208 ) 

209 sizing_result = _recommended_sizing_result( 

210 driver=driver, 

211 cost_curve=cost_curve, 

212 equipment_category=effective_equipment_category, 

213 apply_recommended_sizing_property=apply_recommended_sizing_property, 

214 overwrite_sizing_property=overwrite_sizing_property, 

215 options=options, 

216 ) 

217 driver_input_results = _driver_input_results( 

218 mapping=mapping, 

219 driver=driver, 

220 cost_curve=effective_curve, 

221 equipment_category=effective_equipment_category, 

222 driver_inputs=driver_inputs or {}, 

223 apply_recommended_sizing_property=apply_recommended_sizing_property, 

224 preserve_existing=not explicit_driver_inputs and not overwrite_sizing_property, 

225 options=options, 

226 existing_line=generated_lines.get(mapping.costable_item_id), 

227 ) 

228 

229 if not dry_run: 

230 update_fields: list[str] = [] 

231 if apply_cost_curve and cost_curve is not None and mapping.cost_curve_id != cost_curve.pk: 231 ↛ 240line 231 didn't jump to line 240 because the condition on line 231 was always true

232 mapping.cost_curve = cost_curve 

233 if cost_curve.equipment_category: 233 ↛ 236line 233 didn't jump to line 236 because the condition on line 233 was always true

234 mapping.equipment_category = cost_curve.equipment_category 

235 update_fields.append("equipment_category") 

236 if cost_curve.equipment_subtype: 

237 mapping.equipment_subtype = cost_curve.equipment_subtype 

238 update_fields.append("equipment_subtype") 

239 update_fields.extend(["cost_curve", "updated_at"]) 

240 if update_fields: 240 ↛ 243line 240 didn't jump to line 243 because the condition on line 240 was always true

241 mapping.updated_at = timezone.now() 

242 mapping_updates.append(mapping) 

243 if driver is not None and sizing_result.option is not None: 

244 property_info = property_infos_by_id[sizing_result.option.property_info] 

245 updates = normalize_property_cost_driver(driver, property_info) 

246 for field_name, value in updates.items(): 

247 setattr(driver, field_name, value) 

248 driver.updated_at = timezone.now() 

249 driver_updates.append(driver) 

250 

251 results.append( 

252 BulkEquipmentSetupResult( 

253 mapping=mapping.pk, 

254 costable_item=mapping.costable_item_id, 

255 unit_name=mapping.costable_item.name, 

256 cost_curve=cost_curve.pk if cost_curve is not None else mapping.cost_curve_id, 

257 cost_curve_name=cost_curve.name if cost_curve is not None else (mapping.cost_curve.name if mapping.cost_curve else ""), 

258 curve_status=curve_status, 

259 curve_overwrite=curve_overwrite, 

260 cost_driver=driver.pk if driver is not None else None, 

261 sizing_status=sizing_result.status, 

262 sizing_property=sizing_result.option.property_info if sizing_result.option is not None else None, 

263 sizing_property_label=sizing_result.option.display_name if sizing_result.option is not None else "", 

264 sizing_property_unit=sizing_result.option.unit if sizing_result.option is not None else "", 

265 sizing_property_scope=sizing_result.option.scope if sizing_result.option is not None else "", 

266 sizing_property_object_name=sizing_result.option.object_name if sizing_result.option is not None else "", 

267 sizing_overwrite=sizing_result.overwrite, 

268 driver_input_results=driver_input_results, 

269 message=sizing_result.message, 

270 ) 

271 ) 

272 if mapping_updates: 

273 EquipmentMapping.objects.bulk_update( 

274 mapping_updates, 

275 fields=("cost_curve", "equipment_category", "equipment_subtype", "updated_at"), 

276 ) 

277 if driver_updates: 

278 CostDriver.objects.bulk_update( 

279 driver_updates, 

280 fields=( 

281 "source", "property_info", "manual_property_info", "sizing_mode", 

282 "canonical_unit", "design_value", "unresolved_reason_code", 

283 "warning_payload", "updated_at", 

284 ), 

285 ) 

286 return results 

287 

288 

289@dataclass(frozen=True) 

290class _RecommendedSizingResult: 

291 status: str 

292 option: CostDriverPropertyOption | None = None 

293 overwrite: bool = False 

294 message: str = "" 

295 

296 

297def _curve_status(mapping: EquipmentMapping, cost_curve: CostCurve | None, apply_cost_curve: bool) -> str: 

298 if not apply_cost_curve or cost_curve is None: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true

299 return "not_requested" 

300 if mapping.cost_curve_id == cost_curve.pk: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true

301 return "unchanged" 

302 return "ready" 

303 

304 

305def _recommended_sizing_result( 

306 *, 

307 driver: CostDriver | None, 

308 cost_curve: CostCurve | None, 

309 equipment_category: str, 

310 apply_recommended_sizing_property: bool, 

311 overwrite_sizing_property: bool, 

312 options: list[CostDriverPropertyOption] | None = None, 

313) -> _RecommendedSizingResult: 

314 if not apply_recommended_sizing_property: 

315 return _RecommendedSizingResult(status="not_requested") 

316 if driver is None: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true

317 return _RecommendedSizingResult(status="no_driver", message="No cost driver is available.") 

318 has_existing_sizing = bool(driver.property_info_id) or driver.design_value is not None 

319 if has_existing_sizing and not overwrite_sizing_property: 

320 return _RecommendedSizingResult( 

321 status="preserved", 

322 message="Existing sizing property or custom value will be preserved.", 

323 ) 

324 

325 option = _recommended_property_option( 

326 driver, 

327 cost_curve, 

328 equipment_category=equipment_category, 

329 options=options, 

330 ) 

331 if option is None: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true

332 return _RecommendedSizingResult( 

333 status="no_recommendation", 

334 message="No compatible sizing property was found.", 

335 ) 

336 return _RecommendedSizingResult( 

337 status="ready", 

338 option=option, 

339 overwrite=has_existing_sizing, 

340 ) 

341 

342 

343def _driver_input_results( 

344 *, 

345 mapping: EquipmentMapping, 

346 driver: CostDriver | None, 

347 cost_curve: CostCurve | None, 

348 equipment_category: str, 

349 driver_inputs: BulkDriverInputSetup, 

350 apply_recommended_sizing_property: bool, 

351 preserve_existing: bool, 

352 options: list[CostDriverPropertyOption] | None = None, 

353 existing_line: CapitalCostLine | None = None, 

354) -> dict[str, BulkEquipmentDriverInputResult]: 

355 if cost_curve is None: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true

356 return {} 

357 try: 

358 specs = parse_required_driver_specs(cost_curve.required_driver_specs) 

359 except ValueError: 

360 return {} 

361 existing_inputs = _generated_line_driver_inputs(mapping, line=existing_line) 

362 return { 

363 spec.key: _driver_input_result( 

364 mapping=mapping, 

365 driver=driver, 

366 spec=spec, 

367 equipment_category=equipment_category, 

368 setup=driver_inputs.get(spec.key), 

369 existing_input=existing_inputs.get(spec.key), 

370 apply_recommended_sizing_property=apply_recommended_sizing_property, 

371 preserve_existing=preserve_existing, 

372 options=options, 

373 ) 

374 for spec in specs 

375 } 

376 

377 

378def _driver_input_result( 

379 *, 

380 mapping: EquipmentMapping, 

381 driver: CostDriver | None, 

382 spec: CostCurveDriverSpec, 

383 equipment_category: str, 

384 setup: dict[str, str] | None, 

385 existing_input: dict[str, Any] | None, 

386 apply_recommended_sizing_property: bool, 

387 preserve_existing: bool, 

388 options: list[CostDriverPropertyOption] | None = None, 

389) -> BulkEquipmentDriverInputResult: 

390 mode = (setup or {}).get("mode") or _default_bulk_driver_input_mode( 

391 spec, 

392 apply_recommended_sizing_property=apply_recommended_sizing_property, 

393 ) 

394 existing_has_value = _driver_input_has_value(existing_input) 

395 if mode == "keep": 

396 return BulkEquipmentDriverInputResult( 

397 key=spec.key, 

398 label=spec.label, 

399 unit=spec.unit, 

400 mode=mode, 

401 status="preserved" if existing_has_value else "missing", 

402 message="" if existing_has_value else "No existing driver input is selected.", 

403 ) 

404 if mode == "manual": 

405 if "manual" not in spec.source_options: 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true

406 return BulkEquipmentDriverInputResult( 

407 key=spec.key, 

408 label=spec.label, 

409 unit=spec.unit, 

410 mode=mode, 

411 status="unsupported", 

412 message="Manual values are not allowed for this driver input.", 

413 ) 

414 manual_value = (setup or {}).get("manual_value", "").strip() 

415 return BulkEquipmentDriverInputResult( 

416 key=spec.key, 

417 label=spec.label, 

418 unit=spec.unit, 

419 mode=mode, 

420 status="ready" if manual_value else "missing", 

421 manual_value=manual_value, 

422 overwrite=existing_has_value, 

423 message="" if manual_value else "Enter a manual value.", 

424 ) 

425 if mode != "auto_property": 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true

426 return BulkEquipmentDriverInputResult( 

427 key=spec.key, 

428 label=spec.label, 

429 unit=spec.unit, 

430 mode=mode, 

431 status="unsupported", 

432 message="Unsupported driver input setup mode.", 

433 ) 

434 if "property" not in spec.source_options: 434 ↛ 435line 434 didn't jump to line 435 because the condition on line 434 was never true

435 return BulkEquipmentDriverInputResult( 

436 key=spec.key, 

437 label=spec.label, 

438 unit=spec.unit, 

439 mode=mode, 

440 status="unsupported", 

441 message="Properties are not allowed for this driver input.", 

442 ) 

443 if existing_has_value and preserve_existing: 443 ↛ 444line 443 didn't jump to line 444 because the condition on line 443 was never true

444 return BulkEquipmentDriverInputResult( 

445 key=spec.key, 

446 label=spec.label, 

447 unit=spec.unit, 

448 mode=mode, 

449 status="preserved", 

450 overwrite=False, 

451 message="Existing driver input will be preserved.", 

452 ) 

453 if driver is None: 453 ↛ 454line 453 didn't jump to line 454 because the condition on line 453 was never true

454 return BulkEquipmentDriverInputResult( 

455 key=spec.key, 

456 label=spec.label, 

457 unit=spec.unit, 

458 mode=mode, 

459 status="no_driver", 

460 message="No cost driver is available.", 

461 ) 

462 option = _recommended_property_option_for_unit( 

463 driver, 

464 unit=spec.unit, 

465 equipment_category=equipment_category, 

466 options=options, 

467 ) 

468 if option is None: 

469 return BulkEquipmentDriverInputResult( 

470 key=spec.key, 

471 label=spec.label, 

472 unit=spec.unit, 

473 mode=mode, 

474 status="no_recommendation", 

475 message="No compatible property was found.", 

476 ) 

477 return BulkEquipmentDriverInputResult( 

478 key=spec.key, 

479 label=spec.label, 

480 unit=spec.unit, 

481 mode=mode, 

482 status="ready", 

483 property=option.property_info, 

484 property_label=option.display_name, 

485 property_unit=option.unit, 

486 property_scope=option.scope, 

487 property_object_name=option.object_name, 

488 overwrite=existing_has_value, 

489 ) 

490 

491 

492def apply_bulk_driver_inputs( 

493 *, 

494 mappings: list[EquipmentMapping], 

495 cost_curve: CostCurve | None, 

496 apply_cost_curve: bool, 

497 apply_recommended_sizing_property: bool, 

498 overwrite_sizing_property: bool, 

499 driver_inputs: BulkDriverInputSetup | None, 

500 setup_results: list[BulkEquipmentSetupResult] | None = None, 

501) -> bool: 

502 changed_lines: list[CapitalCostLine] = [] 

503 explicit_driver_inputs = bool(driver_inputs) 

504 generated_lines = {} 

505 for line in CapitalCostLine.objects.filter( 

506 costable_item_id__in=[mapping.costable_item_id for mapping in mappings], 

507 source=GENERATED_CAPITAL_LINE_SOURCE, 

508 ).order_by("pk"): 

509 generated_lines.setdefault(line.costable_item_id, line) 

510 setup_results_by_mapping = { 

511 result.mapping: result.driver_input_results or {} 

512 for result in (setup_results or []) 

513 } 

514 for mapping in mappings: 

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

516 effective_curve = cost_curve if apply_cost_curve and cost_curve is not None else mapping.cost_curve 

517 if effective_curve is None: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true

518 continue 

519 line = generated_lines.get(mapping.costable_item_id) 

520 if line is None: 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true

521 continue 

522 results = setup_results_by_mapping.get(mapping.pk) 

523 if results is None: 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true

524 effective_equipment_category = _bulk_equipment_category( 

525 mapping=mapping, 

526 cost_curve=cost_curve, 

527 apply_cost_curve=apply_cost_curve, 

528 ) 

529 results = _driver_input_results( 

530 mapping=mapping, 

531 driver=driver, 

532 cost_curve=effective_curve, 

533 equipment_category=effective_equipment_category, 

534 driver_inputs=driver_inputs or {}, 

535 apply_recommended_sizing_property=apply_recommended_sizing_property, 

536 preserve_existing=not explicit_driver_inputs and not overwrite_sizing_property, 

537 existing_line=line, 

538 ) 

539 if not results: 539 ↛ 540line 539 didn't jump to line 540 because the condition on line 539 was never true

540 continue 

541 next_inputs = _driver_inputs_payload_for_results(effective_curve, line.driver_inputs, results) 

542 if line.driver_inputs != next_inputs: 

543 line.driver_inputs = next_inputs 

544 line.updated_at = timezone.now() 

545 changed_lines.append(line) 

546 if changed_lines: 

547 CapitalCostLine.objects.bulk_update(changed_lines, fields=("driver_inputs", "updated_at")) 

548 return bool(changed_lines) 

549 

550 

551def _driver_inputs_payload_for_results( 

552 cost_curve: CostCurve, 

553 existing_inputs: Any, 

554 results: dict[str, BulkEquipmentDriverInputResult], 

555) -> CapitalCostDriverInputsPayload: 

556 try: 

557 current_inputs = normalize_capital_cost_driver_inputs(existing_inputs or {}) 

558 except ValueError: 

559 current_inputs = {} 

560 payload: CapitalCostDriverInputsPayload = {} 

561 for spec in parse_required_driver_specs(cost_curve.required_driver_specs): 

562 existing_input = current_inputs.get(spec.key) 

563 result = results.get(spec.key) 

564 if result is None or result.status == "preserved": 564 ↛ 565line 564 didn't jump to line 565 because the condition on line 564 was never true

565 payload[spec.key] = existing_input or _blank_driver_input(spec) 

566 elif result.status == "ready" and result.property is not None: 

567 payload[spec.key] = capital_cost_driver_input_payload( 

568 CapitalCostDriverInput( 

569 source="property", 

570 property_info=result.property, 

571 manual_value="", 

572 unit=spec.unit, 

573 ) 

574 ) 

575 elif result.status == "ready" and result.mode == "manual": 

576 payload[spec.key] = capital_cost_driver_input_payload( 

577 CapitalCostDriverInput( 

578 source="manual", 

579 property_info=None, 

580 manual_value=result.manual_value, 

581 unit=spec.unit, 

582 ) 

583 ) 

584 else: 

585 payload[spec.key] = _blank_driver_input(spec) 

586 return payload 

587 

588 

589def _blank_driver_input(spec: CostCurveDriverSpec): 

590 return capital_cost_driver_input_payload( 

591 CapitalCostDriverInput(source="", property_info=None, manual_value="", unit=spec.unit) 

592 ) 

593 

594 

595def _default_bulk_driver_input_mode( 

596 spec: CostCurveDriverSpec, 

597 *, 

598 apply_recommended_sizing_property: bool, 

599) -> str: 

600 if not apply_recommended_sizing_property: 

601 return "keep" 

602 if "property" in spec.source_options: 602 ↛ 604line 602 didn't jump to line 604 because the condition on line 602 was always true

603 return "auto_property" 

604 if "manual" in spec.source_options: 

605 return "manual" 

606 return "keep" 

607 

608 

609def _recommended_property_option( 

610 driver: CostDriver, 

611 cost_curve: CostCurve | None, 

612 *, 

613 equipment_category: str, 

614 options: list[CostDriverPropertyOption] | None = None, 

615) -> CostDriverPropertyOption | None: 

616 curve_unit = _primary_formula_input_unit(cost_curve) 

617 return _recommended_property_option_for_unit( 

618 driver, 

619 unit=curve_unit, 

620 equipment_category=equipment_category, 

621 options=options, 

622 ) 

623 

624 

625def _recommended_property_option_for_unit( 

626 driver: CostDriver, 

627 *, 

628 unit: str | None, 

629 equipment_category: str, 

630 options: list[CostDriverPropertyOption] | None = None, 

631) -> CostDriverPropertyOption | None: 

632 compatible_fallback: CostDriverPropertyOption | None = None 

633 candidates = options if options is not None else _cost_driver_property_options( 

634 driver, equipment_category=equipment_category 

635 ) 

636 for option in candidates: 

637 if unit and not cost_curve_units_compatible(option.unit, unit): 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true

638 continue 

639 if compatible_fallback is None: 

640 compatible_fallback = option 

641 if not option.recommended: 

642 continue 

643 return option 

644 return compatible_fallback 

645 

646 

647def _generated_line_for_mapping(mapping: EquipmentMapping) -> CapitalCostLine | None: 

648 return ( 

649 CapitalCostLine.objects.filter( 

650 flowsheet_state=mapping.flowsheet_state, 

651 study=mapping.costable_item.study, 

652 costable_item=mapping.costable_item, 

653 source=GENERATED_CAPITAL_LINE_SOURCE, 

654 ) 

655 .order_by("pk") 

656 .first() 

657 ) 

658 

659 

660def _generated_line_driver_inputs( 

661 mapping: EquipmentMapping, 

662 *, 

663 line: CapitalCostLine | None | object = _GENERATED_LINE_NOT_LOADED, 

664) -> CapitalCostDriverInputsPayload: 

665 if line is _GENERATED_LINE_NOT_LOADED: 665 ↛ 666line 665 didn't jump to line 666 because the condition on line 665 was never true

666 line = _generated_line_for_mapping(mapping) 

667 if line is None: 

668 return {} 

669 try: 

670 return normalize_capital_cost_driver_inputs(line.driver_inputs or {}) 

671 except ValueError: 

672 return {} 

673 

674 

675def _driver_input_has_value(driver_input: dict[str, Any] | None) -> bool: 

676 if not driver_input: 676 ↛ 678line 676 didn't jump to line 678 because the condition on line 676 was always true

677 return False 

678 source = driver_input.get("source") 

679 if source == "property": 

680 return driver_input.get("property_info") is not None 

681 if source == "manual": 

682 return bool(str(driver_input.get("manual_value") or "").strip()) 

683 return False 

684 

685 

686def _primary_formula_input_unit(cost_curve: CostCurve | None) -> str | None: 

687 """Return the recommendation unit for setup-only cost-driver suggestions.""" 

688 if cost_curve is None: 688 ↛ 689line 688 didn't jump to line 689 because the condition on line 688 was never true

689 return None 

690 try: 

691 specs = parse_required_driver_specs(cost_curve.required_driver_specs) 

692 except ValueError: 

693 return None 

694 primary_spec = next((spec for spec in specs if spec.role == "formula_input" and spec.primary), None) 

695 return None if primary_spec is None else primary_spec.unit 

696 

697 

698def _bulk_equipment_category( 

699 *, 

700 mapping: EquipmentMapping, 

701 cost_curve: CostCurve | None, 

702 apply_cost_curve: bool, 

703) -> str: 

704 if apply_cost_curve and cost_curve is not None and cost_curve.equipment_category: 704 ↛ 706line 704 didn't jump to line 706 because the condition on line 704 was always true

705 return cost_curve.equipment_category 

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

707 if driver is None: 

708 return mapping.equipment_category 

709 return mapping.equipment_category or _driver_equipment_category(driver) 

710 

711 

712def validate_cost_driver_property(driver: CostDriver, property_info: PropertyInfo) -> None: 

713 available_ids = {option.property_info for option in cost_driver_property_options(driver)} 

714 if property_info.pk not in available_ids: 

715 raise ValueError( 

716 "Cost driver property must belong to the unit operation, an input stream, or an output stream." 

717 ) 

718 

719 

720def normalize_property_cost_driver(driver: CostDriver, property_info: PropertyInfo | None) -> dict[str, Any]: 

721 """Return model field updates implied by selecting a driver property.""" 

722 if property_info is None: 

723 if driver.design_value is not None: 723 ↛ 729line 723 didn't jump to line 729 because the condition on line 723 was always true

724 return { 

725 "source": CostDriverSource.MANUAL_OVERRIDE, 

726 "property_info": None, 

727 "sizing_mode": "manual", 

728 } 

729 return { 

730 "source": CostDriverSource.UNRESOLVED, 

731 "property_info": None, 

732 "sizing_mode": "", 

733 } 

734 

735 payload = dict(driver.warning_payload or {}) 

736 payload.update( 

737 { 

738 "driver_property_id": property_info.pk, 

739 "driver_property_key": property_info.key, 

740 "driver_property_name": property_info.displayName, 

741 "driver_property_unit": property_info.unit, 

742 "warnings": [], 

743 "design_value_basis": "user_selected_unit_or_connected_stream_property", 

744 } 

745 ) 

746 payload.update(_registry_payload_for_property(driver, property_info)) 

747 return { 

748 "source": CostDriverSource.PROPERTY, 

749 "property_info": property_info, 

750 "canonical_unit": property_info.unit or "", 

751 "design_value": None, 

752 "sizing_mode": "property", 

753 "unresolved_reason_code": "", 

754 "warning_payload": payload, 

755 } 

756 

757 

758def apply_recommended_property_for_mapping(mapping: EquipmentMapping) -> CostDriver | None: 

759 """Select the first recommended property after equipment-category changes.""" 

760 driver = ( 

761 CostDriver.objects.filter(costable_item=mapping.costable_item) 

762 .select_related("costable_item", "costable_item__simulation_object") 

763 .first() 

764 ) 

765 if driver is None or driver.property_info_id is not None or driver.design_value is not None: 

766 return None 

767 

768 for option in cost_driver_property_options(driver): 

769 if not option.recommended or not option.has_value: 

770 continue 

771 property_info = PropertyInfo.objects.get(pk=option.property_info) 

772 updates = normalize_property_cost_driver(driver, property_info) 

773 for field_name, value in updates.items(): 

774 setattr(driver, field_name, value) 

775 driver.save(update_fields=[*updates.keys(), "updated_at"]) 

776 return driver 

777 return None 

778 

779 

780def _registry_payload_for_property(driver: CostDriver, property_info: PropertyInfo) -> dict[str, str]: 

781 simulation_object = driver.costable_item.simulation_object 

782 if simulation_object is None: 782 ↛ 783line 782 didn't jump to line 783 because the condition on line 782 was never true

783 return {} 

784 for rule in COST_DRIVER_RULES: 

785 if simulation_object.objectType not in rule.compatible_object_types: 

786 continue 

787 preferred_properties = rule.preferred_properties + rule.preferred_input_stream_properties 

788 for preferred_property in preferred_properties: 

789 if property_info.key != preferred_property.key or property_info.unitType != preferred_property.unit_type: 

790 continue 

791 return { 

792 "equipment_category": rule.equipment_category, 

793 "curve_input_variable": preferred_property.curve_input_variable or rule.curve_input_variable, 

794 } 

795 return {} 

796 

797 

798def _connected_streams(simulation_object, *, direction: str): 

799 ports = simulation_object.ports.filter(direction=direction).select_related("stream").order_by("index", "pk")