Coverage for backend/django/Economics/formulas/native_properties/sync.py: 97%

1from __future__ import annotations 

2 

3import re 

4 

5from django.db import transaction 

6 

7from Economics.formulas.models import EconomicsMetricFormula 

8from Economics.studies.models import EconomicsStudy 

9from Economics.results.services.financial_metrics.contracts import FinancialMetricsError 

10from Economics.results.services.financial_metrics.inputs import calculate_study_financial_metrics 

11from Economics.formulas.builders.native import native_property_expression 

12from Economics.formulas.builders.native_property_formulas import NativePropertyExpression 

13from Economics.formulas.property_state import apply_economics_property_state 

14from Economics.costing.line_properties.sync import sync_economics_line_properties_for_study 

15from Economics.formulas.native_properties.materialize import materialize_economics_native_properties 

16from Economics.formulas.native_properties.specs import EconomicsNativePropertySpec, native_property_specs 

17from Economics.shared.choices import EconomicsScheduleMode 

18 

19PROPERTY_MENTION_RE = re.compile(r"@\[[^\]]+\]\(prop(\d+)\)") 

20SCHEDULE_OBJECTIVE_BLOCK_REASON = ( 

21 "Economics metrics are temporarily unsupported as optimisation objectives " 

22 "when production scheduling is selected for the study." 

23) 

24 

25 

26def sync_economics_native_properties_for_study(study: EconomicsStudy) -> dict[str, int]: 

27 """Materialize native properties and refresh display/formula fields.""" 

28 

29 with transaction.atomic(): 

30 values = materialize_economics_native_properties(study) 

31 specs_by_key = {spec.field_key: spec for spec in native_property_specs(study)} 

32 sync_economics_line_properties_for_study(study) 

33 synced: dict[str, int] = {} 

34 solve_visible_reference_ids: set[int] = set() 

35 blocked_reference_reasons: dict[int, str] = {} 

36 financial_metrics = None 

37 financial_metrics_error = None 

38 if any(spec.result_metric_key for spec in specs_by_key.values()): 38 ↛ 46line 38 didn't jump to line 46 because the condition on line 38 was always true

39 try: 

40 financial_metrics = calculate_study_financial_metrics(study) 

41 except FinancialMetricsError as exc: 

42 financial_metrics_error = exc 

43 # Native property specs are emitted in dependency order. The reference 

44 # guard below can therefore block formulas that compose from earlier 

45 # generated properties that were not solve-visible. 

46 for field_key, property_value in values.items(): 

47 expression = native_property_expression( 

48 study, 

49 field_key, 

50 financial_metrics=financial_metrics, 

51 financial_metrics_error=financial_metrics_error, 

52 ) 

53 expression = _block_expression_with_invalid_references( 

54 expression, 

55 solve_visible_reference_ids=solve_visible_reference_ids, 

56 blocked_reference_reasons=blocked_reference_reasons, 

57 ) 

58 spec = specs_by_key[field_key] 

59 solve_visible = expression.solve_visible and spec.solve_visible 

60 display_value = None if expression.value is None else str(expression.value) 

61 objective_block_reason = _schedule_objective_block_reason(study) 

62 property_incomplete = not solve_visible or bool(objective_block_reason) 

63 incomplete_reason = "" if not property_incomplete else ( 

64 expression.blocked_reason 

65 or objective_block_reason 

66 or f"`{property_value.property.displayName}` is incomplete." 

67 ) 

68 apply_economics_property_state( 

69 property_value.property, 

70 editable=False, 

71 formula_incomplete=property_incomplete, 

72 formula_incomplete_reason=incomplete_reason, 

73 ) 

74 changed_fields = [] 

75 if property_value.value != display_value: 

76 property_value.value = display_value 

77 property_value.displayValue = display_value 

78 changed_fields.extend(["value", "displayValue"]) 

79 if property_value.formula != expression.formula: 

80 property_value.formula = expression.formula 

81 changed_fields.append("formula") 

82 if changed_fields: 

83 property_value.save(update_fields=changed_fields) 

84 if solve_visible: 

85 solve_visible_reference_ids.add(property_value.pk) 

86 blocked_reference_reasons.pop(property_value.pk, None) 

87 else: 

88 solve_visible_reference_ids.discard(property_value.pk) 

89 blocked_reference_reasons[property_value.pk] = ( 

90 expression.blocked_reason 

91 or f"`{property_value.property.displayName}` is not solve-visible." 

92 ) 

93 formula_record_id = expression.formula_record_id 

94 metric_key = spec.result_metric_key 

95 if formula_record_id is None and financial_metrics is not None and metric_key: 

96 metric = financial_metrics.metrics.get(metric_key) 

97 formula_record_id = metric.formula_record_id if metric is not None else None 

98 _link_native_formula_property( 

99 study=study, 

100 spec=spec, 

101 property_value=property_value, 

102 expression=expression, 

103 formula_record_id=formula_record_id, 

104 solve_visible=solve_visible, 

105 ) 

106 synced[field_key] = property_value.pk 

107 return synced 

108 

109 

110def _link_native_formula_property( 

111 *, 

112 study: EconomicsStudy, 

113 spec: EconomicsNativePropertySpec, 

114 property_value, 

115 expression: NativePropertyExpression, 

116 formula_record_id: int | None, 

117 solve_visible: bool, 

118) -> None: 

119 if formula_record_id is not None: 

120 EconomicsMetricFormula.objects.filter( 

121 pk=formula_record_id, 

122 flowsheet_state=study.flowsheet_state, 

123 study=study, 

124 ).update(property_value=property_value) 

125 return 

126 

127 status = "calculated" if expression.value is not None and solve_visible else "unavailable" 

128 EconomicsMetricFormula.objects.update_or_create( 

129 flowsheet_state=study.flowsheet_state, 

130 study=study, 

131 metric_key=_native_metric_key(spec), 

132 defaults={ 

133 "property_value": property_value, 

134 "formula_key": spec.field_key, 

135 "formula": expression.formula, 

136 "property_formula": expression.formula, 

137 "unit": property_value.property.unit, 

138 "value": str(expression.value) if expression.value is not None else None, 

139 "status": status, 

140 "formula_audit": { 

141 "formula_key": spec.field_key, 

142 "formula": expression.formula, 

143 "value": str(expression.value) if expression.value is not None else None, 

144 }, 

145 "blocked_reason": "" if solve_visible else expression.blocked_reason, 

146 }, 

147 ) 

148 

149 

150def _native_metric_key(spec: EconomicsNativePropertySpec) -> str: 

151 return spec.result_metric_key or spec.field_key 

152 

153 

154def _schedule_objective_block_reason(study: EconomicsStudy) -> str: 

155 if study.schedule_mode in {EconomicsScheduleMode.SCENARIO, EconomicsScheduleMode.COMPOSITE}: 

156 return SCHEDULE_OBJECTIVE_BLOCK_REASON 

157 return "" 

158 

159 

160def _block_expression_with_invalid_references( 

161 expression: NativePropertyExpression, 

162 *, 

163 solve_visible_reference_ids: set[int], 

164 blocked_reference_reasons: dict[int, str], 

165) -> NativePropertyExpression: 

166 """Block generated formulas that reference blocked generated children.""" 

167 

168 if not expression.solve_visible or not expression.formula: 

169 return expression 

170 reference_ids = sorted( 

171 {int(match.group(1)) for match in PROPERTY_MENTION_RE.finditer(expression.formula)} 

172 ) 

173 if not reference_ids: 

174 return expression 

175 for reference_id in reference_ids: 

176 if reference_id in solve_visible_reference_ids: 

177 continue 

178 reason = blocked_reference_reasons.get(reference_id) 

179 if reason: 

180 return _blocked_reference_expression(expression, reason) 

181 return expression 

182 

183 

184def _blocked_reference_expression( 

185 expression: NativePropertyExpression, 

186 reason: str, 

187) -> NativePropertyExpression: 

188 return NativePropertyExpression( 

189 "", 

190 False, 

191 reason, 

192 expression.value, 

193 expression.formula_record_id, 

194 )