Coverage for backend/django/Economics/formulas/builders/operating.py: 78%
260 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3from dataclasses import dataclass
4from decimal import Decimal, InvalidOperation
6import sympy
8from Economics.shared.choices import (
10 DefaultRateReviewStatus,
12 DefaultRateType,
14 DefaultRateValueKind,
16 OperatingLineCategory,
18 OperatingLineEconomicEffect,
20)
22from Economics.reference_data.models import EconomicsDefaultRate
24from Economics.studies.models import EconomicsStudy
26from Economics.costing.models import OperatingCostLine
27from Economics.costing.operating.line_calculation import (
28 AnnualBasisQuantity,
29 annualization_requires_operating_hours,
30 annualized_basis_quantity,
31 convert_quantity,
32 strip_single_time_denominator,
33)
34from Economics.settings_profiles.services.settings_profiles import get_settings_profile
36from Economics.formulas.engine.core import EconomicsFormula, FormulaError, FormulaInput, FormulaStep, decimal_to_sympy
37from core.auxiliary.formula_units import formula_unit_expression
40SUPPORTED_OPERATING_METHODS = frozenset({"rate_times_quantity", "work_to_cost"})
41PROCESS_ENERGY_QUANTITY_UNIT = "MWh"
42OPERATING_LINE_RATE_QUANTUM = Decimal("0.00000001")
43STEAM_MASS_RATE_DENOMINATORS = frozenset({"t"})
44STEAM_ENERGY_BASIS_UNIT = "GJ"
47@dataclass(frozen=True)
48class OperatingLineFormula:
49 formula: EconomicsFormula
50 annual_basis: AnnualBasisQuantity
51 basis_quantity: Decimal
52 basis_unit: str
53 target_basis_unit: str
54 annualization_factor: Decimal
55 rate_amount: Decimal
56 rate_unit: str
58 def evaluate(self) -> Decimal | None:
59 return self.formula.evaluate({self.formula.inputs[0].key: self.basis_quantity})
62@dataclass(frozen=True)
63class OperatingLineRateBasis:
64 rate_amount: Decimal
65 rate_unit: str
66 source_basis_unit: str
67 target_basis_unit: str
70@dataclass(frozen=True)
71class OperatingAggregateFormula:
72 formula: EconomicsFormula
73 bindings: dict[str, Decimal]
75 def evaluate(self) -> Decimal | None:
76 return self.formula.evaluate(self.bindings)
79@dataclass(frozen=True)
80class ProcessEnergyContributionFormula:
81 formula: EconomicsFormula
82 source_quantity: Decimal
83 source_unit: str
85 def evaluate(self) -> Decimal | None:
86 return self.formula.evaluate({})
89@dataclass(frozen=True)
90class DefaultRateFormula:
91 formula: EconomicsFormula
92 bindings: dict[str, Decimal]
94 def evaluate(self) -> Decimal | None:
95 value = self.formula.evaluate(self.bindings)
96 if value is None: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 return None
98 return value.quantize(OPERATING_LINE_RATE_QUANTUM)
101def build_operating_line_formula(line: OperatingCostLine, *, study: EconomicsStudy) -> OperatingLineFormula:
102 """Build the canonical annual amount formula for one operating line."""
103 if line.calculation_method not in SUPPORTED_OPERATING_METHODS:
104 raise FormulaError(
105 "unsupported_operating_line_method",
106 "uses an unsupported calculation method.",
107 context={"operating_line_id": line.pk, "calculation_method": line.calculation_method},
108 )
109 if line.basis_quantity is None:
110 raise FormulaError(
111 "missing_operating_line_basis",
112 "has no basis quantity.",
113 context={"operating_line_id": line.pk},
114 )
115 rate_basis = resolve_operating_line_rate_basis(line)
116 annualization_factor = _annualization_factor_for_line(
117 line,
118 study=study,
119 target_basis_unit=rate_basis.target_basis_unit,
120 )
121 if annualization_factor is None:
122 context = {
123 "operating_line_id": line.pk,
124 "source_unit": line.basis_unit,
125 "target_unit": rate_basis.target_basis_unit,
126 }
127 if _missing_annual_operating_hours_for_annualization( 127 ↛ 137line 127 didn't jump to line 137 because the condition on line 127 was always true
128 line,
129 study=study,
130 target_basis_unit=rate_basis.target_basis_unit,
131 ):
132 raise FormulaError(
133 "missing_annual_operating_hours",
134 _missing_annual_operating_hours_message(line, rate_basis.rate_unit),
135 context=context,
136 )
137 raise FormulaError(
138 "unsupported_operating_line_unit_conversion",
139 _unsupported_annual_cost_conversion_message(line, rate_basis.rate_unit),
140 context=context,
141 )
142 annual_basis_quantity = line.basis_quantity * annualization_factor
143 basis_symbol = sympy.Symbol(_basis_input_key(line))
144 expression = _multiply_terms(
145 basis_symbol,
146 annualization_factor,
147 rate_basis.rate_amount,
148 )
149 formula = EconomicsFormula(
150 key=f"operating_line:{line.pk or 'unsaved'}",
151 expression=expression,
152 unit=f"{line.currency}/year",
153 inputs=(
154 FormulaInput(
155 key=str(basis_symbol),
156 label="Basis quantity",
157 unit=line.basis_unit,
158 source_property_info_id=line.source_property_info_id,
159 ),
160 ),
161 steps=(
162 FormulaStep(
163 kind="annualization_factor",
164 label="Annualization factor",
165 expression=str(annualization_factor),
166 amount=annualization_factor,
167 unit=rate_basis.target_basis_unit,
168 ),
169 FormulaStep(
170 kind="rate",
171 label="Rate",
172 expression=str(rate_basis.rate_amount),
173 amount=rate_basis.rate_amount,
174 unit=rate_basis.rate_unit,
175 ),
176 ),
177 )
178 return OperatingLineFormula(
179 formula=formula,
180 annual_basis=AnnualBasisQuantity(quantity=annual_basis_quantity, unit=rate_basis.target_basis_unit),
181 basis_quantity=line.basis_quantity,
182 basis_unit=line.basis_unit,
183 target_basis_unit=rate_basis.target_basis_unit,
184 annualization_factor=annualization_factor,
185 rate_amount=rate_basis.rate_amount,
186 rate_unit=rate_basis.rate_unit,
187 )
190def resolve_operating_line_rate_basis(
191 line: OperatingCostLine,
192 *,
193 source_basis_unit: str | None = None,
194) -> OperatingLineRateBasis:
195 """Resolve the price and physical basis implied by one operating line."""
197 rate_amount = _effective_rate_amount(line)
198 if rate_amount is None: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 raise FormulaError(
200 "missing_operating_line_rate",
201 "has no rate.",
202 context={"operating_line_id": line.pk},
203 )
204 rate_unit = _effective_rate_unit(line)
205 basis_unit = source_basis_unit or line.basis_unit
206 target_basis_unit = rate_denominator_unit(
207 rate_unit=rate_unit,
208 source_basis_unit=basis_unit,
209 )
210 if not target_basis_unit: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 target_basis_unit = strip_single_time_denominator(basis_unit)
212 return OperatingLineRateBasis(
213 rate_amount=rate_amount,
214 rate_unit=rate_unit,
215 source_basis_unit=basis_unit,
216 target_basis_unit=target_basis_unit,
217 )
220def build_derived_steam_rate_formula(
221 default_rate: EconomicsDefaultRate,
222 *,
223 override: dict | None = None,
224) -> DefaultRateFormula:
225 """Build the derived steam default-rate formula from fuel, steam energy, and efficiency."""
226 if default_rate.rate_type != DefaultRateType.STEAM or default_rate.value_kind != DefaultRateValueKind.DERIVED_TEMPLATE: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 raise FormulaError(
228 "unsupported_default_rate_formula",
229 "Only derived steam templates expose a default-rate formula.",
230 context={"default_rate_id": default_rate.pk},
231 )
232 metadata = default_rate.metadata if isinstance(default_rate.metadata, dict) else {}
233 fuel_price = _metadata_decimal(metadata.get("fuel_price_nzd_per_gj"))
234 steam_energy = _override_or_metadata_decimal(
235 override,
236 "steam_energy_gj_per_t",
237 metadata.get("steam_energy_gj_per_t"),
238 )
239 efficiency_percent = _override_or_metadata_decimal(
240 override,
241 "boiler_efficiency_percent",
242 metadata.get("default_boiler_efficiency_percent"),
243 )
244 if fuel_price is None or steam_energy is None or efficiency_percent is None or efficiency_percent <= 0: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise FormulaError(
246 "blocked_derived_steam_rate",
247 "Derived steam rate requires fuel price, steam energy, and positive boiler efficiency.",
248 context={"default_rate_id": default_rate.pk},
249 )
250 expression = (
251 sympy.Symbol("fuel_price_nzd_per_gj")
252 * sympy.Symbol("steam_energy_gj_per_t")
253 / (sympy.Symbol("boiler_efficiency_percent") / decimal_to_sympy(Decimal("100")))
254 )
255 return DefaultRateFormula(
256 formula=EconomicsFormula(
257 key=f"default_rate:derived_steam:{default_rate.pk or default_rate.key}",
258 expression=expression,
259 unit=default_rate.display_unit,
260 inputs=(
261 FormulaInput("fuel_price_nzd_per_gj", "Fuel price", "NZD/GJ"),
262 FormulaInput("steam_energy_gj_per_t", "Steam energy basis", "GJ/t"),
263 FormulaInput("boiler_efficiency_percent", "Boiler efficiency", "percent"),
264 ),
265 ),
266 bindings={
267 "fuel_price_nzd_per_gj": fuel_price,
268 "steam_energy_gj_per_t": steam_energy,
269 "boiler_efficiency_percent": efficiency_percent,
270 },
271 )
274def build_process_energy_contribution_formula(
275 line: OperatingCostLine,
276 *,
277 study: EconomicsStudy,
278) -> ProcessEnergyContributionFormula:
279 """Build the target-process-energy contribution formula for one operating line."""
280 if line.basis_quantity is None: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true
281 raise FormulaError(
282 "missing_process_energy_basis",
283 "Process-energy contribution requires a basis quantity.",
284 context={"operating_line_id": line.pk},
285 )
286 direct_energy = annualized_basis_quantity(
287 line.basis_quantity,
288 source_unit=line.basis_unit,
289 target_unit=PROCESS_ENERGY_QUANTITY_UNIT,
290 study=study,
291 )
292 if direct_energy is not None: 292 ↛ 300line 292 didn't jump to line 300 because the condition on line 292 was always true
293 return _process_energy_formula(
294 line=line,
295 target_quantity=direct_energy,
296 source_quantity=line.basis_quantity,
297 source_unit=line.basis_unit,
298 )
300 annual_basis = build_operating_line_formula(line, study=study).annual_basis
301 source_unit = strip_single_time_denominator(annual_basis.unit)
302 if not source_unit:
303 raise FormulaError(
304 "unsupported_process_energy_unit",
305 "Process-energy contribution requires an energy-compatible annual basis.",
306 context={"operating_line_id": line.pk, "annual_basis_unit": annual_basis.unit},
307 )
308 target_quantity = convert_quantity(
309 value=annual_basis.quantity,
310 source_unit=source_unit,
311 target_unit=PROCESS_ENERGY_QUANTITY_UNIT,
312 )
313 if target_quantity is None:
314 raise FormulaError(
315 "unsupported_process_energy_unit",
316 "Process-energy contribution requires an energy-compatible annual basis.",
317 context={"operating_line_id": line.pk, "annual_basis_unit": annual_basis.unit},
318 )
319 return _process_energy_formula(
320 line=line,
321 target_quantity=target_quantity,
322 source_quantity=annual_basis.quantity,
323 source_unit=annual_basis.unit,
324 )
327def build_annual_operating_expense_formula(study: EconomicsStudy) -> OperatingAggregateFormula:
328 """Build annual OPEX from included non-revenue operating line formulas."""
329 return _build_operating_total_formula(
330 study,
331 key="annual_operating_expense",
332 label="Annual operating expense",
333 include_revenue=False,
334 )
337def build_annual_operating_revenue_formula(study: EconomicsStudy) -> OperatingAggregateFormula:
338 """Build annual revenue from included output-revenue operating line formulas."""
339 return _build_operating_total_formula(
340 study,
341 key="annual_operating_revenue",
342 label="Annual operating revenue",
343 include_revenue=True,
344 )
347def render_operating_line_property_formula(
348 operating_formula: OperatingLineFormula,
349 *,
350 source_property_formula: str | None,
351) -> str:
352 """Render an operating-line formula as an IDAES-visible annual-cost formula."""
353 rate_unit_expression = formula_unit_expression(operating_formula.rate_unit)
354 target_basis_unit_expression = formula_unit_expression(operating_formula.target_basis_unit)
355 if rate_unit_expression is None or target_basis_unit_expression is None: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true
356 raise FormulaError(
357 "unsupported_operating_line_unit_rendering",
358 "uses a unit conversion that is not solve-visible yet.",
359 )
361 input_key = operating_formula.formula.inputs[0].key
362 if source_property_formula is None:
363 rendered = operating_formula.formula.render_property_formula(
364 {input_key: _decimal_literal(operating_formula.basis_quantity)}
365 )
366 return (
367 f"({rendered} * ({target_basis_unit_expression}) "
368 f"* ({rate_unit_expression}) / year)"
369 )
371 basis_unit_expression = formula_unit_expression(operating_formula.basis_unit)
372 if basis_unit_expression is None: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 raise FormulaError(
374 "unsupported_operating_line_unit_rendering",
375 "uses a unit conversion that is not solve-visible yet.",
376 )
377 converted_source = f"convert({source_property_formula}, {basis_unit_expression})"
378 rendered = operating_formula.formula.render_property_formula({input_key: converted_source})
379 return (
380 f"({rendered} * ({rate_unit_expression}) "
381 f"* ({target_basis_unit_expression}) / ({basis_unit_expression}) / year)"
382 )
385def rate_denominator_unit(*, rate_unit: str, source_basis_unit: str) -> str:
386 """Return the physical basis denominator implied by a rate unit."""
387 if "/" not in rate_unit: 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 return ""
389 denominator = rate_unit.split("/", 1)[1].strip()
390 if denominator == source_basis_unit.strip():
391 return strip_single_time_denominator(denominator)
392 return denominator
395def _annualization_factor_for_line(
396 line: OperatingCostLine,
397 *,
398 study: EconomicsStudy,
399 target_basis_unit: str,
400) -> Decimal | None:
401 annualization_factor = annualized_basis_quantity(
402 Decimal("1"),
403 source_unit=line.basis_unit,
404 target_unit=target_basis_unit,
405 study=study,
406 )
407 if annualization_factor is not None:
408 return annualization_factor
409 return _steam_mass_annualization_factor(
410 line,
411 study=study,
412 target_basis_unit=target_basis_unit,
413 )
416def _missing_annual_operating_hours_for_annualization(
417 line: OperatingCostLine,
418 *,
419 study: EconomicsStudy,
420 target_basis_unit: str,
421) -> bool:
422 settings_profile = get_settings_profile(study)
423 if settings_profile is not None and settings_profile.annual_operating_hours is not None: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true
424 return False
425 if (
426 target_basis_unit.strip().lower() in STEAM_MASS_RATE_DENOMINATORS
427 and line.rate_type == DefaultRateType.STEAM
428 and _steam_energy_basis_gj_per_t(line)
429 ):
430 return annualization_requires_operating_hours(
431 source_unit=line.basis_unit,
432 target_unit=STEAM_ENERGY_BASIS_UNIT,
433 )
434 return annualization_requires_operating_hours(
435 source_unit=line.basis_unit,
436 target_unit=target_basis_unit,
437 )
440def _steam_mass_annualization_factor(
441 line: OperatingCostLine,
442 *,
443 study: EconomicsStudy,
444 target_basis_unit: str,
445) -> Decimal | None:
446 if target_basis_unit.strip().lower() not in STEAM_MASS_RATE_DENOMINATORS:
447 return None
448 if line.rate_type != DefaultRateType.STEAM: 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true
449 return None
450 steam_energy_basis = _steam_energy_basis_gj_per_t(line)
451 if steam_energy_basis is None or steam_energy_basis <= 0: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 return None
453 annual_energy = annualized_basis_quantity(
454 Decimal("1"),
455 source_unit=line.basis_unit,
456 target_unit=STEAM_ENERGY_BASIS_UNIT,
457 study=study,
458 )
459 if annual_energy is None:
460 return None
461 return annual_energy / steam_energy_basis
464def _steam_energy_basis_gj_per_t(line: OperatingCostLine) -> Decimal | None:
465 if not line.source_default_rate_id: 465 ↛ 466line 465 didn't jump to line 466 because the condition on line 465 was never true
466 return None
467 metadata = line.source_default_rate.metadata if isinstance(line.source_default_rate.metadata, dict) else {}
468 return _metadata_decimal(metadata.get("steam_energy_gj_per_t"))
471def _build_operating_total_formula(
472 study: EconomicsStudy,
473 *,
474 key: str,
475 label: str,
476 include_revenue: bool,
477) -> OperatingAggregateFormula:
478 terms: list[sympy.Expr] = []
479 bindings: dict[str, Decimal] = {}
480 inputs: list[FormulaInput] = []
481 blocked_children: list[dict[str, str]] = []
482 for line in study.operating_lines.filter(included=True).select_related("source_default_rate").order_by("pk"):
483 if operating_line_is_revenue(line) != include_revenue:
484 continue
485 try:
486 amount = build_operating_line_formula(line, study=study).evaluate()
487 except FormulaError as exc:
488 amount = None
489 blocked_children.append(
490 {
491 "key": f"operating_line:{line.pk}",
492 "code": exc.code,
493 "reason": exc.message,
494 }
495 )
496 if amount is not None:
497 symbol_key = operating_line_input_key(line.pk)
498 terms.append(sympy.Symbol(symbol_key))
499 bindings[symbol_key] = amount
500 inputs.append(
501 FormulaInput(
502 key=symbol_key,
503 label=line.label,
504 unit=f"{line.currency}/year",
505 )
506 )
507 return OperatingAggregateFormula(
508 formula=EconomicsFormula(
509 key=key,
510 expression=_sum_expressions(terms),
511 unit=f"{_study_currency(study)}/year",
512 inputs=tuple(inputs),
513 steps=(
514 FormulaStep(
515 kind=key,
516 label=label,
517 expression="sum(included operating line formulas)",
518 unit=f"{_study_currency(study)}/year",
519 ),
520 ),
521 missing_child_policy="strict_after_revenue_classification",
522 blocked_children=tuple(blocked_children),
523 blocked_reason=_operating_total_blocked_reason(label, blocked_children),
524 ),
525 bindings=bindings,
526 )
529def _operating_total_blocked_reason(label: str, blocked_children: list[dict[str, str]]) -> str:
530 if not blocked_children:
531 return ""
532 if all(child.get("code") == "missing_annual_operating_hours" for child in blocked_children):
533 return "Set annual operating hours in Settings, then recalculate Project overview."
534 return f"{label} has blocked operating lines."
537def _missing_annual_operating_hours_message(line: OperatingCostLine, rate_unit: str) -> str:
538 unit_pair = _operating_line_unit_pair(line, rate_unit)
539 if unit_pair: 539 ↛ 541line 539 didn't jump to line 541 because the condition on line 539 was always true
540 return f"{unit_pair} needs annual operating hours in Settings."
541 return "Needs annual operating hours in Settings."
544def _unsupported_annual_cost_conversion_message(line: OperatingCostLine, rate_unit: str) -> str:
545 unit_pair = _operating_line_unit_pair(line, rate_unit)
546 if unit_pair:
547 return f"{unit_pair} cannot be converted to an annual cost. Check this line's basis and rate units."
548 return "This line's basis and rate units cannot be converted to an annual cost."
551def _operating_line_unit_pair(line: OperatingCostLine, rate_unit: str) -> str:
552 basis_unit = line.basis_unit.strip()
553 rate_unit = rate_unit.strip()
554 if basis_unit and rate_unit: 554 ↛ 556line 554 didn't jump to line 556 because the condition on line 554 was always true
555 return f"{basis_unit} with {rate_unit}"
556 if basis_unit:
557 return f"{basis_unit} basis"
558 if rate_unit:
559 return f"{rate_unit} rate"
560 return ""
563def operating_line_is_revenue(line: OperatingCostLine) -> bool:
564 """Return whether an operating line is a credit against operating cost."""
565 return (
566 line.economic_effect == OperatingLineEconomicEffect.REVENUE
567 or line.category == OperatingLineCategory.OUTPUT_REVENUE
568 )
571def operating_line_input_key(line_id: int) -> str:
572 return f"operating_line_{line_id}"
575def _process_energy_formula(
576 *,
577 line: OperatingCostLine,
578 target_quantity: Decimal,
579 source_quantity: Decimal,
580 source_unit: str,
581) -> ProcessEnergyContributionFormula:
582 return ProcessEnergyContributionFormula(
583 formula=EconomicsFormula(
584 key=f"process_energy_contribution:{line.pk or 'unsaved'}",
585 expression=decimal_to_sympy(target_quantity),
586 unit=PROCESS_ENERGY_QUANTITY_UNIT,
587 inputs=(),
588 steps=(
589 FormulaStep(
590 kind="process_energy_contribution",
591 label="Process-energy contribution",
592 expression=f"{source_quantity} {source_unit} -> {PROCESS_ENERGY_QUANTITY_UNIT}",
593 amount=target_quantity,
594 unit=PROCESS_ENERGY_QUANTITY_UNIT,
595 ),
596 ),
597 ),
598 source_quantity=source_quantity,
599 source_unit=source_unit,
600 )
603def _multiply_terms(basis_symbol: sympy.Symbol, annualization_factor: Decimal, rate_amount: Decimal) -> sympy.Expr:
604 terms: list[sympy.Expr] = [basis_symbol]
605 if annualization_factor != Decimal("1"):
606 terms.append(decimal_to_sympy(annualization_factor))
607 terms.append(decimal_to_sympy(rate_amount))
608 return sympy.Mul(*terms, evaluate=False)
611def _sum_expressions(terms: list[sympy.Expr]) -> sympy.Expr:
612 if not terms:
613 return decimal_to_sympy(Decimal("0"))
614 if len(terms) == 1:
615 return terms[0]
616 return sympy.Add(*terms, evaluate=False)
619def _study_currency(study: EconomicsStudy) -> str:
620 assumptions = get_settings_profile(study)
621 if assumptions is None:
622 return "NZD"
623 return assumptions.currency or "NZD"
626def _effective_rate_amount(line: OperatingCostLine) -> Decimal | None:
627 if line.rate_amount is not None: 627 ↛ 629line 627 didn't jump to line 629 because the condition on line 627 was always true
628 return line.rate_amount
629 if (
630 line.source_default_rate_id
631 and line.source_default_rate.value is not None
632 and line.source_default_rate.review_status == DefaultRateReviewStatus.REVIEWED
633 and line.source_default_rate.value_kind == DefaultRateValueKind.REVIEWED_DEFAULT
634 ):
635 return line.source_default_rate.value
636 return None
639def _effective_rate_unit(line: OperatingCostLine) -> str:
640 if line.rate_unit: 640 ↛ 642line 640 didn't jump to line 642 because the condition on line 640 was always true
641 return line.rate_unit
642 if line.source_default_rate_id:
643 return line.source_default_rate.display_unit
644 return ""
647def _basis_input_key(line: OperatingCostLine) -> str:
648 return f"basis_{line.pk or 'unsaved'}"
651def _metadata_decimal(value) -> Decimal | None:
652 if value in (None, ""): 652 ↛ 653line 652 didn't jump to line 653 because the condition on line 652 was never true
653 return None
654 try:
655 amount = Decimal(str(value))
656 except (InvalidOperation, ValueError, TypeError):
657 return None
658 return amount if amount.is_finite() else None
661def _override_or_metadata_decimal(override: dict | None, key: str, metadata_value) -> Decimal | None:
662 if not override or key not in override or override.get(key) in (None, ""):
663 return _metadata_decimal(metadata_value)
664 return _metadata_decimal(override.get(key))
667def _decimal_literal(value: Decimal) -> str:
668 return format(value.normalize(), "f")