Coverage for backend/django/Economics/studies/viewsets.py: 89%
217 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from django.core.exceptions import ObjectDoesNotExist
2from django.db import transaction
3from drf_spectacular.utils import OpenApiParameter, OpenApiTypes, extend_schema
4from rest_framework import status
5from rest_framework.decorators import action
6from rest_framework.exceptions import NotFound
7from rest_framework.response import Response
9from core.viewset import ModelViewSet
10from core.validation import get_current_flowsheet
11from Economics.costing.costable_items.serializers import CostableItemSerializer
12from Economics.costing.operating.serializers import (
13 OperatingCostLineSerializer,
14 OperatingLinesFromPropertiesRequestSerializer,
15 OperatingLineFromPropertyRequestSerializer,
16 OperatingStreamPropertyOptionSerializer,
17)
18from Economics.results.serializers import EconomicsChartDatasetSerializer, EconomicsResultRunSerializer, RecalculateRequestSerializer
19from Economics.settings_profiles.serializers import EconomicsAssumptionsSerializer, EconomicsBaselineSerializer
20from Economics.studies.api_mutations import _bad_request_from_error, _mark_study_stale
21from Economics.shared.access import _has_write_access, _require_write_access
22from Economics.studies.models import EconomicsStudy
23from Economics.studies.serializers import DuplicateStudyRequestSerializer, EconomicsStudyFullSerializer, EconomicsStudySerializer, EnableCostingRequestSerializer
24from Economics.studies.services.study_copy import duplicate_study
25from Economics.costing.costable_items.items import enable_costing_for_simulation_object
26from Economics.costing.costable_items.deleted_objects import delete_economics_lines_for_deleted_simulation_objects
27from Economics.costing.operating.stream_properties import (
28 create_operating_line_from_output_property,
29 output_stream_property_options,
30 sync_project_default_operating_line_rates_for_study,
31)
32from Economics.results.services.lifecycle.lifecycle import recalculate_presentation_results
33from Economics.costing.capital.generated_lines import _sync_generated_capital_lines
34from Economics.settings_profiles.services.settings_profiles import (
35 assumptions_from_settings_profile,
36 baseline_from_settings_profile,
37 get_or_create_default_settings_profile,
38 update_settings_profile_from_assumptions,
39 update_settings_profile_from_baseline,
40)
41from core.auxiliary.models.PropertyInfo import PropertyInfo
42from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
45class EconomicsStudyViewSet(ModelViewSet):
46 """CRUD and workflow actions for v1 economics studies."""
48 serializer_class = EconomicsStudySerializer
50 def get_queryset(self):
51 return EconomicsStudy.objects.select_related("settings_profile").prefetch_related(
52 "costable_items__cost_driver",
53 "costable_items__equipment_mapping",
54 "capital_lines",
55 "operating_lines",
56 "result_runs",
57 )
59 def get_serializer_class(self):
60 if self.action in {"retrieve", "full", "recalculate", "duplicate"}:
61 return EconomicsStudyFullSerializer
62 return EconomicsStudySerializer
64 @extend_schema(parameters=[OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT)])
65 def list(self, request):
66 return super().list(request)
68 @extend_schema(parameters=[OpenApiParameter(name="flowsheet", required=True, type=OpenApiTypes.INT)])
69 def create(self, request, *args, **kwargs):
70 return super().create(request, *args, **kwargs)
72 def perform_create(self, serializer):
73 context = get_current_flowsheet() or {}
74 flowsheet_id = context.get("flowsheet")
75 if serializer.validated_data.get("settings_profile") is None and flowsheet_id is not None: 75 ↛ 81line 75 didn't jump to line 81 because the condition on line 75 was always true
76 from core.auxiliary.models import Flowsheet
78 flowsheet = Flowsheet.objects.get(pk=flowsheet_id)
79 serializer.save(settings_profile=get_or_create_default_settings_profile(flowsheet))
80 return
81 serializer.save()
83 def perform_update(self, serializer):
84 previous_profile_id = serializer.instance.settings_profile_id
85 study = serializer.save()
86 if previous_profile_id != study.settings_profile_id: 86 ↛ exitline 86 didn't return from function 'perform_update' because the condition on line 86 was always true
87 sync_project_default_operating_line_rates_for_study(study, reason="economics_settings_profile_selected")
88 _mark_study_stale(study, reason="economics_settings_profile_selected")
90 @extend_schema(responses=EconomicsStudyFullSerializer)
91 @action(detail=True, methods=["get"])
92 def full(self, request, pk=None):
93 study = self.get_object()
94 if _has_write_access(request.user):
95 delete_economics_lines_for_deleted_simulation_objects(study)
96 refreshed_study = self.get_queryset().get(pk=study.pk)
97 return Response(EconomicsStudyFullSerializer(refreshed_study, context=self.get_serializer_context()).data)
99 @extend_schema(responses=OperatingStreamPropertyOptionSerializer(many=True))
100 @action(detail=True, methods=["get"], url_path="operating-property-options")
101 def operating_property_options(self, request, pk=None):
102 study = self.get_object()
103 serializer = OperatingStreamPropertyOptionSerializer(output_stream_property_options(study), many=True)
104 return Response(serializer.data)
106 @extend_schema(
107 request=OperatingLineFromPropertyRequestSerializer,
108 responses={status.HTTP_201_CREATED: OperatingCostLineSerializer},
109 )
110 @action(detail=True, methods=["post"], url_path="operating-lines/from-property")
111 def create_operating_line_from_property(self, request, pk=None):
112 _require_write_access(request.user)
113 study = self.get_object()
114 request_serializer = OperatingLineFromPropertyRequestSerializer(data=request.data)
115 request_serializer.is_valid(raise_exception=True)
116 try:
117 property_info = PropertyInfo.objects.get(pk=request_serializer.validated_data["property_info"])
118 except PropertyInfo.DoesNotExist as error:
119 raise NotFound("Selected property was not found.") from error
120 try:
121 line = create_operating_line_from_output_property(
122 study=study,
123 property_info=property_info,
124 category=request_serializer.validated_data["category"],
125 economic_effect=request_serializer.validated_data["economic_effect"],
126 outlet_stream_disposition=request_serializer.validated_data.get("outlet_stream_disposition", ""),
127 rate_type=request_serializer.validated_data.get("rate_type", None),
128 )
129 except ValueError as error:
130 raise _bad_request_from_error(error) from error
131 _mark_study_stale(study, reason="economics_operating_line_property_selected")
132 return Response(
133 OperatingCostLineSerializer(line, context=self.get_serializer_context()).data,
134 status=status.HTTP_201_CREATED,
135 )
137 @extend_schema(
138 request=OperatingLinesFromPropertiesRequestSerializer,
139 responses={
140 status.HTTP_201_CREATED: OperatingCostLineSerializer(many=True)
141 },
142 )
143 @action(detail=True, methods=["post"], url_path="operating-lines/from-properties")
144 def create_operating_lines_from_properties(self, request, pk=None):
145 _require_write_access(request.user)
146 study = self.get_object()
147 request_serializer = OperatingLinesFromPropertiesRequestSerializer(
148 data=request.data
149 )
150 request_serializer.is_valid(raise_exception=True)
151 line_requests = request_serializer.validated_data["lines"]
152 property_infos = PropertyInfo.objects.in_bulk(
153 [line["property_info"] for line in line_requests]
154 )
155 missing_property_ids = [
156 line["property_info"]
157 for line in line_requests
158 if line["property_info"] not in property_infos
159 ]
160 if missing_property_ids: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 raise NotFound("Selected property was not found.")
162 try:
163 with transaction.atomic():
164 lines = [
165 create_operating_line_from_output_property(
166 study=study,
167 property_info=property_infos[
168 line_request["property_info"]
169 ],
170 category=line_request["category"],
171 economic_effect=line_request["economic_effect"],
172 outlet_stream_disposition=line_request.get(
173 "outlet_stream_disposition", ""
174 ),
175 rate_type=line_request.get("rate_type", None),
176 )
177 for line_request in line_requests
178 ]
179 except ValueError as error:
180 raise _bad_request_from_error(error) from error
181 _mark_study_stale(study, reason="economics_operating_line_property_selected")
182 return Response(
183 OperatingCostLineSerializer(
184 lines, many=True, context=self.get_serializer_context()
185 ).data,
186 status=status.HTTP_201_CREATED,
187 )
189 @extend_schema(methods=["GET"], responses=EconomicsAssumptionsSerializer)
190 @extend_schema(methods=["PUT"], request=EconomicsAssumptionsSerializer, responses=EconomicsAssumptionsSerializer)
191 @extend_schema(methods=["PATCH"], request=EconomicsAssumptionsSerializer(partial=True), responses=EconomicsAssumptionsSerializer)
192 @action(detail=True, methods=["get", "put", "patch"])
193 def assumptions(self, request, pk=None):
194 study = self.get_object()
195 if request.method == "GET":
196 assumptions = assumptions_from_settings_profile(study)
197 if assumptions is not None: 197 ↛ 199line 197 didn't jump to line 199 because the condition on line 197 was always true
198 return Response(EconomicsAssumptionsSerializer(assumptions, context=self.get_serializer_context()).data)
199 try:
200 assumptions = study.assumptions
201 except ObjectDoesNotExist as exc:
202 raise NotFound("Economics assumptions have not been configured for this study.") from exc
203 return Response(EconomicsAssumptionsSerializer(assumptions, context=self.get_serializer_context()).data)
205 _require_write_access(request.user)
206 try:
207 instance = study.assumptions
208 except ObjectDoesNotExist:
209 instance = None
210 data = {**request.data, "study": study.pk}
211 serializer = EconomicsAssumptionsSerializer(instance, data=data, partial=request.method == "PATCH", context=self.get_serializer_context())
212 serializer.is_valid(raise_exception=True)
213 legacy_field_names = None if request.method != "PATCH" else set(serializer.validated_data)
214 assumptions = serializer.save()
215 update_settings_profile_from_assumptions(assumptions, field_names=legacy_field_names)
216 sync_project_default_operating_line_rates_for_study(study, reason="economics_assumptions_api_saved")
217 _mark_study_stale(study, reason="economics_assumptions_api_saved")
218 return Response(EconomicsAssumptionsSerializer(assumptions, context=self.get_serializer_context()).data)
220 @extend_schema(methods=["GET"], responses=EconomicsBaselineSerializer)
221 @extend_schema(methods=["PUT"], request=EconomicsBaselineSerializer, responses=EconomicsBaselineSerializer)
222 @extend_schema(methods=["PATCH"], request=EconomicsBaselineSerializer(partial=True), responses=EconomicsBaselineSerializer)
223 @action(detail=True, methods=["get", "put", "patch"])
224 def baseline(self, request, pk=None):
225 study = self.get_object()
226 if request.method == "GET":
227 baseline = baseline_from_settings_profile(study)
228 if baseline is not None: 228 ↛ 230line 228 didn't jump to line 230 because the condition on line 228 was always true
229 return Response(EconomicsBaselineSerializer(baseline, context=self.get_serializer_context()).data)
230 try:
231 baseline = study.baseline
232 except ObjectDoesNotExist as exc:
233 raise NotFound("Economics baseline has not been configured for this study.") from exc
234 return Response(EconomicsBaselineSerializer(baseline, context=self.get_serializer_context()).data)
236 _require_write_access(request.user)
237 try:
238 instance = study.baseline
239 except ObjectDoesNotExist:
240 instance = None
241 data = {**request.data, "study": study.pk}
242 serializer = EconomicsBaselineSerializer(instance, data=data, partial=request.method == "PATCH", context=self.get_serializer_context())
243 serializer.is_valid(raise_exception=True)
244 legacy_field_names = None if request.method != "PATCH" else set(serializer.validated_data)
245 baseline = serializer.save()
246 update_settings_profile_from_baseline(baseline, field_names=legacy_field_names)
247 _mark_study_stale(study, reason="economics_baseline_api_saved")
248 return Response(EconomicsBaselineSerializer(baseline, context=self.get_serializer_context()).data)
250 @extend_schema(request=EnableCostingRequestSerializer, responses=CostableItemSerializer)
251 @action(detail=True, methods=["post"], url_path="enable-costing")
252 def enable_costing(self, request, pk=None):
253 _require_write_access(request.user)
254 study = self.get_object()
255 serializer = EnableCostingRequestSerializer(data=request.data)
256 serializer.is_valid(raise_exception=True)
257 try:
258 simulation_object = SimulationObject.objects.get(pk=serializer.validated_data["simulation_object"])
259 result = enable_costing_for_simulation_object(study=study, simulation_object=simulation_object)
260 except (SimulationObject.DoesNotExist, ValueError) as exc:
261 raise _bad_request_from_error(exc) from exc
263 _sync_generated_capital_lines(study)
264 _mark_study_stale(study, reason="economics_enable_costing_api", requires_solve=True)
265 response = CostableItemSerializer(result.costable_item, context=self.get_serializer_context()).data
266 response["supported"] = result.supported
267 return Response(response, status=status.HTTP_200_OK)
269 @extend_schema(request=RecalculateRequestSerializer, responses=EconomicsResultRunSerializer)
270 @action(detail=True, methods=["post"])
271 def recalculate(self, request, pk=None):
272 _require_write_access(request.user)
273 study = self.get_object()
274 serializer = RecalculateRequestSerializer(data=request.data)
275 serializer.is_valid(raise_exception=True)
276 try:
277 result_run = recalculate_presentation_results(study, reason=serializer.validated_data["reason"])
278 except ValueError as exc:
279 raise _bad_request_from_error(exc) from exc
280 return Response(EconomicsResultRunSerializer(result_run, context=self.get_serializer_context()).data)
282 @extend_schema(responses=EconomicsResultRunSerializer)
283 @action(detail=True, methods=["get"], url_path="current-result")
284 def current_result(self, request, pk=None):
285 study = self.get_object()
286 run = study.result_runs.filter(status="current").order_by("-created_at", "-pk").first()
287 if run is None:
288 raise NotFound("No current economics result run exists for this study.")
289 return Response(EconomicsResultRunSerializer(run, context=self.get_serializer_context()).data)
291 @extend_schema(responses=EconomicsChartDatasetSerializer(many=True))
292 @action(detail=True, methods=["get"], url_path="chart-datasets")
293 def chart_datasets(self, request, pk=None):
294 study = self.get_object()
295 run = study.result_runs.filter(status="current").order_by("-created_at", "-pk").first()
296 if run is None: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 return Response([])
298 serializer = EconomicsChartDatasetSerializer(run.chart_datasets.order_by("chart_key"), many=True, context=self.get_serializer_context())
299 return Response(serializer.data)
301 @extend_schema(request=DuplicateStudyRequestSerializer, responses={status.HTTP_201_CREATED: EconomicsStudyFullSerializer})
302 @action(detail=True, methods=["post"])
303 def duplicate(self, request, pk=None):
304 _require_write_access(request.user)
305 source = self.get_object()
306 serializer = DuplicateStudyRequestSerializer(data=request.data)
307 serializer.is_valid(raise_exception=True)
308 study = duplicate_study(source, requested_name=serializer.validated_data.get("name", ""))
309 return Response(EconomicsStudyFullSerializer(study, context=self.get_serializer_context()).data, status=status.HTTP_201_CREATED)