vrp.py

# vrp.py
from typing import Dict, List, Tuple
from dataclasses import dataclass
from graph import Graph
import math
import random


@dataclass
class Vehicle:
    id: int
    capacity: float
    max_time: float = float("inf")  # Tk
    # route/load/time tracked externally


# ---------- utilities ----------
def compress_route(route: List[int]) -> List[int]:
    if not route:
        return route
    comp = [route[0]]
    for x in route[1:]:
        if x != comp[-1]:
            comp.append(x)
    i = 0
    while i < len(comp) - 2:
        if comp[i] == comp[i + 2]:
            del comp[i + 1]
            if i > 0:
                i -= 1
        else:
            i += 1
    return comp


def two_opt_route_improve(graph: Graph, route: List[int], weight_lambda: float = 0.0) -> List[int]:
    if len(route) < 4:
        return route
    def route_cost(r):
        c = 0.0
        for i in range(len(r) - 1):
            u, v = r[i], r[i + 1]
            cost, rel = graph.get_edge_cost_rel(u, v)
            if math.isinf(cost):
                return float("inf")
            c += cost + weight_lambda * (1.0 - rel)
        return c

    best = route[:]
    best_cost = route_cost(best)
    improved = True
    while improved:
        improved = False
        n = len(best)
        for i in range(1, n - 2):
            for j in range(i + 1, n - 1):
                new_route = best[:i] + best[i:j+1][::-1] + best[j+1:]
                nc = route_cost(new_route)
                if nc + 1e-9 < best_cost:
                    best = new_route
                    best_cost = nc
                    improved = True
                    break
            if improved:
                break
    return best


# ---------- priority-aware VRP ----------
def priority_aware_vrp(graph: Graph, depot_id: int, vehicles: List[Vehicle], nodes_info: Dict[int, Dict],
                       alpha_time: float = 1.0, weight_lambda: float = 0.0, theta: float = float("inf"),
                       do_2opt: bool = True) -> Tuple[Dict[int, List[int]], Dict[int, float], Dict[int, float]]:
    """
    Priority-aware assignment:
    - Assign high-priority nodes first (priority 5 highest).
    - For each node (descending priority), assign to the vehicle that can serve it feasibly
      (capacity & time) with minimum incremental time (using shortest-path expansions).
    - Build routes by appending actual shortest-path sequences between nodes.
    """
    # initialize
    routes = {v.id: [depot_id] for v in vehicles}
    loads = {v.id: 0.0 for v in vehicles}
    times = {v.id: 0.0 for v in vehicles}

    # helper: dijkstra cache per source
    dijkstra_cache = {}

    def dijkstra_cached(src):
        key = (src, weight_lambda)
        if key in dijkstra_cache:
            return dijkstra_cache[key]
        dijkstra_cache[key] = graph.dijkstra(src, weight_lambda=weight_lambda)
        return dijkstra_cache[key]

    # mark which customers left to serve
    customers = [nid for nid in nodes_info.keys() if nid != depot_id]
    # sort by priority desc, then demand desc
    customers.sort(key=lambda x: (-nodes_info[x]['priority'], -nodes_info[x]['demand']))

    for nid in customers:
        assigned = False
        # attempt to assign to best vehicle (min incremental cost) while enforcing theta for high priority
        best_v = None
        best_incr = float("inf")
        best_path = None
        for v in vehicles:
            if loads[v.id] + nodes_info[nid]['demand'] > v.capacity:
                continue
            last = routes[v.id][-1]
            dist_last, parent_last = dijkstra_cached(last)
            to_n = dist_last.get(nid, float("inf"))
            if math.isinf(to_n):
                continue
            # estimate time to get back to depot after adding node
            dist_n, parent_n = dijkstra_cached(nid)
            to_depot_from_n = dist_n.get(depot_id, float("inf"))
            to_depot_from_last = dist_last.get(depot_id, float("inf"))
            if math.isinf(to_depot_from_n):
                continue
            incr = (to_n + to_depot_from_n) - (to_depot_from_last if not math.isinf(to_depot_from_last) else 0.0)
            # Enforce theta for high-priority nodes (if priority high and theta finite)
            if nodes_info[nid]['priority'] >= 4 and incr > theta:
                continue
            if times[v.id] + incr <= v.max_time and incr < best_incr:
                best_incr = incr
                best_v = v
                best_path = graph.path_from_parent(parent_last, nid)
        if best_v is not None and best_path is not None:
            # append path (excluding starting node)
            if len(best_path) >= 2:
                routes[best_v.id].extend(best_path[1:])
            else:
                routes[best_v.id].append(nid)
            loads[best_v.id] += nodes_info[nid]['demand']
            times[best_v.id] += best_incr
            assigned = True
        # if not assigned, node remains unserved for now
        if not assigned:
            # try last-resort: assign ignoring theta but respecting capacity and connectivity
            for v in vehicles:
                if loads[v.id] + nodes_info[nid]['demand'] > v.capacity:
                    continue
                last = routes[v.id][-1]
                dist_last, parent_last = dijkstra_cached(last)
                if math.isinf(dist_last.get(nid, float("inf"))):
                    continue
                path = graph.path_from_parent(parent_last, nid)
                if len(path) >= 2:
                    routes[v.id].extend(path[1:])
                else:
                    routes[v.id].append(nid)
                loads[v.id] += nodes_info[nid]['demand']
                assigned = True
                break
        # continue to next customer

    # close routes back to depot
    for v in vehicles:
        if routes[v.id][-1] != depot_id:
            last = routes[v.id][-1]
            dist_last, parent_last = dijkstra_cached(last)
            if not math.isinf(dist_last.get(depot_id, float("inf"))):
                path_back = graph.path_from_parent(parent_last, depot_id)
                if len(path_back) >= 2:
                    routes[v.id].extend(path_back[1:])
                else:
                    routes[v.id].append(depot_id)
            else:
                # disconnected -> append depot (cost will be inf)
                routes[v.id].append(depot_id)

    # cleanup: compress and optional 2-opt
    for v in vehicles:
        r = compress_route(routes[v.id])
        if do_2opt:
            r = two_opt_route_improve(graph, r, weight_lambda=weight_lambda)
        if r[0] != depot_id:
            r.insert(0, depot_id)
        if r[-1] != depot_id:
            r.append(depot_id)
        routes[v.id] = r

    return routes, loads, times


# ---------- correct evaluation function ----------
def evaluate_solution(graph: Graph, routes: Dict[int, List[int]], vehicles: List[Vehicle],
                      nodes_info: Dict[int, Dict], alpha: float = 1.0, beta: float = 1.0, gamma: float = 1.0,
                      weight_lambda: float = 0.0) -> Dict:
    """
    Correct objective:
      alpha * sum_i p_i * t_i
      + beta * sum_edges (1 - r(u,v)) over all used edges
      + gamma * sum_k idle(k) where idle(k) = capacity_k - sum(d_i in route)
    Returns detailed metrics.
    """
    total_priority_time = 0.0
    total_unreliability = 0.0
    total_idle = 0.0
    total_edges_used = 0
    delivered_per_vehicle = {v.id: 0.0 for v in vehicles}

    # map vehicle id -> capacity
    cap_map = {v.id: v.capacity for v in vehicles}

    for vid, route in routes.items():
        prefix_time = 0.0
        load = 0.0
        for i in range(len(route) - 1):
            u, v = route[i], route[i + 1]
            cost, rel = graph.get_edge_cost_rel(u, v)
            if math.isinf(cost):
                # disconnected edge -> treat cost as inf and unreliability max
                prefix_time = float("inf")
                total_unreliability += 1.0
            else:
                prefix_time += cost + weight_lambda * (1.0 - rel)
                total_unreliability += (1.0 - rel)
            total_edges_used += 1
            # if v is a customer (not depot), accumulate priority*time and load
            if v in nodes_info and v != list(nodes_info.keys())[0]:
                # nodes_info assumed to include depot as first key maybe: we rely on explicit check v != depot if needed by caller
                total_priority_time += nodes_info[v]['priority'] * prefix_time
                load += nodes_info[v]['demand']
        delivered_per_vehicle[vid] = load
        # idle = leftover capacity
        total_idle += max(0.0, cap_map.get(vid, 0.0) - load)

    avg_reliability = 1.0
    if total_edges_used > 0:
        avg_reliability = max(0.0, 1.0 - (total_unreliability / total_edges_used))

    objective = alpha * total_priority_time + beta * total_unreliability + gamma * total_idle

    return {
        "alpha_priority_time": total_priority_time,
        "beta_edge_unreliability": total_unreliability,
        "gamma_total_idle": total_idle,
        "avg_reliability": avg_reliability,
        "objective_value": objective,
        "delivered_per_vehicle": delivered_per_vehicle,
        "routes": routes
    }


# ---------- dynamic replanning ----------
def replan_on_blocked_edge(graph: Graph, blocked_edge: Tuple[int, int], depot_id: int, vehicles: List[Vehicle],
                           current_routes: Dict[int, List[int]], nodes_info: Dict[int, Dict],
                           weight_lambda: float = 0.0) -> Dict[int, List[int]]:
    """
    Simple replanning: remove the blocked edge, then recompute routes for vehicles
    from their current position to serve remaining unserved customers (greedy priority-aware).
    This is a practical approach - for heavy dynamic loads use incremental algorithms (D* Lite).
    """
    u, v = blocked_edge
    # remove edge from graph
    graph.remove_edge(u, v)

    # determine which customers are already served (nodes present before current position in route)
    served = set()
    for vid, route in current_routes.items():
        # assume everything in route up to current point is served; in a real system we'd know current position/time
        for node in route:
            if node != depot_id:
                served.add(node)

    # compute new routes by re-assigning unserved customers
    unserved_nodes = {nid: nodes_info[nid] for nid in nodes_info if nid != depot_id and nid not in served}
    # call priority_aware_vrp with only the unserved customers by building a nodes_info subset
    # We'll keep same vehicles but start them at depot for simplicity (or one could start from current position)
    new_routes, loads, times = priority_aware_vrp(graph, depot_id, vehicles, nodes_info, weight_lambda=weight_lambda)
    return new_routes