Coverage for backend/django/Economics/results/serializers.py: 95%
108 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from drf_spectacular.utils import extend_schema_field
2from rest_framework import serializers
4from Economics.formulas.serializers import FormulaAuditSerializer
5from Economics.results.models import EconomicsChartDataset, EconomicsResultLine, EconomicsResultRun
6from Economics.results.services.financial_metrics.metric_catalog import MetricComparisonDirection
7from Economics.shared.serializer_base import FlowsheetScopedSerializer
8from Economics.shared.serializers import UnitOptionSerializer
9from Economics.results.services.lifecycle.runs import classify_result_run
10from Economics.results.services.resource_metrics import resource_metric_payloads_for_lines
11from Economics.results.unit_options import annual_result_basis_quantities_by_unit, annual_result_basis_unit_options
14class EconomicsResultLineSerializer(FlowsheetScopedSerializer):
15 formula_audit = serializers.SerializerMethodField()
16 annual_basis_unit_options = serializers.SerializerMethodField()
17 annual_basis_quantities_by_unit = serializers.SerializerMethodField()
19 class Meta:
20 model = EconomicsResultLine
21 fields = (
22 "id",
23 "kind",
24 "group",
25 "label",
26 "row_key",
27 "amount",
28 "unit",
29 "source_costable_item",
30 "source_cost_curve",
31 "source_capital_line",
32 "source_operating_line",
33 "source_index_value",
34 "source_property_info",
35 "source_row_key",
36 "source_label",
37 "source_note",
38 "resource_source_kind",
39 "resource_source_object_name",
40 "resource_source_object_type",
41 "resource_property_name",
42 "resource_breakdown_category",
43 "resource_classification",
44 "resource_metric_quantity",
45 "resource_metric_unit",
46 "annual_basis_quantity",
47 "annual_basis_unit",
48 "annual_basis_unit_options",
49 "annual_basis_quantities_by_unit",
50 "warning_payload",
51 "formula_audit",
52 "sort_order",
53 "created_at",
54 )
56 @extend_schema_field(FormulaAuditSerializer(allow_null=True))
57 def get_formula_audit(self, instance) -> dict | None:
58 payload = instance.warning_payload if isinstance(instance.warning_payload, dict) else {}
59 formula_audit = payload.get("formula_audit")
60 return formula_audit if isinstance(formula_audit, dict) else None
62 @extend_schema_field(UnitOptionSerializer(many=True))
63 def get_annual_basis_unit_options(self, instance) -> list[dict[str, str]]:
64 if instance.annual_basis_quantity is None:
65 return []
66 return annual_result_basis_unit_options(instance.annual_basis_unit)
68 @extend_schema_field(serializers.DictField(child=serializers.CharField()))
69 def get_annual_basis_quantities_by_unit(self, instance) -> dict[str, str]:
70 return annual_result_basis_quantities_by_unit(
71 quantity=instance.annual_basis_quantity,
72 current_unit=instance.annual_basis_unit,
73 )
76class EconomicsChartDatasetSerializer(FlowsheetScopedSerializer):
77 class Meta:
78 model = EconomicsChartDataset
79 fields = (
80 "id",
81 "result_run",
82 "chart_key",
83 "title",
84 "chart_type",
85 "source_row_keys",
86 "chart_data",
87 "rendering_metadata",
88 "created_at",
89 )
92class WarningSummarySerializer(serializers.Serializer):
93 """Public warning shape for API callers."""
95 code = serializers.CharField()
96 severity = serializers.CharField()
97 message = serializers.CharField()
98 context = serializers.DictField(required=False)
101class ResourceMetricSerializer(serializers.Serializer):
102 """High-level resource metric shown in result and comparison summaries."""
104 row_key = serializers.CharField()
105 label = serializers.CharField()
106 amount = serializers.CharField(allow_null=True)
107 unit = serializers.CharField()
108 status = serializers.CharField()
109 maximum_fraction_digits = serializers.IntegerField()
110 comparison_direction = serializers.ChoiceField(
111 choices=[direction.value for direction in MetricComparisonDirection]
112 )
115def _warning_summaries(payload: object) -> list[dict]:
116 if not isinstance(payload, dict): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 return []
118 warnings = payload.get("warnings", [])
119 if not isinstance(warnings, list): 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 return []
122 summaries: list[dict[str, str]] = []
123 for warning in warnings:
124 if not isinstance(warning, dict): 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 continue
126 summaries.append(
127 {
128 "code": str(warning.get("code", "")),
129 "severity": str(warning.get("severity", "")),
130 "message": str(warning.get("message", "")),
131 "context": (
132 warning.get("context", {})
133 if isinstance(warning.get("context"), dict)
134 else {}
135 ),
136 }
137 )
138 return summaries
141class EconomicsResultRunSerializer(FlowsheetScopedSerializer):
142 lines = EconomicsResultLineSerializer(many=True, read_only=True)
143 chart_datasets = EconomicsChartDatasetSerializer(many=True, read_only=True)
144 resource_metrics = serializers.SerializerMethodField()
145 classification = serializers.SerializerMethodField()
146 warnings = serializers.SerializerMethodField()
147 requires_solve = serializers.SerializerMethodField()
148 latest_stale_reason = serializers.SerializerMethodField()
150 class Meta:
151 model = EconomicsResultRun
152 fields = (
153 "id",
154 "flowsheet",
155 "study",
156 "status",
157 "classification",
158 "result_currency",
159 "result_basis_date",
160 "warnings",
161 "requires_solve",
162 "latest_stale_reason",
163 "created_at",
164 "completed_at",
165 "lines",
166 "resource_metrics",
167 "chart_datasets",
168 )
169 read_only_fields = fields
171 @extend_schema_field(serializers.CharField)
172 def get_classification(self, obj: EconomicsResultRun) -> str:
173 return serializer_result_run_classification(obj, self.context)
175 @extend_schema_field(WarningSummarySerializer(many=True))
176 def get_warnings(self, obj: EconomicsResultRun) -> list[dict[str, str]]:
177 return _warning_summaries(obj.warning_payload)
179 @extend_schema_field(serializers.BooleanField)
180 def get_requires_solve(self, obj: EconomicsResultRun) -> bool:
181 payload = obj.warning_payload if isinstance(obj.warning_payload, dict) else {}
182 return bool(payload.get("requires_solve", False))
184 @extend_schema_field(serializers.CharField)
185 def get_latest_stale_reason(self, obj: EconomicsResultRun) -> str:
186 payload = obj.warning_payload if isinstance(obj.warning_payload, dict) else {}
187 return str(payload.get("latest_stale_reason", ""))
189 @extend_schema_field(ResourceMetricSerializer(many=True))
190 def get_resource_metrics(self, obj: EconomicsResultRun) -> list[dict[str, str | None]]:
191 prefetched_lines = getattr(obj, "_prefetched_objects_cache", {}).get("lines")
192 if prefetched_lines is None:
193 lines = obj.lines.filter(group="operating_lines").select_related(
194 "source_operating_line",
195 "source_operating_line__source_default_rate",
196 )
197 else:
198 lines = [line for line in prefetched_lines if line.group == "operating_lines"]
199 return [
200 payload.model_dump(mode="json")
201 for payload in resource_metric_payloads_for_lines(lines)
202 ]
205class ResultStatePayloadSerializer(serializers.Serializer):
206 """Compact current/stale state for presentation results without exposing fingerprints."""
208 run_id = serializers.IntegerField(allow_null=True)
209 status = serializers.CharField()
210 classification = serializers.CharField()
211 completed_at = serializers.DateTimeField(required=False, allow_null=True)
212 warnings = WarningSummarySerializer(many=True)
213 requires_solve = serializers.BooleanField()
214 latest_stale_reason = serializers.CharField(required=False, allow_blank=True)
217def serializer_result_run_classification(
218 result_run: EconomicsResultRun,
219 serializer_context: dict,
220) -> str:
221 """Classify a run once when nested serializers share one request context."""
223 classifications = serializer_context.setdefault("_economics_result_classifications", {})
224 cache_key = (result_run.flowsheet_state_id, result_run.pk, result_run.status)
225 if cache_key not in classifications:
226 classifications[cache_key] = classify_result_run(result_run)
227 return classifications[cache_key]
230class RecalculateRequestSerializer(serializers.Serializer):
231 reason = serializers.CharField(required=False, allow_blank=True, default="api_recalculate")