import sys import json import collections import math from pyvrp import Model from pyvrp.stop import MaxRuntime def build_and_solve(data, mode_settings, time_limit=20): """ Builds the VRP model and solves it. Args: data: Parsed input data dict mode_settings: Dict with fairness/force keys: - fixed_cost: Cost per vehicle (negative to force usage) - metric: 'distance' or 'stops' (defines what 'duration' represents) - limit: Max duration limit (constraint) time_limit: Max runtime in seconds Returns: (result object, total_distance, num_clients) """ distance_matrix = data.get('distance_matrix', []) demands = data.get('demands', []) vehicle_capacities = data.get('vehicle_capacities', []) locations = data.get('locations', []) strict_capacity = bool(data.get('strict_capacity', False)) fixed_cost = int(mode_settings.get('fixed_cost', 0)) metric = mode_settings.get('metric', 'distance') limit = mode_settings.get('limit', None) SCALE = 1000 demands_int = [int(d * SCALE) for d in demands] caps_int = [int(c * SCALE) for c in vehicle_capacities] m = Model() # Add Depot depot_loc = locations[0] if locations and len(locations) > 0 else {'lat': 0, 'lon': 0} depot = m.add_depot(x=float(depot_loc.get('lon', 0)), y=float(depot_loc.get('lat', 0)), name="0") # Add Clients clients = [] for i in range(1, len(demands)): d = demands_int[i] loc = locations[i] if locations and i < len(locations) else {'lat': 0, 'lon': 0} # Define Service Time based on metric # If balancing stops, service time = 1, else 0 service_time = 1 if metric == 'stops' else 0 client = m.add_client( x=float(loc.get('lon', 0)), y=float(loc.get('lat', 0)), delivery=d, service_duration=service_time, name=str(i), required=False, prize=10_000_000 ) clients.append(client) all_nodes = [depot] + clients # Add Edges for i in range(len(distance_matrix)): for j in range(len(distance_matrix[i])): cost = int(distance_matrix[i][j]) # Define Duration based on metric # If balancing stops, travel duration = 0 # If balancing distance, travel duration = cost duration = 0 if metric == 'stops' else cost m.add_edge(all_nodes[i], all_nodes[j], distance=cost, duration=duration) # Add Vehicle Types cap_counts = collections.Counter(caps_int) for cap, count in cap_counts.items(): # Apply limits # If metric is 'stops', limit applies to shift_duration (where duration=stops) # If metric is 'distance', limit applies to max_distance shift_dur = 2_000_000_000 max_dist = 2_000_000_000 if limit is not None: if metric == 'stops': shift_dur = int(limit) elif metric == 'distance': max_dist = int(limit) effective_cap = cap if not strict_capacity: # Allow 5% tolerance if strict capacity is OFF effective_cap = int(cap * 1.05) m.add_vehicle_type( capacity=effective_cap, num_available=count, fixed_cost=fixed_cost, shift_duration=shift_dur, max_distance=max_dist ) res = m.solve(stop=MaxRuntime(time_limit), display=False) # Calculate stats manually best_sol = res.best total_dist = 0 # Calculate total distance using input matrix for route in best_sol.routes(): route_indices = [0] + [c for c in route] + [0] for k in range(len(route_indices) - 1): total_dist += distance_matrix[route_indices[k]][route_indices[k+1]] return res, total_dist, best_sol.num_clients() def main(): try: # sys.stderr.write("Reading input...\n") input_data = sys.stdin.read() # sys.stderr.write(f"Read {len(input_data)} bytes\n") if not input_data: return data = json.loads(input_data) if not data.get('distance_matrix') or not data.get('demands') or not data.get('vehicle_capacities'): raise ValueError("Missing required data") fairness_mode = data.get('fairness_mode', 'none') vehicle_strategy = data.get('vehicle_strategy', 'balanced') max_route_distance_km = float(data.get('max_route_distance_km', 0) or 0) max_route_stops = int(data.get('max_route_stops', 0) or 0) strict_capacity = bool(data.get('strict_capacity', False)) manual_fixed_cost = data.get('fixed_cost') vehicle_capacities = data.get('vehicle_capacities', []) force_usage = bool(data.get('force_usage', False)) pass1_fixed_cost = int(manual_fixed_cost) if manual_fixed_cost is not None else 0 # Debug sys.stderr.write(f"DEBUG: fixed_cost={manual_fixed_cost}, force_usage={force_usage}, pass1={pass1_fixed_cost}\n") # PASS 1: Base Optimization (distance metric, no fairness constraints) base_limit = None if max_route_distance_km > 0: base_limit = max_route_distance_km * 1000.0 res1, dist1, clients1 = build_and_solve( data, mode_settings={'fixed_cost': pass1_fixed_cost, 'metric': 'distance', 'limit': base_limit}, time_limit=15 ) final_res = res1 final_dist = dist1 used_vehicles = res1.best.num_routes() target_vehicles = used_vehicles if used_vehicles > 0 else 1 # If forcing usage, set target to ALL available vehicles if force_usage: target_vehicles = len(vehicle_capacities) # Strategy 1: Squeeze Capacity to force distribution # We want to ensure that (Target-1) vehicles CANNOT hold the total demand. # So (Target-1) * NewCap < TotalDemand # NewCap < TotalDemand / (Target-1) if target_vehicles > 1: demands = data.get('demands', []) total_demand = sum(demands) max_single_demand = max(demands) # Calculate theoretical cap needed to force split # Use 0.95 factor to ensure strictly less than limit_cap = (total_demand / (target_vehicles - 1)) * 0.95 # Ensure we can still carry the biggest item effective_cap = max(max_single_demand, limit_cap) # Apply to all vehicles (taking min of original and calculated) # Note: This assumes homogeneous fleet for simplicity in this logic, # or applies the squeeze to all. new_caps = [] for cap in vehicle_capacities: new_caps.append(min(cap, int(effective_cap))) vehicle_capacities = new_caps data['vehicle_capacities'] = new_caps # Debug sys.stderr.write(f"DEBUG: Force Usage Capacity Squeeze. Total={total_demand}, Target={target_vehicles}, Limit={limit_cap}, NewCaps={new_caps}\n") # Strategy 2: Distance Limit (Relaxed) # Previous logic was too strict (avg * 1.2). # We'll use a looser limit or rely on capacity squeeze. # limit = avg_dist * 2.0 pass2_caps = vehicle_capacities fixed_cost = 0 if manual_fixed_cost is not None: fixed_cost = int(manual_fixed_cost) elif force_usage: fixed_cost = 0 # Ensure no penalty for vehicles elif vehicle_strategy == 'min_vehicles': fixed_cost = int(max(1, dist1 / max(1, target_vehicles * 3))) elif vehicle_strategy == 'balanced': fixed_cost = int(max(0, dist1 / max(1, target_vehicles * 10))) else: fixed_cost = 0 if clients1 > 0 and (fairness_mode != 'none' or (fixed_cost != 0 and fixed_cost != pass1_fixed_cost) or force_usage): if target_vehicles > 1: effective_mode = fairness_mode if force_usage and effective_mode == 'none': effective_mode = 'distance' # Default to distance balancing if forced metric = 'distance' limit = base_limit # If force usage, we need a TIGHTER limit to force vehicles # Calculate ideal average if effective_mode == 'distance': metric = 'distance' avg_dist = dist1 / target_vehicles factor = 1.10 if force_usage: # If forced, do NOT apply tight distance limit, # because geometry might require long routes. # Capacity squeeze handles the forcing. limit = None else: limit = avg_dist * factor elif effective_mode == 'stops': metric = 'stops' avg_stops = clients1 / target_vehicles factor = 1.10 limit = math.ceil(avg_stops * factor) if metric == 'stops' and max_route_stops > 0: limit = min(limit, max_route_stops) # Ensure base_limit is respected if base_limit is not None and metric == 'distance': limit = min(limit, base_limit) res2, dist2, clients2 = build_and_solve( data, mode_settings={'fixed_cost': fixed_cost, 'metric': metric, 'limit': limit}, time_limit=15 ) # If force usage, prefer result with MORE vehicles (routes) even if distance is worse accept = False if force_usage: if res2.best.num_routes() > used_vehicles: accept = True elif res2.best.num_routes() == used_vehicles and dist2 < dist1: accept = True else: if clients2 >= clients1 and dist2 <= dist1 * 1.25: accept = True if accept: final_res = res2 final_dist = dist2 # Construct Output output_routes = [] for route in final_res.best.routes(): route_indices = [0] + [c for c in route] + [0] output_routes.append(route_indices) output = { "routes": output_routes, "total_distance": final_dist, "status": "OPTIMAL" if final_res.is_feasible() else "INFEASIBLE" } print(json.dumps(output)) except Exception as e: error_out = { "error": str(e), "routes": [], "total_distance": 0 } print(json.dumps(error_out)) sys.exit(1) if __name__ == "__main__": # sys.stderr.write("Starting Python Solver...\n") # sys.stderr.flush() main()