zwitschi
1feae7ff85
- Introduced OpexValidationError for handling validation errors in processing opex calculations. - Implemented ProjectProcessingOpexRepository and ScenarioProcessingOpexRepository for managing project and scenario-level processing opex snapshots. - Enhanced UnitOfWork to include repositories for processing opex. - Updated sidebar navigation and scenario detail templates to include links to the new Processing Opex Planner. - Created a new template for the Processing Opex Planner with form handling for input components and parameters. - Developed integration tests for processing opex calculations, covering HTML and JSON flows, including validation for currency mismatches and unsupported frequencies. - Added unit tests for the calculation logic, ensuring correct handling of various scenarios and edge cases.
347 lines
10 KiB
Python
347 lines
10 KiB
Python
"""Pydantic schemas for calculation workflows."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Optional
|
|
|
|
from pydantic import BaseModel, Field, PositiveFloat, ValidationError, field_validator
|
|
|
|
from services.pricing import PricingResult
|
|
|
|
|
|
class ImpurityInput(BaseModel):
|
|
"""Impurity configuration row supplied by the client."""
|
|
|
|
name: str = Field(..., min_length=1)
|
|
value: float | None = Field(None, ge=0)
|
|
threshold: float | None = Field(None, ge=0)
|
|
penalty: float | None = Field(None)
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _normalise_name(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
|
|
class ProfitabilityCalculationRequest(BaseModel):
|
|
"""Request payload for profitability calculations."""
|
|
|
|
metal: str = Field(..., min_length=1)
|
|
ore_tonnage: PositiveFloat
|
|
head_grade_pct: float = Field(..., gt=0, le=100)
|
|
recovery_pct: float = Field(..., gt=0, le=100)
|
|
payable_pct: float | None = Field(None, gt=0, le=100)
|
|
reference_price: PositiveFloat
|
|
treatment_charge: float = Field(0, ge=0)
|
|
smelting_charge: float = Field(0, ge=0)
|
|
moisture_pct: float = Field(0, ge=0, le=100)
|
|
moisture_threshold_pct: float | None = Field(None, ge=0, le=100)
|
|
moisture_penalty_per_pct: float | None = None
|
|
premiums: float = Field(0)
|
|
fx_rate: PositiveFloat = Field(1)
|
|
currency_code: str | None = Field(None, min_length=3, max_length=3)
|
|
processing_opex: float = Field(0, ge=0)
|
|
sustaining_capex: float = Field(0, ge=0)
|
|
initial_capex: float = Field(0, ge=0)
|
|
discount_rate: float | None = Field(None, ge=0, le=100)
|
|
periods: int = Field(10, ge=1, le=120)
|
|
impurities: List[ImpurityInput] = Field(default_factory=list)
|
|
|
|
@field_validator("currency_code")
|
|
@classmethod
|
|
def _uppercase_currency(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.strip().upper()
|
|
|
|
@field_validator("metal")
|
|
@classmethod
|
|
def _normalise_metal(cls, value: str) -> str:
|
|
return value.strip().lower()
|
|
|
|
|
|
class ProfitabilityCosts(BaseModel):
|
|
"""Aggregated cost components for profitability output."""
|
|
|
|
processing_opex_total: float
|
|
sustaining_capex_total: float
|
|
initial_capex: float
|
|
|
|
|
|
class ProfitabilityMetrics(BaseModel):
|
|
"""Financial KPIs yielded by the profitability calculation."""
|
|
|
|
npv: float | None
|
|
irr: float | None
|
|
payback_period: float | None
|
|
margin: float | None
|
|
|
|
|
|
class CashFlowEntry(BaseModel):
|
|
"""Normalized cash flow row for reporting and charting."""
|
|
|
|
period: int
|
|
revenue: float
|
|
processing_opex: float
|
|
sustaining_capex: float
|
|
net: float
|
|
|
|
|
|
class ProfitabilityCalculationResult(BaseModel):
|
|
"""Response body summarizing profitability calculation outputs."""
|
|
|
|
pricing: PricingResult
|
|
costs: ProfitabilityCosts
|
|
metrics: ProfitabilityMetrics
|
|
cash_flows: list[CashFlowEntry]
|
|
currency: str | None
|
|
|
|
|
|
class CapexComponentInput(BaseModel):
|
|
"""Capex component entry supplied by the UI."""
|
|
|
|
id: int | None = Field(default=None, ge=1)
|
|
name: str = Field(..., min_length=1)
|
|
category: str = Field(..., min_length=1)
|
|
amount: float = Field(..., ge=0)
|
|
currency: str | None = Field(None, min_length=3, max_length=3)
|
|
spend_year: int | None = Field(None, ge=0, le=120)
|
|
notes: str | None = Field(None, max_length=500)
|
|
|
|
@field_validator("currency")
|
|
@classmethod
|
|
def _uppercase_currency(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.strip().upper()
|
|
|
|
@field_validator("category")
|
|
@classmethod
|
|
def _normalise_category(cls, value: str) -> str:
|
|
return value.strip().lower()
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _trim_name(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
|
|
class CapexParameters(BaseModel):
|
|
"""Global parameters applied to capex calculations."""
|
|
|
|
currency_code: str | None = Field(None, min_length=3, max_length=3)
|
|
contingency_pct: float | None = Field(0, ge=0, le=100)
|
|
discount_rate_pct: float | None = Field(None, ge=0, le=100)
|
|
evaluation_horizon_years: int | None = Field(10, ge=1, le=100)
|
|
|
|
@field_validator("currency_code")
|
|
@classmethod
|
|
def _uppercase_currency(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.strip().upper()
|
|
|
|
|
|
class CapexCalculationOptions(BaseModel):
|
|
"""Optional behaviour flags for capex calculations."""
|
|
|
|
persist: bool = False
|
|
|
|
|
|
class CapexCalculationRequest(BaseModel):
|
|
"""Request payload for capex aggregation."""
|
|
|
|
components: List[CapexComponentInput] = Field(default_factory=list)
|
|
parameters: CapexParameters = Field(
|
|
default_factory=CapexParameters, # type: ignore[arg-type]
|
|
)
|
|
options: CapexCalculationOptions = Field(
|
|
default_factory=CapexCalculationOptions, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
class CapexCategoryBreakdown(BaseModel):
|
|
"""Breakdown entry describing category totals."""
|
|
|
|
category: str
|
|
amount: float = Field(..., ge=0)
|
|
share: float | None = Field(None, ge=0, le=100)
|
|
|
|
|
|
class CapexTotals(BaseModel):
|
|
"""Aggregated totals for capex workflows."""
|
|
|
|
overall: float = Field(..., ge=0)
|
|
contingency_pct: float = Field(0, ge=0, le=100)
|
|
contingency_amount: float = Field(..., ge=0)
|
|
with_contingency: float = Field(..., ge=0)
|
|
by_category: List[CapexCategoryBreakdown] = Field(default_factory=list)
|
|
|
|
|
|
class CapexTimelineEntry(BaseModel):
|
|
"""Spend profile entry grouped by year."""
|
|
|
|
year: int
|
|
spend: float = Field(..., ge=0)
|
|
cumulative: float = Field(..., ge=0)
|
|
|
|
|
|
class CapexCalculationResult(BaseModel):
|
|
"""Response body for capex calculations."""
|
|
|
|
totals: CapexTotals
|
|
timeline: List[CapexTimelineEntry] = Field(default_factory=list)
|
|
components: List[CapexComponentInput] = Field(default_factory=list)
|
|
parameters: CapexParameters
|
|
options: CapexCalculationOptions
|
|
currency: str | None
|
|
|
|
|
|
class ProcessingOpexComponentInput(BaseModel):
|
|
"""Processing opex component entry supplied by the UI."""
|
|
|
|
id: int | None = Field(default=None, ge=1)
|
|
name: str = Field(..., min_length=1)
|
|
category: str = Field(..., min_length=1)
|
|
unit_cost: float = Field(..., ge=0)
|
|
quantity: float = Field(..., ge=0)
|
|
frequency: str = Field(..., min_length=1)
|
|
currency: str | None = Field(None, min_length=3, max_length=3)
|
|
period_start: int | None = Field(None, ge=0, le=240)
|
|
period_end: int | None = Field(None, ge=0, le=240)
|
|
notes: str | None = Field(None, max_length=500)
|
|
|
|
@field_validator("currency")
|
|
@classmethod
|
|
def _uppercase_currency(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.strip().upper()
|
|
|
|
@field_validator("category")
|
|
@classmethod
|
|
def _normalise_category(cls, value: str) -> str:
|
|
return value.strip().lower()
|
|
|
|
@field_validator("frequency")
|
|
@classmethod
|
|
def _normalise_frequency(cls, value: str) -> str:
|
|
return value.strip().lower()
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _trim_name(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
|
|
class ProcessingOpexParameters(BaseModel):
|
|
"""Global parameters applied to processing opex calculations."""
|
|
|
|
currency_code: str | None = Field(None, min_length=3, max_length=3)
|
|
escalation_pct: float | None = Field(None, ge=0, le=100)
|
|
discount_rate_pct: float | None = Field(None, ge=0, le=100)
|
|
evaluation_horizon_years: int | None = Field(10, ge=1, le=100)
|
|
apply_escalation: bool = True
|
|
|
|
@field_validator("currency_code")
|
|
@classmethod
|
|
def _uppercase_currency(cls, value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return value.strip().upper()
|
|
|
|
|
|
class ProcessingOpexOptions(BaseModel):
|
|
"""Optional behaviour flags for opex calculations."""
|
|
|
|
persist: bool = False
|
|
snapshot_notes: str | None = Field(None, max_length=500)
|
|
|
|
|
|
class ProcessingOpexCalculationRequest(BaseModel):
|
|
"""Request payload for processing opex aggregation."""
|
|
|
|
components: List[ProcessingOpexComponentInput] = Field(
|
|
default_factory=list)
|
|
parameters: ProcessingOpexParameters = Field(
|
|
default_factory=ProcessingOpexParameters, # type: ignore[arg-type]
|
|
)
|
|
options: ProcessingOpexOptions = Field(
|
|
default_factory=ProcessingOpexOptions, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
class ProcessingOpexCategoryBreakdown(BaseModel):
|
|
"""Category breakdown for processing opex totals."""
|
|
|
|
category: str
|
|
annual_cost: float = Field(..., ge=0)
|
|
share: float | None = Field(None, ge=0, le=100)
|
|
|
|
|
|
class ProcessingOpexTimelineEntry(BaseModel):
|
|
"""Timeline entry representing cost over evaluation periods."""
|
|
|
|
period: int
|
|
base_cost: float = Field(..., ge=0)
|
|
escalated_cost: float | None = Field(None, ge=0)
|
|
|
|
|
|
class ProcessingOpexMetrics(BaseModel):
|
|
"""Derived KPIs for processing opex outputs."""
|
|
|
|
annual_average: float | None
|
|
cost_per_ton: float | None
|
|
|
|
|
|
class ProcessingOpexTotals(BaseModel):
|
|
"""Aggregated totals for processing opex."""
|
|
|
|
overall_annual: float = Field(..., ge=0)
|
|
escalated_total: float | None = Field(None, ge=0)
|
|
escalation_pct: float | None = Field(None, ge=0, le=100)
|
|
by_category: List[ProcessingOpexCategoryBreakdown] = Field(
|
|
default_factory=list
|
|
)
|
|
|
|
|
|
class ProcessingOpexCalculationResult(BaseModel):
|
|
"""Response body summarising processing opex calculations."""
|
|
|
|
totals: ProcessingOpexTotals
|
|
timeline: List[ProcessingOpexTimelineEntry] = Field(default_factory=list)
|
|
metrics: ProcessingOpexMetrics
|
|
components: List[ProcessingOpexComponentInput] = Field(
|
|
default_factory=list)
|
|
parameters: ProcessingOpexParameters
|
|
options: ProcessingOpexOptions
|
|
currency: str | None
|
|
|
|
|
|
__all__ = [
|
|
"ImpurityInput",
|
|
"ProfitabilityCalculationRequest",
|
|
"ProfitabilityCosts",
|
|
"ProfitabilityMetrics",
|
|
"CashFlowEntry",
|
|
"ProfitabilityCalculationResult",
|
|
"CapexComponentInput",
|
|
"CapexParameters",
|
|
"CapexCalculationOptions",
|
|
"CapexCalculationRequest",
|
|
"CapexCategoryBreakdown",
|
|
"CapexTotals",
|
|
"CapexTimelineEntry",
|
|
"CapexCalculationResult",
|
|
"ProcessingOpexComponentInput",
|
|
"ProcessingOpexParameters",
|
|
"ProcessingOpexOptions",
|
|
"ProcessingOpexCalculationRequest",
|
|
"ProcessingOpexCategoryBreakdown",
|
|
"ProcessingOpexTimelineEntry",
|
|
"ProcessingOpexMetrics",
|
|
"ProcessingOpexTotals",
|
|
"ProcessingOpexCalculationResult",
|
|
"ValidationError",
|
|
]
|