allucanget/calminer
2
0
Fork 0
Code Pull Requests Actions Packages Releases Activity

feat: add scenarios list page with metrics and quick actions
Some checks failed
CI / test (push) Has been skipped
Details
CI / build (push) Has been skipped
Details
CI / lint (push) Failing after 15s
Details
CI / deploy (push) Has been skipped
Details

Browse Source 
- Introduced a new template for listing scenarios associated with a project.
- Added metrics for total, active, draft, and archived scenarios.
- Implemented quick actions for creating new scenarios and reviewing project overview.
- Enhanced navigation with breadcrumbs for better user experience.

refactor: update Opex and Profitability templates for consistency

- Changed titles and button labels for clarity in Opex and Profitability templates.
- Updated form IDs and action URLs for better alignment with new naming conventions.
- Improved navigation links to include scenario and project overviews.

test: add integration tests for Opex calculations

- Created new tests for Opex calculation HTML and JSON flows.
- Validated successful calculations and ensured correct data persistence.
- Implemented tests for currency mismatch and unsupported frequency scenarios.

test: enhance project and scenario route tests

- Added tests to verify scenario list rendering and calculator shortcuts.
- Ensured scenario detail pages link back to the portfolio correctly.
- Validated project detail pages show associated scenarios accurately.
This commit is contained in:
zwitschi
2025-11-13 16:21:36 +01:00
parent 4f00bf0d3c
commit 522b1e4105
54 changed files with 3419 additions and 700 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
services/calculations.py
View File

@@ -29,14 +29,14 @@ from schemas.calculations import (
CapexTotals,
CapexTimelineEntry,
CashFlowEntry,
ProcessingOpexCalculationRequest,
ProcessingOpexCalculationResult,
ProcessingOpexCategoryBreakdown,
ProcessingOpexComponentInput,
ProcessingOpexMetrics,
ProcessingOpexParameters,
ProcessingOpexTotals,
ProcessingOpexTimelineEntry,
OpexCalculationRequest,
OpexCalculationResult,
OpexCategoryBreakdown,
OpexComponentInput,
OpexMetrics,
OpexParameters,
OpexTotals,
OpexTimelineEntry,
ProfitabilityCalculationRequest,
ProfitabilityCalculationResult,
ProfitabilityCosts,
@@ -101,20 +101,20 @@ def _generate_cash_flows(
*,
periods: int,
net_per_period: float,
initial_capex: float,
capex: float,
) -> tuple[list[CashFlow], list[CashFlowEntry]]:
"""Create cash flow structures for financial metric calculations."""
cash_flow_models: list[CashFlow] = [
CashFlow(amount=-initial_capex, period_index=0)
CashFlow(amount=-capex, period_index=0)
]
cash_flow_entries: list[CashFlowEntry] = [
CashFlowEntry(
period=0,
revenue=0.0,
processing_opex=0.0,
opex=0.0,
sustaining_capex=0.0,
net=-initial_capex,
net=-capex,
)
]
@@ -125,7 +125,7 @@ def _generate_cash_flows(
CashFlowEntry(
period=period,
revenue=0.0,
processing_opex=0.0,
opex=0.0,
sustaining_capex=0.0,
net=net_per_period,
)
@@ -159,26 +159,26 @@ def calculate_profitability(
revenue_total = float(pricing_result.net_revenue)
revenue_per_period = revenue_total / periods
processing_total = float(request.processing_opex) * periods
processing_total = float(request.opex) * periods
sustaining_total = float(request.sustaining_capex) * periods
initial_capex = float(request.initial_capex)
capex = float(request.capex)
net_per_period = (
revenue_per_period
- float(request.processing_opex)
- float(request.opex)
- float(request.sustaining_capex)
)
cash_flow_models, cash_flow_entries = _generate_cash_flows(
periods=periods,
net_per_period=net_per_period,
initial_capex=initial_capex,
capex=capex,
)
# Update per-period entries to include explicit costs for presentation
for entry in cash_flow_entries[1:]:
entry.revenue = revenue_per_period
entry.processing_opex = float(request.processing_opex)
entry.opex = float(request.opex)
entry.sustaining_capex = float(request.sustaining_capex)
entry.net = net_per_period
@@ -196,7 +196,7 @@ def calculate_profitability(
except (ValueError, PaybackNotReachedError):
payback_value = None
total_costs = processing_total + sustaining_total + initial_capex
total_costs = processing_total + sustaining_total + capex
total_net = revenue_total - total_costs
if revenue_total == 0:
@@ -212,9 +212,9 @@ def calculate_profitability(
str(exc), ["currency_code"]) from exc
costs = ProfitabilityCosts(
processing_opex_total=processing_total,
opex_total=processing_total,
sustaining_capex_total=sustaining_total,
initial_capex=initial_capex,
capex=capex,
)
metrics = ProfitabilityMetrics(
@@ -354,18 +354,18 @@ def calculate_initial_capex(
)
def calculate_processing_opex(
request: ProcessingOpexCalculationRequest,
) -> ProcessingOpexCalculationResult:
"""Aggregate processing opex components into annual totals and timeline."""
def calculate_opex(
request: OpexCalculationRequest,
) -> OpexCalculationResult:
"""Aggregate opex components into annual totals and timeline."""
if not request.components:
raise OpexValidationError(
"At least one processing opex component is required for calculation.",
"At least one opex component is required for calculation.",
["components"],
)
parameters: ProcessingOpexParameters = request.parameters
parameters: OpexParameters = request.parameters
base_currency = parameters.currency_code
if base_currency:
try:
@@ -388,7 +388,7 @@ def calculate_processing_opex(
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] = []
normalised_components: list[OpexComponentInput] = []
max_period_end = evaluation_horizon
@@ -448,7 +448,7 @@ def calculate_processing_opex(
timeline_totals[period] += annual_cost
normalised_components.append(
ProcessingOpexComponentInput(
OpexComponentInput(
id=component.id,
name=component.name,
category=component.category,
@@ -471,7 +471,7 @@ def calculate_processing_opex(
str(exc), ["parameters.currency_code"]
) from exc
timeline_entries: list[ProcessingOpexTimelineEntry] = []
timeline_entries: list[OpexTimelineEntry] = []
escalated_values: list[float] = []
overall_annual = timeline_totals.get(1, 0.0)
escalated_total = 0.0
@@ -486,7 +486,7 @@ def calculate_processing_opex(
timeline_escalated[period] = escalated_cost
escalated_total += escalated_cost
timeline_entries.append(
ProcessingOpexTimelineEntry(
OpexTimelineEntry(
period=period,
base_cost=base_cost,
escalated_cost=escalated_cost if apply_escalation else None,
@@ -494,31 +494,31 @@ def calculate_processing_opex(
)
escalated_values.append(escalated_cost)
category_breakdowns: list[ProcessingOpexCategoryBreakdown] = []
category_breakdowns: list[OpexCategoryBreakdown] = []
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(
OpexCategoryBreakdown(
category=category,
annual_cost=total,
share=share,
)
)
metrics = ProcessingOpexMetrics(
metrics = OpexMetrics(
annual_average=fmean(escalated_values) if escalated_values else None,
cost_per_ton=None,
)
totals = ProcessingOpexTotals(
totals = OpexTotals(
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(
return OpexCalculationResult(
totals=totals,
timeline=timeline_entries,
metrics=metrics,
@@ -532,5 +532,5 @@ def calculate_processing_opex(
__all__ = [
"calculate_profitability",
"calculate_initial_capex",
"calculate_processing_opex",
"calculate_opex",
]

188
services/navigation.py Normal file
View File

@@ -0,0 +1,188 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Iterable, List, Optional, Sequence
from fastapi import Request
from models.navigation import NavigationGroup, NavigationLink
from services.repositories import NavigationRepository
from services.session import AuthSession
@dataclass(slots=True)
class NavigationLinkDTO:
id: int
label: str
href: str
match_prefix: str | None
icon: str | None
tooltip: str | None
is_external: bool
children: List["NavigationLinkDTO"] = field(default_factory=list)
@dataclass(slots=True)
class NavigationGroupDTO:
id: int
label: str
icon: str | None
tooltip: str | None
links: List[NavigationLinkDTO] = field(default_factory=list)
@dataclass(slots=True)
class NavigationSidebarDTO:
groups: List[NavigationGroupDTO]
roles: tuple[str, ...]
class NavigationService:
"""Build navigation payloads filtered for the current session."""
def __init__(self, repository: NavigationRepository) -> None:
self._repository = repository
def build_sidebar(
self,
*,
session: AuthSession,
request: Request | None = None,
include_disabled: bool = False,
) -> NavigationSidebarDTO:
roles = self._collect_roles(session)
groups = self._repository.list_groups_with_links(
include_disabled=include_disabled
)
context = self._derive_context(request)
mapped_groups: List[NavigationGroupDTO] = []
for group in groups:
if not include_disabled and not group.is_enabled:
continue
mapped_links = self._map_links(
group.links,
roles,
request=request,
include_disabled=include_disabled,
context=context,
)
if not mapped_links and not include_disabled:
continue
mapped_groups.append(
NavigationGroupDTO(
id=group.id,
label=group.label,
icon=group.icon,
tooltip=group.tooltip,
links=mapped_links,
)
)
return NavigationSidebarDTO(groups=mapped_groups, roles=roles)
def _map_links(
self,
links: Sequence[NavigationLink],
roles: Iterable[str],
*,
request: Request | None,
include_disabled: bool,
context: dict[str, str | None],
) -> List[NavigationLinkDTO]:
resolved_roles = tuple(roles)
mapped: List[NavigationLinkDTO] = []
for link in sorted(links, key=lambda l: (l.sort_order, l.id)):
if not include_disabled and (not link.is_enabled):
continue
if not self._link_visible(link, resolved_roles, include_disabled):
continue
href = self._resolve_href(link, request=request, context=context)
if not href:
continue
children = self._map_links(
link.children,
resolved_roles,
request=request,
include_disabled=include_disabled,
context=context,
)
match_prefix = link.match_prefix or href
mapped.append(
NavigationLinkDTO(
id=link.id,
label=link.label,
href=href,
match_prefix=match_prefix,
icon=link.icon,
tooltip=link.tooltip,
is_external=link.is_external,
children=children,
)
)
return mapped
@staticmethod
def _collect_roles(session: AuthSession) -> tuple[str, ...]:
roles = tuple((session.role_slugs or ()) if session else ())
if session and session.is_authenticated:
return roles
if "anonymous" in roles:
return roles
return roles + ("anonymous",)
@staticmethod
def _derive_context(request: Request | None) -> dict[str, str | None]:
if request is None:
return {"project_id": None, "scenario_id": None}
project_id = request.path_params.get(
"project_id") if hasattr(request, "path_params") else None
scenario_id = request.path_params.get(
"scenario_id") if hasattr(request, "path_params") else None
if not project_id:
project_id = request.query_params.get("project_id")
if not scenario_id:
scenario_id = request.query_params.get("scenario_id")
return {"project_id": project_id, "scenario_id": scenario_id}
def _resolve_href(
self,
link: NavigationLink,
*,
request: Request | None,
context: dict[str, str | None],
) -> str | None:
if link.route_name:
if request is None:
# Fallback to route name when no request is available
return f"/{link.route_name.replace('.', '/')}"
if link.slug in {"profitability", "profitability-calculator"}:
project_id = context.get("project_id")
scenario_id = context.get("scenario_id")
if project_id and scenario_id:
try:
return request.url_for(
link.route_name,
project_id=project_id,
scenario_id=scenario_id,
)
except Exception: # pragma: no cover - defensive
pass
try:
return request.url_for(link.route_name)
except Exception: # pragma: no cover - defensive
return link.href_override
return link.href_override
@staticmethod
def _link_visible(
link: NavigationLink,
roles: Iterable[str],
include_disabled: bool,
) -> bool:
role_tuple = tuple(roles)
if not include_disabled and not link.is_enabled:
return False
if not link.required_roles:
return True
role_set = set(role_tuple)
return any(role in role_set for role in link.required_roles)

114
services/repositories.py
View File

@@ -17,12 +17,14 @@ from models import (
PricingSettings,
ProjectCapexSnapshot,
ProjectProfitability,
ProjectProcessingOpexSnapshot,
ProjectOpexSnapshot,
NavigationGroup,
NavigationLink,
Role,
Scenario,
ScenarioCapexSnapshot,
ScenarioProfitability,
ScenarioProcessingOpexSnapshot,
ScenarioOpexSnapshot,
ScenarioStatus,
SimulationParameter,
User,
@@ -38,6 +40,54 @@ def _enum_value(e):
return getattr(e, "value", e)
class NavigationRepository:
"""Persistence operations for navigation metadata."""
def __init__(self, session: Session) -> None:
self.session = session
def list_groups_with_links(
self,
*,
include_disabled: bool = False,
) -> Sequence[NavigationGroup]:
stmt = (
select(NavigationGroup)
.options(
selectinload(NavigationGroup.links)
.selectinload(NavigationLink.children)
)
.order_by(NavigationGroup.sort_order, NavigationGroup.id)
)
if not include_disabled:
stmt = stmt.where(NavigationGroup.is_enabled.is_(True))
return self.session.execute(stmt).scalars().all()
def get_group_by_slug(self, slug: str) -> NavigationGroup | None:
stmt = select(NavigationGroup).where(NavigationGroup.slug == slug)
return self.session.execute(stmt).scalar_one_or_none()
def get_link_by_slug(
self,
slug: str,
*,
group_id: int | None = None,
) -> NavigationLink | None:
stmt = select(NavigationLink).where(NavigationLink.slug == slug)
if group_id is not None:
stmt = stmt.where(NavigationLink.group_id == group_id)
return self.session.execute(stmt).scalar_one_or_none()
def add_group(self, group: NavigationGroup) -> NavigationGroup:
self.session.add(group)
self.session.flush()
return group
def add_link(self, link: NavigationLink) -> NavigationLink:
self.session.add(link)
self.session.flush()
return link
class ProjectRepository:
"""Persistence operations for Project entities."""
@@ -573,15 +623,15 @@ class ScenarioCapexRepository:
self.session.delete(entity)
class ProjectProcessingOpexRepository:
"""Persistence operations for project-level processing opex snapshots."""
class ProjectOpexRepository:
"""Persistence operations for project-level opex snapshots."""
def __init__(self, session: Session) -> None:
self.session = session
def create(
self, snapshot: ProjectProcessingOpexSnapshot
) -> ProjectProcessingOpexSnapshot:
self, snapshot: ProjectOpexSnapshot
) -> ProjectOpexSnapshot:
self.session.add(snapshot)
self.session.flush()
return snapshot
@@ -591,11 +641,11 @@ class ProjectProcessingOpexRepository:
project_id: int,
*,
limit: int | None = None,
) -> Sequence[ProjectProcessingOpexSnapshot]:
) -> Sequence[ProjectOpexSnapshot]:
stmt = (
select(ProjectProcessingOpexSnapshot)
.where(ProjectProcessingOpexSnapshot.project_id == project_id)
.order_by(ProjectProcessingOpexSnapshot.calculated_at.desc())
select(ProjectOpexSnapshot)
.where(ProjectOpexSnapshot.project_id == project_id)
.order_by(ProjectOpexSnapshot.calculated_at.desc())
)
if limit is not None:
stmt = stmt.limit(limit)
@@ -604,36 +654,36 @@ class ProjectProcessingOpexRepository:
def latest_for_project(
self,
project_id: int,
) -> ProjectProcessingOpexSnapshot | None:
) -> ProjectOpexSnapshot | None:
stmt = (
select(ProjectProcessingOpexSnapshot)
.where(ProjectProcessingOpexSnapshot.project_id == project_id)
.order_by(ProjectProcessingOpexSnapshot.calculated_at.desc())
select(ProjectOpexSnapshot)
.where(ProjectOpexSnapshot.project_id == project_id)
.order_by(ProjectOpexSnapshot.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
stmt = select(ProjectOpexSnapshot).where(
ProjectOpexSnapshot.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"
f"Project opex snapshot {snapshot_id} not found"
)
self.session.delete(entity)
class ScenarioProcessingOpexRepository:
"""Persistence operations for scenario-level processing opex snapshots."""
class ScenarioOpexRepository:
"""Persistence operations for scenario-level opex snapshots."""
def __init__(self, session: Session) -> None:
self.session = session
def create(
self, snapshot: ScenarioProcessingOpexSnapshot
) -> ScenarioProcessingOpexSnapshot:
self, snapshot: ScenarioOpexSnapshot
) -> ScenarioOpexSnapshot:
self.session.add(snapshot)
self.session.flush()
return snapshot
@@ -643,11 +693,11 @@ class ScenarioProcessingOpexRepository:
scenario_id: int,
*,
limit: int | None = None,
) -> Sequence[ScenarioProcessingOpexSnapshot]:
) -> Sequence[ScenarioOpexSnapshot]:
stmt = (
select(ScenarioProcessingOpexSnapshot)
.where(ScenarioProcessingOpexSnapshot.scenario_id == scenario_id)
.order_by(ScenarioProcessingOpexSnapshot.calculated_at.desc())
select(ScenarioOpexSnapshot)
.where(ScenarioOpexSnapshot.scenario_id == scenario_id)
.order_by(ScenarioOpexSnapshot.calculated_at.desc())
)
if limit is not None:
stmt = stmt.limit(limit)
@@ -656,23 +706,23 @@ class ScenarioProcessingOpexRepository:
def latest_for_scenario(
self,
scenario_id: int,
) -> ScenarioProcessingOpexSnapshot | None:
) -> ScenarioOpexSnapshot | None:
stmt = (
select(ScenarioProcessingOpexSnapshot)
.where(ScenarioProcessingOpexSnapshot.scenario_id == scenario_id)
.order_by(ScenarioProcessingOpexSnapshot.calculated_at.desc())
select(ScenarioOpexSnapshot)
.where(ScenarioOpexSnapshot.scenario_id == scenario_id)
.order_by(ScenarioOpexSnapshot.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
stmt = select(ScenarioOpexSnapshot).where(
ScenarioOpexSnapshot.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"
f"Scenario opex snapshot {snapshot_id} not found"
)
self.session.delete(entity)

12
services/session.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal, Optional, TYPE_CHECKING
from typing import Iterable, Literal, Optional, TYPE_CHECKING
from fastapi import Request, Response
@@ -67,6 +67,7 @@ class AuthSession:
tokens: SessionTokens
user: Optional["User"] = None
scopes: tuple[str, ...] = ()
role_slugs: tuple[str, ...] = ()
issued_access_token: Optional[str] = None
issued_refresh_token: Optional[str] = None
clear_cookies: bool = False
@@ -77,7 +78,10 @@ class AuthSession:
@classmethod
def anonymous(cls) -> "AuthSession":
return cls(tokens=SessionTokens(access_token=None, refresh_token=None))
return cls(
tokens=SessionTokens(access_token=None, refresh_token=None),
role_slugs=(),
)
def issue_tokens(
self,
@@ -100,6 +104,10 @@ class AuthSession:
self.tokens = SessionTokens(access_token=None, refresh_token=None)
self.user = None
self.scopes = ()
self.role_slugs = ()
def set_role_slugs(self, roles: Iterable[str]) -> None:
self.role_slugs = tuple(dict.fromkeys(role.strip().lower() for role in roles if role))
def extract_session_tokens(request: Request, strategy: SessionStrategy) -> SessionTokens:

20
services/unit_of_work.py
View File

@@ -14,12 +14,12 @@ from services.repositories import (
PricingSettingsSeedResult,
ProjectRepository,
ProjectProfitabilityRepository,
ProjectProcessingOpexRepository,
ProjectOpexRepository,
ProjectCapexRepository,
RoleRepository,
ScenarioRepository,
ScenarioProfitabilityRepository,
ScenarioProcessingOpexRepository,
ScenarioOpexRepository,
ScenarioCapexRepository,
SimulationParameterRepository,
UserRepository,
@@ -27,6 +27,7 @@ from services.repositories import (
ensure_default_pricing_settings,
ensure_default_roles,
pricing_settings_to_metadata,
NavigationRepository,
)
from services.scenario_validation import ScenarioComparisonValidator
@@ -44,13 +45,14 @@ 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.project_opex: ProjectOpexRepository | None = None
self.scenario_profitability: ScenarioProfitabilityRepository | None = None
self.scenario_capex: ScenarioCapexRepository | None = None
self.scenario_processing_opex: ScenarioProcessingOpexRepository | None = None
self.scenario_opex: ScenarioOpexRepository | None = None
self.users: UserRepository | None = None
self.roles: RoleRepository | None = None
self.pricing_settings: PricingSettingsRepository | None = None
self.navigation: NavigationRepository | None = None
def __enter__(self) -> "UnitOfWork":
self.session = self._session_factory()
@@ -62,17 +64,18 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]):
self.project_profitability = ProjectProfitabilityRepository(
self.session)
self.project_capex = ProjectCapexRepository(self.session)
self.project_processing_opex = ProjectProcessingOpexRepository(
self.project_opex = ProjectOpexRepository(
self.session)
self.scenario_profitability = ScenarioProfitabilityRepository(
self.session
)
self.scenario_capex = ScenarioCapexRepository(self.session)
self.scenario_processing_opex = ScenarioProcessingOpexRepository(
self.scenario_opex = ScenarioOpexRepository(
self.session)
self.users = UserRepository(self.session)
self.roles = RoleRepository(self.session)
self.pricing_settings = PricingSettingsRepository(self.session)
self.navigation = NavigationRepository(self.session)
self._scenario_validator = ScenarioComparisonValidator()
return self
@@ -90,13 +93,14 @@ class UnitOfWork(AbstractContextManager["UnitOfWork"]):
self.simulation_parameters = None
self.project_profitability = None
self.project_capex = None
self.project_processing_opex = None
self.project_opex = None
self.scenario_profitability = None
self.scenario_capex = None
self.scenario_processing_opex = None
self.scenario_opex = None
self.users = None
self.roles = None
self.pricing_settings = None
self.navigation = None
def flush(self) -> None:
if not self.session: