Files
calminer/routes/ui.py
zwitschi 5a84445e90 feat: Enhance equipment, maintenance, production, and simulation management interfaces
- Updated equipment.html to include scenario filtering, equipment addition form, and dynamic equipment listing.
- Enhanced maintenance.html with scenario filtering, maintenance entry form, and dynamic equipment selection.
- Improved production.html to allow scenario filtering and production output entry.
- Revamped reporting.html to display scenario KPI summaries with refresh functionality.
- Expanded simulations.html to support scenario selection, simulation run history, and detailed results display.
- Refactored base_header.html for improved navigation structure and active link highlighting.
2025-10-21 00:09:00 +02:00

595 lines
22 KiB
Python

from collections import defaultdict
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from models.capex import Capex
from models.consumption import Consumption
from models.equipment import Equipment
from models.maintenance import Maintenance
from models.opex import Opex
from models.parameters import Parameter
from models.production_output import ProductionOutput
from models.scenario import Scenario
from models.simulation_result import SimulationResult
from routes.dependencies import get_db
from services.reporting import generate_report
router = APIRouter()
# Set up Jinja2 templates directory
templates = Jinja2Templates(directory="templates")
def _context(request: Request, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"request": request,
"current_year": datetime.now(timezone.utc).year,
}
if extra:
payload.update(extra)
return payload
def _render(
request: Request,
template_name: str,
extra: Optional[Dict[str, Any]] = None,
):
return templates.TemplateResponse(template_name, _context(request, extra))
def _format_currency(value: float) -> str:
return f"${value:,.2f}"
def _format_decimal(value: float) -> str:
return f"{value:,.2f}"
def _format_int(value: int) -> str:
return f"{value:,}"
def _load_scenarios(db: Session) -> Dict[str, Any]:
scenarios: list[Dict[str, Any]] = [
{
"id": item.id,
"name": item.name,
"description": item.description,
}
for item in db.query(Scenario).order_by(Scenario.name).all()
]
return {"scenarios": scenarios}
def _load_parameters(db: Session) -> Dict[str, Any]:
grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for param in db.query(Parameter).order_by(Parameter.scenario_id, Parameter.id):
grouped[param.scenario_id].append(
{
"id": param.id,
"name": param.name,
"value": param.value,
"distribution_type": param.distribution_type,
"distribution_parameters": param.distribution_parameters,
}
)
return {"parameters_by_scenario": dict(grouped)}
def _load_costs(db: Session) -> Dict[str, Any]:
capex_grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for capex in (
db.query(Capex)
.order_by(Capex.scenario_id, Capex.id)
.all()
):
capex_grouped[int(getattr(capex, "scenario_id"))].append(
{
"id": int(getattr(capex, "id")),
"scenario_id": int(getattr(capex, "scenario_id")),
"amount": float(getattr(capex, "amount", 0.0)),
"description": getattr(capex, "description", "") or "",
}
)
opex_grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for opex in (
db.query(Opex)
.order_by(Opex.scenario_id, Opex.id)
.all()
):
opex_grouped[int(getattr(opex, "scenario_id"))].append(
{
"id": int(getattr(opex, "id")),
"scenario_id": int(getattr(opex, "scenario_id")),
"amount": float(getattr(opex, "amount", 0.0)),
"description": getattr(opex, "description", "") or "",
}
)
return {
"capex_by_scenario": dict(capex_grouped),
"opex_by_scenario": dict(opex_grouped),
}
def _load_consumption(db: Session) -> Dict[str, Any]:
grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for record in (
db.query(Consumption)
.order_by(Consumption.scenario_id, Consumption.id)
.all()
):
record_id = int(getattr(record, "id"))
scenario_id = int(getattr(record, "scenario_id"))
amount_value = float(getattr(record, "amount", 0.0))
description = getattr(record, "description", "") or ""
grouped[scenario_id].append(
{
"id": record_id,
"scenario_id": scenario_id,
"amount": amount_value,
"description": description,
}
)
return {"consumption_by_scenario": dict(grouped)}
def _load_production(db: Session) -> Dict[str, Any]:
grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for record in (
db.query(ProductionOutput)
.order_by(ProductionOutput.scenario_id, ProductionOutput.id)
.all()
):
record_id = int(getattr(record, "id"))
scenario_id = int(getattr(record, "scenario_id"))
amount_value = float(getattr(record, "amount", 0.0))
description = getattr(record, "description", "") or ""
grouped[scenario_id].append(
{
"id": record_id,
"scenario_id": scenario_id,
"amount": amount_value,
"description": description,
}
)
return {"production_by_scenario": dict(grouped)}
def _load_equipment(db: Session) -> Dict[str, Any]:
grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for record in (
db.query(Equipment)
.order_by(Equipment.scenario_id, Equipment.id)
.all()
):
record_id = int(getattr(record, "id"))
scenario_id = int(getattr(record, "scenario_id"))
name_value = getattr(record, "name", "") or ""
description = getattr(record, "description", "") or ""
grouped[scenario_id].append(
{
"id": record_id,
"scenario_id": scenario_id,
"name": name_value,
"description": description,
}
)
return {"equipment_by_scenario": dict(grouped)}
def _load_maintenance(db: Session) -> Dict[str, Any]:
grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for record in (
db.query(Maintenance)
.order_by(Maintenance.scenario_id, Maintenance.maintenance_date)
.all()
):
record_id = int(getattr(record, "id"))
scenario_id = int(getattr(record, "scenario_id"))
equipment_id = int(getattr(record, "equipment_id"))
equipment_obj = getattr(record, "equipment", None)
equipment_name = getattr(
equipment_obj, "name", "") if equipment_obj else ""
maintenance_date = getattr(record, "maintenance_date", None)
cost_value = float(getattr(record, "cost", 0.0))
description = getattr(record, "description", "") or ""
grouped[scenario_id].append(
{
"id": record_id,
"scenario_id": scenario_id,
"equipment_id": equipment_id,
"equipment_name": equipment_name,
"maintenance_date": maintenance_date.isoformat() if maintenance_date else "",
"cost": cost_value,
"description": description,
}
)
return {"maintenance_by_scenario": dict(grouped)}
def _load_simulations(db: Session) -> Dict[str, Any]:
scenarios: list[Dict[str, Any]] = [
{
"id": item.id,
"name": item.name,
}
for item in db.query(Scenario).order_by(Scenario.name).all()
]
results_grouped: defaultdict[int, list[Dict[str, Any]]] = defaultdict(list)
for record in (
db.query(SimulationResult)
.order_by(SimulationResult.scenario_id, SimulationResult.iteration)
.all()
):
scenario_id = int(getattr(record, "scenario_id"))
results_grouped[scenario_id].append(
{
"iteration": int(getattr(record, "iteration")),
"result": float(getattr(record, "result", 0.0)),
}
)
runs: list[Dict[str, Any]] = []
sample_limit = 20
for item in scenarios:
scenario_id = int(item["id"])
scenario_results = results_grouped.get(scenario_id, [])
summary = generate_report(
scenario_results) if scenario_results else generate_report([])
runs.append(
{
"scenario_id": scenario_id,
"scenario_name": item["name"],
"iterations": int(summary.get("count", 0)),
"summary": summary,
"sample_results": scenario_results[:sample_limit],
}
)
return {
"simulation_scenarios": scenarios,
"simulation_runs": runs,
}
def _load_reporting(db: Session) -> Dict[str, Any]:
scenarios = _load_scenarios(db)["scenarios"]
runs = _load_simulations(db)["simulation_runs"]
summaries: list[Dict[str, Any]] = []
runs_by_scenario = {run["scenario_id"]: run for run in runs}
for scenario in scenarios:
scenario_id = scenario["id"]
run = runs_by_scenario.get(scenario_id)
summary = run["summary"] if run else generate_report([])
summaries.append(
{
"scenario_id": scenario_id,
"scenario_name": scenario["name"],
"summary": summary,
"iterations": run["iterations"] if run else 0,
}
)
return {
"report_summaries": summaries,
}
def _load_dashboard(db: Session) -> Dict[str, Any]:
scenarios = _load_scenarios(db)["scenarios"]
parameters_by_scenario = _load_parameters(db)["parameters_by_scenario"]
costs_context = _load_costs(db)
capex_by_scenario = costs_context["capex_by_scenario"]
opex_by_scenario = costs_context["opex_by_scenario"]
consumption_by_scenario = _load_consumption(db)["consumption_by_scenario"]
production_by_scenario = _load_production(db)["production_by_scenario"]
equipment_by_scenario = _load_equipment(db)["equipment_by_scenario"]
maintenance_by_scenario = _load_maintenance(db)["maintenance_by_scenario"]
simulation_context = _load_simulations(db)
simulation_runs = simulation_context["simulation_runs"]
runs_by_scenario = {
run["scenario_id"]: run for run in simulation_runs
}
def sum_amounts(grouped: Dict[int, list[Dict[str, Any]]], field: str = "amount") -> float:
total = 0.0
for items in grouped.values():
for item in items:
value = item.get(field, 0.0)
if isinstance(value, (int, float)):
total += float(value)
return total
total_capex = sum_amounts(capex_by_scenario)
total_opex = sum_amounts(opex_by_scenario)
total_consumption = sum_amounts(consumption_by_scenario)
total_production = sum_amounts(production_by_scenario)
total_maintenance_cost = sum_amounts(maintenance_by_scenario, field="cost")
total_parameters = sum(len(items)
for items in parameters_by_scenario.values())
total_equipment = sum(len(items)
for items in equipment_by_scenario.values())
total_maintenance_events = sum(len(items)
for items in maintenance_by_scenario.values())
total_simulation_iterations = sum(
run["iterations"] for run in simulation_runs)
scenario_rows: list[Dict[str, Any]] = []
scenario_labels: list[str] = []
scenario_capex: list[float] = []
scenario_opex: list[float] = []
activity_labels: list[str] = []
activity_production: list[float] = []
activity_consumption: list[float] = []
for scenario in scenarios:
scenario_id = scenario["id"]
scenario_name = scenario["name"]
param_count = len(parameters_by_scenario.get(scenario_id, []))
equipment_count = len(equipment_by_scenario.get(scenario_id, []))
maintenance_count = len(maintenance_by_scenario.get(scenario_id, []))
capex_total = sum(
float(item.get("amount", 0.0))
for item in capex_by_scenario.get(scenario_id, [])
)
opex_total = sum(
float(item.get("amount", 0.0))
for item in opex_by_scenario.get(scenario_id, [])
)
consumption_total = sum(
float(item.get("amount", 0.0))
for item in consumption_by_scenario.get(scenario_id, [])
)
production_total = sum(
float(item.get("amount", 0.0))
for item in production_by_scenario.get(scenario_id, [])
)
run = runs_by_scenario.get(scenario_id)
summary = run["summary"] if run else generate_report([])
iterations = run["iterations"] if run else 0
mean_value = float(summary.get("mean", 0.0))
scenario_rows.append(
{
"scenario_name": scenario_name,
"parameter_count": param_count,
"parameter_display": _format_int(param_count),
"equipment_count": equipment_count,
"equipment_display": _format_int(equipment_count),
"capex_total": capex_total,
"capex_display": _format_currency(capex_total),
"opex_total": opex_total,
"opex_display": _format_currency(opex_total),
"production_total": production_total,
"production_display": _format_decimal(production_total),
"consumption_total": consumption_total,
"consumption_display": _format_decimal(consumption_total),
"maintenance_count": maintenance_count,
"maintenance_display": _format_int(maintenance_count),
"iterations": iterations,
"iterations_display": _format_int(iterations),
"simulation_mean": mean_value,
"simulation_mean_display": _format_decimal(mean_value),
}
)
scenario_labels.append(scenario_name)
scenario_capex.append(capex_total)
scenario_opex.append(opex_total)
activity_labels.append(scenario_name)
activity_production.append(production_total)
activity_consumption.append(consumption_total)
scenario_rows.sort(key=lambda row: row["scenario_name"].lower())
all_simulation_results = [
{"result": float(getattr(item, "result", 0.0))}
for item in db.query(SimulationResult).all()
]
overall_report = generate_report(all_simulation_results)
overall_report_metrics = [
{"label": "Runs", "value": _format_int(
int(overall_report.get("count", 0)))},
{"label": "Mean", "value": _format_decimal(
float(overall_report.get("mean", 0.0)))},
{"label": "Median", "value": _format_decimal(
float(overall_report.get("median", 0.0)))},
{"label": "Std Dev", "value": _format_decimal(
float(overall_report.get("std_dev", 0.0)))},
{"label": "95th Percentile", "value": _format_decimal(
float(overall_report.get("percentile_95", 0.0)))},
{"label": "VaR (95%)", "value": _format_decimal(
float(overall_report.get("value_at_risk_95", 0.0)))},
{"label": "Expected Shortfall (95%)", "value": _format_decimal(
float(overall_report.get("expected_shortfall_95", 0.0)))},
]
recent_simulations: list[Dict[str, Any]] = [
{
"scenario_name": run["scenario_name"],
"iterations": run["iterations"],
"iterations_display": _format_int(run["iterations"]),
"mean_display": _format_decimal(float(run["summary"].get("mean", 0.0))),
"p95_display": _format_decimal(float(run["summary"].get("percentile_95", 0.0))),
}
for run in simulation_runs
if run["iterations"] > 0
]
recent_simulations.sort(key=lambda item: item["iterations"], reverse=True)
recent_simulations = recent_simulations[:5]
upcoming_maintenance = []
for record in (
db.query(Maintenance)
.order_by(Maintenance.maintenance_date.asc())
.limit(5)
.all()
):
maintenance_date = getattr(record, "maintenance_date", None)
upcoming_maintenance.append(
{
"scenario_name": getattr(getattr(record, "scenario", None), "name", "Unknown"),
"equipment_name": getattr(getattr(record, "equipment", None), "name", "Unknown"),
"date_display": maintenance_date.strftime("%Y-%m-%d") if maintenance_date else "",
"cost_display": _format_currency(float(getattr(record, "cost", 0.0))),
"description": getattr(record, "description", "") or "",
}
)
cost_chart_has_data = any(value > 0 for value in scenario_capex) or any(
value > 0 for value in scenario_opex
)
activity_chart_has_data = any(value > 0 for value in activity_production) or any(
value > 0 for value in activity_consumption
)
scenario_cost_chart: Dict[str, list[Any]] = {
"labels": scenario_labels,
"capex": scenario_capex,
"opex": scenario_opex,
}
scenario_activity_chart: Dict[str, list[Any]] = {
"labels": activity_labels,
"production": activity_production,
"consumption": activity_consumption,
}
summary_metrics = [
{"label": "Active Scenarios", "value": _format_int(len(scenarios))},
{"label": "Parameters", "value": _format_int(total_parameters)},
{"label": "CAPEX Total", "value": _format_currency(total_capex)},
{"label": "OPEX Total", "value": _format_currency(total_opex)},
{"label": "Equipment Assets", "value": _format_int(total_equipment)},
{"label": "Maintenance Events",
"value": _format_int(total_maintenance_events)},
{"label": "Consumption", "value": _format_decimal(total_consumption)},
{"label": "Production", "value": _format_decimal(total_production)},
{"label": "Simulation Iterations",
"value": _format_int(total_simulation_iterations)},
{"label": "Maintenance Cost",
"value": _format_currency(total_maintenance_cost)},
]
return {
"summary_metrics": summary_metrics,
"scenario_rows": scenario_rows,
"overall_report_metrics": overall_report_metrics,
"recent_simulations": recent_simulations,
"upcoming_maintenance": upcoming_maintenance,
"scenario_cost_chart": scenario_cost_chart,
"scenario_activity_chart": scenario_activity_chart,
"cost_chart_has_data": cost_chart_has_data,
"activity_chart_has_data": activity_chart_has_data,
"report_available": overall_report.get("count", 0) > 0,
}
@router.get("/", response_class=HTMLResponse)
async def dashboard_root(request: Request, db: Session = Depends(get_db)):
"""Render the primary dashboard landing page."""
return _render(request, "Dashboard.html", _load_dashboard(db))
@router.get("/ui/dashboard", response_class=HTMLResponse)
async def dashboard(request: Request, db: Session = Depends(get_db)):
"""Render the legacy dashboard route for backward compatibility."""
return _render(request, "Dashboard.html", _load_dashboard(db))
@router.get("/ui/dashboard/data", response_class=JSONResponse)
async def dashboard_data(db: Session = Depends(get_db)) -> JSONResponse:
"""Expose dashboard aggregates as JSON for client-side refreshes."""
return JSONResponse(_load_dashboard(db))
@router.get("/ui/scenarios", response_class=HTMLResponse)
async def scenario_form(request: Request, db: Session = Depends(get_db)):
"""Render the scenario creation form."""
context = _load_scenarios(db)
return _render(request, "ScenarioForm.html", context)
@router.get("/ui/parameters", response_class=HTMLResponse)
async def parameter_form(request: Request, db: Session = Depends(get_db)):
"""Render the parameter input form."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_parameters(db))
return _render(request, "ParameterInput.html", context)
@router.get("/ui/costs", response_class=HTMLResponse)
async def costs_view(request: Request, db: Session = Depends(get_db)):
"""Render the costs view with CAPEX and OPEX data."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_costs(db))
return _render(request, "costs.html", context)
@router.get("/ui/consumption", response_class=HTMLResponse)
async def consumption_view(request: Request, db: Session = Depends(get_db)):
"""Render the consumption view with scenario consumption data."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_consumption(db))
return _render(request, "consumption.html", context)
@router.get("/ui/production", response_class=HTMLResponse)
async def production_view(request: Request, db: Session = Depends(get_db)):
"""Render the production view with scenario production data."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_production(db))
return _render(request, "production.html", context)
@router.get("/ui/equipment", response_class=HTMLResponse)
async def equipment_view(request: Request, db: Session = Depends(get_db)):
"""Render the equipment view with scenario equipment data."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_equipment(db))
return _render(request, "equipment.html", context)
@router.get("/ui/maintenance", response_class=HTMLResponse)
async def maintenance_view(request: Request, db: Session = Depends(get_db)):
"""Render the maintenance view with scenario maintenance data."""
context: Dict[str, Any] = {}
context.update(_load_scenarios(db))
context.update(_load_equipment(db))
context.update(_load_maintenance(db))
return _render(request, "maintenance.html", context)
@router.get("/ui/simulations", response_class=HTMLResponse)
async def simulations_view(request: Request, db: Session = Depends(get_db)):
"""Render the simulations view with scenario information and recent runs."""
return _render(request, "simulations.html", _load_simulations(db))
@router.get("/ui/reporting", response_class=HTMLResponse)
async def reporting_view(request: Request, db: Session = Depends(get_db)):
"""Render the reporting view with scenario KPI summaries."""
return _render(request, "reporting.html", _load_reporting(db))