From 1feae7ff85349b6cb207ce9fb2a5ad7f3aab5a27 Mon Sep 17 00:00:00 2001 From: zwitschi Date: Thu, 13 Nov 2025 09:26:57 +0100 Subject: [PATCH] feat: Add Processing Opex functionality - 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. --- changelog.md | 18 +- models/__init__.py | 6 + models/processing_opex_snapshot.py | 109 ++++ models/project.py | 16 + models/scenario.py | 16 + routes/calculations.py | 536 +++++++++++++++++- schemas/calculations.py | 130 +++++ services/calculations.py | 205 ++++++- services/exceptions.py | 11 + services/repositories.py | 106 ++++ services/unit_of_work.py | 10 + templates/partials/sidebar_nav.html | 2 +- templates/scenarios/detail.html | 3 + templates/scenarios/opex.html | 305 ++++++++++ .../test_processing_opex_calculations.py | 310 ++++++++++ .../test_calculations_processing_opex.py | 159 ++++++ 16 files changed, 1931 insertions(+), 11 deletions(-) create mode 100644 models/processing_opex_snapshot.py create mode 100644 templates/scenarios/opex.html create mode 100644 tests/integration/test_processing_opex_calculations.py create mode 100644 tests/services/test_calculations_processing_opex.py diff --git a/changelog.md b/changelog.md index f38dd0f..c1f9ff1 100644 --- a/changelog.md +++ b/changelog.md @@ -1,13 +1,21 @@ # Changelog +## 2025-11-13 + +- Delivered the initial capex planner end-to-end: added scaffolded UI in `templates/scenarios/capex.html`, wired GET/POST handlers through `routes/calculations.py`, implemented calculation logic plus snapshot persistence in `services/calculations.py` and `models/capex_snapshot.py`, updated navigation links, and introduced unit tests in `tests/services/test_calculations_capex.py`. +- Updated UI navigation to surface the processing opex planner by adding the sidebar link in `templates/partials/sidebar_nav.html`, wiring a scenario detail action in `templates/scenarios/detail.html`. +- Completed manual validation of the Initial Capex Planner UI flows (sidebar entry, scenario deep link, validation errors, successful calculation) with results captured in `manual_tests/initial_capex.md`, documented snapshot verification steps, and noted the optional JSON client check for future follow-up. +- Added processing opex calculation unit tests in `tests/services/test_calculations_processing_opex.py` covering success metrics, currency validation, frequency enforcement, and evaluation horizon extension. +- Documented the Processing Opex Planner workflow in `calminer-docs/userguide/processing_opex_planner.md`, linked it from the user guide index, extended `calminer-docs/architecture/08_concepts/02_data_model.md` with snapshot coverage, and captured the completion in `.github/instructions/DONE.md`. +- Implemented processing opex integration coverage in `tests/integration/test_processing_opex_calculations.py`, exercising HTML and JSON flows, verifying snapshot persistence, and asserting currency mismatch handling for form and API submissions. +- Executed the full pytest suite with coverage (211 tests) to confirm no regressions or warnings after the processing opex documentation updates. + ## 2025-11-12 -- Persisted initial capex calculations to project and scenario snapshot tables via shared helper in `routes/calculations.py`, centralised Jinja filter registration, and expanded integration tests to assert HTML/JSON flows create snapshots with correct totals and payload metadata (requires full-suite pytest run to satisfy 80% coverage gate). -- Resolved test suite regressions by registering the UI router in test fixtures, restoring `TABLE_DDLS` for enum validation checks, hardening token tamper detection, and reran the full pytest suite to confirm green builds. - Fixed critical 500 error in reporting dashboard by correcting route reference in reporting.html template - changed 'reports.project_list_page' to 'projects.project_list_page' to resolve NoMatchFound error when accessing /ui/reporting. - Completed navigation validation by inventorying all sidebar navigation links, identifying missing routes for simulations, reporting, settings, themes, and currencies, created new UI routes in routes/ui.py with proper authentication guards, built corresponding templates (simulations.html, reporting.html, settings.html, theme_settings.html, currencies.html), registered the UI router in main.py, updated sidebar navigation to use route names instead of hardcoded URLs, and enhanced navigation.js to use dynamic URL resolution for proper route handling. -- Fixed critical template rendering error in sidebar_nav.html where URL objects from request.url_for() were being used with string methods, causing TypeError. Added |string filters to convert URL objects to strings for proper template rendering. -- Integrated Plotly charting for interactive visualizations in reporting templates, added chart generation methods to ReportingService (\_generate_npv_comparison_chart, \_generate_distribution_histogram), updated project summary and scenario distribution contexts to include chart JSON data, enhanced templates with chart containers and JavaScript rendering, added chart-container CSS styling, and validated all reporting tests pass. +- Fixed critical template rendering error in sidebar_nav.html where URL objects from `request.url_for()` were being used with string methods, causing TypeError. Added `|string` filters to convert URL objects to strings for proper template rendering. +- Integrated Plotly charting for interactive visualizations in reporting templates, added chart generation methods to ReportingService (`generate_npv_comparison_chart`, `generate_distribution_histogram`), updated project summary and scenario distribution contexts to include chart JSON data, enhanced templates with chart containers and JavaScript rendering, added chart-container CSS styling, and validated all reporting tests pass. - Completed local run verification: started application with `uvicorn main:app --reload` without errors, verified authenticated routes (/login, /, /projects/ui, /projects) load correctly with seeded data, and summarized findings for deployment pipeline readiness. - Fixed docker-compose.override.yml command array to remove duplicate "uvicorn" entry, enabling successful container startup with uvicorn reload in development mode. @@ -17,7 +25,6 @@ - Replaced the Alembic migration workflow with the idempotent Pydantic-backed initializer (`scripts/init_db.py`), added a guarded reset utility (`scripts/reset_db.py`), removed migration artifacts/tooling (Alembic directory, config, Docker entrypoint), refreshed the container entrypoint to invoke `uvicorn` directly, and updated installation/architecture docs plus the README to direct developers to the new seeding/reset flow. - Eliminated Bandit hardcoded-secret findings by replacing literal JWT tokens and passwords across auth/security tests with randomized helpers drawn from `tests/utils/security.py`, ensuring fixtures still assert expected behaviours. - Centralized Bandit configuration in `pyproject.toml`, reran `bandit -c pyproject.toml -r calminer tests`, and verified the scan now reports zero issues. -- Updated `.github/instructions/TODO.md` and `.github/instructions/DONE.md` to reflect the completed security scan remediation workflow. - Diagnosed admin bootstrap failure caused by legacy `roles` schema, added Alembic migration `20251112_00_add_roles_metadata_columns.py` to backfill `display_name`, `description`, `created_at`, and `updated_at`, and verified the migration via full pytest run in the activated `.venv`. - Resolved Ruff E402 warnings by moving module docstrings ahead of `from __future__ import annotations` across currency and pricing service modules, dropped the unused `HTTPException` import in `monitoring/__init__.py`, and confirmed a clean `ruff check .` run. - Enhanced the deploy job in `.gitea/workflows/cicache.yml` to capture Kubernetes pod, deployment, and container logs into `/logs/deployment/` for staging/production rollouts and publish them via a `deployment-logs` artifact, updating CI/CD documentation with retrieval instructions. @@ -32,7 +39,6 @@ - Updated header and footer templates to consistently use `logo_big.png` image instead of text logo, with appropriate CSS styling for sizing. - Centralised ISO-4217 currency validation across scenarios, imports, and export filters (`models/scenario.py`, `routes/scenarios.py`, `schemas/scenario.py`, `schemas/imports.py`, `services/export_query.py`) so malformed codes are rejected consistently at every entry point. - Updated scenario services and UI flows to surface friendly validation errors and added regression coverage for imports, exports, API creation, and lifecycle flows ensuring currencies are normalised end-to-end. -- Recorded the completed “Ensure currency is used consistently” work in `.github/instructions/DONE.md` and ran the full pytest suite (150 tests) to verify the refactor. - Linked projects to their pricing settings by updating SQLAlchemy models, repositories, seeding utilities, and migrations, and added regression tests to cover the new association and default backfill. - Bootstrapped database-stored pricing settings at application startup, aligned initial data seeding with the database-first metadata flow, and added tests covering pricing bootstrap creation, project assignment, and idempotency. - Extended pricing configuration support to prefer persisted metadata via `dependencies.get_pricing_metadata`, added retrieval tests for project/default fallbacks, and refreshed docs (`calminer-docs/specifications/price_calculation.md`, `pricing_settings_data_model.md`) to describe the database-backed workflow and bootstrap behaviour. diff --git a/models/__init__.py b/models/__init__.py index bd4a9f4..761de96 100644 --- a/models/__init__.py +++ b/models/__init__.py @@ -29,6 +29,10 @@ 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 +from .processing_opex_snapshot import ( + ProjectProcessingOpexSnapshot, + ScenarioProcessingOpexSnapshot, +) __all__ = [ "FinancialCategory", @@ -37,12 +41,14 @@ __all__ = [ "Project", "ProjectProfitability", "ProjectCapexSnapshot", + "ProjectProcessingOpexSnapshot", "PricingSettings", "PricingMetalSettings", "PricingImpuritySettings", "Scenario", "ScenarioProfitability", "ScenarioCapexSnapshot", + "ScenarioProcessingOpexSnapshot", "ScenarioStatus", "DistributionType", "SimulationParameter", diff --git a/models/processing_opex_snapshot.py b/models/processing_opex_snapshot.py new file mode 100644 index 0000000..f1c1353 --- /dev/null +++ b/models/processing_opex_snapshot.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import JSON, Boolean, 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 ProjectProcessingOpexSnapshot(Base): + """Snapshot of recurring processing opex metrics at the project level.""" + + __tablename__ = "project_processing_opex_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) + overall_annual: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + escalated_total: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + annual_average: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + evaluation_horizon_years: Mapped[int | None] = mapped_column(Integer, nullable=True) + escalation_pct: Mapped[float | None] = mapped_column(Numeric(12, 6), nullable=True) + apply_escalation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=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="processing_opex_snapshots" + ) + created_by: Mapped[User | None] = relationship("User") + + def __repr__(self) -> str: # pragma: no cover + return ( + "ProjectProcessingOpexSnapshot(id={id!r}, project_id={project_id!r}, overall_annual={overall_annual!r})".format( + id=self.id, + project_id=self.project_id, + overall_annual=self.overall_annual, + ) + ) + + +class ScenarioProcessingOpexSnapshot(Base): + """Snapshot of processing opex metrics for an individual scenario.""" + + __tablename__ = "scenario_processing_opex_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) + overall_annual: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + escalated_total: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + annual_average: Mapped[float | None] = mapped_column(Numeric(18, 2), nullable=True) + evaluation_horizon_years: Mapped[int | None] = mapped_column(Integer, nullable=True) + escalation_pct: Mapped[float | None] = mapped_column(Numeric(12, 6), nullable=True) + apply_escalation: Mapped[bool] = mapped_column(Boolean, nullable=False, default=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="processing_opex_snapshots" + ) + created_by: Mapped[User | None] = relationship("User") + + def __repr__(self) -> str: # pragma: no cover + return ( + "ScenarioProcessingOpexSnapshot(id={id!r}, scenario_id={scenario_id!r}, overall_annual={overall_annual!r})".format( + id=self.id, + scenario_id=self.scenario_id, + overall_annual=self.overall_annual, + ) + ) diff --git a/models/project.py b/models/project.py index af8a680..8eecfcb 100644 --- a/models/project.py +++ b/models/project.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, List from .enums import MiningOperationType, sql_enum from .profitability_snapshot import ProjectProfitability from .capex_snapshot import ProjectCapexSnapshot +from .processing_opex_snapshot import ProjectProcessingOpexSnapshot from sqlalchemy import DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -67,6 +68,13 @@ class Project(Base): order_by=lambda: ProjectCapexSnapshot.calculated_at.desc(), passive_deletes=True, ) + processing_opex_snapshots: Mapped[List["ProjectProcessingOpexSnapshot"]] = relationship( + "ProjectProcessingOpexSnapshot", + back_populates="project", + cascade="all, delete-orphan", + order_by=lambda: ProjectProcessingOpexSnapshot.calculated_at.desc(), + passive_deletes=True, + ) @property def latest_profitability(self) -> "ProjectProfitability | None": @@ -84,5 +92,13 @@ class Project(Base): return None return self.capex_snapshots[0] + @property + def latest_processing_opex(self) -> "ProjectProcessingOpexSnapshot | None": + """Return the most recent processing opex snapshot, if any.""" + + if not self.processing_opex_snapshots: + return None + return self.processing_opex_snapshots[0] + def __repr__(self) -> str: # pragma: no cover - helpful for debugging return f"Project(id={self.id!r}, name={self.name!r})" diff --git a/models/scenario.py b/models/scenario.py index 7234a25..f20dbda 100644 --- a/models/scenario.py +++ b/models/scenario.py @@ -21,6 +21,7 @@ from services.currency import normalise_currency from .enums import ResourceType, ScenarioStatus, sql_enum from .profitability_snapshot import ScenarioProfitability from .capex_snapshot import ScenarioCapexSnapshot +from .processing_opex_snapshot import ScenarioProcessingOpexSnapshot if TYPE_CHECKING: # pragma: no cover from .financial_input import FinancialInput @@ -91,6 +92,13 @@ class Scenario(Base): order_by=lambda: ScenarioCapexSnapshot.calculated_at.desc(), passive_deletes=True, ) + processing_opex_snapshots: Mapped[List["ScenarioProcessingOpexSnapshot"]] = relationship( + "ScenarioProcessingOpexSnapshot", + back_populates="scenario", + cascade="all, delete-orphan", + order_by=lambda: ScenarioProcessingOpexSnapshot.calculated_at.desc(), + passive_deletes=True, + ) @validates("currency") def _normalise_currency(self, key: str, value: str | None) -> str | None: @@ -115,3 +123,11 @@ class Scenario(Base): if not self.capex_snapshots: return None return self.capex_snapshots[0] + + @property + def latest_processing_opex(self) -> "ScenarioProcessingOpexSnapshot | None": + """Return the most recent processing opex snapshot for this scenario.""" + + if not self.processing_opex_snapshots: + return None + return self.processing_opex_snapshots[0] diff --git a/routes/calculations.py b/routes/calculations.py index b31ea05..94452e5 100644 --- a/routes/calculations.py +++ b/routes/calculations.py @@ -15,9 +15,11 @@ from dependencies import get_pricing_metadata, get_unit_of_work, require_authent from models import ( Project, ProjectCapexSnapshot, + ProjectProcessingOpexSnapshot, ProjectProfitability, Scenario, ScenarioCapexSnapshot, + ScenarioProcessingOpexSnapshot, ScenarioProfitability, User, ) @@ -27,11 +29,25 @@ from schemas.calculations import ( CapexCalculationResult, CapexComponentInput, CapexParameters, + ProcessingOpexCalculationRequest, + ProcessingOpexCalculationResult, + ProcessingOpexComponentInput, + ProcessingOpexOptions, + ProcessingOpexParameters, ProfitabilityCalculationRequest, ProfitabilityCalculationResult, ) -from services.calculations import calculate_initial_capex, calculate_profitability -from services.exceptions import CapexValidationError, EntityNotFoundError, ProfitabilityValidationError +from services.calculations import ( + calculate_initial_capex, + calculate_processing_opex, + calculate_profitability, +) +from services.exceptions import ( + CapexValidationError, + EntityNotFoundError, + OpexValidationError, + ProfitabilityValidationError, +) from services.pricing import PricingMetadata from services.unit_of_work import UnitOfWork from routes.template_filters import register_common_filters @@ -56,6 +72,26 @@ _CAPEX_CATEGORY_OPTIONS: tuple[dict[str, str], ...] = ( ) _DEFAULT_CAPEX_HORIZON_YEARS = 5 +_OPEX_CATEGORY_OPTIONS: tuple[dict[str, str], ...] = ( + {"value": "labor", "label": "Labor"}, + {"value": "materials", "label": "Materials"}, + {"value": "energy", "label": "Energy"}, + {"value": "maintenance", "label": "Maintenance"}, + {"value": "other", "label": "Other"}, +) + +_OPEX_FREQUENCY_OPTIONS: tuple[dict[str, str], ...] = ( + {"value": "daily", "label": "Daily"}, + {"value": "weekly", "label": "Weekly"}, + {"value": "monthly", "label": "Monthly"}, + {"value": "quarterly", "label": "Quarterly"}, + {"value": "annually", "label": "Annually"}, +) + +_DEFAULT_OPEX_HORIZON_YEARS = 5 + +_PROCESSING_OPEX_TEMPLATE = "scenarios/opex.html" + def _combine_impurity_metadata(metadata: PricingMetadata) -> list[dict[str, object]]: """Build impurity rows combining thresholds and penalties.""" @@ -366,6 +402,171 @@ def _prepare_capex_context( } +def _serialise_opex_component_entry(component: Any) -> dict[str, Any]: + if isinstance(component, ProcessingOpexComponentInput): + 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), + "unit_cost": getattr(component, "unit_cost", None), + "quantity": getattr(component, "quantity", None), + "frequency": getattr(component, "frequency", None), + "currency": getattr(component, "currency", None), + "period_start": getattr(component, "period_start", None), + "period_end": getattr(component, "period_end", None), + "notes": getattr(component, "notes", None), + } + + return { + "id": raw.get("id"), + "name": _value_or_blank(raw.get("name")), + "category": raw.get("category") or "labor", + "unit_cost": _value_or_blank(raw.get("unit_cost")), + "quantity": _value_or_blank(raw.get("quantity")), + "frequency": raw.get("frequency") or "monthly", + "currency": _value_or_blank(raw.get("currency")), + "period_start": _value_or_blank(raw.get("period_start")), + "period_end": _value_or_blank(raw.get("period_end")), + "notes": _value_or_blank(raw.get("notes")), + } + + +def _serialise_opex_parameters(parameters: Any) -> dict[str, Any]: + if isinstance(parameters, ProcessingOpexParameters): + raw = parameters.model_dump() + elif isinstance(parameters, dict): + raw = dict(parameters) + else: + raw = {} + + return { + "currency_code": _value_or_blank(raw.get("currency_code")), + "escalation_pct": _value_or_blank(raw.get("escalation_pct")), + "discount_rate_pct": _value_or_blank(raw.get("discount_rate_pct")), + "evaluation_horizon_years": _value_or_blank( + raw.get("evaluation_horizon_years") + ), + "apply_escalation": _coerce_bool(raw.get("apply_escalation", True)), + } + + +def _serialise_opex_options(options: Any) -> dict[str, Any]: + if isinstance(options, ProcessingOpexOptions): + raw = options.model_dump() + elif isinstance(options, dict): + raw = dict(options) + else: + raw = {} + + return { + "persist": _coerce_bool(raw.get("persist", False)), + "snapshot_notes": _value_or_blank(raw.get("snapshot_notes")), + } + + +def _build_opex_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) + + last_updated_at = getattr(scenario, "opex_updated_at", None) + + return { + "components": [], + "parameters": { + "currency_code": currency or None, + "escalation_pct": None, + "discount_rate_pct": discount_rate, + "evaluation_horizon_years": _DEFAULT_OPEX_HORIZON_YEARS, + "apply_escalation": True, + }, + "options": { + "persist": bool(scenario or project), + "snapshot_notes": None, + }, + "currency_code": currency or None, + "default_horizon": _DEFAULT_OPEX_HORIZON_YEARS, + "last_updated_at": last_updated_at, + } + + +def _prepare_opex_context( + request: Request, + *, + project: Project | None, + scenario: Scenario | None, + form_data: dict[str, Any] | None = None, + result: ProcessingOpexCalculationResult | 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_opex_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_opex_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_opex_parameters(form_data.get("parameters")) + ) + parameters = _serialise_opex_parameters(raw_parameters) + + raw_options = defaults["options"].copy() + if form_data and form_data.get("options"): + raw_options.update(_serialise_opex_options(form_data.get("options"))) + options = _serialise_opex_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": _OPEX_CATEGORY_OPTIONS, + "frequency_options": _OPEX_FREQUENCY_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: @@ -406,6 +607,87 @@ def _partition_capex_error_messages( return general, component_specific +def _partition_opex_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 _opex_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] + if field == "apply_escalation": + parameters[field] = _coerce_bool(normalised_value) + else: + 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: + if "persist" in options: + options["persist"] = _coerce_bool(options.get("persist")) + data["options"] = options + + return data + + def _capex_form_to_payload(form: FormData) -> dict[str, Any]: data: dict[str, Any] = {} components: dict[int, dict[str, Any]] = {} @@ -458,6 +740,15 @@ def _capex_form_to_payload(form: FormData) -> dict[str, Any]: return data +async def _extract_opex_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 _opex_form_to_payload(form) + + async def _extract_capex_payload(request: Request) -> dict[str, Any]: content_type = request.headers.get("content-type", "").lower() if content_type.startswith("application/json"): @@ -772,6 +1063,247 @@ def _persist_capex_snapshots( uow.project_capex.create(project_snapshot) +def _should_persist_opex( + *, + project: Project | None, + scenario: Scenario | None, + request_model: ProcessingOpexCalculationRequest, +) -> bool: + persist_requested = bool( + getattr(request_model, "options", None) + and request_model.options.persist + ) + return persist_requested and bool(project or scenario) + + +def _persist_opex_snapshots( + *, + uow: UnitOfWork, + project: Project | None, + scenario: Scenario | None, + user: User | None, + request_model: ProcessingOpexCalculationRequest, + result: ProcessingOpexCalculationResult, +) -> None: + if not _should_persist_opex( + project=project, + scenario=scenario, + request_model=request_model, + ): + return + + created_by_id = getattr(user, "id", None) + totals = result.totals + metrics = result.metrics + parameters = result.parameters + + overall_annual = float(totals.overall_annual) + escalated_total = ( + float(totals.escalated_total) + if totals.escalated_total is not None + else None + ) + annual_average = ( + float(metrics.annual_average) + if metrics.annual_average is not None + else None + ) + evaluation_horizon = ( + int(parameters.evaluation_horizon_years) + if parameters.evaluation_horizon_years is not None + else None + ) + escalation_pct = ( + float(totals.escalation_pct) + if totals.escalation_pct is not None + else ( + float(parameters.escalation_pct) + if parameters.escalation_pct is not None and parameters.apply_escalation + else None + ) + ) + apply_escalation = bool(parameters.apply_escalation) + component_count = len(result.components) + + payload = { + "request": request_model.model_dump(mode="json"), + "result": result.model_dump(), + } + + if scenario and uow.scenario_processing_opex: + scenario_snapshot = ScenarioProcessingOpexSnapshot( + scenario_id=scenario.id, + created_by_id=created_by_id, + calculation_source="calculations.processing_opex", + currency_code=result.currency, + overall_annual=overall_annual, + escalated_total=escalated_total, + annual_average=annual_average, + evaluation_horizon_years=evaluation_horizon, + escalation_pct=escalation_pct, + apply_escalation=apply_escalation, + component_count=component_count, + payload=payload, + ) + uow.scenario_processing_opex.create(scenario_snapshot) + + if project and uow.project_processing_opex: + project_snapshot = ProjectProcessingOpexSnapshot( + project_id=project.id, + created_by_id=created_by_id, + calculation_source="calculations.processing_opex", + currency_code=result.currency, + overall_annual=overall_annual, + escalated_total=escalated_total, + annual_average=annual_average, + evaluation_horizon_years=evaluation_horizon, + escalation_pct=escalation_pct, + apply_escalation=apply_escalation, + component_count=component_count, + payload=payload, + ) + uow.project_processing_opex.create(project_snapshot) + + +@router.get( + "/processing-opex", + response_class=HTMLResponse, + name="calculations.processing_opex_form", +) +def processing_opex_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 processing opex planner with default context.""" + + project, scenario = _load_project_and_scenario( + uow=uow, project_id=project_id, scenario_id=scenario_id + ) + context = _prepare_opex_context( + request, + project=project, + scenario=scenario, + ) + return templates.TemplateResponse(_PROCESSING_OPEX_TEMPLATE, context) + + +@router.post( + "/processing-opex", + name="calculations.processing_opex_submit", +) +async def processing_opex_submit( + request: Request, + current_user: 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: + """Handle processing opex submissions and respond with HTML or JSON.""" + + wants_json = _is_json_request(request) + payload_data = await _extract_opex_payload(request) + + try: + request_model = ProcessingOpexCalculationRequest.model_validate( + payload_data + ) + result = calculate_processing_opex(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_opex_error_messages( + exc.errors() + ) + context = _prepare_opex_context( + request, + project=project, + scenario=scenario, + form_data=payload_data, + errors=general_errors, + component_errors=component_errors, + ) + return templates.TemplateResponse( + _PROCESSING_OPEX_TEMPLATE, + context, + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + ) + except OpexValidationError 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_opex_context( + request, + project=project, + scenario=scenario, + form_data=payload_data, + errors=errors, + ) + return templates.TemplateResponse( + _PROCESSING_OPEX_TEMPLATE, + context, + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + ) + + project, scenario = _load_project_and_scenario( + uow=uow, project_id=project_id, scenario_id=scenario_id + ) + + _persist_opex_snapshots( + uow=uow, + project=project, + scenario=scenario, + user=current_user, + request_model=request_model, + result=result, + ) + + if wants_json: + return JSONResponse( + status_code=status.HTTP_200_OK, + content=result.model_dump(), + ) + + context = _prepare_opex_context( + request, + project=project, + scenario=scenario, + form_data=request_model.model_dump(mode="json"), + result=result, + ) + notices = _list_from_context(context, "notices") + notices.append("Processing opex calculation completed successfully.") + + return templates.TemplateResponse( + _PROCESSING_OPEX_TEMPLATE, + context, + status_code=status.HTTP_200_OK, + ) + + @router.get( "/capex", response_class=HTMLResponse, diff --git a/schemas/calculations.py b/schemas/calculations.py index 638b16e..32d7637 100644 --- a/schemas/calculations.py +++ b/schemas/calculations.py @@ -197,6 +197,127 @@ class CapexCalculationResult(BaseModel): 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", @@ -212,5 +333,14 @@ __all__ = [ "CapexTotals", "CapexTimelineEntry", "CapexCalculationResult", + "ProcessingOpexComponentInput", + "ProcessingOpexParameters", + "ProcessingOpexOptions", + "ProcessingOpexCalculationRequest", + "ProcessingOpexCategoryBreakdown", + "ProcessingOpexTimelineEntry", + "ProcessingOpexMetrics", + "ProcessingOpexTotals", + "ProcessingOpexCalculationResult", "ValidationError", ] diff --git a/services/calculations.py b/services/calculations.py index 8d071f7..dae1dfa 100644 --- a/services/calculations.py +++ b/services/calculations.py @@ -3,9 +3,14 @@ from __future__ import annotations from collections import defaultdict +from statistics import fmean from services.currency import CurrencyValidationError, normalise_currency -from services.exceptions import CapexValidationError, ProfitabilityValidationError +from services.exceptions import ( + CapexValidationError, + OpexValidationError, + ProfitabilityValidationError, +) from services.financial import ( CashFlow, ConvergenceError, @@ -24,6 +29,14 @@ from schemas.calculations import ( CapexTotals, CapexTimelineEntry, CashFlowEntry, + ProcessingOpexCalculationRequest, + ProcessingOpexCalculationResult, + ProcessingOpexCategoryBreakdown, + ProcessingOpexComponentInput, + ProcessingOpexMetrics, + ProcessingOpexParameters, + ProcessingOpexTotals, + ProcessingOpexTimelineEntry, ProfitabilityCalculationRequest, ProfitabilityCalculationResult, ProfitabilityCosts, @@ -31,6 +44,15 @@ from schemas.calculations import ( ) +_FREQUENCY_MULTIPLIER = { + "daily": 365, + "weekly": 52, + "monthly": 12, + "quarterly": 4, + "annually": 1, +} + + def _build_pricing_input( request: ProfitabilityCalculationRequest, ) -> PricingInput: @@ -332,4 +354,183 @@ def calculate_initial_capex( ) -__all__ = ["calculate_profitability", "calculate_initial_capex"] +def calculate_processing_opex( + request: ProcessingOpexCalculationRequest, +) -> ProcessingOpexCalculationResult: + """Aggregate processing opex components into annual totals and timeline.""" + + if not request.components: + raise OpexValidationError( + "At least one processing opex component is required for calculation.", + ["components"], + ) + + parameters: ProcessingOpexParameters = request.parameters + base_currency = parameters.currency_code + if base_currency: + try: + base_currency = normalise_currency(base_currency) + except CurrencyValidationError as exc: + raise OpexValidationError( + str(exc), ["parameters.currency_code"] + ) from exc + + evaluation_horizon = parameters.evaluation_horizon_years or 1 + if evaluation_horizon <= 0: + raise OpexValidationError( + "Evaluation horizon must be at least 1 year.", + ["parameters.evaluation_horizon_years"], + ) + + escalation_pct = float(parameters.escalation_pct or 0.0) + apply_escalation = bool(parameters.apply_escalation) + + category_totals: dict[str, float] = defaultdict(float) + timeline_totals: dict[int, float] = defaultdict(float) + timeline_escalated: dict[int, float] = defaultdict(float) + normalised_components: list[ProcessingOpexComponentInput] = [] + + max_period_end = evaluation_horizon + + for index, component in enumerate(request.components): + frequency = component.frequency.lower() + multiplier = _FREQUENCY_MULTIPLIER.get(frequency) + if multiplier is None: + raise OpexValidationError( + f"Unsupported frequency '{component.frequency}'.", + [f"components[{index}].frequency"], + ) + + unit_cost = float(component.unit_cost) + quantity = float(component.quantity) + annual_cost = unit_cost * quantity * multiplier + + period_start = component.period_start or 1 + period_end = component.period_end or evaluation_horizon + if period_end < period_start: + raise OpexValidationError( + ( + "Component period_end must be greater than or equal to " + "period_start." + ), + [f"components[{index}].period_end"], + ) + + max_period_end = max(max_period_end, period_end) + + component_currency = component.currency + if component_currency: + try: + component_currency = normalise_currency(component_currency) + except CurrencyValidationError as exc: + raise OpexValidationError( + 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 OpexValidationError( + ( + "Component currency does not match the global currency. " + f"Expected {base_currency}, got {component_currency}." + ), + [f"components[{index}].currency"], + ) + + category_totals[component.category] += annual_cost + + for period in range(period_start, period_end + 1): + timeline_totals[period] += annual_cost + + normalised_components.append( + ProcessingOpexComponentInput( + id=component.id, + name=component.name, + category=component.category, + unit_cost=unit_cost, + quantity=quantity, + frequency=frequency, + currency=component_currency, + period_start=period_start, + period_end=period_end, + notes=component.notes, + ) + ) + + evaluation_horizon = max(evaluation_horizon, max_period_end) + + try: + currency = normalise_currency(base_currency) if base_currency else None + except CurrencyValidationError as exc: + raise OpexValidationError( + str(exc), ["parameters.currency_code"] + ) from exc + + timeline_entries: list[ProcessingOpexTimelineEntry] = [] + escalated_values: list[float] = [] + overall_annual = timeline_totals.get(1, 0.0) + escalated_total = 0.0 + + for period in range(1, evaluation_horizon + 1): + base_cost = timeline_totals.get(period, 0.0) + if apply_escalation: + factor = (1 + escalation_pct / 100.0) ** (period - 1) + else: + factor = 1.0 + escalated_cost = base_cost * factor + timeline_escalated[period] = escalated_cost + escalated_total += escalated_cost + timeline_entries.append( + ProcessingOpexTimelineEntry( + period=period, + base_cost=base_cost, + escalated_cost=escalated_cost if apply_escalation else None, + ) + ) + escalated_values.append(escalated_cost) + + category_breakdowns: list[ProcessingOpexCategoryBreakdown] = [] + total_base = sum(category_totals.values()) + for category, total in sorted(category_totals.items()): + share = (total / total_base * 100.0) if total_base else None + category_breakdowns.append( + ProcessingOpexCategoryBreakdown( + category=category, + annual_cost=total, + share=share, + ) + ) + + metrics = ProcessingOpexMetrics( + annual_average=fmean(escalated_values) if escalated_values else None, + cost_per_ton=None, + ) + + totals = ProcessingOpexTotals( + overall_annual=overall_annual, + escalated_total=escalated_total if apply_escalation else None, + escalation_pct=escalation_pct if apply_escalation else None, + by_category=category_breakdowns, + ) + + return ProcessingOpexCalculationResult( + totals=totals, + timeline=timeline_entries, + metrics=metrics, + components=normalised_components, + parameters=parameters, + options=request.options, + currency=currency, + ) + + +__all__ = [ + "calculate_profitability", + "calculate_initial_capex", + "calculate_processing_opex", +] diff --git a/services/exceptions.py b/services/exceptions.py index a22a784..0eb3b6a 100644 --- a/services/exceptions.py +++ b/services/exceptions.py @@ -48,3 +48,14 @@ class CapexValidationError(Exception): def __str__(self) -> str: # pragma: no cover - mirrors message for logging return self.message + + +@dataclass(eq=False) +class OpexValidationError(Exception): + """Raised when opex 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 diff --git a/services/repositories.py b/services/repositories.py index 79ec2f1..28eafd5 100644 --- a/services/repositories.py +++ b/services/repositories.py @@ -17,10 +17,12 @@ from models import ( PricingSettings, ProjectCapexSnapshot, ProjectProfitability, + ProjectProcessingOpexSnapshot, Role, Scenario, ScenarioCapexSnapshot, ScenarioProfitability, + ScenarioProcessingOpexSnapshot, ScenarioStatus, SimulationParameter, User, @@ -571,6 +573,110 @@ class ScenarioCapexRepository: self.session.delete(entity) +class ProjectProcessingOpexRepository: + """Persistence operations for project-level processing opex snapshots.""" + + def __init__(self, session: Session) -> None: + self.session = session + + def create( + self, snapshot: ProjectProcessingOpexSnapshot + ) -> ProjectProcessingOpexSnapshot: + self.session.add(snapshot) + self.session.flush() + return snapshot + + def list_for_project( + self, + project_id: int, + *, + limit: int | None = None, + ) -> Sequence[ProjectProcessingOpexSnapshot]: + stmt = ( + select(ProjectProcessingOpexSnapshot) + .where(ProjectProcessingOpexSnapshot.project_id == project_id) + .order_by(ProjectProcessingOpexSnapshot.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, + ) -> ProjectProcessingOpexSnapshot | None: + stmt = ( + select(ProjectProcessingOpexSnapshot) + .where(ProjectProcessingOpexSnapshot.project_id == project_id) + .order_by(ProjectProcessingOpexSnapshot.calculated_at.desc()) + .limit(1) + ) + return self.session.execute(stmt).scalar_one_or_none() + + def delete(self, snapshot_id: int) -> None: + stmt = select(ProjectProcessingOpexSnapshot).where( + ProjectProcessingOpexSnapshot.id == snapshot_id + ) + entity = self.session.execute(stmt).scalar_one_or_none() + if entity is None: + raise EntityNotFoundError( + f"Project processing opex snapshot {snapshot_id} not found" + ) + self.session.delete(entity) + + +class ScenarioProcessingOpexRepository: + """Persistence operations for scenario-level processing opex snapshots.""" + + def __init__(self, session: Session) -> None: + self.session = session + + def create( + self, snapshot: ScenarioProcessingOpexSnapshot + ) -> ScenarioProcessingOpexSnapshot: + self.session.add(snapshot) + self.session.flush() + return snapshot + + def list_for_scenario( + self, + scenario_id: int, + *, + limit: int | None = None, + ) -> Sequence[ScenarioProcessingOpexSnapshot]: + stmt = ( + select(ScenarioProcessingOpexSnapshot) + .where(ScenarioProcessingOpexSnapshot.scenario_id == scenario_id) + .order_by(ScenarioProcessingOpexSnapshot.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, + ) -> ScenarioProcessingOpexSnapshot | None: + stmt = ( + select(ScenarioProcessingOpexSnapshot) + .where(ScenarioProcessingOpexSnapshot.scenario_id == scenario_id) + .order_by(ScenarioProcessingOpexSnapshot.calculated_at.desc()) + .limit(1) + ) + return self.session.execute(stmt).scalar_one_or_none() + + def delete(self, snapshot_id: int) -> None: + stmt = select(ScenarioProcessingOpexSnapshot).where( + ScenarioProcessingOpexSnapshot.id == snapshot_id + ) + entity = self.session.execute(stmt).scalar_one_or_none() + if entity is None: + raise EntityNotFoundError( + f"Scenario processing opex snapshot {snapshot_id} not found" + ) + self.session.delete(entity) + + class FinancialInputRepository: """Persistence operations for FinancialInput entities.""" diff --git a/services/unit_of_work.py b/services/unit_of_work.py index 821a34b..d57b98a 100644 --- a/services/unit_of_work.py +++ b/services/unit_of_work.py @@ -14,10 +14,12 @@ from services.repositories import ( PricingSettingsSeedResult, ProjectRepository, ProjectProfitabilityRepository, + ProjectProcessingOpexRepository, ProjectCapexRepository, RoleRepository, ScenarioRepository, ScenarioProfitabilityRepository, + ScenarioProcessingOpexRepository, ScenarioCapexRepository, SimulationParameterRepository, UserRepository, @@ -42,8 +44,10 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]): self.simulation_parameters: SimulationParameterRepository | None = None self.project_profitability: ProjectProfitabilityRepository | None = None self.project_capex: ProjectCapexRepository | None = None + self.project_processing_opex: ProjectProcessingOpexRepository | None = None self.scenario_profitability: ScenarioProfitabilityRepository | None = None self.scenario_capex: ScenarioCapexRepository | None = None + self.scenario_processing_opex: ScenarioProcessingOpexRepository | None = None self.users: UserRepository | None = None self.roles: RoleRepository | None = None self.pricing_settings: PricingSettingsRepository | None = None @@ -58,10 +62,14 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]): self.project_profitability = ProjectProfitabilityRepository( self.session) self.project_capex = ProjectCapexRepository(self.session) + self.project_processing_opex = ProjectProcessingOpexRepository( + self.session) self.scenario_profitability = ScenarioProfitabilityRepository( self.session ) self.scenario_capex = ScenarioCapexRepository(self.session) + self.scenario_processing_opex = ScenarioProcessingOpexRepository( + self.session) self.users = UserRepository(self.session) self.roles = RoleRepository(self.session) self.pricing_settings = PricingSettingsRepository(self.session) @@ -82,8 +90,10 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]): self.simulation_parameters = None self.project_profitability = None self.project_capex = None + self.project_processing_opex = None self.scenario_profitability = None self.scenario_capex = None + self.scenario_processing_opex = None self.users = None self.roles = None self.pricing_settings = None diff --git a/templates/partials/sidebar_nav.html b/templates/partials/sidebar_nav.html index e9fa0d8..0236548 100644 --- a/templates/partials/sidebar_nav.html +++ b/templates/partials/sidebar_nav.html @@ -19,7 +19,7 @@ request.url_for('auth.password_reset_request_form') if request else "match_prefix": "/"}, {"href": projects_href, "label": "Projects", "match_prefix": "/projects"}, {"href": project_create_href, "label": "New Project", "match_prefix": "/projects/create"}, {"href": "/imports/ui", "label": -"Imports", "match_prefix": "/imports"}, {"href": request.url_for('calculations.profitability_form') if request else '/calculations/profitability', "label": "Profitability Calculator", "match_prefix": "/calculations/profitability"}, {"href": request.url_for('calculations.capex_form') if request else '/calculations/capex', "label": "Initial Capex Planner", "match_prefix": "/calculations/capex"} ] }, { "label": "Insights", "links": [ +"Imports", "match_prefix": "/imports"}, {"href": request.url_for('calculations.profitability_form') if request else '/calculations/profitability', "label": "Profitability Calculator", "match_prefix": "/calculations/profitability"}, {"href": request.url_for('calculations.processing_opex_form') if request else '/calculations/processing-opex', "label": "Processing Opex Planner", "match_prefix": "/calculations/processing-opex"}, {"href": request.url_for('calculations.capex_form') if request else '/calculations/capex', "label": "Initial Capex Planner", "match_prefix": "/calculations/capex"} ] }, { "label": "Insights", "links": [ {"href": "/ui/simulations", "label": "Simulations"}, {"href": "/ui/reporting", "label": "Reporting"} ] }, { "label": "Configuration", "links": [ { "href": "/ui/settings", "label": "Settings", "children": [ {"href": "/theme-settings", diff --git a/templates/scenarios/detail.html b/templates/scenarios/detail.html index 7441fae..fc1e823 100644 --- a/templates/scenarios/detail.html +++ b/templates/scenarios/detail.html @@ -14,9 +14,11 @@
{% set profitability_href = url_for('calculations.profitability_form') %} + {% set processing_opex_href = url_for('calculations.processing_opex_form') %} {% set capex_href = url_for('calculations.capex_form') %} {% if project and scenario %} {% set profitability_href = profitability_href ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %} + {% set processing_opex_href = processing_opex_href ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %} {% set capex_href = capex_href ~ '?project_id=' ~ project.id ~ '&scenario_id=' ~ scenario.id %} {% endif %}
@@ -26,6 +28,7 @@
Back to Project Profitability Calculator + Processing Opex Planner Initial Capex Planner Edit Scenario
diff --git a/templates/scenarios/opex.html b/templates/scenarios/opex.html new file mode 100644 index 0000000..8d0946b --- /dev/null +++ b/templates/scenarios/opex.html @@ -0,0 +1,305 @@ +{% extends "base.html" %} +{% block title %}Processing Opex Planner · CalMiner{% endblock %} + +{% block content %} + + +
+
+

Processing Opex Planner

+

Capture recurring operational costs and review annual totals with escalation assumptions.

+
+
+ {% if cancel_url %} + Cancel + {% endif %} + +
+
+ + {% if errors %} +
+

Submission errors

+
    + {% for message in errors %} +
  • {{ message }}
    • + {% endfor %} +
+
+ {% endif %} + + {% if notices %} +
+
    + {% for message in notices %} +
  • {{ message }}
    • + {% endfor %} +
+
+ {% endif %} + +
+ + + +
+
+
+

Opex Components

+

List recurring cost items with frequency, unit cost, and quantities.

+
+ +
+ +
+ + {% if component_errors %} + + {% endif %} + + {% if component_notices %} +
+
    + {% for message in component_notices %} +
  • {{ message }}
    • + {% endfor %} +
+
+ {% endif %} + +
+ + + + + + + + + + + + + + + {% set component_entries = (components if components else [{}]) %} + {% for component in component_entries %} + + + + + + + + + + + + {% endfor %} + +
CategoryComponentUnit CostQuantityFrequencyCurrencyStart PeriodEnd Period
+ + + + + + + + + + + + + + + + + + +
+
+ +

Use start and end periods to indicate when the cost applies within the evaluation horizon.

+
+ + +
+
+ +
+
+

Opex Summary

+

Annual totals, escalation impacts, and category breakdowns.

+
+ + {% if result %} +
+
+

Annual Opex Total

+

+ {{ result.totals.overall_annual | currency_display(result.currency) }} +

+
+
+

Escalated Total

+

+ + {% if result.totals.escalated_total is not none %} + {{ result.totals.escalated_total | currency_display(result.currency) }} + {% else %} + — + {% endif %} + +

+
+
+

Annual Average (Escalated)

+

+ + {% if result.metrics.annual_average is not none %} + {{ result.metrics.annual_average | currency_display(result.currency) }} + {% else %} + — + {% endif %} + +

+
+
+ +
+

Category Breakdown

+
+ + + + + + + + + + {% for entry in result.totals.by_category %} + + + + + + {% endfor %} + +
CategoryAnnual CostShare (%)
{{ entry.category | title }}{{ entry.annual_cost | currency_display(result.currency) }}{% if entry.share is not none %}{{ entry.share | round(2) }}{% else %}—{% endif %}
+
+
+ +
+

Timeline

+
+ + + + + + + + + + {% for entry in result.timeline %} + + + + + + {% endfor %} + +
PeriodBase CostEscalated Cost
{{ entry.period }}{{ entry.base_cost | currency_display(result.currency) }} + {% if entry.escalated_cost is not none %} + {{ entry.escalated_cost | currency_display(result.currency) }} + {% else %} + — + {% endif %} +
+
+
+ {% else %} +

Run the calculation to populate summary metrics and timeline insights.

+ {% endif %} +
+{% endblock %} diff --git a/tests/integration/test_processing_opex_calculations.py b/tests/integration/test_processing_opex_calculations.py new file mode 100644 index 0000000..1aff197 --- /dev/null +++ b/tests/integration/test_processing_opex_calculations.py @@ -0,0 +1,310 @@ +from __future__ import annotations + +from collections.abc import Callable + +import pytest +from fastapi.testclient import TestClient + +from services.unit_of_work import UnitOfWork + + +def _create_project(client: TestClient, name: str) -> int: + response = client.post( + "/projects", + json={ + "name": name, + "location": "Nevada", + "operation_type": "open_pit", + "description": "Project for processing opex 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": "Processing opex scenario", + "status": "draft", + "currency": "usd", + "primary_resource": "diesel", + }, + ) + assert response.status_code == 201 + return response.json()["id"] + + +def test_processing_opex_calculation_html_flow( + client: TestClient, + unit_of_work_factory: Callable[[], UnitOfWork], +) -> None: + project_id = _create_project(client, "Opex HTML Project") + scenario_id = _create_scenario(client, project_id, "Opex HTML Scenario") + + form_page = client.get( + f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}" + ) + assert form_page.status_code == 200 + assert "Processing Opex Planner" in form_page.text + + response = client.post( + f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}", + data={ + "components[0][name]": "Power", + "components[0][category]": "energy", + "components[0][unit_cost]": "1000", + "components[0][quantity]": "1", + "components[0][frequency]": "monthly", + "components[0][currency]": "USD", + "components[0][period_start]": "1", + "components[0][period_end]": "3", + "components[1][name]": "Maintenance", + "components[1][category]": "maintenance", + "components[1][unit_cost]": "2500", + "components[1][quantity]": "1", + "components[1][frequency]": "quarterly", + "components[1][currency]": "USD", + "components[1][period_start]": "1", + "components[1][period_end]": "2", + "parameters[currency_code]": "USD", + "parameters[escalation_pct]": "5", + "parameters[discount_rate_pct]": "3", + "parameters[evaluation_horizon_years]": "3", + "parameters[apply_escalation]": "1", + "options[persist]": "1", + "options[snapshot_notes]": "Processing opex HTML flow", + }, + ) + assert response.status_code == 200 + assert "Processing opex calculation completed successfully." in response.text + assert "Opex Summary" in response.text + assert "$22,000.00" in response.text or "22,000" in response.text + + with unit_of_work_factory() as uow: + assert uow.project_processing_opex is not None + assert uow.scenario_processing_opex is not None + + project_snapshots = uow.project_processing_opex.list_for_project( + project_id) + scenario_snapshots = uow.scenario_processing_opex.list_for_scenario( + scenario_id) + + assert len(project_snapshots) == 1 + assert len(scenario_snapshots) == 1 + + project_snapshot = project_snapshots[0] + scenario_snapshot = scenario_snapshots[0] + + assert project_snapshot.overall_annual is not None + assert float( + project_snapshot.overall_annual) == pytest.approx(22_000.0) + assert project_snapshot.escalated_total is not None + assert float( + project_snapshot.escalated_total) == pytest.approx(58_330.0) + assert project_snapshot.apply_escalation is True + assert project_snapshot.component_count == 2 + assert project_snapshot.currency_code == "USD" + + assert scenario_snapshot.overall_annual is not None + assert float( + scenario_snapshot.overall_annual) == pytest.approx(22_000.0) + assert scenario_snapshot.escalated_total is not None + assert float( + scenario_snapshot.escalated_total) == pytest.approx(58_330.0) + assert scenario_snapshot.apply_escalation is True + assert scenario_snapshot.component_count == 2 + assert scenario_snapshot.currency_code == "USD" + + +def test_processing_opex_calculation_json_flow( + client: TestClient, + unit_of_work_factory: Callable[[], UnitOfWork], +) -> None: + project_id = _create_project(client, "Opex JSON Project") + scenario_id = _create_scenario(client, project_id, "Opex JSON Scenario") + + payload = { + "components": [ + { + "name": "Reagents", + "category": "materials", + "unit_cost": 400, + "quantity": 10, + "frequency": "monthly", + "currency": "USD", + "period_start": 1, + "period_end": 3, + }, + { + "name": "Labor", + "category": "labor", + "unit_cost": 1500, + "quantity": 4, + "frequency": "weekly", + "currency": "USD", + "period_start": 1, + "period_end": 3, + }, + { + "name": "Maintenance", + "category": "maintenance", + "unit_cost": 12000, + "quantity": 1, + "frequency": "annually", + "currency": "USD", + "period_start": 1, + "period_end": 3, + }, + ], + "parameters": { + "currency_code": "USD", + "escalation_pct": 4, + "discount_rate_pct": 2, + "evaluation_horizon_years": 3, + "apply_escalation": True, + }, + "options": {"persist": True, "snapshot_notes": "Processing opex JSON flow"}, + } + + response = client.post( + f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}", + json=payload, + ) + assert response.status_code == 200 + + data = response.json() + assert data["currency"] == "USD" + + expected_overall = 372_000.0 + escalation_factor = 1 + (payload["parameters"]["escalation_pct"] / 100.0) + expected_timeline = [ + expected_overall * (escalation_factor ** i) for i in range(payload["parameters"]["evaluation_horizon_years"]) + ] + expected_escalated_total = sum(expected_timeline) + expected_average = expected_escalated_total / len(expected_timeline) + + assert data["totals"]["overall_annual"] == pytest.approx(expected_overall) + assert data["totals"]["escalated_total"] == pytest.approx( + expected_escalated_total) + assert data["totals"]["escalation_pct"] == pytest.approx(4.0) + + by_category = {entry["category"] : entry for entry in data["totals"]["by_category"]} + assert by_category["materials"]["annual_cost"] == pytest.approx(48_000.0) + assert by_category["labor"]["annual_cost"] == pytest.approx(312_000.0) + assert by_category["maintenance"]["annual_cost"] == pytest.approx(12_000.0) + + assert len(data["timeline"]) == 3 + for index, entry in enumerate(data["timeline"], start=0): + assert entry["period"] == index + 1 + assert entry["base_cost"] == pytest.approx(expected_overall) + assert entry["escalated_cost"] == pytest.approx( + expected_timeline[index]) + + assert data["metrics"]["annual_average"] == pytest.approx(expected_average) + + with unit_of_work_factory() as uow: + assert uow.project_processing_opex is not None + assert uow.scenario_processing_opex is not None + + project_snapshot = uow.project_processing_opex.latest_for_project( + project_id) + scenario_snapshot = uow.scenario_processing_opex.latest_for_scenario( + scenario_id) + + assert project_snapshot is not None + assert scenario_snapshot is not None + + assert project_snapshot.overall_annual is not None + assert float(project_snapshot.overall_annual) == pytest.approx( + expected_overall) + assert project_snapshot.escalated_total is not None + assert float(project_snapshot.escalated_total) == pytest.approx( + expected_escalated_total) + assert project_snapshot.apply_escalation is True + + assert scenario_snapshot.annual_average is not None + assert float(scenario_snapshot.annual_average) == pytest.approx( + expected_average) + assert scenario_snapshot.apply_escalation is True + + +@pytest.mark.parametrize("content_type", ["form", "json"]) +def test_processing_opex_calculation_currency_mismatch( + client: TestClient, + unit_of_work_factory: Callable[[], UnitOfWork], + content_type: str, +) -> None: + project_id = _create_project( + client, f"Opex {content_type.title()} Error Project") + scenario_id = _create_scenario( + client, project_id, f"Opex {content_type.title()} Error Scenario") + + if content_type == "json": + payload = { + "components": [ + { + "name": "Power", + "category": "energy", + "unit_cost": 500, + "quantity": 1, + "frequency": "monthly", + "currency": "USD", + "period_start": 1, + "period_end": 2, + } + ], + "parameters": {"currency_code": "CAD"}, + "options": {"persist": True}, + } + response = client.post( + f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}", + json=payload, + ) + assert response.status_code == 422 + body = response.json() + assert "Component currency does not match" in body.get("message", "") + assert any( + "components[0].currency" in entry for entry in body.get("errors", [])) + else: + response = client.post( + f"/calculations/processing-opex?project_id={project_id}&scenario_id={scenario_id}", + data={ + "components[0][name]": "Power", + "components[0][category]": "energy", + "components[0][unit_cost]": "500", + "components[0][quantity]": "1", + "components[0][frequency]": "monthly", + "components[0][currency]": "USD", + "components[0][period_start]": "1", + "components[0][period_end]": "2", + "parameters[currency_code]": "CAD", + "options[persist]": "1", + }, + ) + assert response.status_code == 422 + assert hasattr(response, "context") + context = getattr(response, "context", {}) or {} + combined_errors = [ + str(entry) + for entry in ( + (context.get("errors") or []) + + (context.get("component_errors") or []) + ) + ] + assert any( + "components[0].currency" in entry for entry in combined_errors) + + with unit_of_work_factory() as uow: + assert uow.project_processing_opex is not None + assert uow.scenario_processing_opex is not None + + project_snapshots = uow.project_processing_opex.list_for_project( + project_id) + scenario_snapshots = uow.scenario_processing_opex.list_for_scenario( + scenario_id) + + assert project_snapshots == [] + assert scenario_snapshots == [] diff --git a/tests/services/test_calculations_processing_opex.py b/tests/services/test_calculations_processing_opex.py new file mode 100644 index 0000000..f04a8db --- /dev/null +++ b/tests/services/test_calculations_processing_opex.py @@ -0,0 +1,159 @@ +import pytest + +from schemas.calculations import ( + ProcessingOpexCalculationRequest, + ProcessingOpexComponentInput, + ProcessingOpexOptions, + ProcessingOpexParameters, +) +from services.calculations import calculate_processing_opex +from services.exceptions import OpexValidationError + + +def _component(**overrides) -> ProcessingOpexComponentInput: + defaults = { + "id": None, + "name": "Component", + "category": "energy", + "unit_cost": 1000.0, + "quantity": 1.0, + "frequency": "monthly", + "currency": "USD", + "period_start": 1, + "period_end": 1, + "notes": None, + } + defaults.update(overrides) + return ProcessingOpexComponentInput(**defaults) + + +def test_calculate_processing_opex_success(): + request = ProcessingOpexCalculationRequest( + components=[ + _component( + name="Power", + category="energy", + unit_cost=1000.0, + quantity=1, + frequency="monthly", + period_start=1, + period_end=3, + ), + _component( + name="Maintenance", + category="maintenance", + unit_cost=2500.0, + quantity=1, + frequency="quarterly", + period_start=1, + period_end=2, + ), + ], + parameters=ProcessingOpexParameters( + currency_code="USD", + escalation_pct=5, + discount_rate_pct=None, + evaluation_horizon_years=2, + apply_escalation=True, + ), + options=ProcessingOpexOptions(persist=True, snapshot_notes=None), + ) + + result = calculate_processing_opex(request) + + assert result.currency == "USD" + assert result.options.persist is True + + assert result.totals.overall_annual == pytest.approx(22_000.0) + assert result.totals.escalated_total == pytest.approx(58_330.0, rel=1e-4) + assert result.totals.escalation_pct == pytest.approx(5.0) + + categories = {entry.category: entry for entry in result.totals.by_category} + assert categories["energy"].annual_cost == pytest.approx(12_000.0) + assert categories["maintenance"].annual_cost == pytest.approx(10_000.0) + + assert len(result.timeline) == 3 + timeline = {entry.period: entry for entry in result.timeline} + assert timeline[1].base_cost == pytest.approx(22_000.0) + assert timeline[2].base_cost == pytest.approx(22_000.0) + assert timeline[3].base_cost == pytest.approx(12_000.0) + assert timeline[1].escalated_cost == pytest.approx(22_000.0) + assert timeline[2].escalated_cost == pytest.approx(23_100.0, rel=1e-4) + assert timeline[3].escalated_cost == pytest.approx(13_230.0, rel=1e-4) + + assert result.metrics.annual_average == pytest.approx( + 19_443.3333, rel=1e-4) + + assert len(result.components) == 2 + assert result.components[0].frequency == "monthly" + assert result.components[1].frequency == "quarterly" + + +def test_calculate_processing_opex_currency_mismatch(): + request = ProcessingOpexCalculationRequest( + components=[_component(currency="USD")], + parameters=ProcessingOpexParameters( + currency_code="CAD", + escalation_pct=None, + discount_rate_pct=None, + evaluation_horizon_years=10, + ), + ) + + with pytest.raises(OpexValidationError) as exc: + calculate_processing_opex(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_processing_opex_unsupported_frequency(): + request = ProcessingOpexCalculationRequest( + components=[_component(frequency="biweekly")], + parameters=ProcessingOpexParameters( + currency_code="USD", + escalation_pct=None, + discount_rate_pct=None, + evaluation_horizon_years=2, + ), + ) + + with pytest.raises(OpexValidationError) as exc: + calculate_processing_opex(request) + + assert "Unsupported frequency" in exc.value.message + assert exc.value.field_errors and "components[0].frequency" in exc.value.field_errors[0] + + +def test_calculate_processing_opex_requires_components(): + request = ProcessingOpexCalculationRequest(components=[]) + + with pytest.raises(OpexValidationError) as exc: + calculate_processing_opex(request) + + assert "At least one processing opex component" in exc.value.message + assert exc.value.field_errors and "components" in exc.value.field_errors[0] + + +def test_calculate_processing_opex_extends_evaluation_horizon(): + request = ProcessingOpexCalculationRequest( + components=[ + _component(period_start=1, period_end=4), + ], + parameters=ProcessingOpexParameters( + currency_code="USD", + discount_rate_pct=0, + escalation_pct=0, + evaluation_horizon_years=2, + apply_escalation=False, + ), + ) + + result = calculate_processing_opex(request) + + assert len(result.timeline) == 4 + assert result.timeline[-1].period == 4 + assert all(entry.escalated_cost is None for entry in result.timeline) + assert result.timeline[-1].base_cost == pytest.approx(12_000.0) + assert result.metrics.annual_average == pytest.approx( + 12_000.0, rel=1e-4)