feat: Implement initial capex calculation feature
- Added CapexComponentInput, CapexParameters, CapexCalculationRequest, CapexCalculationResult, and related schemas for capex calculations. - Introduced calculate_initial_capex function to aggregate capex components and compute totals and timelines. - Created ProjectCapexRepository and ScenarioCapexRepository for managing capex snapshots in the database. - Developed capex.html template for capturing and displaying initial capex data. - Registered common Jinja2 filters for formatting currency and percentages. - Implemented unit and integration tests for capex calculation functionality. - Updated unit of work to include new repositories for capex management.
This commit is contained in:
@@ -28,6 +28,7 @@ from .scenario import Scenario
|
||||
from .simulation_parameter import SimulationParameter
|
||||
from .user import Role, User, UserRole, password_context
|
||||
from .profitability_snapshot import ProjectProfitability, ScenarioProfitability
|
||||
from .capex_snapshot import ProjectCapexSnapshot, ScenarioCapexSnapshot
|
||||
|
||||
__all__ = [
|
||||
"FinancialCategory",
|
||||
@@ -35,11 +36,13 @@ __all__ = [
|
||||
"MiningOperationType",
|
||||
"Project",
|
||||
"ProjectProfitability",
|
||||
"ProjectCapexSnapshot",
|
||||
"PricingSettings",
|
||||
"PricingMetalSettings",
|
||||
"PricingImpuritySettings",
|
||||
"Scenario",
|
||||
"ScenarioProfitability",
|
||||
"ScenarioCapexSnapshot",
|
||||
"ScenarioStatus",
|
||||
"DistributionType",
|
||||
"SimulationParameter",
|
||||
|
||||
111
models/capex_snapshot.py
Normal file
111
models/capex_snapshot.py
Normal file
@@ -0,0 +1,111 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import JSON, DateTime, ForeignKey, Integer, Numeric, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from config.database import Base
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from .project import Project
|
||||
from .scenario import Scenario
|
||||
from .user import User
|
||||
|
||||
|
||||
class ProjectCapexSnapshot(Base):
|
||||
"""Snapshot of aggregated initial capex metrics at the project level."""
|
||||
|
||||
__tablename__ = "project_capex_snapshots"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
project_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True
|
||||
)
|
||||
created_by_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True
|
||||
)
|
||||
calculation_source: Mapped[str | None] = mapped_column(
|
||||
String(64), nullable=True)
|
||||
calculated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now()
|
||||
)
|
||||
currency_code: Mapped[str | None] = mapped_column(String(3), nullable=True)
|
||||
total_capex: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
contingency_pct: Mapped[float | None] = mapped_column(
|
||||
Numeric(12, 6), nullable=True)
|
||||
contingency_amount: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
total_with_contingency: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
component_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
payload: Mapped[dict | None] = mapped_column(JSON, nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
|
||||
project: Mapped[Project] = relationship(
|
||||
"Project", back_populates="capex_snapshots"
|
||||
)
|
||||
created_by: Mapped[User | None] = relationship("User")
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return (
|
||||
"ProjectCapexSnapshot(id={id!r}, project_id={project_id!r}, total_capex={total_capex!r})".format(
|
||||
id=self.id, project_id=self.project_id, total_capex=self.total_capex
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class ScenarioCapexSnapshot(Base):
|
||||
"""Snapshot of initial capex metrics for an individual scenario."""
|
||||
|
||||
__tablename__ = "scenario_capex_snapshots"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
scenario_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("scenarios.id", ondelete="CASCADE"), nullable=False, index=True
|
||||
)
|
||||
created_by_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True
|
||||
)
|
||||
calculation_source: Mapped[str | None] = mapped_column(
|
||||
String(64), nullable=True)
|
||||
calculated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now()
|
||||
)
|
||||
currency_code: Mapped[str | None] = mapped_column(String(3), nullable=True)
|
||||
total_capex: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
contingency_pct: Mapped[float | None] = mapped_column(
|
||||
Numeric(12, 6), nullable=True)
|
||||
contingency_amount: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
total_with_contingency: Mapped[float | None] = mapped_column(
|
||||
Numeric(18, 2), nullable=True)
|
||||
component_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
payload: Mapped[dict | None] = mapped_column(JSON, nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
|
||||
scenario: Mapped[Scenario] = relationship(
|
||||
"Scenario", back_populates="capex_snapshots"
|
||||
)
|
||||
created_by: Mapped[User | None] = relationship("User")
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover
|
||||
return (
|
||||
"ScenarioCapexSnapshot(id={id!r}, scenario_id={scenario_id!r}, total_capex={total_capex!r})".format(
|
||||
id=self.id, scenario_id=self.scenario_id, total_capex=self.total_capex
|
||||
)
|
||||
)
|
||||
@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, List
|
||||
|
||||
from .enums import MiningOperationType, sql_enum
|
||||
from .profitability_snapshot import ProjectProfitability
|
||||
from .capex_snapshot import ProjectCapexSnapshot
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
@@ -59,6 +60,13 @@ class Project(Base):
|
||||
order_by=lambda: ProjectProfitability.calculated_at.desc(),
|
||||
passive_deletes=True,
|
||||
)
|
||||
capex_snapshots: Mapped[List["ProjectCapexSnapshot"]] = relationship(
|
||||
"ProjectCapexSnapshot",
|
||||
back_populates="project",
|
||||
cascade="all, delete-orphan",
|
||||
order_by=lambda: ProjectCapexSnapshot.calculated_at.desc(),
|
||||
passive_deletes=True,
|
||||
)
|
||||
|
||||
@property
|
||||
def latest_profitability(self) -> "ProjectProfitability | None":
|
||||
@@ -68,5 +76,13 @@ class Project(Base):
|
||||
return None
|
||||
return self.profitability_snapshots[0]
|
||||
|
||||
@property
|
||||
def latest_capex(self) -> "ProjectCapexSnapshot | None":
|
||||
"""Return the most recent capex snapshot, if any."""
|
||||
|
||||
if not self.capex_snapshots:
|
||||
return None
|
||||
return self.capex_snapshots[0]
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover - helpful for debugging
|
||||
return f"Project(id={self.id!r}, name={self.name!r})"
|
||||
|
||||
@@ -20,6 +20,7 @@ from config.database import Base
|
||||
from services.currency import normalise_currency
|
||||
from .enums import ResourceType, ScenarioStatus, sql_enum
|
||||
from .profitability_snapshot import ScenarioProfitability
|
||||
from .capex_snapshot import ScenarioCapexSnapshot
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from .financial_input import FinancialInput
|
||||
@@ -83,6 +84,13 @@ class Scenario(Base):
|
||||
order_by=lambda: ScenarioProfitability.calculated_at.desc(),
|
||||
passive_deletes=True,
|
||||
)
|
||||
capex_snapshots: Mapped[List["ScenarioCapexSnapshot"]] = relationship(
|
||||
"ScenarioCapexSnapshot",
|
||||
back_populates="scenario",
|
||||
cascade="all, delete-orphan",
|
||||
order_by=lambda: ScenarioCapexSnapshot.calculated_at.desc(),
|
||||
passive_deletes=True,
|
||||
)
|
||||
|
||||
@validates("currency")
|
||||
def _normalise_currency(self, key: str, value: str | None) -> str | None:
|
||||
@@ -99,3 +107,11 @@ class Scenario(Base):
|
||||
if not self.profitability_snapshots:
|
||||
return None
|
||||
return self.profitability_snapshots[0]
|
||||
|
||||
@property
|
||||
def latest_capex(self) -> "ScenarioCapexSnapshot | None":
|
||||
"""Return the most recent capex snapshot for this scenario."""
|
||||
|
||||
if not self.capex_snapshots:
|
||||
return None
|
||||
return self.capex_snapshots[0]
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
from typing import Any, Sequence
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, Request, status
|
||||
from fastapi.responses import HTMLResponse, JSONResponse, Response
|
||||
@@ -14,22 +14,31 @@ from starlette.datastructures import FormData
|
||||
from dependencies import get_pricing_metadata, get_unit_of_work, require_authenticated_user
|
||||
from models import (
|
||||
Project,
|
||||
ProjectCapexSnapshot,
|
||||
ProjectProfitability,
|
||||
Scenario,
|
||||
ScenarioCapexSnapshot,
|
||||
ScenarioProfitability,
|
||||
User,
|
||||
)
|
||||
from schemas.calculations import (
|
||||
CapexCalculationOptions,
|
||||
CapexCalculationRequest,
|
||||
CapexCalculationResult,
|
||||
CapexComponentInput,
|
||||
CapexParameters,
|
||||
ProfitabilityCalculationRequest,
|
||||
ProfitabilityCalculationResult,
|
||||
)
|
||||
from services.calculations import calculate_profitability
|
||||
from services.exceptions import EntityNotFoundError, ProfitabilityValidationError
|
||||
from services.calculations import calculate_initial_capex, calculate_profitability
|
||||
from services.exceptions import CapexValidationError, EntityNotFoundError, ProfitabilityValidationError
|
||||
from services.pricing import PricingMetadata
|
||||
from services.unit_of_work import UnitOfWork
|
||||
from routes.template_filters import register_common_filters
|
||||
|
||||
router = APIRouter(prefix="/calculations", tags=["Calculations"])
|
||||
templates = Jinja2Templates(directory="templates")
|
||||
register_common_filters(templates)
|
||||
|
||||
_SUPPORTED_METALS: tuple[dict[str, str], ...] = (
|
||||
{"value": "copper", "label": "Copper"},
|
||||
@@ -39,6 +48,14 @@ _SUPPORTED_METALS: tuple[dict[str, str], ...] = (
|
||||
_SUPPORTED_METAL_VALUES = {entry["value"] for entry in _SUPPORTED_METALS}
|
||||
_DEFAULT_EVALUATION_PERIODS = 10
|
||||
|
||||
_CAPEX_CATEGORY_OPTIONS: tuple[dict[str, str], ...] = (
|
||||
{"value": "equipment", "label": "Equipment"},
|
||||
{"value": "infrastructure", "label": "Infrastructure"},
|
||||
{"value": "land", "label": "Land & Property"},
|
||||
{"value": "miscellaneous", "label": "Miscellaneous"},
|
||||
)
|
||||
_DEFAULT_CAPEX_HORIZON_YEARS = 5
|
||||
|
||||
|
||||
def _combine_impurity_metadata(metadata: PricingMetadata) -> list[dict[str, object]]:
|
||||
"""Build impurity rows combining thresholds and penalties."""
|
||||
@@ -190,6 +207,266 @@ def _prepare_form_data_for_display(
|
||||
return data
|
||||
|
||||
|
||||
def _coerce_bool(value: Any) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
lowered = value.strip().lower()
|
||||
return lowered in {"1", "true", "yes", "on"}
|
||||
return bool(value)
|
||||
|
||||
|
||||
def _serialise_capex_component_entry(component: Any) -> dict[str, Any]:
|
||||
if isinstance(component, CapexComponentInput):
|
||||
raw = component.model_dump()
|
||||
elif isinstance(component, dict):
|
||||
raw = dict(component)
|
||||
else:
|
||||
raw = {
|
||||
"id": getattr(component, "id", None),
|
||||
"name": getattr(component, "name", None),
|
||||
"category": getattr(component, "category", None),
|
||||
"amount": getattr(component, "amount", None),
|
||||
"currency": getattr(component, "currency", None),
|
||||
"spend_year": getattr(component, "spend_year", None),
|
||||
"notes": getattr(component, "notes", None),
|
||||
}
|
||||
|
||||
return {
|
||||
"id": raw.get("id"),
|
||||
"name": _value_or_blank(raw.get("name")),
|
||||
"category": raw.get("category") or "equipment",
|
||||
"amount": _value_or_blank(raw.get("amount")),
|
||||
"currency": _value_or_blank(raw.get("currency")),
|
||||
"spend_year": _value_or_blank(raw.get("spend_year")),
|
||||
"notes": _value_or_blank(raw.get("notes")),
|
||||
}
|
||||
|
||||
|
||||
def _serialise_capex_parameters(parameters: Any) -> dict[str, Any]:
|
||||
if isinstance(parameters, CapexParameters):
|
||||
raw = parameters.model_dump()
|
||||
elif isinstance(parameters, dict):
|
||||
raw = dict(parameters)
|
||||
else:
|
||||
raw = {}
|
||||
|
||||
return {
|
||||
"currency_code": _value_or_blank(raw.get("currency_code")),
|
||||
"contingency_pct": _value_or_blank(raw.get("contingency_pct")),
|
||||
"discount_rate_pct": _value_or_blank(raw.get("discount_rate_pct")),
|
||||
"evaluation_horizon_years": _value_or_blank(
|
||||
raw.get("evaluation_horizon_years")
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _serialise_capex_options(options: Any) -> dict[str, Any]:
|
||||
if isinstance(options, CapexCalculationOptions):
|
||||
raw = options.model_dump()
|
||||
elif isinstance(options, dict):
|
||||
raw = dict(options)
|
||||
else:
|
||||
raw = {}
|
||||
|
||||
return {"persist": _coerce_bool(raw.get("persist", False))}
|
||||
|
||||
|
||||
def _build_capex_defaults(
|
||||
*,
|
||||
project: Project | None,
|
||||
scenario: Scenario | None,
|
||||
) -> dict[str, Any]:
|
||||
currency = ""
|
||||
if scenario and getattr(scenario, "currency", None):
|
||||
currency = str(scenario.currency).upper()
|
||||
elif project and getattr(project, "currency", None):
|
||||
currency = str(project.currency).upper()
|
||||
|
||||
discount_rate = ""
|
||||
scenario_discount = getattr(scenario, "discount_rate", None)
|
||||
if scenario_discount is not None:
|
||||
discount_rate = float(scenario_discount)
|
||||
|
||||
return {
|
||||
"components": [],
|
||||
"parameters": {
|
||||
"currency_code": currency or None,
|
||||
"contingency_pct": None,
|
||||
"discount_rate_pct": discount_rate,
|
||||
"evaluation_horizon_years": _DEFAULT_CAPEX_HORIZON_YEARS,
|
||||
},
|
||||
"options": {
|
||||
"persist": bool(scenario or project),
|
||||
},
|
||||
"currency_code": currency or None,
|
||||
"default_horizon": _DEFAULT_CAPEX_HORIZON_YEARS,
|
||||
"last_updated_at": getattr(scenario, "capex_updated_at", None),
|
||||
}
|
||||
|
||||
|
||||
def _prepare_capex_context(
|
||||
request: Request,
|
||||
*,
|
||||
project: Project | None,
|
||||
scenario: Scenario | None,
|
||||
form_data: dict[str, Any] | None = None,
|
||||
result: CapexCalculationResult | None = None,
|
||||
errors: list[str] | None = None,
|
||||
notices: list[str] | None = None,
|
||||
component_errors: list[str] | None = None,
|
||||
component_notices: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
if form_data is not None and hasattr(form_data, "model_dump"):
|
||||
form_data = form_data.model_dump() # type: ignore[assignment]
|
||||
|
||||
defaults = _build_capex_defaults(project=project, scenario=scenario)
|
||||
|
||||
raw_components: list[Any] = []
|
||||
if form_data and "components" in form_data:
|
||||
raw_components = list(form_data.get("components") or [])
|
||||
components = [
|
||||
_serialise_capex_component_entry(component) for component in raw_components
|
||||
]
|
||||
|
||||
raw_parameters = defaults["parameters"].copy()
|
||||
if form_data and form_data.get("parameters"):
|
||||
raw_parameters.update(
|
||||
_serialise_capex_parameters(form_data.get("parameters"))
|
||||
)
|
||||
parameters = _serialise_capex_parameters(raw_parameters)
|
||||
|
||||
raw_options = defaults["options"].copy()
|
||||
if form_data and form_data.get("options"):
|
||||
raw_options.update(_serialise_capex_options(form_data.get("options")))
|
||||
options = _serialise_capex_options(raw_options)
|
||||
|
||||
currency_code = parameters.get(
|
||||
"currency_code") or defaults["currency_code"]
|
||||
|
||||
return {
|
||||
"request": request,
|
||||
"project": project,
|
||||
"scenario": scenario,
|
||||
"components": components,
|
||||
"parameters": parameters,
|
||||
"options": options,
|
||||
"currency_code": currency_code,
|
||||
"category_options": _CAPEX_CATEGORY_OPTIONS,
|
||||
"default_horizon": defaults["default_horizon"],
|
||||
"last_updated_at": defaults["last_updated_at"],
|
||||
"result": result,
|
||||
"errors": errors or [],
|
||||
"notices": notices or [],
|
||||
"component_errors": component_errors or [],
|
||||
"component_notices": component_notices or [],
|
||||
"cancel_url": request.headers.get("Referer"),
|
||||
"form_action": request.url.path,
|
||||
"csrf_token": None,
|
||||
}
|
||||
|
||||
|
||||
def _format_error_location(location: tuple[Any, ...]) -> str:
|
||||
path = ""
|
||||
for part in location:
|
||||
if isinstance(part, int):
|
||||
path += f"[{part}]"
|
||||
else:
|
||||
if path:
|
||||
path += f".{part}"
|
||||
else:
|
||||
path = str(part)
|
||||
return path or "(input)"
|
||||
|
||||
|
||||
def _partition_capex_error_messages(
|
||||
errors: Sequence[Any],
|
||||
) -> tuple[list[str], list[str]]:
|
||||
general: list[str] = []
|
||||
component_specific: list[str] = []
|
||||
|
||||
for error in errors:
|
||||
if isinstance(error, dict):
|
||||
mapping = error
|
||||
else:
|
||||
try:
|
||||
mapping = dict(error)
|
||||
except TypeError:
|
||||
mapping = {}
|
||||
|
||||
location = tuple(mapping.get("loc", ()))
|
||||
message = mapping.get("msg", "Invalid value")
|
||||
formatted_location = _format_error_location(location)
|
||||
entry = f"{formatted_location} - {message}"
|
||||
if location and location[0] == "components":
|
||||
component_specific.append(entry)
|
||||
else:
|
||||
general.append(entry)
|
||||
|
||||
return general, component_specific
|
||||
|
||||
|
||||
def _capex_form_to_payload(form: FormData) -> dict[str, Any]:
|
||||
data: dict[str, Any] = {}
|
||||
components: dict[int, dict[str, Any]] = {}
|
||||
parameters: dict[str, Any] = {}
|
||||
options: dict[str, Any] = {}
|
||||
|
||||
for key, value in form.multi_items():
|
||||
normalised_value = _normalise_form_value(value)
|
||||
|
||||
if key.startswith("components["):
|
||||
try:
|
||||
index_part = key[len("components["):]
|
||||
index_str, remainder = index_part.split("]", 1)
|
||||
field = remainder.strip()[1:-1]
|
||||
index = int(index_str)
|
||||
except (ValueError, IndexError):
|
||||
continue
|
||||
entry = components.setdefault(index, {})
|
||||
entry[field] = normalised_value
|
||||
continue
|
||||
|
||||
if key.startswith("parameters["):
|
||||
field = key[len("parameters["):-1]
|
||||
parameters[field] = normalised_value
|
||||
continue
|
||||
|
||||
if key.startswith("options["):
|
||||
field = key[len("options["):-1]
|
||||
options[field] = normalised_value
|
||||
continue
|
||||
|
||||
if key == "csrf_token":
|
||||
continue
|
||||
|
||||
data[key] = normalised_value
|
||||
|
||||
if components:
|
||||
ordered = [
|
||||
components[index] for index in sorted(components.keys())
|
||||
]
|
||||
data["components"] = ordered
|
||||
|
||||
if parameters:
|
||||
data["parameters"] = parameters
|
||||
|
||||
if options:
|
||||
options["persist"] = _coerce_bool(options.get("persist"))
|
||||
data["options"] = options
|
||||
|
||||
return data
|
||||
|
||||
|
||||
async def _extract_capex_payload(request: Request) -> dict[str, Any]:
|
||||
content_type = request.headers.get("content-type", "").lower()
|
||||
if content_type.startswith("application/json"):
|
||||
body = await request.json()
|
||||
return body if isinstance(body, dict) else {}
|
||||
form = await request.form()
|
||||
return _capex_form_to_payload(form)
|
||||
|
||||
|
||||
def _prepare_default_context(
|
||||
request: Request,
|
||||
*,
|
||||
@@ -424,6 +701,205 @@ def _persist_profitability_snapshots(
|
||||
uow.project_profitability.create(project_snapshot)
|
||||
|
||||
|
||||
def _should_persist_capex(
|
||||
*,
|
||||
project: Project | None,
|
||||
scenario: Scenario | None,
|
||||
request_model: CapexCalculationRequest,
|
||||
) -> bool:
|
||||
"""Determine whether capex snapshots should be stored."""
|
||||
|
||||
persist_requested = bool(
|
||||
getattr(request_model, "options", None)
|
||||
and request_model.options.persist
|
||||
)
|
||||
return persist_requested and bool(project or scenario)
|
||||
|
||||
|
||||
def _persist_capex_snapshots(
|
||||
*,
|
||||
uow: UnitOfWork,
|
||||
project: Project | None,
|
||||
scenario: Scenario | None,
|
||||
user: User | None,
|
||||
request_model: CapexCalculationRequest,
|
||||
result: CapexCalculationResult,
|
||||
) -> None:
|
||||
if not _should_persist_capex(
|
||||
project=project,
|
||||
scenario=scenario,
|
||||
request_model=request_model,
|
||||
):
|
||||
return
|
||||
|
||||
created_by_id = getattr(user, "id", None)
|
||||
totals = result.totals
|
||||
component_count = len(result.components)
|
||||
|
||||
payload = {
|
||||
"request": request_model.model_dump(mode="json"),
|
||||
"result": result.model_dump(),
|
||||
}
|
||||
|
||||
if scenario and uow.scenario_capex:
|
||||
scenario_snapshot = ScenarioCapexSnapshot(
|
||||
scenario_id=scenario.id,
|
||||
created_by_id=created_by_id,
|
||||
calculation_source="calculations.capex",
|
||||
currency_code=result.currency,
|
||||
total_capex=float(totals.overall),
|
||||
contingency_pct=float(totals.contingency_pct),
|
||||
contingency_amount=float(totals.contingency_amount),
|
||||
total_with_contingency=float(totals.with_contingency),
|
||||
component_count=component_count,
|
||||
payload=payload,
|
||||
)
|
||||
uow.scenario_capex.create(scenario_snapshot)
|
||||
|
||||
if project and uow.project_capex:
|
||||
project_snapshot = ProjectCapexSnapshot(
|
||||
project_id=project.id,
|
||||
created_by_id=created_by_id,
|
||||
calculation_source="calculations.capex",
|
||||
currency_code=result.currency,
|
||||
total_capex=float(totals.overall),
|
||||
contingency_pct=float(totals.contingency_pct),
|
||||
contingency_amount=float(totals.contingency_amount),
|
||||
total_with_contingency=float(totals.with_contingency),
|
||||
component_count=component_count,
|
||||
payload=payload,
|
||||
)
|
||||
uow.project_capex.create(project_snapshot)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/capex",
|
||||
response_class=HTMLResponse,
|
||||
name="calculations.capex_form",
|
||||
)
|
||||
def capex_form(
|
||||
request: Request,
|
||||
_: User = Depends(require_authenticated_user),
|
||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||
project_id: int | None = Query(
|
||||
None, description="Optional project identifier"),
|
||||
scenario_id: int | None = Query(
|
||||
None, description="Optional scenario identifier"),
|
||||
) -> HTMLResponse:
|
||||
"""Render the initial capex planner template with defaults."""
|
||||
|
||||
project, scenario = _load_project_and_scenario(
|
||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||
)
|
||||
context = _prepare_capex_context(
|
||||
request,
|
||||
project=project,
|
||||
scenario=scenario,
|
||||
)
|
||||
return templates.TemplateResponse("scenarios/capex.html", context)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/capex",
|
||||
name="calculations.capex_submit",
|
||||
)
|
||||
async def capex_submit(
|
||||
request: Request,
|
||||
_: User = Depends(require_authenticated_user),
|
||||
uow: UnitOfWork = Depends(get_unit_of_work),
|
||||
project_id: int | None = Query(
|
||||
None, description="Optional project identifier"),
|
||||
scenario_id: int | None = Query(
|
||||
None, description="Optional scenario identifier"),
|
||||
) -> Response:
|
||||
"""Process capex submissions and return aggregated results."""
|
||||
|
||||
wants_json = _is_json_request(request)
|
||||
payload_data = await _extract_capex_payload(request)
|
||||
|
||||
try:
|
||||
request_model = CapexCalculationRequest.model_validate(payload_data)
|
||||
result = calculate_initial_capex(request_model)
|
||||
except ValidationError as exc:
|
||||
if wants_json:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
content={"errors": exc.errors()},
|
||||
)
|
||||
|
||||
project, scenario = _load_project_and_scenario(
|
||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||
)
|
||||
general_errors, component_errors = _partition_capex_error_messages(
|
||||
exc.errors()
|
||||
)
|
||||
context = _prepare_capex_context(
|
||||
request,
|
||||
project=project,
|
||||
scenario=scenario,
|
||||
form_data=payload_data,
|
||||
errors=general_errors,
|
||||
component_errors=component_errors,
|
||||
)
|
||||
return templates.TemplateResponse(
|
||||
"scenarios/capex.html",
|
||||
context,
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
)
|
||||
except CapexValidationError as exc:
|
||||
if wants_json:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
content={
|
||||
"errors": list(exc.field_errors or []),
|
||||
"message": exc.message,
|
||||
},
|
||||
)
|
||||
|
||||
project, scenario = _load_project_and_scenario(
|
||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||
)
|
||||
errors = list(exc.field_errors or []) or [exc.message]
|
||||
context = _prepare_capex_context(
|
||||
request,
|
||||
project=project,
|
||||
scenario=scenario,
|
||||
form_data=payload_data,
|
||||
errors=errors,
|
||||
)
|
||||
return templates.TemplateResponse(
|
||||
"scenarios/capex.html",
|
||||
context,
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
)
|
||||
|
||||
project, scenario = _load_project_and_scenario(
|
||||
uow=uow, project_id=project_id, scenario_id=scenario_id
|
||||
)
|
||||
|
||||
if wants_json:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_200_OK,
|
||||
content=result.model_dump(),
|
||||
)
|
||||
|
||||
context = _prepare_capex_context(
|
||||
request,
|
||||
project=project,
|
||||
scenario=scenario,
|
||||
form_data=request_model.model_dump(mode="json"),
|
||||
result=result,
|
||||
)
|
||||
notices = _list_from_context(context, "notices")
|
||||
notices.append("Initial capex calculation completed successfully.")
|
||||
|
||||
return templates.TemplateResponse(
|
||||
"scenarios/capex.html",
|
||||
context,
|
||||
status_code=status.HTTP_200_OK,
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/profitability",
|
||||
response_class=HTMLResponse,
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime
|
||||
from datetime import date
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
@@ -24,96 +24,11 @@ from services.reporting import (
|
||||
validate_percentiles,
|
||||
)
|
||||
from services.unit_of_work import UnitOfWork
|
||||
from routes.template_filters import register_common_filters
|
||||
|
||||
router = APIRouter(prefix="/reports", tags=["Reports"])
|
||||
templates = Jinja2Templates(directory="templates")
|
||||
|
||||
# Add custom Jinja2 filters
|
||||
|
||||
|
||||
def format_datetime(value):
|
||||
"""Format a datetime object for display in templates."""
|
||||
if not isinstance(value, datetime):
|
||||
return ""
|
||||
if value.tzinfo is None:
|
||||
# Assume UTC if no timezone
|
||||
from datetime import timezone
|
||||
value = value.replace(tzinfo=timezone.utc)
|
||||
# Format as readable date/time
|
||||
return value.strftime("%Y-%m-%d %H:%M UTC")
|
||||
|
||||
|
||||
def currency_display(value, currency_code):
|
||||
"""Format a numeric value with currency symbol/code."""
|
||||
if value is None:
|
||||
return "—"
|
||||
|
||||
# Format the number
|
||||
if isinstance(value, (int, float)):
|
||||
formatted_value = f"{value:,.2f}"
|
||||
else:
|
||||
formatted_value = str(value)
|
||||
|
||||
# Add currency code
|
||||
if currency_code:
|
||||
return f"{currency_code} {formatted_value}"
|
||||
return formatted_value
|
||||
|
||||
|
||||
def format_metric(value, metric_name, currency_code=None):
|
||||
"""Format metric values appropriately based on metric type."""
|
||||
if value is None:
|
||||
return "—"
|
||||
|
||||
# For currency-related metrics, use currency formatting
|
||||
currency_metrics = {'npv', 'inflows', 'outflows',
|
||||
'net', 'total_inflows', 'total_outflows', 'total_net'}
|
||||
if metric_name in currency_metrics and currency_code:
|
||||
return currency_display(value, currency_code)
|
||||
|
||||
# For percentage metrics
|
||||
percentage_metrics = {'irr', 'payback_period'}
|
||||
if metric_name in percentage_metrics:
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:.2f}%"
|
||||
return f"{value}%"
|
||||
|
||||
# Default numeric formatting
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:,.2f}"
|
||||
|
||||
return str(value)
|
||||
|
||||
|
||||
def percentage_display(value):
|
||||
"""Format a value as a percentage."""
|
||||
if value is None:
|
||||
return "—"
|
||||
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:.2f}%"
|
||||
|
||||
return f"{value}%"
|
||||
|
||||
|
||||
def period_display(value):
|
||||
"""Format a period value (like payback period)."""
|
||||
if value is None:
|
||||
return "—"
|
||||
|
||||
if isinstance(value, (int, float)):
|
||||
if value == int(value):
|
||||
return f"{int(value)} years"
|
||||
return f"{value:.1f} years"
|
||||
|
||||
return str(value)
|
||||
|
||||
|
||||
templates.env.filters['format_datetime'] = format_datetime
|
||||
templates.env.filters['currency_display'] = currency_display
|
||||
templates.env.filters['format_metric'] = format_metric
|
||||
templates.env.filters['percentage_display'] = percentage_display
|
||||
templates.env.filters['period_display'] = period_display
|
||||
register_common_filters(templates)
|
||||
|
||||
READ_ROLES = ("viewer", "analyst", "project_manager", "admin")
|
||||
MANAGE_ROLES = ("project_manager", "admin")
|
||||
|
||||
95
routes/template_filters.py
Normal file
95
routes/template_filters.py
Normal file
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
|
||||
def format_datetime(value: Any) -> str:
|
||||
"""Render datetime values consistently for templates."""
|
||||
if not isinstance(value, datetime):
|
||||
return ""
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=timezone.utc)
|
||||
return value.strftime("%Y-%m-%d %H:%M UTC")
|
||||
|
||||
|
||||
def currency_display(value: Any, currency_code: str | None) -> str:
|
||||
"""Format numeric values with currency context."""
|
||||
if value is None:
|
||||
return "—"
|
||||
if isinstance(value, (int, float)):
|
||||
formatted_value = f"{value:,.2f}"
|
||||
else:
|
||||
formatted_value = str(value)
|
||||
if currency_code:
|
||||
return f"{currency_code} {formatted_value}"
|
||||
return formatted_value
|
||||
|
||||
|
||||
def format_metric(value: Any, metric_name: str, currency_code: str | None = None) -> str:
|
||||
"""Format metrics according to their semantic type."""
|
||||
if value is None:
|
||||
return "—"
|
||||
|
||||
currency_metrics = {
|
||||
"npv",
|
||||
"inflows",
|
||||
"outflows",
|
||||
"net",
|
||||
"total_inflows",
|
||||
"total_outflows",
|
||||
"total_net",
|
||||
}
|
||||
if metric_name in currency_metrics and currency_code:
|
||||
return currency_display(value, currency_code)
|
||||
|
||||
percentage_metrics = {"irr", "payback_period"}
|
||||
if metric_name in percentage_metrics:
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:.2f}%"
|
||||
return f"{value}%"
|
||||
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:,.2f}"
|
||||
|
||||
return str(value)
|
||||
|
||||
|
||||
def percentage_display(value: Any) -> str:
|
||||
"""Format numeric values as percentages."""
|
||||
if value is None:
|
||||
return "—"
|
||||
if isinstance(value, (int, float)):
|
||||
return f"{value:.2f}%"
|
||||
return f"{value}%"
|
||||
|
||||
|
||||
def period_display(value: Any) -> str:
|
||||
"""Format period values in years."""
|
||||
if value is None:
|
||||
return "—"
|
||||
if isinstance(value, (int, float)):
|
||||
if value == int(value):
|
||||
return f"{int(value)} years"
|
||||
return f"{value:.1f} years"
|
||||
return str(value)
|
||||
|
||||
|
||||
def register_common_filters(templates: Jinja2Templates) -> None:
|
||||
templates.env.filters["format_datetime"] = format_datetime
|
||||
templates.env.filters["currency_display"] = currency_display
|
||||
templates.env.filters["format_metric"] = format_metric
|
||||
templates.env.filters["percentage_display"] = percentage_display
|
||||
templates.env.filters["period_display"] = period_display
|
||||
|
||||
|
||||
__all__ = [
|
||||
"format_datetime",
|
||||
"currency_display",
|
||||
"format_metric",
|
||||
"percentage_display",
|
||||
"period_display",
|
||||
"register_common_filters",
|
||||
]
|
||||
@@ -97,6 +97,106 @@ class ProfitabilityCalculationResult(BaseModel):
|
||||
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
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ImpurityInput",
|
||||
"ProfitabilityCalculationRequest",
|
||||
@@ -104,5 +204,13 @@ __all__ = [
|
||||
"ProfitabilityMetrics",
|
||||
"CashFlowEntry",
|
||||
"ProfitabilityCalculationResult",
|
||||
"CapexComponentInput",
|
||||
"CapexParameters",
|
||||
"CapexCalculationOptions",
|
||||
"CapexCalculationRequest",
|
||||
"CapexCategoryBreakdown",
|
||||
"CapexTotals",
|
||||
"CapexTimelineEntry",
|
||||
"CapexCalculationResult",
|
||||
"ValidationError",
|
||||
]
|
||||
|
||||
@@ -2,8 +2,10 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
from services.currency import CurrencyValidationError, normalise_currency
|
||||
from services.exceptions import ProfitabilityValidationError
|
||||
from services.exceptions import CapexValidationError, ProfitabilityValidationError
|
||||
from services.financial import (
|
||||
CashFlow,
|
||||
ConvergenceError,
|
||||
@@ -14,6 +16,13 @@ from services.financial import (
|
||||
)
|
||||
from services.pricing import PricingInput, PricingMetadata, PricingResult, calculate_pricing
|
||||
from schemas.calculations import (
|
||||
CapexCalculationRequest,
|
||||
CapexCalculationResult,
|
||||
CapexCategoryBreakdown,
|
||||
CapexComponentInput,
|
||||
CapexParameters,
|
||||
CapexTotals,
|
||||
CapexTimelineEntry,
|
||||
CashFlowEntry,
|
||||
ProfitabilityCalculationRequest,
|
||||
ProfitabilityCalculationResult,
|
||||
@@ -202,4 +211,125 @@ def calculate_profitability(
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["calculate_profitability"]
|
||||
def calculate_initial_capex(
|
||||
request: CapexCalculationRequest,
|
||||
) -> CapexCalculationResult:
|
||||
"""Aggregate capex components into totals and timelines."""
|
||||
|
||||
if not request.components:
|
||||
raise CapexValidationError(
|
||||
"At least one capex component is required for calculation.",
|
||||
["components"],
|
||||
)
|
||||
|
||||
parameters = request.parameters
|
||||
|
||||
base_currency = parameters.currency_code
|
||||
if base_currency:
|
||||
try:
|
||||
base_currency = normalise_currency(base_currency)
|
||||
except CurrencyValidationError as exc:
|
||||
raise CapexValidationError(
|
||||
str(exc), ["parameters.currency_code"]
|
||||
) from exc
|
||||
|
||||
overall = 0.0
|
||||
category_totals: dict[str, float] = defaultdict(float)
|
||||
timeline_totals: dict[int, float] = defaultdict(float)
|
||||
normalised_components: list[CapexComponentInput] = []
|
||||
|
||||
for index, component in enumerate(request.components):
|
||||
amount = float(component.amount)
|
||||
overall += amount
|
||||
|
||||
category_totals[component.category] += amount
|
||||
|
||||
spend_year = component.spend_year or 0
|
||||
timeline_totals[spend_year] += amount
|
||||
|
||||
component_currency = component.currency
|
||||
if component_currency:
|
||||
try:
|
||||
component_currency = normalise_currency(component_currency)
|
||||
except CurrencyValidationError as exc:
|
||||
raise CapexValidationError(
|
||||
str(exc), [f"components[{index}].currency"]
|
||||
) from exc
|
||||
|
||||
if base_currency is None and component_currency:
|
||||
base_currency = component_currency
|
||||
elif (
|
||||
base_currency is not None
|
||||
and component_currency is not None
|
||||
and component_currency != base_currency
|
||||
):
|
||||
raise CapexValidationError(
|
||||
(
|
||||
"Component currency does not match the global currency. "
|
||||
f"Expected {base_currency}, got {component_currency}."
|
||||
),
|
||||
[f"components[{index}].currency"],
|
||||
)
|
||||
|
||||
normalised_components.append(
|
||||
CapexComponentInput(
|
||||
id=component.id,
|
||||
name=component.name,
|
||||
category=component.category,
|
||||
amount=amount,
|
||||
currency=component_currency,
|
||||
spend_year=component.spend_year,
|
||||
notes=component.notes,
|
||||
)
|
||||
)
|
||||
|
||||
contingency_pct = float(parameters.contingency_pct or 0.0)
|
||||
contingency_amount = overall * (contingency_pct / 100.0)
|
||||
grand_total = overall + contingency_amount
|
||||
|
||||
category_breakdowns: list[CapexCategoryBreakdown] = []
|
||||
if category_totals:
|
||||
for category, total in sorted(category_totals.items()):
|
||||
share = (total / overall * 100.0) if overall else None
|
||||
category_breakdowns.append(
|
||||
CapexCategoryBreakdown(
|
||||
category=category,
|
||||
amount=total,
|
||||
share=share,
|
||||
)
|
||||
)
|
||||
|
||||
cumulative = 0.0
|
||||
timeline_entries: list[CapexTimelineEntry] = []
|
||||
for year, spend in sorted(timeline_totals.items()):
|
||||
cumulative += spend
|
||||
timeline_entries.append(
|
||||
CapexTimelineEntry(year=year, spend=spend, cumulative=cumulative)
|
||||
)
|
||||
|
||||
try:
|
||||
currency = normalise_currency(base_currency) if base_currency else None
|
||||
except CurrencyValidationError as exc:
|
||||
raise CapexValidationError(
|
||||
str(exc), ["parameters.currency_code"]
|
||||
) from exc
|
||||
|
||||
totals = CapexTotals(
|
||||
overall=overall,
|
||||
contingency_pct=contingency_pct,
|
||||
contingency_amount=contingency_amount,
|
||||
with_contingency=grand_total,
|
||||
by_category=category_breakdowns,
|
||||
)
|
||||
|
||||
return CapexCalculationResult(
|
||||
totals=totals,
|
||||
timeline=timeline_entries,
|
||||
components=normalised_components,
|
||||
parameters=parameters,
|
||||
options=request.options,
|
||||
currency=currency,
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["calculate_profitability", "calculate_initial_capex"]
|
||||
|
||||
@@ -37,3 +37,14 @@ class ProfitabilityValidationError(Exception):
|
||||
|
||||
def __str__(self) -> str: # pragma: no cover - mirrors message for logging
|
||||
return self.message
|
||||
|
||||
|
||||
@dataclass(eq=False)
|
||||
class CapexValidationError(Exception):
|
||||
"""Raised when capex calculation inputs fail domain validation."""
|
||||
|
||||
message: str
|
||||
field_errors: Sequence[str] | None = None
|
||||
|
||||
def __str__(self) -> str: # pragma: no cover - mirrors message for logging
|
||||
return self.message
|
||||
|
||||
@@ -15,9 +15,11 @@ from models import (
|
||||
PricingImpuritySettings,
|
||||
PricingMetalSettings,
|
||||
PricingSettings,
|
||||
ProjectCapexSnapshot,
|
||||
ProjectProfitability,
|
||||
Role,
|
||||
Scenario,
|
||||
ScenarioCapexSnapshot,
|
||||
ScenarioProfitability,
|
||||
ScenarioStatus,
|
||||
SimulationParameter,
|
||||
@@ -469,6 +471,106 @@ class ScenarioProfitabilityRepository:
|
||||
self.session.delete(entity)
|
||||
|
||||
|
||||
class ProjectCapexRepository:
|
||||
"""Persistence operations for project-level capex snapshots."""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self.session = session
|
||||
|
||||
def create(self, snapshot: ProjectCapexSnapshot) -> ProjectCapexSnapshot:
|
||||
self.session.add(snapshot)
|
||||
self.session.flush()
|
||||
return snapshot
|
||||
|
||||
def list_for_project(
|
||||
self,
|
||||
project_id: int,
|
||||
*,
|
||||
limit: int | None = None,
|
||||
) -> Sequence[ProjectCapexSnapshot]:
|
||||
stmt = (
|
||||
select(ProjectCapexSnapshot)
|
||||
.where(ProjectCapexSnapshot.project_id == project_id)
|
||||
.order_by(ProjectCapexSnapshot.calculated_at.desc())
|
||||
)
|
||||
if limit is not None:
|
||||
stmt = stmt.limit(limit)
|
||||
return self.session.execute(stmt).scalars().all()
|
||||
|
||||
def latest_for_project(
|
||||
self,
|
||||
project_id: int,
|
||||
) -> ProjectCapexSnapshot | None:
|
||||
stmt = (
|
||||
select(ProjectCapexSnapshot)
|
||||
.where(ProjectCapexSnapshot.project_id == project_id)
|
||||
.order_by(ProjectCapexSnapshot.calculated_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
return self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
def delete(self, snapshot_id: int) -> None:
|
||||
stmt = select(ProjectCapexSnapshot).where(
|
||||
ProjectCapexSnapshot.id == snapshot_id
|
||||
)
|
||||
entity = self.session.execute(stmt).scalar_one_or_none()
|
||||
if entity is None:
|
||||
raise EntityNotFoundError(
|
||||
f"Project capex snapshot {snapshot_id} not found"
|
||||
)
|
||||
self.session.delete(entity)
|
||||
|
||||
|
||||
class ScenarioCapexRepository:
|
||||
"""Persistence operations for scenario-level capex snapshots."""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self.session = session
|
||||
|
||||
def create(self, snapshot: ScenarioCapexSnapshot) -> ScenarioCapexSnapshot:
|
||||
self.session.add(snapshot)
|
||||
self.session.flush()
|
||||
return snapshot
|
||||
|
||||
def list_for_scenario(
|
||||
self,
|
||||
scenario_id: int,
|
||||
*,
|
||||
limit: int | None = None,
|
||||
) -> Sequence[ScenarioCapexSnapshot]:
|
||||
stmt = (
|
||||
select(ScenarioCapexSnapshot)
|
||||
.where(ScenarioCapexSnapshot.scenario_id == scenario_id)
|
||||
.order_by(ScenarioCapexSnapshot.calculated_at.desc())
|
||||
)
|
||||
if limit is not None:
|
||||
stmt = stmt.limit(limit)
|
||||
return self.session.execute(stmt).scalars().all()
|
||||
|
||||
def latest_for_scenario(
|
||||
self,
|
||||
scenario_id: int,
|
||||
) -> ScenarioCapexSnapshot | None:
|
||||
stmt = (
|
||||
select(ScenarioCapexSnapshot)
|
||||
.where(ScenarioCapexSnapshot.scenario_id == scenario_id)
|
||||
.order_by(ScenarioCapexSnapshot.calculated_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
return self.session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
def delete(self, snapshot_id: int) -> None:
|
||||
stmt = select(ScenarioCapexSnapshot).where(
|
||||
ScenarioCapexSnapshot.id == snapshot_id
|
||||
)
|
||||
entity = self.session.execute(stmt).scalar_one_or_none()
|
||||
if entity is None:
|
||||
raise EntityNotFoundError(
|
||||
f"Scenario capex snapshot {snapshot_id} not found"
|
||||
)
|
||||
self.session.delete(entity)
|
||||
|
||||
|
||||
class FinancialInputRepository:
|
||||
"""Persistence operations for FinancialInput entities."""
|
||||
|
||||
|
||||
@@ -14,9 +14,11 @@ from services.repositories import (
|
||||
PricingSettingsSeedResult,
|
||||
ProjectRepository,
|
||||
ProjectProfitabilityRepository,
|
||||
ProjectCapexRepository,
|
||||
RoleRepository,
|
||||
ScenarioRepository,
|
||||
ScenarioProfitabilityRepository,
|
||||
ScenarioCapexRepository,
|
||||
SimulationParameterRepository,
|
||||
UserRepository,
|
||||
ensure_admin_user as ensure_admin_user_record,
|
||||
@@ -39,7 +41,9 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]):
|
||||
self.financial_inputs: FinancialInputRepository | None = None
|
||||
self.simulation_parameters: SimulationParameterRepository | None = None
|
||||
self.project_profitability: ProjectProfitabilityRepository | None = None
|
||||
self.project_capex: ProjectCapexRepository | None = None
|
||||
self.scenario_profitability: ScenarioProfitabilityRepository | None = None
|
||||
self.scenario_capex: ScenarioCapexRepository | None = None
|
||||
self.users: UserRepository | None = None
|
||||
self.roles: RoleRepository | None = None
|
||||
self.pricing_settings: PricingSettingsRepository | None = None
|
||||
@@ -53,9 +57,11 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]):
|
||||
self.session)
|
||||
self.project_profitability = ProjectProfitabilityRepository(
|
||||
self.session)
|
||||
self.project_capex = ProjectCapexRepository(self.session)
|
||||
self.scenario_profitability = ScenarioProfitabilityRepository(
|
||||
self.session
|
||||
)
|
||||
self.scenario_capex = ScenarioCapexRepository(self.session)
|
||||
self.users = UserRepository(self.session)
|
||||
self.roles = RoleRepository(self.session)
|
||||
self.pricing_settings = PricingSettingsRepository(self.session)
|
||||
@@ -75,7 +81,9 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]):
|
||||
self.financial_inputs = None
|
||||
self.simulation_parameters = None
|
||||
self.project_profitability = None
|
||||
self.project_capex = None
|
||||
self.scenario_profitability = None
|
||||
self.scenario_capex = None
|
||||
self.users = None
|
||||
self.roles = None
|
||||
self.pricing_settings = None
|
||||
|
||||
264
templates/scenarios/capex.html
Normal file
264
templates/scenarios/capex.html
Normal file
@@ -0,0 +1,264 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}Initial Capex Planner · CalMiner{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<nav class="breadcrumb">
|
||||
<a href="{{ url_for('projects.project_list_page') }}">Projects</a>
|
||||
{% if project %}
|
||||
<a href="{{ url_for('projects.view_project', project_id=project.id) }}">{{ project.name }}</a>
|
||||
{% endif %}
|
||||
{% if scenario %}
|
||||
<a href="{{ url_for('scenarios.view_scenario', scenario_id=scenario.id) }}">{{ scenario.name }}</a>
|
||||
{% endif %}
|
||||
<span aria-current="page">Initial Capex</span>
|
||||
</nav>
|
||||
|
||||
<header class="page-header">
|
||||
<div>
|
||||
<h1>Initial Capex Planner</h1>
|
||||
<p class="text-muted">Capture upfront capital requirements and review categorized totals.</p>
|
||||
</div>
|
||||
<div class="header-actions">
|
||||
{% if cancel_url %}
|
||||
<a class="btn" href="{{ cancel_url }}">Cancel</a>
|
||||
{% endif %}
|
||||
<button class="btn primary" type="submit" form="capex-form">Save & Calculate</button>
|
||||
</div>
|
||||
</header>
|
||||
|
||||
{% if errors %}
|
||||
<div class="alert alert-error">
|
||||
<h2 class="sr-only">Submission errors</h2>
|
||||
<ul>
|
||||
{% for message in errors %}
|
||||
<li>{{ message }}</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if notices %}
|
||||
<div class="alert alert-info">
|
||||
<ul>
|
||||
{% for message in notices %}
|
||||
<li>{{ message }}</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<form id="capex-form" class="form scenario-form" method="post" action="{{ form_action }}">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token or '' }}" />
|
||||
<input type="hidden" name="options[persist]" value="{{ '1' if options is defined and options and options.persist else '' }}" />
|
||||
|
||||
<div class="layout-two-column stackable">
|
||||
<section class="panel">
|
||||
<header class="section-header">
|
||||
<h2>Capex Components</h2>
|
||||
<p class="section-subtitle">Break down initial capital items with category, amount, and timing.</p>
|
||||
</header>
|
||||
|
||||
<div class="table-actions">
|
||||
<button class="btn secondary" type="button" data-action="add-component">Add Component</button>
|
||||
</div>
|
||||
|
||||
{% if component_errors is defined and component_errors %}
|
||||
<div class="alert alert-error slim" role="alert" aria-live="polite">
|
||||
<h3 class="sr-only">Component issues</h3>
|
||||
<ul>
|
||||
{% for message in component_errors %}
|
||||
<li>{{ message }}</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if component_notices is defined and component_notices %}
|
||||
<div class="alert alert-info slim" role="status" aria-live="polite">
|
||||
<ul>
|
||||
{% for message in component_notices %}
|
||||
<li>{{ message }}</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<div class="table-responsive horizontal-scroll">
|
||||
<table class="input-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th scope="col">Category</th>
|
||||
<th scope="col">Component</th>
|
||||
<th scope="col">Amount</th>
|
||||
<th scope="col">Currency</th>
|
||||
<th scope="col">Spend Year</th>
|
||||
<th scope="col">Notes</th>
|
||||
<th class="sr-only" scope="col">Actions</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% set component_entries = (components if components is defined and components else [{}]) %}
|
||||
{% for component in component_entries %}
|
||||
<tr data-row-index="{{ loop.index0 }}">
|
||||
<td>
|
||||
<input type="hidden" name="components[{{ loop.index0 }}][id]" value="{{ component.id or '' }}" />
|
||||
<select name="components[{{ loop.index0 }}][category]">
|
||||
{% for option in category_options %}
|
||||
<option value="{{ option.value }}" {% if component.category == option.value %}selected{% endif %}>{{ option.label }}</option>
|
||||
{% endfor %}
|
||||
</select>
|
||||
</td>
|
||||
<td>
|
||||
<input type="text" name="components[{{ loop.index0 }}][name]" value="{{ component.name or '' }}" />
|
||||
</td>
|
||||
<td>
|
||||
<input type="number" min="0" step="0.01" name="components[{{ loop.index0 }}][amount]" value="{{ component.amount or '' }}" />
|
||||
</td>
|
||||
<td>
|
||||
<input type="text" maxlength="3" name="components[{{ loop.index0 }}][currency]" value="{{ component.currency or currency_code or (scenario.currency if scenario else '') or (project.currency if project else '') }}" />
|
||||
</td>
|
||||
<td>
|
||||
<input type="number" min="0" step="1" name="components[{{ loop.index0 }}][spend_year]" value="{{ component.spend_year or '' }}" />
|
||||
</td>
|
||||
<td>
|
||||
<input type="text" name="components[{{ loop.index0 }}][notes]" value="{{ component.notes or '' }}" />
|
||||
</td>
|
||||
<td class="row-actions">
|
||||
<button class="btn link" type="button" data-action="remove-component">Remove</button>
|
||||
</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
<p class="muted">Add rows for each capital item. Amounts should reflect pre-contingency values.</p>
|
||||
</section>
|
||||
|
||||
<aside class="panel">
|
||||
<header class="section-header">
|
||||
<h2>Global Parameters</h2>
|
||||
<p class="section-subtitle">Configure defaults applied across components.</p>
|
||||
</header>
|
||||
|
||||
<div class="form-grid">
|
||||
<div class="form-group">
|
||||
<label for="capex_currency_code">Default Currency</label>
|
||||
<input id="capex_currency_code" name="parameters[currency_code]" type="text" maxlength="3" value="{{ (parameters.currency_code if parameters is defined and parameters else None) or (currency_code if currency_code is defined else None) or (scenario.currency if scenario else None) or (project.currency if project else '') }}" />
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label for="contingency_pct">Contingency (%)</label>
|
||||
<input id="contingency_pct" name="parameters[contingency_pct]" type="number" min="0" max="100" step="0.01" value="{{ (parameters.contingency_pct if parameters is defined and parameters else '') }}" />
|
||||
<p class="field-help">Applied across component totals.</p>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label for="discount_rate">Discount Rate (%)</label>
|
||||
<input id="discount_rate" name="parameters[discount_rate_pct]" type="number" min="0" max="100" step="0.01" value="{{ (parameters.discount_rate_pct if parameters is defined and parameters else None) or (scenario.discount_rate if scenario else '') }}" />
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label for="evaluation_horizon_years">Evaluation Horizon (years)</label>
|
||||
<input id="evaluation_horizon_years" name="parameters[evaluation_horizon_years]" type="number" min="1" step="1" value="{{ (parameters.evaluation_horizon_years if parameters is defined and parameters else None) or (default_horizon if default_horizon is defined else '') }}" />
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="panel">
|
||||
<h3>Assumptions</h3>
|
||||
<dl class="definition-list">
|
||||
<div>
|
||||
<dt>Categories Configured</dt>
|
||||
<dd>{{ category_options | length }}</dd>
|
||||
</div>
|
||||
<div>
|
||||
<dt>Last Updated</dt>
|
||||
<dd>{{ last_updated_at or '—' }}</dd>
|
||||
</div>
|
||||
</dl>
|
||||
<p class="muted">Defaults reflect scenario preferences. Adjust before calculating.</p>
|
||||
</div>
|
||||
</aside>
|
||||
</div>
|
||||
</form>
|
||||
|
||||
<section class="report-section">
|
||||
<header class="section-header">
|
||||
<h2>Capex Summary</h2>
|
||||
<p class="section-subtitle">Calculated totals and categorized breakdowns.</p>
|
||||
</header>
|
||||
|
||||
{% if result %}
|
||||
<div class="report-grid">
|
||||
<article class="report-card">
|
||||
<h3>Total Initial Capex</h3>
|
||||
<p class="metric">
|
||||
<strong>{{ result.totals.overall | currency_display(result.currency) }}</strong>
|
||||
</p>
|
||||
</article>
|
||||
<article class="report-card">
|
||||
<h3>Contingency Applied</h3>
|
||||
<p class="metric">
|
||||
<strong>{{ result.totals.contingency_amount | currency_display(result.currency) }}</strong>
|
||||
</p>
|
||||
</article>
|
||||
<article class="report-card">
|
||||
<h3>Grand Total</h3>
|
||||
<p class="metric">
|
||||
<strong>{{ result.totals.with_contingency | currency_display(result.currency) }}</strong>
|
||||
</p>
|
||||
</article>
|
||||
</div>
|
||||
|
||||
{% if result.totals.by_category %}
|
||||
<table class="metrics-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th scope="col">Category</th>
|
||||
<th scope="col">Amount</th>
|
||||
<th scope="col">Share</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for row in result.totals.by_category %}
|
||||
<tr>
|
||||
<th scope="row">{{ row.category }}</th>
|
||||
<td>{{ row.amount | currency_display(result.currency) }}</td>
|
||||
<td>{{ row.share | percentage_display }}</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
{% endif %}
|
||||
|
||||
{% if result.timeline %}
|
||||
<table class="metrics-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th scope="col">Year</th>
|
||||
<th scope="col">Spend</th>
|
||||
<th scope="col">Cumulative</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for entry in result.timeline %}
|
||||
<tr>
|
||||
<th scope="row">{{ entry.year }}</th>
|
||||
<td>{{ entry.spend | currency_display(result.currency) }}</td>
|
||||
<td>{{ entry.cumulative | currency_display(result.currency) }}</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
{% endif %}
|
||||
{% else %}
|
||||
<p class="muted">Provide component details and calculate to see capex totals.</p>
|
||||
{% endif %}
|
||||
</section>
|
||||
|
||||
<section class="report-section">
|
||||
<header class="section-header">
|
||||
<h2>Visualisations</h2>
|
||||
<p class="section-subtitle">Charts render after calculations complete.</p>
|
||||
</header>
|
||||
<div id="capex-category-chart" class="chart-container"></div>
|
||||
<div id="capex-timeline-chart" class="chart-container"></div>
|
||||
</section>
|
||||
{% endblock %}
|
||||
@@ -17,6 +17,7 @@ from dependencies import get_auth_session, get_import_ingestion_service, get_uni
|
||||
from models import User
|
||||
from routes.auth import router as auth_router
|
||||
from routes.dashboard import router as dashboard_router
|
||||
from routes.calculations import router as calculations_router
|
||||
from routes.projects import router as projects_router
|
||||
from routes.scenarios import router as scenarios_router
|
||||
from routes.imports import router as imports_router
|
||||
@@ -57,6 +58,7 @@ def app(session_factory: sessionmaker) -> FastAPI:
|
||||
application = FastAPI()
|
||||
application.include_router(auth_router)
|
||||
application.include_router(dashboard_router)
|
||||
application.include_router(calculations_router)
|
||||
application.include_router(projects_router)
|
||||
application.include_router(scenarios_router)
|
||||
application.include_router(imports_router)
|
||||
|
||||
123
tests/integration/test_capex_calculations.py
Normal file
123
tests/integration/test_capex_calculations.py
Normal file
@@ -0,0 +1,123 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
def _create_project(client: TestClient, name: str) -> int:
|
||||
response = client.post(
|
||||
"/projects",
|
||||
json={
|
||||
"name": name,
|
||||
"location": "Nevada",
|
||||
"operation_type": "open_pit",
|
||||
"description": "Project for capex testing",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
return response.json()["id"]
|
||||
|
||||
|
||||
def _create_scenario(client: TestClient, project_id: int, name: str) -> int:
|
||||
response = client.post(
|
||||
f"/projects/{project_id}/scenarios",
|
||||
json={
|
||||
"name": name,
|
||||
"description": "Capex scenario",
|
||||
"status": "draft",
|
||||
"currency": "usd",
|
||||
"primary_resource": "diesel",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
return response.json()["id"]
|
||||
|
||||
|
||||
def test_capex_calculation_html_flow(client: TestClient) -> None:
|
||||
project_id = _create_project(client, "Capex HTML Project")
|
||||
scenario_id = _create_scenario(client, project_id, "Capex HTML Scenario")
|
||||
|
||||
form_page = client.get(
|
||||
f"/calculations/capex?project_id={project_id}&scenario_id={scenario_id}"
|
||||
)
|
||||
assert form_page.status_code == 200
|
||||
assert "Initial Capex Planner" in form_page.text
|
||||
|
||||
response = client.post(
|
||||
f"/calculations/capex?project_id={project_id}&scenario_id={scenario_id}",
|
||||
data={
|
||||
"components[0][name]": "Crusher",
|
||||
"components[0][category]": "equipment",
|
||||
"components[0][amount]": "1200000",
|
||||
"components[0][currency]": "USD",
|
||||
"components[0][spend_year]": "0",
|
||||
"components[1][name]": "Conveyor",
|
||||
"components[1][category]": "equipment",
|
||||
"components[1][amount]": "800000",
|
||||
"components[1][currency]": "USD",
|
||||
"components[1][spend_year]": "1",
|
||||
"parameters[currency_code]": "USD",
|
||||
"parameters[contingency_pct]": "5",
|
||||
"parameters[discount_rate_pct]": "8",
|
||||
"parameters[evaluation_horizon_years]": "5",
|
||||
"options[persist]": "1",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert "Initial capex calculation completed successfully." in response.text
|
||||
assert "Capex Summary" in response.text
|
||||
assert "$1,200,000.00" in response.text or "1,200,000" in response.text
|
||||
assert "USD" in response.text
|
||||
|
||||
|
||||
def test_capex_calculation_json_flow(client: TestClient) -> None:
|
||||
project_id = _create_project(client, "Capex JSON Project")
|
||||
scenario_id = _create_scenario(client, project_id, "Capex JSON Scenario")
|
||||
|
||||
payload = {
|
||||
"components": [
|
||||
{
|
||||
"name": "Camp",
|
||||
"category": "infrastructure",
|
||||
"amount": 600000,
|
||||
"currency": "USD",
|
||||
"spend_year": 0,
|
||||
},
|
||||
{
|
||||
"name": "Power",
|
||||
"category": "infrastructure",
|
||||
"amount": 400000,
|
||||
"currency": "USD",
|
||||
"spend_year": 1,
|
||||
},
|
||||
],
|
||||
"parameters": {
|
||||
"currency_code": "USD",
|
||||
"contingency_pct": 12.5,
|
||||
"discount_rate_pct": 6.5,
|
||||
"evaluation_horizon_years": 4,
|
||||
},
|
||||
"options": {"persist": True},
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
f"/calculations/capex?project_id={project_id}&scenario_id={scenario_id}",
|
||||
json=payload,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["currency"] == "USD"
|
||||
assert data["totals"]["overall"] == 1_000_000
|
||||
assert data["totals"]["contingency_pct"] == pytest.approx(12.5)
|
||||
assert data["totals"]["contingency_amount"] == pytest.approx(125000)
|
||||
assert data["totals"]["with_contingency"] == pytest.approx(1_125_000)
|
||||
|
||||
by_category = {row["category"]: row for row in data["totals"]["by_category"]}
|
||||
assert by_category["infrastructure"]["amount"] == pytest.approx(1_000_000)
|
||||
assert by_category["infrastructure"]["share"] == pytest.approx(100)
|
||||
|
||||
assert len(data["timeline"]) == 2
|
||||
assert data["timeline"][0]["year"] == 0
|
||||
assert data["timeline"][0]["spend"] == pytest.approx(600_000)
|
||||
assert data["timeline"][1]["cumulative"] == pytest.approx(1_000_000)
|
||||
93
tests/services/test_calculations_capex.py
Normal file
93
tests/services/test_calculations_capex.py
Normal file
@@ -0,0 +1,93 @@
|
||||
import pytest
|
||||
|
||||
from schemas.calculations import (
|
||||
CapexCalculationOptions,
|
||||
CapexCalculationRequest,
|
||||
CapexComponentInput,
|
||||
CapexParameters,
|
||||
)
|
||||
from services.calculations import calculate_initial_capex
|
||||
from services.exceptions import CapexValidationError
|
||||
|
||||
|
||||
def _component(**kwargs) -> CapexComponentInput:
|
||||
defaults = {
|
||||
"id": None,
|
||||
"name": "Component",
|
||||
"category": "equipment",
|
||||
"amount": 1_000_000.0,
|
||||
"currency": "USD",
|
||||
"spend_year": 0,
|
||||
"notes": None,
|
||||
}
|
||||
defaults.update(kwargs)
|
||||
return CapexComponentInput(**defaults)
|
||||
|
||||
|
||||
def test_calculate_initial_capex_success():
|
||||
request = CapexCalculationRequest(
|
||||
components=[
|
||||
_component(name="Crusher", category="equipment",
|
||||
amount=1_200_000, spend_year=0),
|
||||
_component(name="Conveyor", category="equipment",
|
||||
amount=800_000, spend_year=1),
|
||||
_component(name="Camp", category="infrastructure",
|
||||
amount=600_000, spend_year=1, currency="usd"),
|
||||
],
|
||||
parameters=CapexParameters(
|
||||
currency_code="USD",
|
||||
contingency_pct=10,
|
||||
discount_rate_pct=8,
|
||||
evaluation_horizon_years=5,
|
||||
),
|
||||
options=CapexCalculationOptions(persist=True),
|
||||
)
|
||||
|
||||
result = calculate_initial_capex(request)
|
||||
|
||||
assert result.currency == "USD"
|
||||
assert result.options.persist is True
|
||||
|
||||
assert result.totals.overall == pytest.approx(2_600_000)
|
||||
assert result.totals.contingency_pct == pytest.approx(10)
|
||||
assert result.totals.contingency_amount == pytest.approx(260_000)
|
||||
assert result.totals.with_contingency == pytest.approx(2_860_000)
|
||||
|
||||
by_category = {row.category: row for row in result.totals.by_category}
|
||||
assert by_category["equipment"].amount == pytest.approx(2_000_000)
|
||||
assert by_category["infrastructure"].amount == pytest.approx(600_000)
|
||||
assert by_category["equipment"].share == pytest.approx(76.923, rel=1e-3)
|
||||
assert by_category["infrastructure"].share == pytest.approx(
|
||||
23.077, rel=1e-3)
|
||||
|
||||
timeline = {(entry.year, entry.spend): entry.cumulative for entry in result.timeline}
|
||||
assert timeline[(0, 1_200_000)] == pytest.approx(1_200_000)
|
||||
assert timeline[(1, 1_400_000)] == pytest.approx(2_600_000)
|
||||
|
||||
assert len(result.components) == 3
|
||||
assert result.components[2].currency == "USD"
|
||||
|
||||
|
||||
def test_calculate_initial_capex_currency_mismatch_raises():
|
||||
request = CapexCalculationRequest(
|
||||
components=[
|
||||
_component(amount=500_000, currency="USD"),
|
||||
],
|
||||
parameters=CapexParameters(currency_code="CAD"),
|
||||
)
|
||||
|
||||
with pytest.raises(CapexValidationError) as exc:
|
||||
calculate_initial_capex(request)
|
||||
|
||||
assert "Component currency does not match" in exc.value.message
|
||||
assert exc.value.field_errors and "components[0].currency" in exc.value.field_errors[0]
|
||||
|
||||
|
||||
def test_calculate_initial_capex_requires_components():
|
||||
request = CapexCalculationRequest(components=[])
|
||||
|
||||
with pytest.raises(CapexValidationError) as exc:
|
||||
calculate_initial_capex(request)
|
||||
|
||||
assert "At least one capex component" in exc.value.message
|
||||
assert exc.value.field_errors and "components" in exc.value.field_errors[0]
|
||||
Reference in New Issue
Block a user