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 import FinancialMetricsError, calculate_study_financial_metrics 

10from Economics.formulas.builders.native import native_property_expression 

11from Economics.formulas.builders.native_property_formulas import NativePropertyExpression 

12from Economics.formulas.property_state import apply_economics_property_state 

13from Economics.costing.line_properties.sync import sync_economics_line_properties_for_study 

14from Economics.formulas.native_properties.materialize import materialize_economics_native_properties 

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

16 

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

18 

19 

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

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

22 

23 with transaction.atomic(): 

24 values = materialize_economics_native_properties(study) 

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

26 sync_economics_line_properties_for_study(study) 

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

28 solve_visible_reference_ids: set[int] = set() 

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

30 financial_metrics = None 

31 financial_metrics_error = None 

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

33 try: 

34 financial_metrics = calculate_study_financial_metrics(study) 

35 except FinancialMetricsError as exc: 

36 financial_metrics_error = exc 

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

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

39 # generated properties that were not solve-visible. 

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

41 expression = native_property_expression( 

42 study, 

43 field_key, 

44 financial_metrics=financial_metrics, 

45 financial_metrics_error=financial_metrics_error, 

46 ) 

47 expression = _block_expression_with_invalid_references( 

48 expression, 

49 solve_visible_reference_ids=solve_visible_reference_ids, 

50 blocked_reference_reasons=blocked_reference_reasons, 

51 ) 

52 spec = specs_by_key[field_key] 

53 solve_visible = expression.solve_visible and spec.solve_visible 

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

55 incomplete_reason = "" if solve_visible else ( 

56 expression.blocked_reason 

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

58 ) 

59 apply_economics_property_state( 

60 property_value.property, 

61 editable=False, 

62 formula_incomplete=not solve_visible, 

63 formula_incomplete_reason=incomplete_reason, 

64 ) 

65 changed_fields = [] 

66 if property_value.value != display_value: 

67 property_value.value = display_value 

68 property_value.displayValue = display_value 

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

70 if property_value.formula != expression.formula: 

71 property_value.formula = expression.formula 

72 changed_fields.append("formula") 

73 if changed_fields: 

74 property_value.save(update_fields=changed_fields) 

75 if solve_visible: 

76 solve_visible_reference_ids.add(property_value.pk) 

77 blocked_reference_reasons.pop(property_value.pk, None) 

78 else: 

79 solve_visible_reference_ids.discard(property_value.pk) 

80 blocked_reference_reasons[property_value.pk] = ( 

81 expression.blocked_reason 

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

83 ) 

84 formula_record_id = expression.formula_record_id 

85 metric_key = spec.result_metric_key 

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

87 metric = financial_metrics.metrics.get(metric_key) 

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

89 _link_native_formula_property( 

90 study=study, 

91 spec=spec, 

92 property_value=property_value, 

93 expression=expression, 

94 formula_record_id=formula_record_id, 

95 solve_visible=solve_visible, 

96 ) 

97 synced[field_key] = property_value.pk 

98 return synced 

99 

100 

101def _link_native_formula_property( 

102 *, 

103 study: EconomicsStudy, 

104 spec: EconomicsNativePropertySpec, 

105 property_value, 

106 expression: NativePropertyExpression, 

107 formula_record_id: int | None, 

108 solve_visible: bool, 

109) -> None: 

110 if formula_record_id is not None: 

111 EconomicsMetricFormula.objects.filter( 

112 pk=formula_record_id, 

113 flowsheet=study.flowsheet, 

114 study=study, 

115 ).update(property_value=property_value) 

116 return 

117 

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

119 EconomicsMetricFormula.objects.update_or_create( 

120 flowsheet=study.flowsheet, 

121 study=study, 

122 metric_key=_native_metric_key(spec), 

123 defaults={ 

124 "property_value": property_value, 

125 "formula_key": spec.field_key, 

126 "formula": expression.formula, 

127 "property_formula": expression.formula, 

128 "unit": property_value.property.unit, 

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

130 "status": status, 

131 "formula_audit": { 

132 "formula_key": spec.field_key, 

133 "formula": expression.formula, 

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

135 }, 

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

137 }, 

138 ) 

139 

140 

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

142 return spec.result_metric_key or spec.field_key 

143 

144 

145def _block_expression_with_invalid_references( 

146 expression: NativePropertyExpression, 

147 *, 

148 solve_visible_reference_ids: set[int], 

149 blocked_reference_reasons: dict[int, str], 

150) -> NativePropertyExpression: 

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

152 

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

154 return expression 

155 reference_ids = sorted( 

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

157 ) 

158 if not reference_ids: 

159 return expression 

160 for reference_id in reference_ids: 

161 if reference_id in solve_visible_reference_ids: 

162 continue 

163 reason = blocked_reference_reasons.get(reference_id) 

164 if reason: 

165 return _blocked_reference_expression(expression, reason) 

166 return expression 

167 

168 

169def _blocked_reference_expression( 

170 expression: NativePropertyExpression, 

171 reason: str, 

172) -> NativePropertyExpression: 

173 return NativePropertyExpression( 

174 "", 

175 False, 

176 reason, 

177 expression.value, 

178 expression.formula_record_id, 

179 )