""" FastAPI Service for GreedyOptim Scheduling Exposes greedyOptim functionality with customizable input data """ from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Dict, List, Any, Optional from datetime import datetime import logging import sys import os # Add parent directory to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Import greedyOptim components from greedyOptim.scheduler import optimize_trainset_schedule, compare_optimization_methods from greedyOptim.models import OptimizationConfig, OptimizationResult from greedyOptim.error_handling import DataValidator from greedyOptim.schedule_generator import generate_schedule_from_result # Import DataService for synthetic data generation (optional) from DataService.enhanced_generator import EnhancedMetroDataGenerator # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Create FastAPI app app = FastAPI( title="GreedyOptim Scheduling API", description="Advanced train scheduling optimization using genetic algorithms, PSO, CMA-ES, and more", version="2.0.0", docs_url="/docs", redoc_url="/redoc" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ============================================================================ # Request/Response Models # ============================================================================ class TrainsetStatusInput(BaseModel): """Single trainset operational status""" trainset_id: str operational_status: str = Field(..., description="IN_SERVICE, STANDBY, MAINTENANCE, OUT_OF_SERVICE, TESTING (or legacy: Available, In-Service, Maintenance, Standby, Out-of-Order)") last_maintenance_date: Optional[str] = None total_mileage_km: Optional[float] = None age_years: Optional[float] = None class FitnessCertificateInput(BaseModel): """Fitness certificate for a trainset""" trainset_id: str department: str = Field(..., description="Safety, Operations, Technical, Electrical, Mechanical") status: str = Field(..., description="ISSUED, EXPIRED, SUSPENDED, PENDING, IN_PROGRESS, REVOKED, RENEWED, CANCELLED (or legacy: Valid, Expired, Expiring-Soon, Suspended)") issue_date: Optional[str] = None expiry_date: Optional[str] = None class JobCardInput(BaseModel): """Job card/work order for trainset""" trainset_id: str job_id: str priority: str = Field(..., description="Critical, High, Medium, Low") status: str = Field(..., description="Open, In-Progress, Closed, Pending-Parts") description: Optional[str] = None estimated_hours: Optional[float] = None class ComponentHealthInput(BaseModel): """Component health status""" trainset_id: str component: str = Field(..., description="Brakes, HVAC, Doors, Propulsion, etc.") status: str = Field(..., description="EXCELLENT, GOOD, FAIR, POOR, CRITICAL, FAILED (or legacy: Good, Fair, Warning, Critical)") wear_level: Optional[float] = Field(None, ge=0, le=100) last_inspection: Optional[str] = None class OptimizationConfigInput(BaseModel): """Configuration for optimization algorithm""" required_service_trains: Optional[int] = Field(15, description="Minimum trains required in service") min_standby: Optional[int] = Field(2, description="Minimum standby trains") # Genetic Algorithm parameters population_size: Optional[int] = Field(50, ge=10, le=200) generations: Optional[int] = Field(100, ge=10, le=1000) mutation_rate: Optional[float] = Field(0.1, ge=0.0, le=1.0) crossover_rate: Optional[float] = Field(0.8, ge=0.0, le=1.0) elite_size: Optional[int] = Field(5, ge=1) class ScheduleOptimizationRequest(BaseModel): """Request for schedule optimization""" trainset_status: List[TrainsetStatusInput] fitness_certificates: List[FitnessCertificateInput] job_cards: Optional[List[JobCardInput]] = Field(default_factory=list, description="Job cards are optional, defaults to empty list") component_health: List[ComponentHealthInput] # Optional metadata metadata: Optional[Dict[str, Any]] = None date: Optional[str] = Field(None, description="Date for schedule (YYYY-MM-DD)") # Optimization configuration config: Optional[OptimizationConfigInput] = None method: str = Field("ga", description="Optimization method: ga, cmaes, pso, sa, nsga2, adaptive, ensemble") # Optional additional data branding_contracts: Optional[List[Dict[str, Any]]] = None maintenance_schedule: Optional[List[Dict[str, Any]]] = None performance_metrics: Optional[List[Dict[str, Any]]] = None class CompareMethodsRequest(BaseModel): """Request to compare multiple optimization methods""" trainset_status: List[TrainsetStatusInput] fitness_certificates: List[FitnessCertificateInput] job_cards: Optional[List[JobCardInput]] = Field(default_factory=list, description="Job cards are optional, defaults to empty list") component_health: List[ComponentHealthInput] metadata: Optional[Dict[str, Any]] = None date: Optional[str] = None config: Optional[OptimizationConfigInput] = None methods: List[str] = Field(["ga", "pso", "cmaes"], description="Methods to compare") class SyntheticDataRequest(BaseModel): """Request to generate synthetic data""" num_trainsets: int = Field(25, ge=5, le=100, description="Number of trainsets to generate") maintenance_rate: float = Field(0.1, ge=0.0, le=0.5, description="Percentage in maintenance") availability_rate: float = Field(0.8, ge=0.5, le=1.0, description="Percentage available for service") class ScheduleOptimizationResponse(BaseModel): """Response from optimization""" success: bool method: str fitness_score: float # Schedule allocation service_trains: List[str] standby_trains: List[str] maintenance_trains: List[str] unavailable_trains: List[str] # Metrics num_service: int num_standby: int num_maintenance: int num_unavailable: int # Detailed scores service_score: float standby_score: float health_score: float certificate_score: float # Metadata execution_time_seconds: Optional[float] = None timestamp: str constraints_satisfied: bool warnings: Optional[List[str]] = None # New models for full schedule response class StationStopResponse(BaseModel): """A single station stop within a trip""" station_code: str station_name: str arrival_time: Optional[str] = None departure_time: Optional[str] = None distance_from_origin_km: float platform: Optional[int] = None class TripResponse(BaseModel): """A single trip from origin to destination with all stops""" trip_id: str trip_number: int direction: str # "UP" or "DOWN" origin: str destination: str departure_time: str arrival_time: str stops: List[StationStopResponse] = [] class ServiceBlockResponse(BaseModel): """Service block with timing details and trips""" block_id: str departure_time: str origin: str destination: str trip_count: int estimated_km: float journey_time_minutes: Optional[float] = None period: Optional[str] = None is_peak: bool = False trips: Optional[List[TripResponse]] = None class TrainsetScheduleResponse(BaseModel): """Complete schedule for a single trainset""" trainset_id: str status: str readiness_score: float daily_km_allocation: float cumulative_km: float assigned_duty: Optional[str] = None priority_rank: Optional[int] = None service_blocks: Optional[List[ServiceBlockResponse]] = None stabling_bay: Optional[str] = None standby_reason: Optional[str] = None maintenance_type: Optional[str] = None ibl_bay: Optional[str] = None estimated_completion: Optional[str] = None alerts: Optional[List[str]] = None class FleetSummaryResponse(BaseModel): """Fleet summary statistics""" total_trainsets: int revenue_service: int standby: int maintenance: int availability_percent: float class OptimizationMetricsResponse(BaseModel): """Optimization metrics""" fitness_score: float method: str mileage_variance_coefficient: float total_planned_km: float optimization_runtime_ms: int class AlertResponse(BaseModel): """Schedule alert""" trainset_id: str severity: str alert_type: str message: str class FullScheduleResponse(BaseModel): """Complete schedule response with service blocks and timing""" schedule_id: str generated_at: str valid_from: str valid_until: str depot: str trainsets: List[TrainsetScheduleResponse] fleet_summary: FleetSummaryResponse optimization_metrics: OptimizationMetricsResponse alerts: List[AlertResponse] # ============================================================================ # Helper Functions # ============================================================================ def convert_pydantic_to_dict(request: ScheduleOptimizationRequest) -> Dict[str, Any]: """Convert Pydantic request model to dict format expected by greedyOptim""" data = { "trainset_status": [ts.dict() for ts in request.trainset_status], "fitness_certificates": [fc.dict() for fc in request.fitness_certificates], "job_cards": [jc.dict() for jc in request.job_cards] if request.job_cards else [], "component_health": [ch.dict() for ch in request.component_health], "metadata": request.metadata or { "generated_at": datetime.now().isoformat(), "system": "Kochi Metro Rail", "date": request.date or datetime.now().strftime("%Y-%m-%d") } } # Add optional data if provided if request.branding_contracts: data["branding_contracts"] = request.branding_contracts if request.maintenance_schedule: data["maintenance_schedule"] = request.maintenance_schedule if request.performance_metrics: data["performance_metrics"] = request.performance_metrics return data def convert_config(config_input: Optional[OptimizationConfigInput]) -> OptimizationConfig: """Convert Pydantic config to OptimizationConfig""" if config_input is None: return OptimizationConfig() return OptimizationConfig( required_service_trains=config_input.required_service_trains or 15, min_standby=config_input.min_standby or 2, population_size=config_input.population_size or 50, generations=config_input.generations or 100, mutation_rate=config_input.mutation_rate or 0.1, crossover_rate=config_input.crossover_rate or 0.8, elite_size=config_input.elite_size or 5 ) def convert_result_to_response( result: OptimizationResult, method: str, execution_time: Optional[float] = None ) -> ScheduleOptimizationResponse: """Convert OptimizationResult to API response""" # Extract objectives objectives = result.objectives # Determine unavailable trains (those not selected, standby, or maintenance) all_trains = set(result.selected_trainsets + result.standby_trainsets + result.maintenance_trainsets) unavailable = [] # We don't have this info in current result structure return ScheduleOptimizationResponse( success=True, method=method, fitness_score=result.fitness_score, service_trains=result.selected_trainsets, standby_trains=result.standby_trainsets, maintenance_trains=result.maintenance_trainsets, unavailable_trains=unavailable, num_service=len(result.selected_trainsets), num_standby=len(result.standby_trainsets), num_maintenance=len(result.maintenance_trainsets), num_unavailable=len(unavailable), service_score=objectives.get('service', 0.0), standby_score=objectives.get('standby', 0.0), health_score=objectives.get('health', 0.0), certificate_score=objectives.get('certificates', 0.0), execution_time_seconds=execution_time, timestamp=datetime.now().isoformat(), constraints_satisfied=len(result.selected_trainsets) >= 10, # Basic check warnings=None ) def convert_schedule_result_to_response(schedule_result) -> FullScheduleResponse: """Convert ScheduleResult to API FullScheduleResponse""" from greedyOptim.models import ScheduleResult trainsets = [] for ts in schedule_result.trainsets: service_blocks_resp = None if ts.service_blocks: service_blocks_resp = [] for sb in ts.service_blocks: # Convert trips with station stops trips_resp = None if sb.trips: trips_resp = [] for trip in sb.trips: stops_resp = [ StationStopResponse( station_code=stop.station_code, station_name=stop.station_name, arrival_time=stop.arrival_time, departure_time=stop.departure_time, distance_from_origin_km=stop.distance_from_origin_km, platform=stop.platform ) for stop in trip.stops ] trips_resp.append(TripResponse( trip_id=trip.trip_id, trip_number=trip.trip_number, direction=trip.direction, origin=trip.origin, destination=trip.destination, departure_time=trip.departure_time, arrival_time=trip.arrival_time, stops=stops_resp )) service_blocks_resp.append(ServiceBlockResponse( block_id=sb.block_id, departure_time=sb.departure_time, origin=sb.origin, destination=sb.destination, trip_count=sb.trip_count, estimated_km=sb.estimated_km, journey_time_minutes=sb.journey_time_minutes, period=sb.period, is_peak=sb.is_peak, trips=trips_resp )) trainsets.append(TrainsetScheduleResponse( trainset_id=ts.trainset_id, status=ts.status.value if hasattr(ts.status, 'value') else ts.status, readiness_score=ts.readiness_score, daily_km_allocation=ts.daily_km_allocation, cumulative_km=ts.cumulative_km, assigned_duty=ts.assigned_duty, priority_rank=ts.priority_rank, service_blocks=service_blocks_resp, stabling_bay=ts.stabling_bay, standby_reason=ts.standby_reason, maintenance_type=ts.maintenance_type.value if ts.maintenance_type and hasattr(ts.maintenance_type, 'value') else ts.maintenance_type, ibl_bay=ts.ibl_bay, estimated_completion=ts.estimated_completion, alerts=ts.alerts )) alerts = [ AlertResponse( trainset_id=a.trainset_id, severity=a.severity.value if hasattr(a.severity, 'value') else a.severity, alert_type=a.alert_type, message=a.message ) for a in schedule_result.alerts ] return FullScheduleResponse( schedule_id=schedule_result.schedule_id, generated_at=schedule_result.generated_at, valid_from=schedule_result.valid_from, valid_until=schedule_result.valid_until, depot=schedule_result.depot, trainsets=trainsets, fleet_summary=FleetSummaryResponse( total_trainsets=schedule_result.fleet_summary.total_trainsets, revenue_service=schedule_result.fleet_summary.revenue_service, standby=schedule_result.fleet_summary.standby, maintenance=schedule_result.fleet_summary.maintenance, availability_percent=schedule_result.fleet_summary.availability_percent ), optimization_metrics=OptimizationMetricsResponse( fitness_score=schedule_result.optimization_metrics.fitness_score, method=schedule_result.optimization_metrics.method, mileage_variance_coefficient=schedule_result.optimization_metrics.mileage_variance_coefficient, total_planned_km=schedule_result.optimization_metrics.total_planned_km, optimization_runtime_ms=schedule_result.optimization_metrics.optimization_runtime_ms ), alerts=alerts ) # ============================================================================ # API Endpoints # ============================================================================ @app.get("/") async def root(): """Root endpoint with API information""" return { "service": "GreedyOptim Scheduling API", "version": "2.0.0", "description": "Advanced train scheduling optimization", "endpoints": { "POST /optimize": "Optimize schedule with custom data (returns trainset allocations)", "POST /schedule": "Generate full schedule with service blocks and timing", "POST /compare": "Compare multiple optimization methods", "POST /generate-synthetic": "Generate synthetic test data", "POST /validate": "Validate input data structure", "GET /health": "Health check", "GET /methods": "List available optimization methods", "GET /docs": "Interactive API documentation" } } @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "timestamp": datetime.now().isoformat(), "service": "greedyoptim-api" } @app.get("/methods") async def list_methods(): """List available optimization methods""" return { "available_methods": { "ga": { "name": "Genetic Algorithm", "description": "Evolutionary optimization using selection, crossover, and mutation", "typical_time": "medium", "solution_quality": "high" }, "cmaes": { "name": "CMA-ES", "description": "Covariance Matrix Adaptation Evolution Strategy", "typical_time": "medium-high", "solution_quality": "very high" }, "pso": { "name": "Particle Swarm Optimization", "description": "Swarm intelligence-based optimization", "typical_time": "medium", "solution_quality": "high" }, "sa": { "name": "Simulated Annealing", "description": "Probabilistic optimization inspired by metallurgy", "typical_time": "medium", "solution_quality": "medium-high" }, "nsga2": { "name": "NSGA-II", "description": "Non-dominated Sorting Genetic Algorithm (multi-objective)", "typical_time": "high", "solution_quality": "very high" }, "adaptive": { "name": "Adaptive Optimizer", "description": "Automatically selects best algorithm", "typical_time": "high", "solution_quality": "very high" }, "ensemble": { "name": "Ensemble Optimizer", "description": "Runs multiple algorithms in parallel", "typical_time": "high", "solution_quality": "highest" } }, "default_method": "ga", "recommended_for_speed": "ga", "recommended_for_quality": "ensemble" } # ============================================================================ # Station & Route Information Endpoints # ============================================================================ @app.get("/stations") async def get_stations(): """Get all metro stations with their details. Returns the complete list of stations from the configured route, including distance information, terminal status, and depot locations. """ try: from greedyOptim.station_loader import get_station_loader loader = get_station_loader() return { "success": True, "route": loader.to_dict(), "summary": { "total_stations": loader.station_count, "total_distance_km": loader.total_distance_km, "terminals": loader.terminals } } except Exception as e: logger.error(f"Failed to get station data: {e}") raise HTTPException(status_code=500, detail=f"Failed to load station data: {str(e)}") @app.get("/stations/{station_identifier}") async def get_station_details(station_identifier: str): """Get details for a specific station by name or code. Args: station_identifier: Station name (e.g., 'Aluva') or code (e.g., 'ALV') """ try: from greedyOptim.station_loader import get_station_loader loader = get_station_loader() # Try by name first, then by code station = loader.get_station_by_name(station_identifier) if not station: station = loader.get_station_by_code(station_identifier) if not station: raise HTTPException( status_code=404, detail=f"Station not found: {station_identifier}" ) return { "success": True, "station": { "sr_no": station.sr_no, "code": station.code, "name": station.name, "distance_from_prev_km": station.distance_from_prev_km, "cumulative_distance_km": station.cumulative_distance_km, "is_terminal": station.is_terminal, "has_depot": station.has_depot, "platform_count": station.platform_count, "depot_name": station.depot_name } } except HTTPException: raise except Exception as e: logger.error(f"Failed to get station details: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/route/journey") async def calculate_journey( origin: str, destination: str, departure_time: str = "07:00" ): """Calculate journey details between two stations. Args: origin: Origin station name or code destination: Destination station name or code departure_time: Departure time in HH:MM format (default: 07:00) Returns: Journey details including intermediate stations with arrival times """ try: from greedyOptim.station_loader import get_station_loader loader = get_station_loader() # Validate stations exist origin_station = loader.get_station_by_name(origin) or loader.get_station_by_code(origin) dest_station = loader.get_station_by_name(destination) or loader.get_station_by_code(destination) if not origin_station: raise HTTPException(status_code=404, detail=f"Origin station not found: {origin}") if not dest_station: raise HTTPException(status_code=404, detail=f"Destination station not found: {destination}") # Get journey details station_sequence = loader.get_station_sequence_for_trip( origin_station.name, dest_station.name, include_times=True, departure_time=departure_time ) distance = loader.get_distance_between(origin_station.name, dest_station.name) journey_time = loader.calculate_journey_time(origin_station.name, dest_station.name) return { "success": True, "journey": { "origin": origin_station.name, "destination": dest_station.name, "distance_km": round(distance, 3), "journey_time_minutes": round(journey_time, 1), "departure_time": departure_time, "num_stops": len(station_sequence) - 1, "stations": station_sequence } } except HTTPException: raise except Exception as e: logger.error(f"Failed to calculate journey: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/route/round-trip") async def get_round_trip_info(): """Get round trip information for the full route. Returns round trip time and distance between terminals. """ try: from greedyOptim.station_loader import get_station_loader loader = get_station_loader() round_trip_time = loader.calculate_round_trip_time() terminals = loader.terminals one_way_distance = loader.total_distance_km return { "success": True, "round_trip": { "terminals": terminals, "one_way_distance_km": round(one_way_distance, 3), "round_trip_distance_km": round(one_way_distance * 2, 3), "round_trip_time_minutes": round(round_trip_time, 1), "round_trip_time_hours": round(round_trip_time / 60, 2) }, "operational_params": loader.route_info.operational_params } except Exception as e: logger.error(f"Failed to get round trip info: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/service-blocks") async def get_service_blocks(): """Get all available service blocks for the day. Returns pre-defined service blocks that can be assigned to trainsets. """ try: from greedyOptim.service_blocks import ServiceBlockGenerator generator = ServiceBlockGenerator() blocks = generator.get_all_service_blocks() # Group by period by_period = {} for block in blocks: period = block['period'] if period not in by_period: by_period[period] = [] by_period[period].append(block) return { "success": True, "total_blocks": len(blocks), "route_length_km": generator.route_length_km, "terminals": generator.terminals, "blocks_by_period": by_period, "all_blocks": blocks } except Exception as e: logger.error(f"Failed to get service blocks: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/optimize", response_model=ScheduleOptimizationResponse) async def optimize_schedule(request: ScheduleOptimizationRequest): """ Optimize train schedule with custom input data. This endpoint accepts detailed trainset data and returns an optimized schedule that maximizes service coverage while respecting all constraints. """ try: import time start_time = time.time() logger.info(f"Received optimization request with {len(request.trainset_status)} trainsets, method: {request.method}") # Convert request to dict format data = convert_pydantic_to_dict(request) # Validate data validation_errors = DataValidator.validate_data(data) if validation_errors: raise HTTPException( status_code=400, detail={ "error": "Data validation failed", "validation_errors": validation_errors, "message": "Please fix the data structure and try again" } ) # Convert config config = convert_config(request.config) # Run optimization result = optimize_trainset_schedule(data, request.method, config) execution_time = time.time() - start_time logger.info(f"Optimization completed in {execution_time:.3f}s, fitness: {result.fitness_score:.4f}") # Convert to response response = convert_result_to_response(result, request.method, execution_time) return response except HTTPException: raise except Exception as e: logger.error(f"Optimization error: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail={ "error": "Optimization failed", "message": str(e), "type": type(e).__name__ } ) @app.post("/schedule", response_model=FullScheduleResponse) async def generate_full_schedule(request: ScheduleOptimizationRequest): """ Generate complete schedule with service blocks and timing. This endpoint returns a full schedule with: - Service blocks with departure times and routes - Bay allocations - Daily km assignments - Fleet summary - Alerts and warnings Use this endpoint when you need operational timetables, not just trainset allocations. """ try: import time start_time = time.time() logger.info(f"Received full schedule request with {len(request.trainset_status)} trainsets, method: {request.method}") # Convert request to dict format data = convert_pydantic_to_dict(request) # Validate data validation_errors = DataValidator.validate_data(data) if validation_errors: raise HTTPException( status_code=400, detail={ "error": "Data validation failed", "validation_errors": validation_errors, "message": "Please fix the data structure and try again" } ) # Convert config config = convert_config(request.config) # Run optimization result = optimize_trainset_schedule(data, request.method, config) execution_time = time.time() - start_time runtime_ms = int(execution_time * 1000) logger.info(f"Optimization completed in {execution_time:.3f}s, fitness: {result.fitness_score:.4f}") # Generate full schedule with service blocks schedule_result = generate_schedule_from_result( data=data, optimization_result=result, method=request.method, runtime_ms=runtime_ms, config=config, date=request.date, depot="Muttom_Depot" ) # Convert to response response = convert_schedule_result_to_response(schedule_result) logger.info(f"Full schedule generated: {schedule_result.schedule_id}") return response except HTTPException: raise except Exception as e: logger.error(f"Schedule generation error: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail={ "error": "Schedule generation failed", "message": str(e), "type": type(e).__name__ } ) @app.post("/compare") async def compare_methods(request: CompareMethodsRequest): """ Compare multiple optimization methods on the same input data. Returns results from all requested methods for comparison. """ try: import time logger.info(f"Comparing methods: {request.methods}") # Create a temporary request object for conversion temp_request = ScheduleOptimizationRequest( trainset_status=request.trainset_status, fitness_certificates=request.fitness_certificates, job_cards=request.job_cards, component_health=request.component_health, metadata=request.metadata, date=request.date, method="ga" # Default method for conversion ) # Convert request to dict format data = convert_pydantic_to_dict(temp_request) # Validate data validation_errors = DataValidator.validate_data(data) if validation_errors: raise HTTPException(status_code=400, detail={"error": "Data validation failed", "details": validation_errors}) # Convert config config = convert_config(request.config) # Compare methods start_time = time.time() results = compare_optimization_methods(data, request.methods, config) total_time = time.time() - start_time # Convert results comparison = { "methods": {}, "summary": { "total_execution_time": total_time, "methods_compared": len(results), "timestamp": datetime.now().isoformat() } } best_score = -float('inf') best_method = None for method, result in results.items(): if result is None: comparison["methods"][method] = { "success": False, "error": "Optimization failed for this method" } continue comparison["methods"][method] = convert_result_to_response( result, method ).dict() if result.fitness_score > best_score: best_score = result.fitness_score best_method = method comparison["summary"]["best_method"] = best_method comparison["summary"]["best_score"] = best_score if best_method else None logger.info(f"Comparison completed, best: {best_method} ({best_score:.4f})") return JSONResponse(content=comparison) except Exception as e: logger.error(f"Comparison error: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail={"error": "Comparison failed", "message": str(e)} ) @app.post("/generate-synthetic") async def generate_synthetic_data(request: SyntheticDataRequest): """ Generate synthetic test data using EnhancedMetroDataGenerator. Useful for testing the optimization API without providing real data. """ try: logger.info(f"Generating synthetic data for {request.num_trainsets} trainsets") # Generate data generator = EnhancedMetroDataGenerator(num_trainsets=request.num_trainsets) data = generator.generate_complete_enhanced_dataset() # Remove trainset_profiles as it contains non-serializable datetime objects # and is not needed for optimization data_for_response = { "trainset_status": data["trainset_status"], "fitness_certificates": data["fitness_certificates"], "job_cards": data["job_cards"], "component_health": data["component_health"], "metadata": data.get("metadata", {}) } logger.info(f"Generated synthetic data with {len(data['trainset_status'])} trainsets") return JSONResponse(content={ "success": True, "data": data_for_response, "metadata": { "num_trainsets": len(data['trainset_status']), "num_fitness_certificates": len(data['fitness_certificates']), "num_job_cards": len(data['job_cards']), "num_component_health": len(data['component_health']), "generated_at": datetime.now().isoformat() } }) except Exception as e: logger.error(f"Synthetic data generation error: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail={"error": "Data generation failed", "message": str(e)} ) @app.post("/validate") async def validate_data(request: ScheduleOptimizationRequest): """ Validate input data structure without running optimization. Returns validation results and suggestions for fixing issues. """ try: # Convert to dict data = convert_pydantic_to_dict(request) # Validate validation_errors = DataValidator.validate_data(data) if not validation_errors: return { "valid": True, "message": "Data structure is valid", "num_trainsets": len(request.trainset_status), "num_certificates": len(request.fitness_certificates), "num_job_cards": len(request.job_cards) if request.job_cards else 0, "num_component_health": len(request.component_health) } return { "valid": False, "validation_errors": validation_errors, "suggestions": [ "Check that all trainset_ids are consistent across sections", "Ensure operational_status values are valid (Available, In-Service, Maintenance, Standby, Out-of-Order)", "Verify certificate expiry dates are in ISO format", "Confirm component wear_level is between 0-100" ] } except Exception as e: raise HTTPException( status_code=400, detail={"error": "Validation failed", "message": str(e)} ) if __name__ == "__main__": import uvicorn uvicorn.run("api.greedyoptim_api:app", host="0.0.0.0", port=7860, reload=True)