blackopsrepl/vehicle-routing-python

1from fastapi import FastAPI, HTTPException, Query2from fastapi.staticfiles import StaticFiles3from fastapi.responses import StreamingResponse4from uuid import uuid45from typing import Dict, List, Optional6from dataclasses import asdict7from enum import Enum8import logging9import json10import asyncio11 12from .domain import VehicleRoutePlan, Location13from .converters import plan_to_model, model_to_plan14from .domain import VehicleRoutePlanModel15from .score_analysis import ConstraintAnalysisDTO, MatchAnalysisDTO16from .demo_data import generate_demo_data, DemoData17from .solver import solver_manager, solution_manager18from .routing import compute_distance_matrix_with_progress, DistanceMatrix19from pydantic import BaseModel, Field20 21 22class RoutingMode(str, Enum):23    """Routing mode for distance calculations."""24    HAVERSINE = "haversine"  # Fast, straight-line estimation25    REAL_ROADS = "real_roads"  # Slower, uses OSMnx for real road routes26 27logger = logging.getLogger(__name__)28 29app = FastAPI(docs_url='/q/swagger-ui')30 31data_sets: Dict[str, VehicleRoutePlan] = {}32 33 34# Request/Response models for recommendation endpoints35class VehicleRecommendation(BaseModel):36    """Recommendation for assigning a visit to a vehicle at a specific index."""37    vehicle_id: str = Field(..., alias="vehicleId")38    index: int39 40    class Config:41        populate_by_name = True42 43 44class RecommendedAssignmentResponse(BaseModel):45    """Response from the recommendation API."""46    proposition: VehicleRecommendation47    score_diff: str = Field(..., alias="scoreDiff")48 49    class Config:50        populate_by_name = True51 52 53class RecommendationRequest(BaseModel):54    """Request for visit assignment recommendations."""55    solution: VehicleRoutePlanModel56    visit_id: str = Field(..., alias="visitId")57 58    class Config:59        populate_by_name = True60 61 62class ApplyRecommendationRequest(BaseModel):63    """Request to apply a recommendation."""64    solution: VehicleRoutePlanModel65    visit_id: str = Field(..., alias="visitId")66    vehicle_id: str = Field(..., alias="vehicleId")67    index: int68 69    class Config:70        populate_by_name = True71 72 73def json_to_vehicle_route_plan(json_data: dict) -> VehicleRoutePlan:74    """Convert JSON data to VehicleRoutePlan using the model converters."""75    plan_model = VehicleRoutePlanModel.model_validate(json_data)76    return model_to_plan(plan_model)77 78 79@app.get("/demo-data")80async def get_demo_data():81    """Get available demo data sets."""82    return [demo.name for demo in DemoData]83 84def _extract_all_locations(plan: VehicleRoutePlan) -> list[Location]:85    """Extract all unique locations from a route plan."""86    locations = []87    seen = set()88 89    for vehicle in plan.vehicles:90        key = (vehicle.home_location.latitude, vehicle.home_location.longitude)91        if key not in seen:92            locations.append(vehicle.home_location)93            seen.add(key)94 95    for visit in plan.visits:96        key = (visit.location.latitude, visit.location.longitude)97        if key not in seen:98            locations.append(visit.location)99            seen.add(key)100 101    return locations102 103 104def _extract_route_geometries(plan: VehicleRoutePlan) -> Dict[str, List[Optional[str]]]:105    """106    Extract route geometries from the distance matrix for all vehicles.107    Returns empty dict if no distance matrix is available.108    """109    distance_matrix = Location.get_distance_matrix()110    if distance_matrix is None:111        return {}112 113    geometries: Dict[str, List[Optional[str]]] = {}114 115    for vehicle in plan.vehicles:116        segments: List[Optional[str]] = []117 118        if not vehicle.visits:119            geometries[vehicle.id] = segments120            continue121 122        # Segment from depot to first visit123        prev_location = vehicle.home_location124        for visit in vehicle.visits:125            geometry = distance_matrix.get_geometry(prev_location, visit.location)126            segments.append(geometry)127            prev_location = visit.location128 129        # Segment from last visit back to depot130        geometry = distance_matrix.get_geometry(prev_location, vehicle.home_location)131        segments.append(geometry)132 133        geometries[vehicle.id] = segments134 135    return geometries136 137 138def _initialize_distance_matrix(139    plan: VehicleRoutePlan,140    use_real_roads: bool = False,141    progress_callback=None142) -> Optional[DistanceMatrix]:143    """144    Initialize the distance matrix for a route plan.145 146    Args:147        plan: The route plan with locations148        use_real_roads: If True, use OSMnx for real road routing (slower)149                       If False, use haversine estimation (fast, default)150        progress_callback: Optional callback for progress updates151 152    Returns the computed matrix, or None if routing failed.153    """154    locations = _extract_all_locations(plan)155    if not locations:156        return None157 158    logger.info(f"Computing distance matrix for {len(locations)} locations (mode: {'real_roads' if use_real_roads else 'haversine'})...")159 160    # Compute bounding box from the plan161    bbox = (162        plan.north_east_corner.latitude,163        plan.south_west_corner.latitude,164        plan.north_east_corner.longitude,165        plan.south_west_corner.longitude,166    )167 168    try:169        matrix = compute_distance_matrix_with_progress(170            locations,171            bbox=bbox,172            use_osm=use_real_roads,173            progress_callback=progress_callback174        )175        Location.set_distance_matrix(matrix)176        logger.info("Distance matrix computed and set successfully")177        return matrix178    except Exception as e:179        logger.warning(f"Failed to compute distance matrix: {e}")180        return None181 182 183@app.get("/demo-data/{demo_name}", response_model=VehicleRoutePlanModel)184async def get_demo_data_by_name(185    demo_name: str,186    routing: RoutingMode = Query(187        default=RoutingMode.HAVERSINE,188        description="Routing mode: 'haversine' (fast, default) or 'real_roads' (slower, accurate)"189    )190) -> VehicleRoutePlanModel:191    """192    Get a specific demo data set.193 194    Args:195        demo_name: Name of the demo dataset (PHILADELPHIA, HARTFORT, FIRENZE)196        routing: Routing mode - 'haversine' (fast default) or 'real_roads' (slower, accurate)197 198    When routing=real_roads, computes the distance matrix using real road network199    data (OSMnx) for accurate routing. The first call may take 5-15 seconds200    to download the OSM network (cached for subsequent calls).201    """202    try:203        demo_data = DemoData[demo_name]204        domain_plan = generate_demo_data(demo_data)205 206        # Initialize distance matrix with selected routing mode207        use_real_roads = routing == RoutingMode.REAL_ROADS208        _initialize_distance_matrix(domain_plan, use_real_roads=use_real_roads)209 210        return plan_to_model(domain_plan)211    except KeyError:212        raise HTTPException(status_code=404, detail=f"Demo data '{demo_name}' not found")213 214 215# Progress tracking for SSE216_progress_queues: Dict[str, asyncio.Queue] = {}217 218 219@app.get("/demo-data/{demo_name}/stream")220async def get_demo_data_with_progress(221    demo_name: str,222    routing: RoutingMode = Query(223        default=RoutingMode.HAVERSINE,224        description="Routing mode: 'haversine' (fast, default) or 'real_roads' (slower, accurate)"225    )226):227    """228    Get demo data with Server-Sent Events (SSE) progress updates.229 230    This endpoint streams progress updates while computing the distance matrix,231    then returns the final solution. Use this when routing=real_roads and you232    want to show progress to the user.233 234    Events emitted:235    - progress: {phase, message, percent, detail}236    - complete: {solution: VehicleRoutePlanModel}237    - error: {message}238    """239    async def generate():240        try:241            demo_data = DemoData[demo_name]242            domain_plan = generate_demo_data(demo_data)243 244            use_real_roads = routing == RoutingMode.REAL_ROADS245 246            if not use_real_roads:247                # Fast path - no progress needed for haversine248                yield f"data: {json.dumps({'event': 'progress', 'phase': 'computing', 'message': 'Computing distances...', 'percent': 50})}\n\n"249                _initialize_distance_matrix(domain_plan, use_real_roads=False)250                yield f"data: {json.dumps({'event': 'progress', 'phase': 'complete', 'message': 'Ready!', 'percent': 100})}\n\n"251                result = plan_to_model(domain_plan)252                # Include geometries (straight lines in haversine mode)253                geometries = _extract_route_geometries(domain_plan)254                yield f"data: {json.dumps({'event': 'complete', 'solution': result.model_dump(by_alias=True), 'geometries': geometries})}\n\n"255            else:256                # Slow path - stream progress for OSMnx257                progress_events = []258 259                def progress_callback(phase: str, message: str, percent: int, detail: str = ""):260                    progress_events.append({261                        'event': 'progress',262                        'phase': phase,263                        'message': message,264                        'percent': percent,265                        'detail': detail266                    })267 268                # Run computation in thread pool to not block269                import concurrent.futures270                with concurrent.futures.ThreadPoolExecutor() as executor:271                    future = executor.submit(272                        _initialize_distance_matrix,273                        domain_plan,274                        use_real_roads=True,275                        progress_callback=progress_callback276                    )277 278                    # Stream progress events while waiting279                    last_sent = 0280                    while not future.done():281                        await asyncio.sleep(0.1)282                        while last_sent < len(progress_events):283                            yield f"data: {json.dumps(progress_events[last_sent])}\n\n"284                            last_sent += 1285 286                    # Send any remaining progress events287                    while last_sent < len(progress_events):288                        yield f"data: {json.dumps(progress_events[last_sent])}\n\n"289                        last_sent += 1290 291                    # Get result (will raise if exception occurred)292                    future.result()293 294                yield f"data: {json.dumps({'event': 'progress', 'phase': 'complete', 'message': 'Ready!', 'percent': 100})}\n\n"295                result = plan_to_model(domain_plan)296 297                # Include geometries in response for real roads mode298                geometries = _extract_route_geometries(domain_plan)299                yield f"data: {json.dumps({'event': 'complete', 'solution': result.model_dump(by_alias=True), 'geometries': geometries})}\n\n"300 301        except KeyError:302            yield f"data: {json.dumps({'event': 'error', 'message': f'Demo data not found: {demo_name}'})}\n\n"303        except Exception as e:304            logger.exception(f"Error in SSE stream: {e}")305            yield f"data: {json.dumps({'event': 'error', 'message': str(e)})}\n\n"306 307    return StreamingResponse(308        generate(),309        media_type="text/event-stream",310        headers={311            "Cache-Control": "no-cache",312            "Connection": "keep-alive",313            "X-Accel-Buffering": "no"314        }315    )316 317 318@app.get("/route-plans/{problem_id}", response_model=VehicleRoutePlanModel, response_model_exclude_none=True)319async def get_route(problem_id: str) -> VehicleRoutePlanModel:320    route = data_sets.get(problem_id)321    if not route:322        raise HTTPException(status_code=404, detail="Route plan not found")323    route.solver_status = solver_manager.get_solver_status(problem_id)324    return plan_to_model(route)325 326@app.post("/route-plans")327async def solve_route(plan_model: VehicleRoutePlanModel) -> str:328    job_id = str(uuid4())329    # Convert to domain model for solver330    domain_plan = model_to_plan(plan_model)331    data_sets[job_id] = domain_plan332    solver_manager.solve_and_listen(333        job_id,334        domain_plan,335        lambda solution: data_sets.update({job_id: solution})336    )337    return job_id338 339@app.put("/route-plans/analyze")340async def analyze_route(plan_model: VehicleRoutePlanModel) -> dict:341    domain_plan = model_to_plan(plan_model)342    analysis = solution_manager.analyze(domain_plan)343    constraints = []344    for constraint in getattr(analysis, 'constraint_analyses', []) or []:345        matches = [346            MatchAnalysisDTO(347                name=str(getattr(getattr(match, 'constraint_ref', None), 'constraint_name', "")),348                score=str(getattr(match, 'score', "0hard/0soft")),349                justification=str(getattr(match, 'justification', ""))350            )351            for match in getattr(constraint, 'matches', []) or []352        ]353        constraints.append(ConstraintAnalysisDTO(354            name=str(getattr(constraint, 'constraint_name', "")),355            weight=str(getattr(constraint, 'weight', "0hard/0soft")),356            score=str(getattr(constraint, 'score', "0hard/0soft")),357            matches=matches358        ))359    return {"constraints": [asdict(constraint) for constraint in constraints]}360 361@app.get("/route-plans")362async def list_route_plans() -> List[str]:363    """List the job IDs of all submitted route plans."""364    return list(data_sets.keys())365 366 367@app.get("/route-plans/{problem_id}/status")368async def get_route_status(problem_id: str) -> dict:369    """Get the route plan status and score for a given job ID."""370    route = data_sets.get(problem_id)371    if not route:372        raise HTTPException(status_code=404, detail="Route plan not found")373    solver_status = solver_manager.get_solver_status(problem_id)374    return {375        "name": route.name,376        "score": str(route.score) if route.score else None,377        "solverStatus": solver_status.name if solver_status else None,378    }379 380 381@app.delete("/route-plans/{problem_id}")382async def stop_solving(problem_id: str) -> VehicleRoutePlanModel:383    """Terminate solving for a given job ID. Returns the best solution so far."""384    solver_manager.terminate_early(problem_id)385    route = data_sets.get(problem_id)386    if not route:387        raise HTTPException(status_code=404, detail="Route plan not found")388    route.solver_status = solver_manager.get_solver_status(problem_id)389    return plan_to_model(route)390 391 392@app.post("/route-plans/recommendation")393async def recommend_assignment(request: RecommendationRequest) -> List[RecommendedAssignmentResponse]:394    """395    Request recommendations for assigning a visit to vehicles.396 397    Returns a list of recommended assignments sorted by score impact.398    """399    domain_plan = model_to_plan(request.solution)400 401    # Find the visit by ID402    visit = None403    for v in domain_plan.visits:404        if v.id == request.visit_id:405            visit = v406            break407 408    if visit is None:409        raise HTTPException(status_code=404, detail=f"Visit {request.visit_id} not found")410 411    # Get recommendations using solution_manager412    try:413        recommendations = solution_manager.recommend_assignment(414            domain_plan,415            visit,416            lambda v: VehicleRecommendation(vehicle_id=v.vehicle.id, index=v.vehicle.visits.index(v))417        )418 419        # Convert to response format (limit to top 5)420        result = []421        for rec in recommendations[:5]:422            result.append(RecommendedAssignmentResponse(423                proposition=rec.proposition,424                score_diff=str(rec.score_diff) if hasattr(rec, 'score_diff') else "0hard/0soft"425            ))426        return result427    except Exception:428        # If recommend_assignment is not available, return empty list429        return []430 431 432@app.post("/route-plans/recommendation/apply")433async def apply_recommendation(request: ApplyRecommendationRequest) -> VehicleRoutePlanModel:434    """435    Apply a recommendation to assign a visit to a vehicle at a specific index.436 437    Returns the updated solution.438    """439    domain_plan = model_to_plan(request.solution)440 441    # Find the vehicle by ID442    vehicle = None443    for v in domain_plan.vehicles:444        if v.id == request.vehicle_id:445            vehicle = v446            break447 448    if vehicle is None:449        raise HTTPException(status_code=404, detail=f"Vehicle {request.vehicle_id} not found")450 451    # Find the visit by ID452    visit = None453    for v in domain_plan.visits:454        if v.id == request.visit_id:455            visit = v456            break457 458    if visit is None:459        raise HTTPException(status_code=404, detail=f"Visit {request.visit_id} not found")460 461    # Insert visit at the specified index462    vehicle.visits.insert(request.index, visit)463 464    # Update the solution to recalculate shadow variables465    solution_manager.update(domain_plan)466 467    return plan_to_model(domain_plan)468 469 470class RouteGeometryResponse(BaseModel):471    """Response containing encoded polyline geometries for all vehicle routes."""472    geometries: Dict[str, List[Optional[str]]]473 474 475@app.get("/route-plans/{problem_id}/geometry", response_model=RouteGeometryResponse)476async def get_route_geometry(problem_id: str) -> RouteGeometryResponse:477    """478    Get route geometries for all vehicle routes in a problem.479 480    Returns encoded polylines (Google polyline format) for each route segment.481    Each vehicle's route is represented as a list of encoded polylines:482    - First segment: depot -> first visit483    - Middle segments: visit -> visit484    - Last segment: last visit -> depot485 486    These can be decoded on the frontend to display actual road routes487    instead of straight lines.488    """489    route = data_sets.get(problem_id)490    if not route:491        raise HTTPException(status_code=404, detail="Route plan not found")492 493    distance_matrix = Location.get_distance_matrix()494    if distance_matrix is None:495        # No distance matrix available - return empty geometries496        return RouteGeometryResponse(geometries={})497 498    geometries: Dict[str, List[Optional[str]]] = {}499 500    for vehicle in route.vehicles:501        segments: List[Optional[str]] = []502 503        if not vehicle.visits:504            # No visits assigned to this vehicle505            geometries[vehicle.id] = segments506            continue507 508        # Segment from depot to first visit509        prev_location = vehicle.home_location510        for visit in vehicle.visits:511            geometry = distance_matrix.get_geometry(prev_location, visit.location)512            segments.append(geometry)513            prev_location = visit.location514 515        # Segment from last visit back to depot516        geometry = distance_matrix.get_geometry(prev_location, vehicle.home_location)517        segments.append(geometry)518 519        geometries[vehicle.id] = segments520 521    return RouteGeometryResponse(geometries=geometries)522 523 524app.mount("/", StaticFiles(directory="static", html=True), name="static")525