Coverage for backend/django/Economics/costing/costable_items/serializers.py: 85%

1from decimal import Decimal 

2 

3from drf_spectacular.utils import extend_schema_field 

4from rest_framework import serializers 

5 

6from Economics.costing.models import CostCurve, CostDriver, CostableItem, EquipmentMapping 

7from Economics.shared.choices import BulkEquipmentDriverInputMode, CostDriverSource 

8from Economics.shared.serializer_base import FlowsheetScopedSerializer 

9from Economics.costing.cost_curves.driver_properties import ( 

10 apply_recommended_property_for_mapping, 

11 manual_property_for_costable_item, 

12 normalize_property_cost_driver, 

13 validate_cost_driver_property, 

14) 

15from Economics.costing.capital.lang_factors import resolve_lang_factor 

16from Economics.reference_data.models import EconomicsLangFactorDefault 

17 

18 

19class EquipmentMappingSerializer(FlowsheetScopedSerializer): 

20 same_flowsheet_fields = ("costable_item",) 

21 lang_factor_label = serializers.SerializerMethodField() 

22 effective_lang_factor = serializers.SerializerMethodField() 

23 lang_factor_default_value = serializers.SerializerMethodField() 

24 lang_factor_source = serializers.SerializerMethodField() 

25 lang_factor_is_custom = serializers.SerializerMethodField() 

26 lang_factor_default_key = serializers.SerializerMethodField() 

27 

28 class Meta: 

29 model = EquipmentMapping 

30 fields = ( 

31 "id", 

32 "flowsheet", 

33 "costable_item", 

34 "cost_curve", 

35 "equipment_category", 

36 "equipment_subtype", 

37 "cost_basis", 

38 "install_factor_profile", 

39 "install_factor", 

40 "use_study_lang_factor", 

41 "lang_factor_label", 

42 "effective_lang_factor", 

43 "lang_factor_default_value", 

44 "lang_factor_source", 

45 "lang_factor_is_custom", 

46 "lang_factor_default_key", 

47 "applicability_notes", 

48 "created_at", 

49 "updated_at", 

50 ) 

51 read_only_fields = ( 

52 "id", 

53 "flowsheet", 

54 "lang_factor_label", 

55 "effective_lang_factor", 

56 "lang_factor_default_value", 

57 "lang_factor_source", 

58 "lang_factor_is_custom", 

59 "lang_factor_default_key", 

60 "created_at", 

61 "updated_at", 

62 ) 

63 

64 @extend_schema_field(serializers.CharField) 

65 def get_lang_factor_label(self, instance) -> str: 

66 return self._lang_factor(instance).label 

67 

68 @extend_schema_field(serializers.DecimalField(max_digits=10, decimal_places=6, allow_null=True)) 

69 def get_effective_lang_factor(self, instance) -> Decimal | None: 

70 return self._lang_factor(instance).effective_value 

71 

72 @extend_schema_field(serializers.DecimalField(max_digits=10, decimal_places=6, allow_null=True)) 

73 def get_lang_factor_default_value(self, instance) -> Decimal | None: 

74 return self._lang_factor(instance).active_default_value 

75 

76 @extend_schema_field(serializers.CharField) 

77 def get_lang_factor_source(self, instance) -> str: 

78 return self._lang_factor(instance).source 

79 

80 @extend_schema_field(serializers.BooleanField) 

81 def get_lang_factor_is_custom(self, instance) -> bool: 

82 return self._lang_factor(instance).is_custom 

83 

84 @extend_schema_field(serializers.CharField) 

85 def get_lang_factor_default_key(self, instance) -> str: 

86 return self._lang_factor(instance).default_key 

87 

88 def _lang_factor(self, instance): 

89 """Resolve each mapping once and share one reference-default lookup per response.""" 

90 resolutions = self.context.setdefault("_economics_lang_factor_resolutions", {}) 

91 if instance.pk not in resolutions: 

92 defaults = self.context.get("_economics_lang_factor_defaults") 

93 if defaults is None: 93 ↛ 98line 93 didn't jump to line 98 because the condition on line 93 was always true

94 defaults = {} 

95 for row in EconomicsLangFactorDefault.objects.order_by("pk"): 

96 defaults.setdefault((row.scope, row.unit_operation_type or ""), row) 

97 self.context["_economics_lang_factor_defaults"] = defaults 

98 resolutions[instance.pk] = resolve_lang_factor(instance, defaults=defaults) 

99 return resolutions[instance.pk] 

100 

101 def validate(self, attrs): 

102 attrs = super().validate(attrs) 

103 costable_item = attrs.get("costable_item") or getattr(self.instance, "costable_item", None) 

104 cost_curve = attrs.get("cost_curve") if "cost_curve" in attrs else getattr(self.instance, "cost_curve", None) 

105 errors = {} 

106 if cost_curve is not None: 

107 self._validate_cost_curve_assignment(errors=errors, costable_item=costable_item, cost_curve=cost_curve, attrs=attrs) 

108 if errors: 

109 raise serializers.ValidationError(errors) 

110 return attrs 

111 

112 def _validate_cost_curve_assignment(self, *, errors: dict, costable_item, cost_curve: CostCurve, attrs: dict) -> None: 

113 if costable_item is None: 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 errors["costable_item"] = "Cost curve assignment requires a costable item." 

115 return 

116 driver = getattr(costable_item, "cost_driver", None) 

117 if driver is None: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 errors["cost_driver"] = "Cost curve assignment requires a selected cost driver." 

119 return 

120 if not cost_curve.cost_basis: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 errors["cost_basis"] = "Cost curve assignment requires a purchase or installed cost basis." 

122 if cost_curve.equipment_category and cost_curve.equipment_category != self._equipment_category(attrs): 

123 errors["equipment_category"] = "Mapping equipment category must match the selected cost curve." 

124 if cost_curve.equipment_subtype and cost_curve.equipment_subtype != self._equipment_subtype(attrs): 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 errors["equipment_subtype"] = "Mapping equipment subtype must match the selected cost curve." 

126 def _equipment_category(self, attrs: dict) -> str: 

127 cost_curve = attrs.get("cost_curve") 

128 if "equipment_category" in attrs and attrs["equipment_category"]: 

129 return attrs["equipment_category"] 

130 if cost_curve is not None and cost_curve.equipment_category: 

131 return cost_curve.equipment_category 

132 if "equipment_category" in attrs: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true

133 return attrs["equipment_category"] 

134 return getattr(self.instance, "equipment_category", "") 

135 

136 def _equipment_subtype(self, attrs: dict) -> str: 

137 cost_curve = attrs.get("cost_curve") 

138 if "equipment_subtype" in attrs and attrs["equipment_subtype"]: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 return attrs["equipment_subtype"] 

140 if cost_curve is not None and cost_curve.equipment_subtype: 140 ↛ 142line 140 didn't jump to line 142 because the condition on line 140 was always true

141 return cost_curve.equipment_subtype 

142 if "equipment_subtype" in attrs: 

143 return attrs["equipment_subtype"] 

144 return getattr(self.instance, "equipment_subtype", "") 

145 

146 def update(self, instance, validated_data): 

147 cost_curve = validated_data.get("cost_curve") 

148 if cost_curve is not None: 

149 if not validated_data.get("equipment_category") and cost_curve.equipment_category: 

150 validated_data["equipment_category"] = cost_curve.equipment_category 

151 if not validated_data.get("equipment_subtype") and cost_curve.equipment_subtype: 

152 validated_data["equipment_subtype"] = cost_curve.equipment_subtype 

153 category_changed = ( 

154 "equipment_category" in validated_data 

155 and validated_data["equipment_category"] != instance.equipment_category 

156 ) 

157 instance = super().update(instance, validated_data) 

158 if category_changed: 

159 apply_recommended_property_for_mapping(instance) 

160 return instance 

161 

162 

163class CostDriverSerializer(FlowsheetScopedSerializer): 

164 same_flowsheet_fields = ("costable_item", "property_info", "manual_property_info") 

165 

166 class Meta: 

167 model = CostDriver 

168 fields = ( 

169 "id", 

170 "flowsheet", 

171 "costable_item", 

172 "source", 

173 "property_info", 

174 "manual_property_info", 

175 "sizing_mode", 

176 "canonical_unit", 

177 "design_value", 

178 "unresolved_reason_code", 

179 "warning_payload", 

180 "created_at", 

181 "updated_at", 

182 ) 

183 read_only_fields = ("id", "flowsheet", "created_at", "updated_at") 

184 

185 def to_representation(self, instance): 

186 data = super().to_representation(instance) 

187 if data.get("sizing_mode") in {"scale", "sum", "max"}: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 data["sizing_mode"] = "" 

189 return data 

190 

191 def validate(self, attrs): 

192 attrs = super().validate(attrs) 

193 property_info = attrs.get("property_info") if "property_info" in attrs else None 

194 if property_info is not None: 

195 driver = self.instance 

196 if driver is None: 

197 costable_item = attrs.get("costable_item") 

198 if costable_item is None: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true

199 raise serializers.ValidationError( 

200 {"costable_item": "Cost driver property selection requires a costable item."} 

201 ) 

202 driver = CostDriver(costable_item=costable_item) 

203 driver.manual_property_info = attrs.get( 

204 "manual_property_info", 

205 manual_property_for_costable_item(costable_item), 

206 ) 

207 try: 

208 validate_cost_driver_property(driver, property_info) 

209 except ValueError as exc: 

210 raise serializers.ValidationError({"property_info": str(exc)}) from exc 

211 return attrs 

212 

213 def create(self, validated_data): 

214 property_info = validated_data.get("property_info") if "property_info" in validated_data else None 

215 costable_item = validated_data.get("costable_item") 

216 if costable_item is not None and "manual_property_info" not in validated_data: 216 ↛ 218line 216 didn't jump to line 218 because the condition on line 216 was always true

217 validated_data["manual_property_info"] = manual_property_for_costable_item(costable_item) 

218 driver = CostDriver( 

219 costable_item=costable_item, 

220 manual_property_info=validated_data.get("manual_property_info"), 

221 design_value=validated_data.get("design_value"), 

222 warning_payload=validated_data.get("warning_payload") or {}, 

223 ) 

224 if property_info is not None: 

225 validated_data.update(normalize_property_cost_driver(driver, property_info)) 

226 elif "property_info" in validated_data: 226 ↛ 228line 226 didn't jump to line 228 because the condition on line 226 was always true

227 validated_data.update(normalize_property_cost_driver(driver, None)) 

228 elif validated_data.get("source") == CostDriverSource.MANUAL_OVERRIDE: 

229 validated_data["property_info"] = None 

230 validated_data.setdefault("sizing_mode", "manual") 

231 return super().create(validated_data) 

232 

233 def update(self, instance, validated_data): 

234 if "property_info" in validated_data: 234 ↛ 236line 234 didn't jump to line 236 because the condition on line 234 was always true

235 validated_data.update(normalize_property_cost_driver(instance, validated_data["property_info"])) 

236 elif validated_data.get("source") == CostDriverSource.MANUAL_OVERRIDE: 

237 validated_data["property_info"] = None 

238 validated_data.setdefault("sizing_mode", "manual") 

239 return super().update(instance, validated_data) 

240 

241 

242class CostDriverPropertyOptionSerializer(serializers.Serializer): 

243 property_info = serializers.IntegerField() 

244 scope = serializers.ChoiceField(choices=("unit", "input_stream", "output_stream")) 

245 object_id = serializers.IntegerField() 

246 object_name = serializers.CharField() 

247 object_type = serializers.CharField() 

248 property_key = serializers.CharField() 

249 display_name = serializers.CharField() 

250 unit = serializers.CharField(allow_blank=True) 

251 unit_type = serializers.CharField(allow_blank=True) 

252 value_preview = serializers.CharField(allow_blank=True) 

253 has_value = serializers.BooleanField() 

254 recommended = serializers.BooleanField() 

255 recommendation_label = serializers.CharField(allow_blank=True) 

256 schedule_varying = serializers.BooleanField() 

257 

258 

259class BulkEquipmentDriverInputSetupSerializer(serializers.Serializer): 

260 mode = serializers.ChoiceField(choices=BulkEquipmentDriverInputMode.choices) 

261 manual_value = serializers.CharField(required=False, allow_blank=True, default="") 

262 

263 def validate_manual_value(self, value): 

264 return str(value or "").strip() 

265 

266 

267class BulkEquipmentDriverInputsField(serializers.DictField): 

268 child = BulkEquipmentDriverInputSetupSerializer() 

269 

270 def to_internal_value(self, data): 

271 if data in (None, ""): 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true

272 return {} 

273 if not isinstance(data, dict): 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true

274 raise serializers.ValidationError("Driver input setup must be an object keyed by cost curve input.") 

275 normalized = super().to_internal_value(data) 

276 if any(not str(key or "").strip() for key in normalized): 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true

277 raise serializers.ValidationError("Driver input keys cannot be blank.") 

278 return {str(key).strip(): value for key, value in normalized.items()} 

279 

280 

281class BulkEquipmentSetupRequestSerializer(serializers.Serializer): 

282 mapping_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False) 

283 cost_curve = serializers.IntegerField(required=False, allow_null=True) 

284 dry_run = serializers.BooleanField(default=True) 

285 apply_cost_curve = serializers.BooleanField(default=True) 

286 apply_recommended_sizing_property = serializers.BooleanField(default=True) 

287 overwrite_sizing_property = serializers.BooleanField(default=True) 

288 driver_inputs = BulkEquipmentDriverInputsField(required=False, default=dict) 

289 

290 

291class BulkEquipmentDriverInputResultSerializer(serializers.Serializer): 

292 key = serializers.CharField() 

293 label = serializers.CharField() 

294 unit = serializers.CharField(allow_blank=True) 

295 mode = serializers.ChoiceField(choices=BulkEquipmentDriverInputMode.choices) 

296 status = serializers.CharField() 

297 property = serializers.IntegerField(allow_null=True) 

298 property_label = serializers.CharField(allow_blank=True) 

299 property_unit = serializers.CharField(allow_blank=True) 

300 property_scope = serializers.CharField(allow_blank=True) 

301 property_object_name = serializers.CharField(allow_blank=True) 

302 manual_value = serializers.CharField(allow_blank=True) 

303 overwrite = serializers.BooleanField() 

304 message = serializers.CharField(allow_blank=True) 

305 

306 

307class BulkEquipmentSetupResultSerializer(serializers.Serializer): 

308 mapping = serializers.IntegerField() 

309 costable_item = serializers.IntegerField() 

310 unit_name = serializers.CharField() 

311 cost_curve = serializers.IntegerField(allow_null=True) 

312 cost_curve_name = serializers.CharField(allow_blank=True) 

313 curve_status = serializers.CharField() 

314 curve_overwrite = serializers.BooleanField() 

315 cost_driver = serializers.IntegerField(allow_null=True) 

316 sizing_status = serializers.CharField() 

317 sizing_property = serializers.IntegerField(allow_null=True) 

318 sizing_property_label = serializers.CharField(allow_blank=True) 

319 sizing_property_unit = serializers.CharField(allow_blank=True) 

320 sizing_property_scope = serializers.CharField(allow_blank=True) 

321 sizing_property_object_name = serializers.CharField(allow_blank=True) 

322 sizing_overwrite = serializers.BooleanField() 

323 driver_input_results = serializers.DictField(child=BulkEquipmentDriverInputResultSerializer()) 

324 message = serializers.CharField(allow_blank=True) 

325 

326 

327class CostableItemSerializer(FlowsheetScopedSerializer): 

328 same_flowsheet_fields = ("study", "simulation_object") 

329 cost_driver = CostDriverSerializer(read_only=True) 

330 equipment_mapping = EquipmentMappingSerializer(read_only=True) 

331 simulation_object_name = serializers.CharField(source="simulation_object.componentName", read_only=True) 

332 simulation_object_type = serializers.CharField(source="simulation_object.objectType", read_only=True) 

333 

334 class Meta: 

335 model = CostableItem 

336 fields = ( 

337 "id", 

338 "flowsheet", 

339 "study", 

340 "item_type", 

341 "simulation_object", 

342 "simulation_object_name", 

343 "simulation_object_type", 

344 "name", 

345 "included", 

346 "manual", 

347 "notes", 

348 "cost_driver", 

349 "equipment_mapping", 

350 "created_at", 

351 "updated_at", 

352 ) 

353 read_only_fields = ("id", "flowsheet", "created_at", "updated_at") 

354 

355 def validate(self, attrs): 

356 attrs = super().validate(attrs) 

357 simulation_object = attrs.get("simulation_object", getattr(self.instance, "simulation_object", None)) 

358 if simulation_object is not None and simulation_object.objectType == "group": 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true

359 raise serializers.ValidationError( 

360 {"simulation_object": "Flowsheet groups cannot be created or shown as v1 costable items."} 

361 ) 

362 return attrs