Coverage for backend/django/Economics/costing/models.py: 86%
212 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from decimal import Decimal
3from django.core.exceptions import ValidationError
4from django.db import models
6from core.managers import AccessControlManager
7from Economics.shared.choices import (
8 CapitalLineBasis,
9 CapitalLineDepreciationMode,
10 CostBasis,
11 CostCurveEvaluationKind,
12 CostDriverSource,
13 CostableItemType,
14 DefaultRateType,
15 OperatingLineBasisQuantitySource,
16 OperatingLineCategory,
17 OperatingLineEconomicEffect,
18 OperatingLineRateSourceMode,
19 OutletStreamDisposition,
20)
21from Economics.shared.model_base import FlowsheetScopedEconomicsModel
24class CostCurve(FlowsheetScopedEconomicsModel):
25 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_cost_curves")
26 curve_key = models.CharField(max_length=128)
27 name = models.CharField(max_length=160)
28 equipment_category = models.CharField(max_length=64)
29 equipment_subtype = models.CharField(max_length=128, blank=True)
30 cost_basis = models.CharField(max_length=32, choices=CostBasis.choices, default=CostBasis.PURCHASE)
31 evaluation_kind = models.CharField(
32 max_length=32,
33 choices=CostCurveEvaluationKind.choices,
34 default=CostCurveEvaluationKind.EXPRESSION,
35 )
36 output_unit = models.CharField(max_length=32, default="NZD")
37 expression_text = models.TextField(blank=True)
38 required_driver_specs = models.JSONField(default=list, blank=True)
39 discrete_variants = models.JSONField(default=list, blank=True)
40 valid_min = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
41 valid_max = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
42 valid_range_note = models.CharField(max_length=255, blank=True)
43 currency = models.CharField(max_length=3, default="NZD")
44 basis_date = models.DateField(null=True, blank=True)
45 basis_index_name = models.CharField(max_length=128, blank=True)
46 basis_index_value = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
47 source_document_title = models.CharField(max_length=255, blank=True)
48 source_page = models.CharField(max_length=64, blank=True)
49 source_figure = models.CharField(max_length=64, blank=True)
50 source_data_origin = models.CharField(max_length=128, blank=True)
51 source_range_precision = models.CharField(max_length=64, blank=True)
52 source_license_status = models.CharField(max_length=64, blank=True)
53 source_reference = models.CharField(max_length=255, blank=True)
54 source_note = models.TextField(blank=True)
55 notes = models.TextField(blank=True)
56 applicability_warning = models.TextField(blank=True)
57 active = models.BooleanField(default=True)
58 created_at = models.DateTimeField(auto_now_add=True)
59 updated_at = models.DateTimeField(auto_now=True)
61 objects = AccessControlManager()
63 class Meta:
64 ordering = ["equipment_category", "equipment_subtype", "curve_key"]
65 constraints = [
66 models.UniqueConstraint(
67 fields=["flowsheet", "curve_key"],
68 name="unique_cost_curve_key_per_flowsheet",
69 ),
70 ]
72 def __str__(self):
73 return self.name
75class CostableItem(FlowsheetScopedEconomicsModel):
76 same_flowsheet_fields = ("study", "simulation_object")
78 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_costable_items")
79 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="costable_items")
80 item_type = models.CharField(
81 max_length=32,
82 choices=CostableItemType.choices,
83 default=CostableItemType.SIMULATION_OBJECT,
84 )
85 simulation_object = models.ForeignKey(
86 "flowsheetInternals_unitops.SimulationObject",
87 on_delete=models.SET_NULL,
88 related_name="economics_costable_items",
89 null=True,
90 blank=True,
91 )
92 name = models.CharField(max_length=128)
93 included = models.BooleanField(default=True)
94 manual = models.BooleanField(default=False)
95 notes = models.TextField(blank=True)
96 created_at = models.DateTimeField(auto_now_add=True)
97 updated_at = models.DateTimeField(auto_now=True)
99 objects = AccessControlManager()
101 class Meta:
102 ordering = ["created_at"]
103 constraints = [
104 models.UniqueConstraint(
105 fields=["study", "simulation_object"],
106 name="unique_costable_simulation_object_per_study",
107 ),
108 ]
110 def __str__(self):
111 return self.name
113 def clean(self):
114 super().clean()
115 if self.simulation_object_id and self.simulation_object.objectType == "group": 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 raise ValidationError({"simulation_object": "Flowsheet groups cannot be costed as v1 costable items."})
118class EquipmentMapping(FlowsheetScopedEconomicsModel):
119 same_flowsheet_fields = ("costable_item", "cost_curve")
121 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_equipment_mappings")
122 costable_item = models.OneToOneField("CostableItem", on_delete=models.CASCADE, related_name="equipment_mapping")
123 cost_curve = models.ForeignKey(
124 "CostCurve",
125 on_delete=models.SET_NULL,
126 related_name="equipment_mappings",
127 null=True,
128 blank=True,
129 )
130 equipment_category = models.CharField(max_length=64)
131 equipment_subtype = models.CharField(max_length=128, blank=True)
132 cost_basis = models.CharField(max_length=32, choices=CostBasis.choices, default=CostBasis.PURCHASE)
133 install_factor_profile = models.CharField(max_length=64, blank=True)
134 install_factor = models.DecimalField(max_digits=10, decimal_places=6, null=True, blank=True)
135 use_study_lang_factor = models.BooleanField(default=True)
136 applicability_notes = models.TextField(blank=True)
137 created_at = models.DateTimeField(auto_now_add=True)
138 updated_at = models.DateTimeField(auto_now=True)
140 objects = AccessControlManager()
142 def __str__(self):
143 if self.equipment_subtype: 143 ↛ 145line 143 didn't jump to line 145 because the condition on line 143 was always true
144 return f"{self.costable_item.name}: {self.equipment_category} / {self.equipment_subtype}"
145 return f"{self.costable_item.name}: {self.equipment_category}"
147class CostDriver(FlowsheetScopedEconomicsModel):
148 same_flowsheet_fields = ("costable_item", "property_info", "manual_property_info")
150 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_cost_drivers")
151 costable_item = models.OneToOneField("CostableItem", on_delete=models.CASCADE, related_name="cost_driver")
152 source = models.CharField(max_length=32, choices=CostDriverSource.choices, default=CostDriverSource.UNRESOLVED)
153 property_info = models.ForeignKey(
154 "core_auxiliary.PropertyInfo",
155 on_delete=models.SET_NULL,
156 related_name="economics_cost_drivers",
157 null=True,
158 blank=True,
159 help_text="Selected solved or configured property used as the cost driver.",
160 )
161 manual_property_info = models.ForeignKey(
162 "core_auxiliary.PropertyInfo",
163 on_delete=models.SET_NULL,
164 related_name="manual_economics_cost_drivers",
165 null=True,
166 blank=True,
167 )
168 sizing_mode = models.CharField(max_length=64, blank=True)
169 canonical_unit = models.CharField(max_length=32, blank=True)
170 design_value = models.DecimalField(max_digits=20, decimal_places=8, null=True, blank=True)
171 unresolved_reason_code = models.CharField(max_length=64, blank=True)
172 warning_payload = models.JSONField(default=dict, blank=True)
173 created_at = models.DateTimeField(auto_now_add=True)
174 updated_at = models.DateTimeField(auto_now=True)
176 objects = AccessControlManager()
178 def __str__(self):
179 return f"{self.costable_item.name} driver ({self.source})"
181class CapitalCostLine(FlowsheetScopedEconomicsModel):
182 same_flowsheet_fields = ("study", "costable_item", "cost_curve")
184 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_capital_lines")
185 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="capital_lines")
186 costable_item = models.ForeignKey("CostableItem", on_delete=models.SET_NULL, related_name="capital_lines", null=True, blank=True)
187 cost_curve = models.ForeignKey("CostCurve", on_delete=models.SET_NULL, related_name="capital_lines", null=True, blank=True)
188 label = models.CharField(max_length=160)
189 line_type = models.CharField(max_length=64)
190 calculation_basis = models.CharField(max_length=32, choices=CapitalLineBasis.choices, default=CapitalLineBasis.FIXED)
191 amount = models.DecimalField(max_digits=18, decimal_places=4, null=True, blank=True)
192 basis_percent = models.DecimalField(max_digits=9, decimal_places=4, null=True, blank=True)
193 depreciation_mode = models.CharField(
194 max_length=32,
195 choices=CapitalLineDepreciationMode.choices,
196 default=CapitalLineDepreciationMode.STUDY_DEFAULT,
197 )
198 depreciation_life_years = models.PositiveIntegerField(null=True, blank=True)
199 depreciation_salvage_percent = models.DecimalField(max_digits=7, decimal_places=4, null=True, blank=True)
200 peak_demand_kw = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
201 minimum_peak_demand_kw = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
202 currency = models.CharField(max_length=3, default="NZD")
203 included = models.BooleanField(default=True)
204 manual = models.BooleanField(default=False)
205 source = models.CharField(max_length=128, blank=True)
206 confidence = models.CharField(max_length=64, blank=True)
207 warning_payload = models.JSONField(default=dict, blank=True)
208 driver_inputs = models.JSONField(default=dict, blank=True)
209 created_at = models.DateTimeField(auto_now_add=True)
210 updated_at = models.DateTimeField(auto_now=True)
212 objects = AccessControlManager()
214 class Meta:
215 ordering = ["created_at"]
217 def clean(self):
218 super().clean()
219 errors = {}
220 if self.costable_item_id and self.costable_item.study_id != self.study_id: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 errors["costable_item"] = "Costable item must belong to the capital line study."
222 if self.calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT:
223 if self.basis_percent is None: 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true
224 errors["basis_percent"] = "Percentage capital lines require a percentage."
225 elif self.basis_percent < 0: 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 errors["basis_percent"] = "Percentage capital lines cannot be negative."
227 elif self.basis_percent is not None: 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true
228 errors["basis_percent"] = "Fixed capital lines do not use a percentage basis."
229 if self.depreciation_mode == CapitalLineDepreciationMode.CUSTOM:
230 if self.depreciation_life_years in (None, 0):
231 errors["depreciation_life_years"] = "Custom depreciation requires an equipment life."
232 elif self.depreciation_life_years is not None:
233 errors["depreciation_life_years"] = "Only custom depreciation uses a line equipment life."
234 if self.depreciation_mode != CapitalLineDepreciationMode.CUSTOM and self.depreciation_salvage_percent is not None:
235 errors["depreciation_salvage_percent"] = "Only custom depreciation uses a line residual value."
236 if (
237 self.depreciation_salvage_percent is not None
238 and not Decimal("0") <= self.depreciation_salvage_percent <= Decimal("100")
239 ):
240 errors["depreciation_salvage_percent"] = "Residual value must be between 0 and 100 percent."
241 if self.manual and self.calculation_basis == CapitalLineBasis.FIXED and self.amount is not None and self.amount < 0: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 errors["amount"] = "Fixed capital lines cannot be negative."
243 if self.peak_demand_kw is not None and self.peak_demand_kw < 0: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 errors["peak_demand_kw"] = "Peak demand cannot be negative."
245 if self.minimum_peak_demand_kw is not None and self.minimum_peak_demand_kw < 0: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 errors["minimum_peak_demand_kw"] = "Minimum peak demand cannot be negative."
247 if ( 247 ↛ 252line 247 didn't jump to line 252 because the condition on line 247 was never true
248 self.peak_demand_kw is not None
249 and self.minimum_peak_demand_kw is not None
250 and self.peak_demand_kw < self.minimum_peak_demand_kw
251 ):
252 errors["peak_demand_kw"] = "Peak demand cannot be below the current flowsheet work."
253 if errors:
254 raise ValidationError(errors)
256 def __str__(self):
257 return self.label
259class OperatingCostLine(FlowsheetScopedEconomicsModel):
260 same_flowsheet_fields = ("study", "costable_item", "source_property_info")
262 flowsheet = models.ForeignKey("core_auxiliary.Flowsheet", on_delete=models.CASCADE, related_name="economics_operating_lines")
263 study = models.ForeignKey("EconomicsStudy", on_delete=models.CASCADE, related_name="operating_lines")
264 costable_item = models.ForeignKey("CostableItem", on_delete=models.SET_NULL, related_name="operating_lines", null=True, blank=True)
265 label = models.CharField(max_length=160)
266 line_type = models.CharField(max_length=64)
267 category = models.CharField(max_length=32, choices=OperatingLineCategory.choices, default=OperatingLineCategory.CUSTOM)
268 economic_effect = models.CharField(
269 max_length=16,
270 choices=OperatingLineEconomicEffect.choices,
271 default=OperatingLineEconomicEffect.COST,
272 )
273 currency = models.CharField(max_length=3, default="NZD")
274 basis_quantity = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
275 basis_unit = models.CharField(max_length=32, blank=True)
276 basis_quantity_source = models.CharField(
277 max_length=32,
278 choices=OperatingLineBasisQuantitySource.choices,
279 default=OperatingLineBasisQuantitySource.MANUAL_OVERRIDE,
280 )
281 rate_amount = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
282 rate_unit = models.CharField(max_length=32, blank=True)
283 rate_type = models.CharField(max_length=32, choices=DefaultRateType.choices, blank=True)
284 rate_source_mode = models.CharField(
285 max_length=32,
286 choices=OperatingLineRateSourceMode.choices,
287 default=OperatingLineRateSourceMode.CUSTOM,
288 )
289 calculation_method = models.CharField(max_length=64, blank=True)
290 source_property_info = models.ForeignKey(
291 "core_auxiliary.PropertyInfo",
292 on_delete=models.SET_NULL,
293 related_name="economics_operating_lines",
294 null=True,
295 blank=True,
296 )
297 source_default_rate = models.ForeignKey(
298 "EconomicsDefaultRate",
299 on_delete=models.SET_NULL,
300 related_name="operating_lines",
301 null=True,
302 blank=True,
303 )
304 outlet_stream_disposition = models.CharField(max_length=16, choices=OutletStreamDisposition.choices, blank=True)
305 included = models.BooleanField(default=True)
306 manual = models.BooleanField(default=False)
307 source = models.CharField(max_length=128, blank=True)
308 warning_payload = models.JSONField(default=dict, blank=True)
309 created_at = models.DateTimeField(auto_now_add=True)
310 updated_at = models.DateTimeField(auto_now=True)
312 objects = AccessControlManager()
314 class Meta:
315 ordering = ["created_at"]
317 def clean(self):
318 super().clean()
319 errors = {}
320 if self.costable_item_id and self.study_id and self.costable_item.study_id != self.study_id: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 errors["costable_item"] = "Operating line costable item must belong to the same economics study."
322 if self.category == OperatingLineCategory.OUTPUT_REVENUE:
323 self.economic_effect = OperatingLineEconomicEffect.REVENUE
324 if self.category == OperatingLineCategory.OUTPUT_REVENUE and self.outlet_stream_disposition not in ( 324 ↛ 328line 324 didn't jump to line 328 because the condition on line 324 was never true
325 "",
326 OutletStreamDisposition.SOLD,
327 ):
328 errors["outlet_stream_disposition"] = "Sold output lines must classify the outlet stream as sold."
329 if self.category == OperatingLineCategory.DISPOSAL and self.outlet_stream_disposition not in ( 329 ↛ 333line 329 didn't jump to line 333 because the condition on line 329 was never true
330 "",
331 OutletStreamDisposition.DISPOSED,
332 ):
333 errors["outlet_stream_disposition"] = "Disposal lines must classify the outlet stream as disposed."
334 if ( 334 ↛ 339line 334 didn't jump to line 339 because the condition on line 334 was never true
335 isinstance(self.warning_payload, dict)
336 and self.warning_payload.get("source") == "outlet_stream_suggestion"
337 and not self.outlet_stream_disposition
338 ):
339 errors["outlet_stream_disposition"] = (
340 "Outlet stream suggestions must be classified as sold, disposed, or ignored before affecting economics."
341 )
342 if self.outlet_stream_disposition == OutletStreamDisposition.IGNORED and self.included: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true
343 errors["included"] = "Ignored outlet stream suggestions cannot be included in economics totals."
344 if errors: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 raise ValidationError(errors)
347 def __str__(self):
348 return self.label