Coverage for backend/django/Economics/formulas/builders/metric_formulas.py: 100%

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from decimal import Decimal 

5from typing import Any 

6 

7from Economics.formulas.models import EconomicsMetricFormula 

8 

9from Economics.studies.models import EconomicsStudy 

10from Economics.formulas.engine.core import FormulaError 

11from Economics.formulas.builders.metrics import BoundMetricFormula 

12from Economics.formulas.builders.metric_property_formulas import metric_property_formula_bindings 

13 

14 

15@dataclass(frozen=True) 

16class PersistedMetricFormula: 

17 """Small immutable handle returned to financial metric contracts.""" 

18 

19 id: int 

20 property_formula: str 

21 formula_audit: dict[str, Any] 

22 blocked_reason: str = "" 

23 

24 

25class MetricFormulaStore: 

26 """Persist formula rows for study-backed financial metric calculations.""" 

27 

28 def __init__(self, study: EconomicsStudy): 

29 self.study = study 

30 self.persisted_metric_keys: set[str] = set() 

31 

32 def persist_metric_formula( 

33 self, 

34 *, 

35 metric_key: str, 

36 formula: BoundMetricFormula, 

37 value: Decimal | None, 

38 status: str, 

39 assumptions: Any, 

40 formula_audit: dict[str, Any], 

41 ) -> PersistedMetricFormula: 

42 """Upsert the formula row used for a single financial metric.""" 

43 

44 self.persisted_metric_keys.add(metric_key) 

45 property_formula = "" 

46 blocked_reason = "" 

47 try: 

48 property_formula = formula.render_property_formula( 

49 metric_property_formula_bindings( 

50 self.study, 

51 metric_key=metric_key, 

52 formula=formula, 

53 assumptions=assumptions, 

54 ) 

55 ) 

56 except FormulaError as exc: 

57 blocked_reason = exc.message 

58 record, _created = EconomicsMetricFormula.objects.update_or_create( 

59 flowsheet=self.study.flowsheet, 

60 study=self.study, 

61 metric_key=metric_key, 

62 defaults={ 

63 "formula_key": formula.formula.key, 

64 "formula": formula_audit.get("formula") or "", 

65 "property_formula": property_formula, 

66 "unit": formula.formula.unit, 

67 "value": str(value) if value is not None else None, 

68 "status": status, 

69 "formula_audit": formula_audit, 

70 "blocked_reason": blocked_reason, 

71 }, 

72 ) 

73 return PersistedMetricFormula( 

74 id=record.pk, 

75 property_formula=record.property_formula, 

76 formula_audit=record.formula_audit, 

77 blocked_reason=record.blocked_reason, 

78 ) 

79 

80 def delete_stale_metric_formulas(self) -> None: 

81 """Remove persisted formulas that were not emitted by the latest calculation.""" 

82 

83 self.study.metric_formulas.exclude(metric_key__in=self.persisted_metric_keys).delete() 

84 

85 def delete_all_metric_formulas(self) -> None: 

86 """Remove all persisted formulas for a study after a failed calculation.""" 

87 

88 self.study.metric_formulas.all().delete()