nessi
58eab24258
Add FastAPI-based monitoring dashboard with support for Codex CLI and Devin API v3 Enterprise consumption tracking. Includes Docker deployment, systemd service configuration, kiosk mode launcher, comprehensive documentation, and demo mode for testing without credentials.
95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.models import ProviderUsage, UsageResponse
|
|
from app.providers.base import GENERIC_ERROR, UsageProvider
|
|
from app.providers.codex import CodexProvider
|
|
from app.providers.demo import DemoProvider
|
|
from app.providers.devin import DevinProvider
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / 'templates'))
|
|
|
|
|
|
def _disabled(provider: str) -> ProviderUsage:
|
|
return ProviderUsage(provider=provider, status='disabled', source='Configuration')
|
|
|
|
|
|
def create_app(settings: Settings | None = None,
|
|
providers: tuple[UsageProvider, UsageProvider] | None = None) -> FastAPI:
|
|
config = settings or get_settings()
|
|
selected = providers or (
|
|
DemoProvider('codex', config) if config.demo_mode else CodexProvider(config),
|
|
DemoProvider('devin', config) if config.demo_mode else DevinProvider(config),
|
|
)
|
|
codex, devin = selected
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
yield
|
|
closed: set[int] = set()
|
|
for provider in (codex, devin):
|
|
if id(provider) not in closed:
|
|
closed.add(id(provider))
|
|
await provider.aclose()
|
|
|
|
application = FastAPI(title='AI Usage Dashboard', lifespan=lifespan)
|
|
application.mount('/static', StaticFiles(directory=str(BASE_DIR / 'static')), name='static')
|
|
|
|
@application.get('/', response_class=HTMLResponse)
|
|
async def index(request: Request):
|
|
response = templates.TemplateResponse(request=request, name='index.html', context={'refresh_interval': config.refresh_interval})
|
|
response.headers['Cache-Control'] = 'no-store'
|
|
return response
|
|
|
|
@application.get('/health')
|
|
async def health():
|
|
return {'status': 'ok'}
|
|
|
|
async def one(provider: UsageProvider, enabled: bool, timeout: float, name: str) -> ProviderUsage:
|
|
if not enabled:
|
|
return _disabled(name)
|
|
try:
|
|
return await asyncio.wait_for(provider.get_usage(), timeout=timeout + 0.25)
|
|
except Exception:
|
|
return ProviderUsage(provider=name, status='error', limits=[], error=GENERIC_ERROR,
|
|
last_successful_update=provider.last_successful_update, source=provider.source)
|
|
|
|
@application.get('/api/usage', response_model=UsageResponse)
|
|
async def usage():
|
|
codex_usage, devin_usage = await asyncio.gather(
|
|
one(codex, config.codex_enabled, config.codex_timeout, 'codex'),
|
|
one(devin, config.devin_enabled, config.devin_timeout, 'devin'))
|
|
response = UsageResponse(codex=codex_usage, devin=devin_usage,
|
|
server_time=datetime.now(timezone.utc), demo_mode=config.demo_mode,
|
|
refresh_interval=config.refresh_interval)
|
|
result = response.model_dump(mode='json')
|
|
return_response = JSONResponse(content=result)
|
|
return_response.headers['Cache-Control'] = 'no-store'
|
|
return return_response
|
|
|
|
return application
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
def run() -> None:
|
|
import uvicorn
|
|
|
|
settings = get_settings()
|
|
uvicorn.run(app, host=settings.host, port=settings.port)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|