zwitschi
97b1c0360b
- Updated test functions in various test files to enhance code clarity by formatting long lines and improving indentation. - Adjusted assertions to use multi-line formatting for better readability. - Added new test cases for theme settings API to ensure proper functionality. - Ensured consistent use of line breaks and spacing across test files for uniformity.
785 lines
28 KiB
Python
785 lines
28 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
|
|
from models.currency import Currency
|
|
from routes.currencies import DEFAULT_CURRENCY_CODE, _ensure_default_currency
|
|
from services.settings import (
|
|
CSS_COLOR_DEFAULTS,
|
|
get_css_color_settings,
|
|
list_css_env_override_rows,
|
|
read_css_color_env_overrides,
|
|
)
|
|
|
|
|
|
CURRENCY_CHOICES: list[Dict[str, Any]] = [
|
|
{"id": "USD", "name": "US Dollar (USD)"},
|
|
{"id": "EUR", "name": "Euro (EUR)"},
|
|
{"id": "CLP", "name": "Chilean Peso (CLP)"},
|
|
{"id": "RMB", "name": "Chinese Yuan (RMB)"},
|
|
{"id": "GBP", "name": "British Pound (GBP)"},
|
|
{"id": "CAD", "name": "Canadian Dollar (CAD)"},
|
|
{"id": "AUD", "name": "Australian Dollar (AUD)"},
|
|
]
|
|
|
|
MEASUREMENT_UNITS: list[Dict[str, Any]] = [
|
|
{"id": "tonnes", "name": "Tonnes", "symbol": "t"},
|
|
{"id": "kilograms", "name": "Kilograms", "symbol": "kg"},
|
|
{"id": "pounds", "name": "Pounds", "symbol": "lb"},
|
|
{"id": "liters", "name": "Liters", "symbol": "L"},
|
|
{"id": "cubic_meters", "name": "Cubic Meters", "symbol": "m3"},
|
|
{"id": "kilowatt_hours", "name": "Kilowatt Hours", "symbol": "kWh"},
|
|
]
|
|
|
|
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,
|
|
):
|
|
context = _context(request, extra)
|
|
return templates.TemplateResponse(request, template_name, context)
|
|
|
|
|
|
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 "",
|
|
"currency_code": getattr(capex, "currency_code", "USD")
|
|
or "USD",
|
|
}
|
|
)
|
|
|
|
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 "",
|
|
"currency_code": getattr(opex, "currency_code", "USD") or "USD",
|
|
}
|
|
)
|
|
|
|
return {
|
|
"capex_by_scenario": dict(capex_grouped),
|
|
"opex_by_scenario": dict(opex_grouped),
|
|
}
|
|
|
|
|
|
def _load_currencies(db: Session) -> Dict[str, Any]:
|
|
items: list[Dict[str, Any]] = []
|
|
for c in (
|
|
db.query(Currency)
|
|
.filter_by(is_active=True)
|
|
.order_by(Currency.code)
|
|
.all()
|
|
):
|
|
items.append(
|
|
{"id": c.code, "name": f"{c.name} ({c.code})", "symbol": c.symbol}
|
|
)
|
|
if not items:
|
|
items.append({"id": "USD", "name": "US Dollar (USD)", "symbol": "$"})
|
|
return {"currency_options": items}
|
|
|
|
|
|
def _load_currency_settings(db: Session) -> Dict[str, Any]:
|
|
_ensure_default_currency(db)
|
|
records = db.query(Currency).order_by(Currency.code).all()
|
|
currencies: list[Dict[str, Any]] = []
|
|
for record in records:
|
|
code_value = getattr(record, "code")
|
|
currencies.append(
|
|
{
|
|
"id": int(getattr(record, "id")),
|
|
"code": code_value,
|
|
"name": getattr(record, "name"),
|
|
"symbol": getattr(record, "symbol"),
|
|
"is_active": bool(getattr(record, "is_active", True)),
|
|
"is_default": code_value == DEFAULT_CURRENCY_CODE,
|
|
}
|
|
)
|
|
|
|
active_count = sum(1 for item in currencies if item["is_active"])
|
|
inactive_count = len(currencies) - active_count
|
|
|
|
return {
|
|
"currencies": currencies,
|
|
"currency_stats": {
|
|
"total": len(currencies),
|
|
"active": active_count,
|
|
"inactive": inactive_count,
|
|
},
|
|
"default_currency_code": DEFAULT_CURRENCY_CODE,
|
|
"currency_api_base": "/api/currencies",
|
|
}
|
|
|
|
|
|
def _load_css_settings(db: Session) -> Dict[str, Any]:
|
|
variables = get_css_color_settings(db)
|
|
env_overrides = read_css_color_env_overrides()
|
|
env_rows = list_css_env_override_rows()
|
|
env_meta = {row["css_key"]: row for row in env_rows}
|
|
return {
|
|
"css_variables": variables,
|
|
"css_defaults": CSS_COLOR_DEFAULTS,
|
|
"css_env_overrides": env_overrides,
|
|
"css_env_override_rows": env_rows,
|
|
"css_env_override_meta": env_meta,
|
|
}
|
|
|
|
|
|
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 ""
|
|
unit_name = getattr(record, "unit_name", None)
|
|
unit_symbol = getattr(record, "unit_symbol", None)
|
|
grouped[scenario_id].append(
|
|
{
|
|
"id": record_id,
|
|
"scenario_id": scenario_id,
|
|
"amount": amount_value,
|
|
"description": description,
|
|
"unit_name": unit_name,
|
|
"unit_symbol": unit_symbol,
|
|
}
|
|
)
|
|
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 ""
|
|
unit_name = getattr(record, "unit_name", None)
|
|
unit_symbol = getattr(record, "unit_symbol", None)
|
|
grouped[scenario_id].append(
|
|
{
|
|
"id": record_id,
|
|
"scenario_id": scenario_id,
|
|
"amount": amount_value,
|
|
"description": description,
|
|
"unit_name": unit_name,
|
|
"unit_symbol": unit_symbol,
|
|
}
|
|
)
|
|
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: list[Dict[str, Any]] = []
|
|
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))
|
|
context.update(_load_currencies(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))
|
|
context["unit_options"] = MEASUREMENT_UNITS
|
|
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))
|
|
context["unit_options"] = MEASUREMENT_UNITS
|
|
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))
|
|
|
|
|
|
@router.get("/ui/settings", response_class=HTMLResponse)
|
|
async def settings_view(request: Request, db: Session = Depends(get_db)):
|
|
"""Render the settings landing page."""
|
|
context = _load_css_settings(db)
|
|
return _render(request, "settings.html", context)
|
|
|
|
|
|
@router.get("/ui/currencies", response_class=HTMLResponse)
|
|
async def currencies_view(request: Request, db: Session = Depends(get_db)):
|
|
"""Render the currency administration page with full currency context."""
|
|
context = _load_currency_settings(db)
|
|
return _render(request, "currencies.html", context)
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
async def login_page(request: Request):
|
|
return _render(request, "login.html")
|
|
|
|
|
|
@router.get("/register", response_class=HTMLResponse)
|
|
async def register_page(request: Request):
|
|
return _render(request, "register.html")
|
|
|
|
|
|
@router.get("/profile", response_class=HTMLResponse)
|
|
async def profile_page(request: Request):
|
|
return _render(request, "profile.html")
|
|
|
|
|
|
@router.get("/forgot-password", response_class=HTMLResponse)
|
|
async def forgot_password_page(request: Request):
|
|
return _render(request, "forgot_password.html")
|
|
|
|
|
|
@router.get("/theme-settings", response_class=HTMLResponse)
|
|
async def theme_settings_page(request: Request, db: Session = Depends(get_db)):
|
|
"""Render the theme settings page."""
|
|
context = _load_css_settings(db)
|
|
return _render(request, "theme_settings.html", context)
|