Coverage for backend/django/Economics/costing/capital/custom_capital_lines.py: 60%

80 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2026-06-23 21:51 +0000

1"""Custom capital-line calculation helpers. 

2 

3Manual fixed capital lines carry their dollar amount directly. Percentage 

4capital lines are intentionally calculated against the generated unit-operation 

5CAPEX subtotal only, so multiple percentage lines each apply to the same stable 

6base rather than compounding on top of each other. 

7""" 

8 

9from __future__ import annotations 

10 

11from decimal import Decimal, ROUND_HALF_UP 

12 

13from Economics.costing.models import CapitalCostLine 

14 

15from Economics.shared.choices import CapitalLineBasis 

16 

17from Economics.studies.models import EconomicsStudy 

18from Economics.formulas.builders.capital import ( 

19 build_custom_capital_line_formula, 

20 build_generated_unit_capex_subtotal_formula, 

21) 

22from Economics.formulas.engine.core import FormulaError, FormulaEvaluation 

23 

24 

25_RESULT_AMOUNT_QUANTUM = Decimal("0.0001") 

26 

27 

28def base_capex_for_custom_percentage_lines(study: EconomicsStudy) -> Decimal: 

29 """Return the generated unit-operation CAPEX subtotal used by percent rows.""" 

30 subtotal_formula = build_generated_unit_capex_subtotal_formula(study) 

31 subtotal = subtotal_formula.evaluate() 

32 if subtotal is None: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 raise FormulaError("blocked_generated_capex_subtotal", subtotal_formula.formula.blocked_reason) 

34 return _result_amount(subtotal) 

35 

36 

37def resolved_capital_line_amount( 

38 line: CapitalCostLine, 

39 *, 

40 base_capex: Decimal, 

41) -> Decimal | None: 

42 """Return the line amount using ``base_capex`` for percentage custom rows.""" 

43 try: 

44 amount = build_custom_capital_line_formula(line, base_capex=base_capex).evaluate() 

45 except FormulaError: 

46 return None 

47 if line.calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT: 

48 return _result_amount(amount) 

49 return amount 

50 

51 

52def sync_custom_capital_lines(study: EconomicsStudy) -> None: 

53 """Persist calculated dollar amounts for custom percentage capital lines.""" 

54 lines = CapitalCostLine.objects.filter( 

55 flowsheet=study.flowsheet, 

56 study=study, 

57 calculation_basis=CapitalLineBasis.BASE_CAPEX_PERCENT, 

58 ) 

59 if not lines.exists(): 

60 return 

61 try: 

62 base_capex = base_capex_for_custom_percentage_lines(study) 

63 except FormulaError as exc: 

64 for line in lines: 

65 warning_payload = { 

66 "calculation_method": "base_capex_percent", 

67 "base_capex": None, 

68 "basis_percent": None if line.basis_percent is None else str(line.basis_percent), 

69 "amount": None, 

70 "formula": {}, 

71 "warnings": [ 

72 { 

73 "code": exc.code, 

74 "severity": "error", 

75 "message": exc.message, 

76 "context": exc.context, 

77 } 

78 ], 

79 } 

80 changed_fields = [] 

81 if line.amount is not None: 

82 line.amount = None 

83 changed_fields.append("amount") 

84 if line.warning_payload != warning_payload: 

85 line.warning_payload = warning_payload 

86 changed_fields.append("warning_payload") 

87 if line.confidence != "blocked": 

88 line.confidence = "blocked" 

89 changed_fields.append("confidence") 

90 if changed_fields: 

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

92 return 

93 for line in lines: 

94 try: 

95 formula = build_custom_capital_line_formula(line, base_capex=base_capex) 

96 amount = _result_amount(formula.evaluate()) 

97 formula_payload = formula.formula.audit_payload( 

98 FormulaEvaluation(value=amount, bindings=formula.bindings) 

99 ) 

100 except FormulaError: 

101 amount = None 

102 formula_payload = {} 

103 warning_payload = _percentage_warning_payload( 

104 line=line, 

105 base_capex=base_capex, 

106 amount=amount, 

107 formula=formula_payload, 

108 ) 

109 changed_fields = [] 

110 if line.amount != amount: 110 ↛ 113line 110 didn't jump to line 113 because the condition on line 110 was always true

111 line.amount = amount 

112 changed_fields.append("amount") 

113 if line.warning_payload != warning_payload: 113 ↛ 116line 113 didn't jump to line 116 because the condition on line 113 was always true

114 line.warning_payload = warning_payload 

115 changed_fields.append("warning_payload") 

116 if line.confidence != "calculated": 116 ↛ 119line 116 didn't jump to line 119 because the condition on line 116 was always true

117 line.confidence = "calculated" 

118 changed_fields.append("confidence") 

119 if changed_fields: 119 ↛ 93line 119 didn't jump to line 93 because the condition on line 119 was always true

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

121 

122 

123def _percentage_warning_payload( 

124 *, 

125 line: CapitalCostLine, 

126 base_capex: Decimal, 

127 amount: Decimal | None, 

128 formula: dict, 

129) -> dict: 

130 percent = line.basis_percent or Decimal("0") 

131 return json_ready( 

132 { 

133 "calculation_method": CapitalLineBasis.BASE_CAPEX_PERCENT, 

134 "base_capex_amount": _result_amount(base_capex), 

135 "basis_percent": percent, 

136 "amount": amount, 

137 "formula": formula, 

138 "capital_factors": [ 

139 { 

140 "kind": "base_capex_percent", 

141 "label": "Base CAPEX percentage", 

142 "factor": str(percent / Decimal("100")), 

143 "percent": str(percent), 

144 "amount": str(amount) if amount is not None else None, 

145 "detail": "Applied to generated unit-operation capital subtotal.", 

146 } 

147 ], 

148 } 

149 ) 

150 

151 

152def _result_amount(value: Decimal | None) -> Decimal | None: 

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

154 return None 

155 return value.quantize(_RESULT_AMOUNT_QUANTUM, rounding=ROUND_HALF_UP) 

156 

157 

158def json_ready(value): 

159 if isinstance(value, Decimal): 

160 return str(value) 

161 if isinstance(value, dict): 

162 return {str(key): json_ready(nested_value) for key, nested_value in value.items()} 

163 if isinstance(value, (list, tuple)): 

164 return [json_ready(nested_value) for nested_value in value] 

165 return value