Coverage for backend/django/Economics/costing/models.py: 87%
229 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from decimal import Decimal
3from django.core.exceptions import ValidationError
4from django.db import models
6from core.managers import (
7 AccessControlManager,
8 AllFlowsheetStatesManager,
9 ProjectAccessControlManager,
10 UserAccessControlManager,
11)
12from Economics.shared.choices import (
13 CapitalLineBasis,
14 CapitalLineDepreciationMode,
15 CostBasis,
16 CostCurveEvaluationKind,
17 CostDriverSource,
18 CostableItemType,
19 DefaultRateType,
20 OperatingLineBasisQuantitySource,
21 OperatingLineCategory,
22 OperatingLineEconomicEffect,
23 OperatingLineRateSourceMode,
24 OutletStreamDisposition,
25)
26from Economics.shared.model_base import FlowsheetScopedEconomicsModel
29COST_CURVE_DEFINITION_FIELDS = (
30 "curve_key",
31 "name",
32 "equipment_category",
33 "equipment_subtype",
34 "cost_basis",
35 "evaluation_kind",
36 "output_unit",
37 "expression_text",
38 "required_driver_specs",
39 "discrete_variants",
40 "valid_min",
41 "valid_max",
42 "valid_range_note",
43 "currency",
44 "basis_date",
45 "basis_index_name",
46 "basis_index_value",
47 "source_document_title",
48 "source_page",
49 "source_figure",
50 "source_data_origin",
51 "source_range_precision",
52 "source_license_status",
53 "source_reference",
54 "source_note",
55 "notes",
56 "applicability_warning",
57 "active",
58)
61class CostCurveDefinition(models.Model):
62 """Shared persisted definition fields for project curves and user templates."""
64 curve_key = models.CharField(max_length=128)
65 name = models.CharField(max_length=160)
66 equipment_category = models.CharField(max_length=64)
67 equipment_subtype = models.CharField(max_length=128, blank=True)
68 cost_basis = models.CharField(max_length=32, choices=CostBasis.choices, default=CostBasis.PURCHASE)
69 evaluation_kind = models.CharField(
70 max_length=32,
71 choices=CostCurveEvaluationKind.choices,
72 default=CostCurveEvaluationKind.EXPRESSION,
73 )
74 output_unit = models.CharField(max_length=32, default="NZD")
75 expression_text = models.TextField(blank=True)
76 required_driver_specs = models.JSONField(default=list, blank=True)
77 discrete_variants = models.JSONField(default=list, blank=True)
78 valid_min = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
79 valid_max = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
80 valid_range_note = models.CharField(max_length=255, blank=True)
81 currency = models.CharField(max_length=3, default="NZD")
82 basis_date = models.DateField(null=True, blank=True)
83 basis_index_name = models.CharField(max_length=128, blank=True)
84 basis_index_value = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
85 source_document_title = models.CharField(max_length=255, blank=True)
86 source_page = models.CharField(max_length=64, blank=True)
87 source_figure = models.CharField(max_length=64, blank=True)
88 source_data_origin = models.CharField(max_length=128, blank=True)
89 source_range_precision = models.CharField(max_length=64, blank=True)
90 source_license_status = models.CharField(max_length=64, blank=True)
91 source_reference = models.CharField(max_length=255, blank=True)
92 source_note = models.TextField(blank=True)
93 notes = models.TextField(blank=True)
94 applicability_warning = models.TextField(blank=True)
95 active = models.BooleanField(default=True)
96 created_at = models.DateTimeField(auto_now_add=True)
97 updated_at = models.DateTimeField(auto_now=True)
99 class Meta:
100 abstract = True
101 ordering = ["equipment_category", "equipment_subtype", "curve_key"]
103 def __str__(self):
104 return self.name
106 def save(self, *args, **kwargs):
107 self.full_clean()
108 return super().save(*args, **kwargs)
111class CostCurve(CostCurveDefinition):
112 """Editable project-owned cost curve used by capital configuration."""
114 project = models.ForeignKey(
115 "core_auxiliary.Project",
116 on_delete=models.CASCADE,
117 related_name="economics_cost_curves",
118 )
120 objects = ProjectAccessControlManager()
122 class Meta(CostCurveDefinition.Meta):
123 constraints = [
124 models.UniqueConstraint(
125 fields=["project", "curve_key"],
126 name="unique_cost_curve_key_per_project",
127 ),
128 ]
131class CostCurveTemplate(CostCurveDefinition):
132 """Reusable cost-curve template visible only to its owning user."""
134 owner = models.ForeignKey(
135 "authentication_user.User",
136 on_delete=models.CASCADE,
137 related_name="economics_cost_curve_templates",
138 )
140 objects = UserAccessControlManager()
142 class Meta(CostCurveDefinition.Meta):
143 constraints = [
144 models.UniqueConstraint(
145 fields=["owner", "curve_key"],
146 name="unique_cost_curve_template_key_per_user",
147 ),
148 ]
151class CostableItem(FlowsheetScopedEconomicsModel):
152 same_flowsheet_fields = ("study", "simulation_object")
154 flowsheet_state = models.ForeignKey("core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="economics_costable_items")
155 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="costable_items")
156 item_type = models.CharField(
157 max_length=32,
158 choices=CostableItemType.choices,
159 default=CostableItemType.SIMULATION_OBJECT,
160 )
161 simulation_object = models.ForeignKey(
162 "flowsheetInternals_unitops.SimulationObject",
163 on_delete=models.SET_NULL,
164 related_name="economics_costable_items",
165 null=True,
166 blank=True,
167 )
168 name = models.CharField(max_length=128)
169 included = models.BooleanField(default=True)
170 manual = models.BooleanField(default=False)
171 notes = models.TextField(blank=True)
172 created_at = models.DateTimeField(auto_now_add=True)
173 updated_at = models.DateTimeField(auto_now=True)
175 objects = AccessControlManager()
176 all_states = AllFlowsheetStatesManager()
178 class Meta:
179 ordering = ["created_at"]
180 constraints = [
181 models.UniqueConstraint(
182 fields=["study", "simulation_object"],
183 name="unique_costable_simulation_object_per_study",
184 ),
185 ]
187 def __str__(self):
188 return self.name
190 def clean(self):
191 super().clean()
192 if self.simulation_object_id and self.simulation_object.objectType == "group": 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 raise ValidationError({"simulation_object": "Flowsheet groups cannot be costed as v1 costable items."})
195class EquipmentMapping(FlowsheetScopedEconomicsModel):
196 same_flowsheet_fields = ("costable_item",)
198 flowsheet_state = models.ForeignKey("core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="economics_equipment_mappings")
199 costable_item = models.OneToOneField("CostableItem", on_delete=models.CASCADE, related_name="equipment_mapping")
200 cost_curve = models.ForeignKey(
201 "CostCurve",
202 on_delete=models.SET_NULL,
203 related_name="equipment_mappings",
204 null=True,
205 blank=True,
206 )
207 equipment_category = models.CharField(max_length=64)
208 equipment_subtype = models.CharField(max_length=128, blank=True)
209 cost_basis = models.CharField(max_length=32, choices=CostBasis.choices, default=CostBasis.PURCHASE)
210 install_factor_profile = models.CharField(max_length=64, blank=True)
211 install_factor = models.DecimalField(max_digits=10, decimal_places=6, null=True, blank=True)
212 use_study_lang_factor = models.BooleanField(default=True)
213 applicability_notes = models.TextField(blank=True)
214 created_at = models.DateTimeField(auto_now_add=True)
215 updated_at = models.DateTimeField(auto_now=True)
217 objects = AccessControlManager()
218 all_states = AllFlowsheetStatesManager()
220 def __str__(self):
221 if self.equipment_subtype: 221 ↛ 223line 221 didn't jump to line 223 because the condition on line 221 was always true
222 return f"{self.costable_item.name}: {self.equipment_category} / {self.equipment_subtype}"
223 return f"{self.costable_item.name}: {self.equipment_category}"
225class CostDriver(FlowsheetScopedEconomicsModel):
226 same_flowsheet_fields = ("costable_item", "property_info", "manual_property_info")
228 flowsheet_state = models.ForeignKey("core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="economics_cost_drivers")
229 costable_item = models.OneToOneField("CostableItem", on_delete=models.CASCADE, related_name="cost_driver")
230 source = models.CharField(max_length=32, choices=CostDriverSource.choices, default=CostDriverSource.UNRESOLVED)
231 property_info = models.ForeignKey(
232 "core_auxiliary.PropertyInfo",
233 on_delete=models.SET_NULL,
234 related_name="economics_cost_drivers",
235 null=True,
236 blank=True,
237 help_text="Selected solved or configured property used as the cost driver.",
238 )
239 manual_property_info = models.ForeignKey(
240 "core_auxiliary.PropertyInfo",
241 on_delete=models.SET_NULL,
242 related_name="manual_economics_cost_drivers",
243 null=True,
244 blank=True,
245 )
246 sizing_mode = models.CharField(max_length=64, blank=True)
247 canonical_unit = models.CharField(max_length=32, blank=True)
248 design_value = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
249 unresolved_reason_code = models.CharField(max_length=64, blank=True)
250 warning_payload = models.JSONField(default=dict, blank=True)
251 created_at = models.DateTimeField(auto_now_add=True)
252 updated_at = models.DateTimeField(auto_now=True)
254 objects = AccessControlManager()
255 all_states = AllFlowsheetStatesManager()
257 def __str__(self):
258 return f"{self.costable_item.name} driver ({self.source})"
260class CapitalCostLine(FlowsheetScopedEconomicsModel):
261 same_flowsheet_fields = ("study", "costable_item")
263 flowsheet_state = models.ForeignKey("core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="economics_capital_lines")
264 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="capital_lines")
265 costable_item = models.ForeignKey("CostableItem", on_delete=models.SET_NULL, related_name="capital_lines", null=True, blank=True)
266 cost_curve = models.ForeignKey("CostCurve", on_delete=models.SET_NULL, related_name="capital_lines", null=True, blank=True)
267 label = models.CharField(max_length=160)
268 line_type = models.CharField(max_length=64)
269 calculation_basis = models.CharField(max_length=32, choices=CapitalLineBasis.choices, default=CapitalLineBasis.FIXED)
270 amount = models.DecimalField(max_digits=18, decimal_places=4, null=True, blank=True)
271 basis_percent = models.DecimalField(max_digits=9, decimal_places=4, null=True, blank=True)
272 depreciation_mode = models.CharField(
273 max_length=32,
274 choices=CapitalLineDepreciationMode.choices,
275 default=CapitalLineDepreciationMode.STUDY_DEFAULT,
276 )
277 depreciation_life_years = models.PositiveIntegerField(null=True, blank=True)
278 depreciation_salvage_percent = models.DecimalField(max_digits=7, decimal_places=4, null=True, blank=True)
279 peak_demand_kw = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
280 minimum_peak_demand_kw = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
281 currency = models.CharField(max_length=3, default="NZD")
282 included = models.BooleanField(default=True)
283 manual = models.BooleanField(default=False)
284 source = models.CharField(max_length=128, blank=True)
285 confidence = models.CharField(max_length=64, blank=True)
286 warning_payload = models.JSONField(default=dict, blank=True)
287 driver_inputs = models.JSONField(default=dict, blank=True)
288 created_at = models.DateTimeField(auto_now_add=True)
289 updated_at = models.DateTimeField(auto_now=True)
291 objects = AccessControlManager()
292 all_states = AllFlowsheetStatesManager()
294 class Meta:
295 ordering = ["created_at"]
297 def clean(self):
298 super().clean()
299 errors = {}
300 if self.costable_item_id and self.costable_item.study_id != self.study_id: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 errors["costable_item"] = "Costable item must belong to the capital line study."
302 if self.calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT:
303 if self.basis_percent is None: 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 errors["basis_percent"] = "Percentage capital lines require a percentage."
305 elif self.basis_percent < 0: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true
306 errors["basis_percent"] = "Percentage capital lines cannot be negative."
307 elif self.basis_percent is not None: 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 errors["basis_percent"] = "Fixed capital lines do not use a percentage basis."
309 if self.depreciation_mode == CapitalLineDepreciationMode.CUSTOM:
310 if self.depreciation_life_years in (None, 0):
311 errors["depreciation_life_years"] = "Custom depreciation requires an equipment life."
312 elif self.depreciation_life_years is not None:
313 errors["depreciation_life_years"] = "Only custom depreciation uses a line equipment life."
314 if self.depreciation_mode != CapitalLineDepreciationMode.CUSTOM and self.depreciation_salvage_percent is not None:
315 errors["depreciation_salvage_percent"] = "Only custom depreciation uses a line residual value."
316 if (
317 self.depreciation_salvage_percent is not None
318 and not Decimal("0") <= self.depreciation_salvage_percent <= Decimal("100")
319 ):
320 errors["depreciation_salvage_percent"] = "Residual value must be between 0 and 100 percent."
321 if self.manual and self.calculation_basis == CapitalLineBasis.FIXED and self.amount is not None and self.amount < 0: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 errors["amount"] = "Fixed capital lines cannot be negative."
323 if self.peak_demand_kw is not None and self.peak_demand_kw < 0: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 errors["peak_demand_kw"] = "Peak demand cannot be negative."
325 if self.minimum_peak_demand_kw is not None and self.minimum_peak_demand_kw < 0: 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true
326 errors["minimum_peak_demand_kw"] = "Minimum peak demand cannot be negative."
327 if ( 327 ↛ 332line 327 didn't jump to line 332 because the condition on line 327 was never true
328 self.peak_demand_kw is not None
329 and self.minimum_peak_demand_kw is not None
330 and self.peak_demand_kw < self.minimum_peak_demand_kw
331 ):
332 errors["peak_demand_kw"] = "Peak demand cannot be below the current flowsheet work."
333 if errors:
334 raise ValidationError(errors)
336 def __str__(self):
337 return self.label
339class OperatingCostLine(FlowsheetScopedEconomicsModel):
340 same_flowsheet_fields = ("study", "costable_item", "source_property_info")
342 flowsheet_state = models.ForeignKey("core_auxiliary.FlowsheetState", on_delete=models.CASCADE, related_name="economics_operating_lines")
343 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="operating_lines")
344 costable_item = models.ForeignKey("CostableItem", on_delete=models.SET_NULL, related_name="operating_lines", null=True, blank=True)
345 label = models.CharField(max_length=160)
346 line_type = models.CharField(max_length=64)
347 category = models.CharField(max_length=32, choices=OperatingLineCategory.choices, default=OperatingLineCategory.CUSTOM)
348 economic_effect = models.CharField(
349 max_length=16,
350 choices=OperatingLineEconomicEffect.choices,
351 default=OperatingLineEconomicEffect.COST,
352 )
353 currency = models.CharField(max_length=3, default="NZD")
354 basis_quantity = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
355 basis_unit = models.CharField(max_length=32, blank=True)
356 basis_quantity_source = models.CharField(
357 max_length=32,
358 choices=OperatingLineBasisQuantitySource.choices,
359 default=OperatingLineBasisQuantitySource.MANUAL_OVERRIDE,
360 )
361 rate_amount = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
362 rate_unit = models.CharField(max_length=32, blank=True)
363 rate_type = models.CharField(max_length=32, choices=DefaultRateType.choices, blank=True)
364 rate_source_mode = models.CharField(
365 max_length=32,
366 choices=OperatingLineRateSourceMode.choices,
367 default=OperatingLineRateSourceMode.CUSTOM,
368 )
369 calculation_method = models.CharField(max_length=64, blank=True)
370 source_property_info = models.ForeignKey(
371 "core_auxiliary.PropertyInfo",
372 on_delete=models.SET_NULL,
373 related_name="economics_operating_lines",
374 null=True,
375 blank=True,
376 )
377 source_default_rate = models.ForeignKey(
378 "EconomicsDefaultRate",
379 on_delete=models.SET_NULL,
380 related_name="operating_lines",
381 null=True,
382 blank=True,
383 )
384 outlet_stream_disposition = models.CharField(max_length=16, choices=OutletStreamDisposition.choices, blank=True)
385 included = models.BooleanField(default=True)
386 manual = models.BooleanField(default=False)
387 source = models.CharField(max_length=128, blank=True)
388 warning_payload = models.JSONField(default=dict, blank=True)
389 created_at = models.DateTimeField(auto_now_add=True)
390 updated_at = models.DateTimeField(auto_now=True)
392 objects = AccessControlManager()
393 all_states = AllFlowsheetStatesManager()
395 class Meta:
396 ordering = ["created_at"]
398 def clean(self):
399 super().clean()
400 errors = {}
401 if self.costable_item_id and self.study_id and self.costable_item.study_id != self.study_id: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 errors["costable_item"] = "Operating line costable item must belong to the same economics study."
403 if self.category == OperatingLineCategory.OUTPUT_REVENUE:
404 self.economic_effect = OperatingLineEconomicEffect.REVENUE
405 if self.category == OperatingLineCategory.OUTPUT_REVENUE and self.outlet_stream_disposition not in ( 405 ↛ 409line 405 didn't jump to line 409 because the condition on line 405 was never true
406 "",
407 OutletStreamDisposition.SOLD,
408 ):
409 errors["outlet_stream_disposition"] = "Sold output lines must classify the outlet stream as sold."
410 if self.category == OperatingLineCategory.DISPOSAL and self.outlet_stream_disposition not in ( 410 ↛ 414line 410 didn't jump to line 414 because the condition on line 410 was never true
411 "",
412 OutletStreamDisposition.DISPOSED,
413 ):
414 errors["outlet_stream_disposition"] = "Disposal lines must classify the outlet stream as disposed."
415 if ( 415 ↛ 420line 415 didn't jump to line 420 because the condition on line 415 was never true
416 isinstance(self.warning_payload, dict)
417 and self.warning_payload.get("source") == "outlet_stream_suggestion"
418 and not self.outlet_stream_disposition
419 ):
420 errors["outlet_stream_disposition"] = (
421 "Outlet stream suggestions must be classified as sold, disposed, or ignored before affecting economics."
422 )
423 if self.outlet_stream_disposition == OutletStreamDisposition.IGNORED and self.included: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true
424 errors["included"] = "Ignored outlet stream suggestions cannot be included in economics totals."
425 if errors: 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true
426 raise ValidationError(errors)
428 def __str__(self):
429 return self.label