#!/usr/bin/env python3 """ Cidadão.AI Models - API Server FastAPI server for ML model inference. """ import os import sys from contextlib import asynccontextmanager from typing import Dict, List, Any, Optional import logging from fastapi import FastAPI, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from prometheus_client import Counter, Histogram, generate_latest # Add parent to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) # Import models from src.models.anomaly_detection import AnomalyDetector from src.models.pattern_analysis import PatternAnalyzer from src.models.spectral_analysis import SpectralAnalyzer # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Prometheus metrics REQUEST_COUNT = Counter('cidadao_models_requests_total', 'Total requests', ['endpoint']) REQUEST_DURATION = Histogram('cidadao_models_request_duration_seconds', 'Request duration') ANOMALIES_DETECTED = Counter('cidadao_models_anomalies_total', 'Total anomalies detected') # Global models models = {} @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan manager.""" logger.info("🤖 Cidadão.AI Models API starting up...") # Initialize models models["anomaly_detector"] = AnomalyDetector() models["pattern_analyzer"] = PatternAnalyzer() models["spectral_analyzer"] = SpectralAnalyzer() logger.info("✅ All models loaded successfully") yield logger.info("🛑 Cidadão.AI Models API shutting down...") # Create FastAPI app app = FastAPI( title="🤖 Cidadão.AI Models API", description="Specialized ML models for Brazilian government transparency analysis", version="1.0.0", lifespan=lifespan ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Request/Response Models class Contract(BaseModel): """Government contract data.""" id: str description: str value: float supplier: str date: str organ: str class AnomalyRequest(BaseModel): """Request for anomaly detection.""" contracts: List[Dict[str, Any]] = Field(..., description="List of contracts to analyze") threshold: Optional[float] = Field(default=0.7, description="Anomaly threshold") class AnomalyResponse(BaseModel): """Response from anomaly detection.""" anomalies: List[Dict[str, Any]] total_analyzed: int anomalies_found: int confidence_score: float model_version: str = "1.0.0" class PatternRequest(BaseModel): """Request for pattern analysis.""" data: Dict[str, Any] = Field(..., description="Data to analyze patterns") analysis_type: str = Field(default="temporal", description="Type of pattern analysis") class PatternResponse(BaseModel): """Response from pattern analysis.""" patterns: List[Dict[str, Any]] pattern_count: int confidence: float insights: List[str] class SpectralRequest(BaseModel): """Request for spectral analysis.""" time_series: List[float] = Field(..., description="Time series data") sampling_rate: Optional[float] = Field(default=1.0, description="Sampling rate") class SpectralResponse(BaseModel): """Response from spectral analysis.""" frequencies: List[float] amplitudes: List[float] dominant_frequency: float periodic_patterns: List[Dict[str, Any]] # Endpoints @app.get("/") async def root(): """Root endpoint with API info.""" REQUEST_COUNT.labels(endpoint="/").inc() return { "api": "Cidadão.AI Models", "version": "1.0.0", "status": "operational", "models": list(models.keys()), "endpoints": { "anomaly_detection": "/v1/detect-anomalies", "pattern_analysis": "/v1/analyze-patterns", "spectral_analysis": "/v1/analyze-spectral", "health": "/health", "metrics": "/metrics" } } @app.get("/health") async def health_check(): """Health check endpoint.""" REQUEST_COUNT.labels(endpoint="/health").inc() return { "status": "healthy", "models_loaded": len(models) == 3, "models": {name: "loaded" for name in models.keys()} } @app.post("/v1/detect-anomalies", response_model=AnomalyResponse) async def detect_anomalies(request: AnomalyRequest): """Detect anomalies in government contracts.""" REQUEST_COUNT.labels(endpoint="/v1/detect-anomalies").inc() try: with REQUEST_DURATION.time(): # Run anomaly detection detector = models["anomaly_detector"] results = await detector.predict(request.contracts) # Count anomalies anomalies = [r for r in results if r.get("is_anomaly", False)] ANOMALIES_DETECTED.inc(len(anomalies)) return AnomalyResponse( anomalies=anomalies, total_analyzed=len(request.contracts), anomalies_found=len(anomalies), confidence_score=0.87 ) except Exception as e: logger.error(f"Anomaly detection error: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Anomaly detection failed: {str(e)}" ) @app.post("/v1/analyze-patterns", response_model=PatternResponse) async def analyze_patterns(request: PatternRequest): """Analyze patterns in government data.""" REQUEST_COUNT.labels(endpoint="/v1/analyze-patterns").inc() try: with REQUEST_DURATION.time(): analyzer = models["pattern_analyzer"] # Mock analysis for now patterns = [ { "type": "temporal", "description": "Peak spending in December", "confidence": 0.92 }, { "type": "vendor_concentration", "description": "High concentration of contracts with few vendors", "confidence": 0.85 } ] return PatternResponse( patterns=patterns, pattern_count=len(patterns), confidence=0.88, insights=[ "Seasonal spending patterns detected", "Vendor concentration above normal threshold" ] ) except Exception as e: logger.error(f"Pattern analysis error: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Pattern analysis failed: {str(e)}" ) @app.post("/v1/analyze-spectral", response_model=SpectralResponse) async def analyze_spectral(request: SpectralRequest): """Perform spectral analysis on time series data.""" REQUEST_COUNT.labels(endpoint="/v1/analyze-spectral").inc() try: with REQUEST_DURATION.time(): analyzer = models["spectral_analyzer"] # Mock spectral analysis return SpectralResponse( frequencies=[0.1, 0.2, 0.5, 1.0], amplitudes=[10.5, 25.3, 5.2, 45.8], dominant_frequency=1.0, periodic_patterns=[ { "frequency": 1.0, "period": "annual", "strength": 0.95 } ] ) except Exception as e: logger.error(f"Spectral analysis error: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Spectral analysis failed: {str(e)}" ) @app.get("/metrics") async def metrics(): """Prometheus metrics endpoint.""" return generate_latest().decode('utf-8') if __name__ == "__main__": import uvicorn port = int(os.getenv("PORT", 8001)) host = os.getenv("HOST", "0.0.0.0") logger.info(f"🚀 Starting Cidadão.AI Models API on {host}:{port}") uvicorn.run( "api_server:app", host=host, port=port, reload=True )