Enterprise AI Engineering & Solutions for Logistics & Supply Chain

import time
from fastapi import FastAPI
from pydantic import BaseModel
from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp

app = FastAPI(title="Logistics Vehicle Route Solver", version="2.0.0")

class RouteRequest(BaseModel):
    distance_matrix: list[list[int]]
    demands: list[int]
    vehicle_capacities: list[int]
    num_vehicles: int
    depot: int = 0

class RouteResponse(BaseModel):
    solve_latency_ms: float
    total_distance_km: float
    vehicle_routes: list[list[int]]

@app.post("/api/v1/solve-routes", response_model=RouteResponse)
async def solve_routes(payload: RouteRequest):
    start_time = time.perf_counter()
    
    # 1. Create Routing Index Manager and Model
    manager = pywrapcp.RoutingIndexManager(
        len(payload.distance_matrix), payload.num_vehicles, payload.depot
    )
    routing = pywrapcp.RoutingModel(manager)

    # 2. Register Transit Distance Callback
    def distance_callback(from_index, to_index):
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        return payload.distance_matrix[from_node][to_node]

    transit_callback_index = routing.RegisterTransitCallback(distance_callback)
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)

    # 3. Add Vehicle Capacity Constraints
    def demand_callback(from_index):
        from_node = manager.IndexToNode(from_index)
        return payload.demands[from_node]

    demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
    routing.AddDimensionWithVehicleCapacity(
        demand_callback_index, 0, payload.vehicle_capacities, True, "Capacity"
    )

    # 4. Set Search Parameters & Solve
    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    solution = routing.SolveWithParameters(search_parameters)

    elapsed_ms = (time.perf_counter() - start_time) * 1000

    routes = []
    total_dist = 0
    if solution:
        for vehicle_id in range(payload.num_vehicles):
            index = routing.Start(vehicle_id)
            route = []
            while not routing.IsEnd(index):
                route.append(manager.IndexToNode(index))
                previous_index = index
                index = solution.Value(routing.NextVar(index))
                total_dist += routing.GetArcCostForVehicle(previous_index, index, vehicle_id)
            route.append(manager.IndexToNode(index))
            routes.append(route)

    return RouteResponse(
        solve_latency_ms=round(elapsed_ms, 2),
        total_distance_km=round(total_dist / 1000.0, 2),
        vehicle_routes=routes
    )