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))