Coverage for backend/django/Economics/formulas/builders/operating.py: 78%

260 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-07-22 05:22 +0000

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from decimal import Decimal, InvalidOperation 

5 

6import sympy 

7 

8from Economics.shared.choices import ( 

9 

10 DefaultRateReviewStatus, 

11 

12 DefaultRateType, 

13 

14 DefaultRateValueKind, 

15 

16 OperatingLineCategory, 

17 

18 OperatingLineEconomicEffect, 

19 

20) 

21 

22from Economics.reference_data.models import EconomicsDefaultRate 

23 

24from Economics.studies.models import EconomicsStudy 

25 

26from Economics.costing.models import OperatingCostLine 

27from Economics.costing.operating.line_calculation import ( 

28 AnnualBasisQuantity, 

29 annualization_requires_operating_hours, 

30 annualized_basis_quantity, 

31 convert_quantity, 

32 strip_single_time_denominator, 

33) 

34from Economics.settings_profiles.services.settings_profiles import get_settings_profile 

35 

36from Economics.formulas.engine.core import EconomicsFormula, FormulaError, FormulaInput, FormulaStep, decimal_to_sympy 

37from core.auxiliary.formula_units import formula_unit_expression 

38 

39 

40SUPPORTED_OPERATING_METHODS = frozenset({"rate_times_quantity", "work_to_cost"}) 

41PROCESS_ENERGY_QUANTITY_UNIT = "MWh" 

42OPERATING_LINE_RATE_QUANTUM = Decimal("0.00000001") 

43STEAM_MASS_RATE_DENOMINATORS = frozenset({"t"}) 

44STEAM_ENERGY_BASIS_UNIT = "GJ" 

45 

46 

47@dataclass(frozen=True) 

48class OperatingLineFormula: 

49 formula: EconomicsFormula 

50 annual_basis: AnnualBasisQuantity 

51 basis_quantity: Decimal 

52 basis_unit: str 

53 target_basis_unit: str 

54 annualization_factor: Decimal 

55 rate_amount: Decimal 

56 rate_unit: str 

57 

58 def evaluate(self) -> Decimal | None: 

59 return self.formula.evaluate({self.formula.inputs[0].key: self.basis_quantity}) 

60 

61 

62@dataclass(frozen=True) 

63class OperatingLineRateBasis: 

64 rate_amount: Decimal 

65 rate_unit: str 

66 source_basis_unit: str 

67 target_basis_unit: str 

68 

69 

70@dataclass(frozen=True) 

71class OperatingAggregateFormula: 

72 formula: EconomicsFormula 

73 bindings: dict[str, Decimal] 

74 

75 def evaluate(self) -> Decimal | None: 

76 return self.formula.evaluate(self.bindings) 

77 

78 

79@dataclass(frozen=True) 

80class ProcessEnergyContributionFormula: 

81 formula: EconomicsFormula 

82 source_quantity: Decimal 

83 source_unit: str 

84 

85 def evaluate(self) -> Decimal | None: 

86 return self.formula.evaluate({}) 

87 

88 

89@dataclass(frozen=True) 

90class DefaultRateFormula: 

91 formula: EconomicsFormula 

92 bindings: dict[str, Decimal] 

93 

94 def evaluate(self) -> Decimal | None: 

95 value = self.formula.evaluate(self.bindings) 

96 if value is None: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true

97 return None 

98 return value.quantize(OPERATING_LINE_RATE_QUANTUM) 

99 

100 

101def build_operating_line_formula(line: OperatingCostLine, *, study: EconomicsStudy) -> OperatingLineFormula: 

102 """Build the canonical annual amount formula for one operating line.""" 

103 if line.calculation_method not in SUPPORTED_OPERATING_METHODS: 

104 raise FormulaError( 

105 "unsupported_operating_line_method", 

106 "uses an unsupported calculation method.", 

107 context={"operating_line_id": line.pk, "calculation_method": line.calculation_method}, 

108 ) 

109 if line.basis_quantity is None: 

110 raise FormulaError( 

111 "missing_operating_line_basis", 

112 "has no basis quantity.", 

113 context={"operating_line_id": line.pk}, 

114 ) 

115 rate_basis = resolve_operating_line_rate_basis(line) 

116 annualization_factor = _annualization_factor_for_line( 

117 line, 

118 study=study, 

119 target_basis_unit=rate_basis.target_basis_unit, 

120 ) 

121 if annualization_factor is None: 

122 context = { 

123 "operating_line_id": line.pk, 

124 "source_unit": line.basis_unit, 

125 "target_unit": rate_basis.target_basis_unit, 

126 } 

127 if _missing_annual_operating_hours_for_annualization( 127 ↛ 137line 127 didn't jump to line 137 because the condition on line 127 was always true

128 line, 

129 study=study, 

130 target_basis_unit=rate_basis.target_basis_unit, 

131 ): 

132 raise FormulaError( 

133 "missing_annual_operating_hours", 

134 _missing_annual_operating_hours_message(line, rate_basis.rate_unit), 

135 context=context, 

136 ) 

137 raise FormulaError( 

138 "unsupported_operating_line_unit_conversion", 

139 _unsupported_annual_cost_conversion_message(line, rate_basis.rate_unit), 

140 context=context, 

141 ) 

142 annual_basis_quantity = line.basis_quantity * annualization_factor 

143 basis_symbol = sympy.Symbol(_basis_input_key(line)) 

144 expression = _multiply_terms( 

145 basis_symbol, 

146 annualization_factor, 

147 rate_basis.rate_amount, 

148 ) 

149 formula = EconomicsFormula( 

150 key=f"operating_line:{line.pk or 'unsaved'}", 

151 expression=expression, 

152 unit=f"{line.currency}/year", 

153 inputs=( 

154 FormulaInput( 

155 key=str(basis_symbol), 

156 label="Basis quantity", 

157 unit=line.basis_unit, 

158 source_property_info_id=line.source_property_info_id, 

159 ), 

160 ), 

161 steps=( 

162 FormulaStep( 

163 kind="annualization_factor", 

164 label="Annualization factor", 

165 expression=str(annualization_factor), 

166 amount=annualization_factor, 

167 unit=rate_basis.target_basis_unit, 

168 ), 

169 FormulaStep( 

170 kind="rate", 

171 label="Rate", 

172 expression=str(rate_basis.rate_amount), 

173 amount=rate_basis.rate_amount, 

174 unit=rate_basis.rate_unit, 

175 ), 

176 ), 

177 ) 

178 return OperatingLineFormula( 

179 formula=formula, 

180 annual_basis=AnnualBasisQuantity(quantity=annual_basis_quantity, unit=rate_basis.target_basis_unit), 

181 basis_quantity=line.basis_quantity, 

182 basis_unit=line.basis_unit, 

183 target_basis_unit=rate_basis.target_basis_unit, 

184 annualization_factor=annualization_factor, 

185 rate_amount=rate_basis.rate_amount, 

186 rate_unit=rate_basis.rate_unit, 

187 ) 

188 

189 

190def resolve_operating_line_rate_basis( 

191 line: OperatingCostLine, 

192 *, 

193 source_basis_unit: str | None = None, 

194) -> OperatingLineRateBasis: 

195 """Resolve the price and physical basis implied by one operating line.""" 

196 

197 rate_amount = _effective_rate_amount(line) 

198 if rate_amount is None: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true

199 raise FormulaError( 

200 "missing_operating_line_rate", 

201 "has no rate.", 

202 context={"operating_line_id": line.pk}, 

203 ) 

204 rate_unit = _effective_rate_unit(line) 

205 basis_unit = source_basis_unit or line.basis_unit 

206 target_basis_unit = rate_denominator_unit( 

207 rate_unit=rate_unit, 

208 source_basis_unit=basis_unit, 

209 ) 

210 if not target_basis_unit: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 target_basis_unit = strip_single_time_denominator(basis_unit) 

212 return OperatingLineRateBasis( 

213 rate_amount=rate_amount, 

214 rate_unit=rate_unit, 

215 source_basis_unit=basis_unit, 

216 target_basis_unit=target_basis_unit, 

217 ) 

218 

219 

220def build_derived_steam_rate_formula( 

221 default_rate: EconomicsDefaultRate, 

222 *, 

223 override: dict | None = None, 

224) -> DefaultRateFormula: 

225 """Build the derived steam default-rate formula from fuel, steam energy, and efficiency.""" 

226 if default_rate.rate_type != DefaultRateType.STEAM or default_rate.value_kind != DefaultRateValueKind.DERIVED_TEMPLATE: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true

227 raise FormulaError( 

228 "unsupported_default_rate_formula", 

229 "Only derived steam templates expose a default-rate formula.", 

230 context={"default_rate_id": default_rate.pk}, 

231 ) 

232 metadata = default_rate.metadata if isinstance(default_rate.metadata, dict) else {} 

233 fuel_price = _metadata_decimal(metadata.get("fuel_price_nzd_per_gj")) 

234 steam_energy = _override_or_metadata_decimal( 

235 override, 

236 "steam_energy_gj_per_t", 

237 metadata.get("steam_energy_gj_per_t"), 

238 ) 

239 efficiency_percent = _override_or_metadata_decimal( 

240 override, 

241 "boiler_efficiency_percent", 

242 metadata.get("default_boiler_efficiency_percent"), 

243 ) 

244 if fuel_price is None or steam_energy is None or efficiency_percent is None or efficiency_percent <= 0: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 raise FormulaError( 

246 "blocked_derived_steam_rate", 

247 "Derived steam rate requires fuel price, steam energy, and positive boiler efficiency.", 

248 context={"default_rate_id": default_rate.pk}, 

249 ) 

250 expression = ( 

251 sympy.Symbol("fuel_price_nzd_per_gj") 

252 * sympy.Symbol("steam_energy_gj_per_t") 

253 / (sympy.Symbol("boiler_efficiency_percent") / decimal_to_sympy(Decimal("100"))) 

254 ) 

255 return DefaultRateFormula( 

256 formula=EconomicsFormula( 

257 key=f"default_rate:derived_steam:{default_rate.pk or default_rate.key}", 

258 expression=expression, 

259 unit=default_rate.display_unit, 

260 inputs=( 

261 FormulaInput("fuel_price_nzd_per_gj", "Fuel price", "NZD/GJ"), 

262 FormulaInput("steam_energy_gj_per_t", "Steam energy basis", "GJ/t"), 

263 FormulaInput("boiler_efficiency_percent", "Boiler efficiency", "percent"), 

264 ), 

265 ), 

266 bindings={ 

267 "fuel_price_nzd_per_gj": fuel_price, 

268 "steam_energy_gj_per_t": steam_energy, 

269 "boiler_efficiency_percent": efficiency_percent, 

270 }, 

271 ) 

272 

273 

274def build_process_energy_contribution_formula( 

275 line: OperatingCostLine, 

276 *, 

277 study: EconomicsStudy, 

278) -> ProcessEnergyContributionFormula: 

279 """Build the target-process-energy contribution formula for one operating line.""" 

280 if line.basis_quantity is None: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 raise FormulaError( 

282 "missing_process_energy_basis", 

283 "Process-energy contribution requires a basis quantity.", 

284 context={"operating_line_id": line.pk}, 

285 ) 

286 direct_energy = annualized_basis_quantity( 

287 line.basis_quantity, 

288 source_unit=line.basis_unit, 

289 target_unit=PROCESS_ENERGY_QUANTITY_UNIT, 

290 study=study, 

291 ) 

292 if direct_energy is not None: 292 ↛ 300line 292 didn't jump to line 300 because the condition on line 292 was always true

293 return _process_energy_formula( 

294 line=line, 

295 target_quantity=direct_energy, 

296 source_quantity=line.basis_quantity, 

297 source_unit=line.basis_unit, 

298 ) 

299 

300 annual_basis = build_operating_line_formula(line, study=study).annual_basis 

301 source_unit = strip_single_time_denominator(annual_basis.unit) 

302 if not source_unit: 

303 raise FormulaError( 

304 "unsupported_process_energy_unit", 

305 "Process-energy contribution requires an energy-compatible annual basis.", 

306 context={"operating_line_id": line.pk, "annual_basis_unit": annual_basis.unit}, 

307 ) 

308 target_quantity = convert_quantity( 

309 value=annual_basis.quantity, 

310 source_unit=source_unit, 

311 target_unit=PROCESS_ENERGY_QUANTITY_UNIT, 

312 ) 

313 if target_quantity is None: 

314 raise FormulaError( 

315 "unsupported_process_energy_unit", 

316 "Process-energy contribution requires an energy-compatible annual basis.", 

317 context={"operating_line_id": line.pk, "annual_basis_unit": annual_basis.unit}, 

318 ) 

319 return _process_energy_formula( 

320 line=line, 

321 target_quantity=target_quantity, 

322 source_quantity=annual_basis.quantity, 

323 source_unit=annual_basis.unit, 

324 ) 

325 

326 

327def build_annual_operating_expense_formula(study: EconomicsStudy) -> OperatingAggregateFormula: 

328 """Build annual OPEX from included non-revenue operating line formulas.""" 

329 return _build_operating_total_formula( 

330 study, 

331 key="annual_operating_expense", 

332 label="Annual operating expense", 

333 include_revenue=False, 

334 ) 

335 

336 

337def build_annual_operating_revenue_formula(study: EconomicsStudy) -> OperatingAggregateFormula: 

338 """Build annual revenue from included output-revenue operating line formulas.""" 

339 return _build_operating_total_formula( 

340 study, 

341 key="annual_operating_revenue", 

342 label="Annual operating revenue", 

343 include_revenue=True, 

344 ) 

345 

346 

347def render_operating_line_property_formula( 

348 operating_formula: OperatingLineFormula, 

349 *, 

350 source_property_formula: str | None, 

351) -> str: 

352 """Render an operating-line formula as an IDAES-visible annual-cost formula.""" 

353 rate_unit_expression = formula_unit_expression(operating_formula.rate_unit) 

354 target_basis_unit_expression = formula_unit_expression(operating_formula.target_basis_unit) 

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

356 raise FormulaError( 

357 "unsupported_operating_line_unit_rendering", 

358 "uses a unit conversion that is not solve-visible yet.", 

359 ) 

360 

361 input_key = operating_formula.formula.inputs[0].key 

362 if source_property_formula is None: 

363 rendered = operating_formula.formula.render_property_formula( 

364 {input_key: _decimal_literal(operating_formula.basis_quantity)} 

365 ) 

366 return ( 

367 f"({rendered} * ({target_basis_unit_expression}) " 

368 f"* ({rate_unit_expression}) / year)" 

369 ) 

370 

371 basis_unit_expression = formula_unit_expression(operating_formula.basis_unit) 

372 if basis_unit_expression is None: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true

373 raise FormulaError( 

374 "unsupported_operating_line_unit_rendering", 

375 "uses a unit conversion that is not solve-visible yet.", 

376 ) 

377 converted_source = f"convert({source_property_formula}, {basis_unit_expression})" 

378 rendered = operating_formula.formula.render_property_formula({input_key: converted_source}) 

379 return ( 

380 f"({rendered} * ({rate_unit_expression}) " 

381 f"* ({target_basis_unit_expression}) / ({basis_unit_expression}) / year)" 

382 ) 

383 

384 

385def rate_denominator_unit(*, rate_unit: str, source_basis_unit: str) -> str: 

386 """Return the physical basis denominator implied by a rate unit.""" 

387 if "/" not in rate_unit: 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true

388 return "" 

389 denominator = rate_unit.split("/", 1)[1].strip() 

390 if denominator == source_basis_unit.strip(): 

391 return strip_single_time_denominator(denominator) 

392 return denominator 

393 

394 

395def _annualization_factor_for_line( 

396 line: OperatingCostLine, 

397 *, 

398 study: EconomicsStudy, 

399 target_basis_unit: str, 

400) -> Decimal | None: 

401 annualization_factor = annualized_basis_quantity( 

402 Decimal("1"), 

403 source_unit=line.basis_unit, 

404 target_unit=target_basis_unit, 

405 study=study, 

406 ) 

407 if annualization_factor is not None: 

408 return annualization_factor 

409 return _steam_mass_annualization_factor( 

410 line, 

411 study=study, 

412 target_basis_unit=target_basis_unit, 

413 ) 

414 

415 

416def _missing_annual_operating_hours_for_annualization( 

417 line: OperatingCostLine, 

418 *, 

419 study: EconomicsStudy, 

420 target_basis_unit: str, 

421) -> bool: 

422 settings_profile = get_settings_profile(study) 

423 if settings_profile is not None and settings_profile.annual_operating_hours is not None: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true

424 return False 

425 if ( 

426 target_basis_unit.strip().lower() in STEAM_MASS_RATE_DENOMINATORS 

427 and line.rate_type == DefaultRateType.STEAM 

428 and _steam_energy_basis_gj_per_t(line) 

429 ): 

430 return annualization_requires_operating_hours( 

431 source_unit=line.basis_unit, 

432 target_unit=STEAM_ENERGY_BASIS_UNIT, 

433 ) 

434 return annualization_requires_operating_hours( 

435 source_unit=line.basis_unit, 

436 target_unit=target_basis_unit, 

437 ) 

438 

439 

440def _steam_mass_annualization_factor( 

441 line: OperatingCostLine, 

442 *, 

443 study: EconomicsStudy, 

444 target_basis_unit: str, 

445) -> Decimal | None: 

446 if target_basis_unit.strip().lower() not in STEAM_MASS_RATE_DENOMINATORS: 

447 return None 

448 if line.rate_type != DefaultRateType.STEAM: 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true

449 return None 

450 steam_energy_basis = _steam_energy_basis_gj_per_t(line) 

451 if steam_energy_basis is None or steam_energy_basis <= 0: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true

452 return None 

453 annual_energy = annualized_basis_quantity( 

454 Decimal("1"), 

455 source_unit=line.basis_unit, 

456 target_unit=STEAM_ENERGY_BASIS_UNIT, 

457 study=study, 

458 ) 

459 if annual_energy is None: 

460 return None 

461 return annual_energy / steam_energy_basis 

462 

463 

464def _steam_energy_basis_gj_per_t(line: OperatingCostLine) -> Decimal | None: 

465 if not line.source_default_rate_id: 465 ↛ 466line 465 didn't jump to line 466 because the condition on line 465 was never true

466 return None 

467 metadata = line.source_default_rate.metadata if isinstance(line.source_default_rate.metadata, dict) else {} 

468 return _metadata_decimal(metadata.get("steam_energy_gj_per_t")) 

469 

470 

471def _build_operating_total_formula( 

472 study: EconomicsStudy, 

473 *, 

474 key: str, 

475 label: str, 

476 include_revenue: bool, 

477) -> OperatingAggregateFormula: 

478 terms: list[sympy.Expr] = [] 

479 bindings: dict[str, Decimal] = {} 

480 inputs: list[FormulaInput] = [] 

481 blocked_children: list[dict[str, str]] = [] 

482 for line in study.operating_lines.filter(included=True).select_related("source_default_rate").order_by("pk"): 

483 if operating_line_is_revenue(line) != include_revenue: 

484 continue 

485 try: 

486 amount = build_operating_line_formula(line, study=study).evaluate() 

487 except FormulaError as exc: 

488 amount = None 

489 blocked_children.append( 

490 { 

491 "key": f"operating_line:{line.pk}", 

492 "code": exc.code, 

493 "reason": exc.message, 

494 } 

495 ) 

496 if amount is not None: 

497 symbol_key = operating_line_input_key(line.pk) 

498 terms.append(sympy.Symbol(symbol_key)) 

499 bindings[symbol_key] = amount 

500 inputs.append( 

501 FormulaInput( 

502 key=symbol_key, 

503 label=line.label, 

504 unit=f"{line.currency}/year", 

505 ) 

506 ) 

507 return OperatingAggregateFormula( 

508 formula=EconomicsFormula( 

509 key=key, 

510 expression=_sum_expressions(terms), 

511 unit=f"{_study_currency(study)}/year", 

512 inputs=tuple(inputs), 

513 steps=( 

514 FormulaStep( 

515 kind=key, 

516 label=label, 

517 expression="sum(included operating line formulas)", 

518 unit=f"{_study_currency(study)}/year", 

519 ), 

520 ), 

521 missing_child_policy="strict_after_revenue_classification", 

522 blocked_children=tuple(blocked_children), 

523 blocked_reason=_operating_total_blocked_reason(label, blocked_children), 

524 ), 

525 bindings=bindings, 

526 ) 

527 

528 

529def _operating_total_blocked_reason(label: str, blocked_children: list[dict[str, str]]) -> str: 

530 if not blocked_children: 

531 return "" 

532 if all(child.get("code") == "missing_annual_operating_hours" for child in blocked_children): 

533 return "Set annual operating hours in Settings, then recalculate Project overview." 

534 return f"{label} has blocked operating lines." 

535 

536 

537def _missing_annual_operating_hours_message(line: OperatingCostLine, rate_unit: str) -> str: 

538 unit_pair = _operating_line_unit_pair(line, rate_unit) 

539 if unit_pair: 539 ↛ 541line 539 didn't jump to line 541 because the condition on line 539 was always true

540 return f"{unit_pair} needs annual operating hours in Settings." 

541 return "Needs annual operating hours in Settings." 

542 

543 

544def _unsupported_annual_cost_conversion_message(line: OperatingCostLine, rate_unit: str) -> str: 

545 unit_pair = _operating_line_unit_pair(line, rate_unit) 

546 if unit_pair: 

547 return f"{unit_pair} cannot be converted to an annual cost. Check this line's basis and rate units." 

548 return "This line's basis and rate units cannot be converted to an annual cost." 

549 

550 

551def _operating_line_unit_pair(line: OperatingCostLine, rate_unit: str) -> str: 

552 basis_unit = line.basis_unit.strip() 

553 rate_unit = rate_unit.strip() 

554 if basis_unit and rate_unit: 554 ↛ 556line 554 didn't jump to line 556 because the condition on line 554 was always true

555 return f"{basis_unit} with {rate_unit}" 

556 if basis_unit: 

557 return f"{basis_unit} basis" 

558 if rate_unit: 

559 return f"{rate_unit} rate" 

560 return "" 

561 

562 

563def operating_line_is_revenue(line: OperatingCostLine) -> bool: 

564 """Return whether an operating line is a credit against operating cost.""" 

565 return ( 

566 line.economic_effect == OperatingLineEconomicEffect.REVENUE 

567 or line.category == OperatingLineCategory.OUTPUT_REVENUE 

568 ) 

569 

570 

571def operating_line_input_key(line_id: int) -> str: 

572 return f"operating_line_{line_id}" 

573 

574 

575def _process_energy_formula( 

576 *, 

577 line: OperatingCostLine, 

578 target_quantity: Decimal, 

579 source_quantity: Decimal, 

580 source_unit: str, 

581) -> ProcessEnergyContributionFormula: 

582 return ProcessEnergyContributionFormula( 

583 formula=EconomicsFormula( 

584 key=f"process_energy_contribution:{line.pk or 'unsaved'}", 

585 expression=decimal_to_sympy(target_quantity), 

586 unit=PROCESS_ENERGY_QUANTITY_UNIT, 

587 inputs=(), 

588 steps=( 

589 FormulaStep( 

590 kind="process_energy_contribution", 

591 label="Process-energy contribution", 

592 expression=f"{source_quantity} {source_unit} -> {PROCESS_ENERGY_QUANTITY_UNIT}", 

593 amount=target_quantity, 

594 unit=PROCESS_ENERGY_QUANTITY_UNIT, 

595 ), 

596 ), 

597 ), 

598 source_quantity=source_quantity, 

599 source_unit=source_unit, 

600 ) 

601 

602 

603def _multiply_terms(basis_symbol: sympy.Symbol, annualization_factor: Decimal, rate_amount: Decimal) -> sympy.Expr: 

604 terms: list[sympy.Expr] = [basis_symbol] 

605 if annualization_factor != Decimal("1"): 

606 terms.append(decimal_to_sympy(annualization_factor)) 

607 terms.append(decimal_to_sympy(rate_amount)) 

608 return sympy.Mul(*terms, evaluate=False) 

609 

610 

611def _sum_expressions(terms: list[sympy.Expr]) -> sympy.Expr: 

612 if not terms: 

613 return decimal_to_sympy(Decimal("0")) 

614 if len(terms) == 1: 

615 return terms[0] 

616 return sympy.Add(*terms, evaluate=False) 

617 

618 

619def _study_currency(study: EconomicsStudy) -> str: 

620 assumptions = get_settings_profile(study) 

621 if assumptions is None: 

622 return "NZD" 

623 return assumptions.currency or "NZD" 

624 

625 

626def _effective_rate_amount(line: OperatingCostLine) -> Decimal | None: 

627 if line.rate_amount is not None: 627 ↛ 629line 627 didn't jump to line 629 because the condition on line 627 was always true

628 return line.rate_amount 

629 if ( 

630 line.source_default_rate_id 

631 and line.source_default_rate.value is not None 

632 and line.source_default_rate.review_status == DefaultRateReviewStatus.REVIEWED 

633 and line.source_default_rate.value_kind == DefaultRateValueKind.REVIEWED_DEFAULT 

634 ): 

635 return line.source_default_rate.value 

636 return None 

637 

638 

639def _effective_rate_unit(line: OperatingCostLine) -> str: 

640 if line.rate_unit: 640 ↛ 642line 640 didn't jump to line 642 because the condition on line 640 was always true

641 return line.rate_unit 

642 if line.source_default_rate_id: 

643 return line.source_default_rate.display_unit 

644 return "" 

645 

646 

647def _basis_input_key(line: OperatingCostLine) -> str: 

648 return f"basis_{line.pk or 'unsaved'}" 

649 

650 

651def _metadata_decimal(value) -> Decimal | None: 

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

653 return None 

654 try: 

655 amount = Decimal(str(value)) 

656 except (InvalidOperation, ValueError, TypeError): 

657 return None 

658 return amount if amount.is_finite() else None 

659 

660 

661def _override_or_metadata_decimal(override: dict | None, key: str, metadata_value) -> Decimal | None: 

662 if not override or key not in override or override.get(key) in (None, ""): 

663 return _metadata_decimal(metadata_value) 

664 return _metadata_decimal(override.get(key)) 

665 

666 

667def _decimal_literal(value: Decimal) -> str: 

668 return format(value.normalize(), "f")