Coverage for backend/django/Economics/results/serializers.py: 94%

1from drf_spectacular.utils import extend_schema_field 

2from rest_framework import serializers 

3 

4from Economics.formulas.serializers import FormulaAuditSerializer 

5from Economics.results.models import EconomicsChartDataset, EconomicsResultLine, EconomicsResultRun 

6from Economics.shared.serializer_base import FlowsheetScopedSerializer 

7from Economics.shared.serializers import UnitOptionSerializer 

8from Economics.results.services.lifecycle.runs import classify_result_run 

9from Economics.results.unit_options import annual_result_basis_quantities_by_unit, annual_result_basis_unit_options 

10 

11 

12class EconomicsResultLineSerializer(FlowsheetScopedSerializer): 

13 formula_audit = serializers.SerializerMethodField() 

14 annual_basis_unit_options = serializers.SerializerMethodField() 

15 annual_basis_quantities_by_unit = serializers.SerializerMethodField() 

16 

17 class Meta: 

18 model = EconomicsResultLine 

19 fields = ( 

20 "id", 

21 "kind", 

22 "group", 

23 "label", 

24 "row_key", 

25 "amount", 

26 "unit", 

27 "source_costable_item", 

28 "source_cost_curve", 

29 "source_capital_line", 

30 "source_operating_line", 

31 "source_index_value", 

32 "source_property_info", 

33 "source_row_key", 

34 "source_label", 

35 "source_note", 

36 "resource_source_kind", 

37 "resource_source_object_name", 

38 "resource_source_object_type", 

39 "resource_property_name", 

40 "resource_breakdown_category", 

41 "annual_basis_quantity", 

42 "annual_basis_unit", 

43 "annual_basis_unit_options", 

44 "annual_basis_quantities_by_unit", 

45 "warning_payload", 

46 "formula_audit", 

47 "sort_order", 

48 "created_at", 

49 ) 

50 

51 @extend_schema_field(FormulaAuditSerializer(allow_null=True)) 

52 def get_formula_audit(self, instance) -> dict | None: 

53 payload = instance.warning_payload if isinstance(instance.warning_payload, dict) else {} 

54 formula_audit = payload.get("formula_audit") 

55 return formula_audit if isinstance(formula_audit, dict) else None 

56 

57 @extend_schema_field(UnitOptionSerializer(many=True)) 

58 def get_annual_basis_unit_options(self, instance) -> list[dict[str, str]]: 

59 if instance.annual_basis_quantity is None: 

60 return [] 

61 return annual_result_basis_unit_options(instance.annual_basis_unit) 

62 

63 @extend_schema_field(serializers.DictField(child=serializers.CharField())) 

64 def get_annual_basis_quantities_by_unit(self, instance) -> dict[str, str]: 

65 return annual_result_basis_quantities_by_unit( 

66 quantity=instance.annual_basis_quantity, 

67 current_unit=instance.annual_basis_unit, 

68 ) 

69 

70 

71class EconomicsChartDatasetSerializer(FlowsheetScopedSerializer): 

72 class Meta: 

73 model = EconomicsChartDataset 

74 fields = ( 

75 "id", 

76 "result_run", 

77 "chart_key", 

78 "title", 

79 "chart_type", 

80 "source_row_keys", 

81 "chart_data", 

82 "rendering_metadata", 

83 "created_at", 

84 ) 

85 

86 

87class WarningSummarySerializer(serializers.Serializer): 

88 """Public warning shape for API callers.""" 

89 

90 code = serializers.CharField() 

91 severity = serializers.CharField() 

92 message = serializers.CharField() 

93 context = serializers.DictField(required=False) 

94 

95 

96def _warning_summaries(payload: object) -> list[dict]: 

97 if not isinstance(payload, dict): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 return [] 

99 warnings = payload.get("warnings", []) 

100 if not isinstance(warnings, list): 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 return [] 

102 

103 summaries: list[dict[str, str]] = [] 

104 for warning in warnings: 

105 if not isinstance(warning, dict): 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true

106 continue 

107 summaries.append( 

108 { 

109 "code": str(warning.get("code", "")), 

110 "severity": str(warning.get("severity", "")), 

111 "message": str(warning.get("message", "")), 

112 "context": ( 

113 warning.get("context", {}) 

114 if isinstance(warning.get("context"), dict) 

115 else {} 

116 ), 

117 } 

118 ) 

119 return summaries 

120 

121 

122class EconomicsResultRunSerializer(FlowsheetScopedSerializer): 

123 lines = EconomicsResultLineSerializer(many=True, read_only=True) 

124 chart_datasets = EconomicsChartDatasetSerializer(many=True, read_only=True) 

125 classification = serializers.SerializerMethodField() 

126 warnings = serializers.SerializerMethodField() 

127 requires_solve = serializers.SerializerMethodField() 

128 latest_stale_reason = serializers.SerializerMethodField() 

129 

130 class Meta: 

131 model = EconomicsResultRun 

132 fields = ( 

133 "id", 

134 "flowsheet", 

135 "study", 

136 "status", 

137 "classification", 

138 "result_currency", 

139 "result_basis_date", 

140 "warnings", 

141 "requires_solve", 

142 "latest_stale_reason", 

143 "created_at", 

144 "completed_at", 

145 "lines", 

146 "chart_datasets", 

147 ) 

148 read_only_fields = fields 

149 

150 @extend_schema_field(serializers.CharField) 

151 def get_classification(self, obj: EconomicsResultRun) -> str: 

152 return classify_result_run(obj) 

153 

154 @extend_schema_field(WarningSummarySerializer(many=True)) 

155 def get_warnings(self, obj: EconomicsResultRun) -> list[dict[str, str]]: 

156 return _warning_summaries(obj.warning_payload) 

157 

158 @extend_schema_field(serializers.BooleanField) 

159 def get_requires_solve(self, obj: EconomicsResultRun) -> bool: 

160 payload = obj.warning_payload if isinstance(obj.warning_payload, dict) else {} 

161 return bool(payload.get("requires_solve", False)) 

162 

163 @extend_schema_field(serializers.CharField) 

164 def get_latest_stale_reason(self, obj: EconomicsResultRun) -> str: 

165 payload = obj.warning_payload if isinstance(obj.warning_payload, dict) else {} 

166 return str(payload.get("latest_stale_reason", "")) 

167 

168 

169class ResultStatePayloadSerializer(serializers.Serializer): 

170 """Compact current/stale state for presentation results without exposing fingerprints.""" 

171 

172 run_id = serializers.IntegerField(allow_null=True) 

173 status = serializers.CharField() 

174 classification = serializers.CharField() 

175 completed_at = serializers.DateTimeField(required=False, allow_null=True) 

176 warnings = WarningSummarySerializer(many=True) 

177 requires_solve = serializers.BooleanField() 

178 latest_stale_reason = serializers.CharField(required=False, allow_blank=True) 

179 

180class RecalculateRequestSerializer(serializers.Serializer): 

181 reason = serializers.CharField(required=False, allow_blank=True, default="api_recalculate")