8b0241673d
route_solver.py : - dimension Temps PAR VÉHICULE (AddDimensionWithVehicleTransits) : service × efficacité(tech,skill) → un tech rapide (facteur bas) tient plus de jobs. - coût d'arc : + efficacité (rapide récompensé) + pénalité de POLYVALENCE (nb de compétences) → les cas simples vont d'abord aux techs les moins polyvalents, réservant les experts. eff_bias=8, versatility_weight=6. roster.js : passe skill_eff + efficiency par véhicule au solveur. Testé en conteneur prod : tech rapide+mono-compétence rafle les jobs, polyvalent réservé. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
241 lines
12 KiB
Python
241 lines
12 KiB
Python
"""
|
||
Dispatch VRP — solveur de TOURNÉES (OR-Tools Routing / CP).
|
||
|
||
Problème : VRPTW + compétences + temps sur place (« field service »).
|
||
• nœuds = jobs ; véhicules = techniciens (chacun part/revient à SON domicile)
|
||
• compétence requise du job = filtre DUR (véhicules autorisés)
|
||
• temps sur place (service) + trajet comptent dans la dimension TEMPS, bornée
|
||
par le quart de chaque tech (fenêtre) ; fenêtre horaire par job optionnelle
|
||
• objectif : MINIMISER le trajet total (+ léger biais spécialiste via l'ordre
|
||
des compétences du tech) ; un job qui ne rentre pas est laissé NON planifié
|
||
(pénalité de « drop ») plutôt que de rendre le modèle infaisable.
|
||
|
||
C'est ce que le greedy ne peut pas faire : optimisation GLOBALE → regroupe les
|
||
jobs d'un même secteur sur le même tech (plus de fractionnement).
|
||
|
||
Entrée (dict) :
|
||
jobs: [{id, lat, lon, service_min, skill, tw_start_min?, tw_end_min?, priority_boost?}]
|
||
vehicles: [{id, name, skills:[...] (ORDRE = priorité), home_lat?, home_lon?,
|
||
shift_start_min, shift_end_min}]
|
||
matrix?: [[minutes]] (N_jobs+N_veh carré, jobs d'abord puis domiciles) — sinon haversine
|
||
speed_kmh?, default_service_min?, rank_weight?, drop_penalty?, max_seconds?
|
||
Sortie : {status, routes:[{vehicle, vehicle_name, stops:[{job_id,arrival_min,service_min}],
|
||
travel_min}], unassigned:[job_id], total_travel_min, objective, solve_ms}
|
||
"""
|
||
from __future__ import annotations
|
||
import math
|
||
from ortools.constraint_solver import pywrapcp, routing_enums_pb2
|
||
|
||
|
||
def _haversine_km(a, b) -> float:
|
||
(lat1, lon1), (lat2, lon2) = a, b
|
||
R = 6371.0
|
||
p1, p2 = math.radians(lat1), math.radians(lat2)
|
||
dphi = math.radians(lat2 - lat1)
|
||
dlmb = math.radians(lon2 - lon1)
|
||
h = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
|
||
return 2 * R * math.asin(min(1.0, math.sqrt(h)))
|
||
|
||
|
||
def solve_route(req: dict) -> dict:
|
||
all_jobs = req.get("jobs", []) or []
|
||
vehicles = req.get("vehicles", []) or []
|
||
if not all_jobs or not vehicles:
|
||
return {"status": "EMPTY", "routes": [], "unassigned": [j["id"] for j in all_jobs],
|
||
"total_travel_min": 0, "objective": None, "solve_ms": 0}
|
||
|
||
speed = float(req.get("speed_kmh") or 45.0)
|
||
default_service = int(req.get("default_service_min") or 60)
|
||
# DISTANCE PRIORITAIRE : la spécialité (ordre de compétence) ne fait que DÉPARTAGER à trajet ~égal.
|
||
# rank_w = « minutes virtuelles » par cran d'ordre (spécialiste=0). Petit → un tech multi-compétences PROCHE
|
||
# passe avant un spécialiste LOIN (ex. un installateur qui a aussi la réparation prend une réparation voisine).
|
||
rank_w = float(req.get("rank_weight") if req.get("rank_weight") is not None else 2.0)
|
||
base_pen = int(req.get("drop_penalty") or 100000)
|
||
max_seconds = float(req.get("max_seconds") or 10)
|
||
# EFFICACITÉ (par compétence) : facteur de DURÉE (bas = rapide) → un tech rapide fait le job en moins de temps
|
||
# (dimension Temps par véhicule) ET coûte moins (biais de coût) → il reçoit plus de travail. Voir eff_bias.
|
||
eff_bias = float(req.get("eff_bias") if req.get("eff_bias") is not None else 8.0)
|
||
# RÉSERVE DES EXPERTS : pénalise les techs POLYVALENTS (beaucoup de compétences) → les cas simples vont d'abord
|
||
# aux moins polyvalents (qui ne pourraient rien faire d'autre), gardant les experts pour les jobs exigeants.
|
||
vers_w = float(req.get("versatility_weight") if req.get("versatility_weight") is not None else 6.0)
|
||
|
||
def eff_of(veh, sk):
|
||
se = veh.get("skill_eff") or {}
|
||
if sk and se.get(sk):
|
||
try:
|
||
return max(0.2, min(3.0, float(se[sk])))
|
||
except Exception:
|
||
pass
|
||
try:
|
||
return max(0.2, min(3.0, float(veh.get("efficiency") or 1.0)))
|
||
except Exception:
|
||
return 1.0
|
||
|
||
def skill_ok(veh, sk):
|
||
return (not sk) or (sk in (veh.get("skills") or []))
|
||
|
||
# Pré-filtre : un job sans AUCUN tech compétent est non planifiable → hors modèle (évite les "allowed" vides).
|
||
jobs, unservable = [], []
|
||
for j in all_jobs:
|
||
if any(skill_ok(v, j.get("skill")) for v in vehicles):
|
||
jobs.append(j)
|
||
else:
|
||
unservable.append(j["id"])
|
||
if not jobs:
|
||
return {"status": "OK", "routes": [], "unassigned": unservable,
|
||
"total_travel_min": 0, "objective": 0, "solve_ms": 0}
|
||
|
||
n_jobs, n_veh = len(jobs), len(vehicles)
|
||
N = n_jobs + n_veh # 0..n_jobs-1 = jobs ; n_jobs+v = domicile du véhicule v
|
||
|
||
coords = [None] * N
|
||
for i, j in enumerate(jobs):
|
||
if j.get("lat") is not None and j.get("lon") is not None:
|
||
coords[i] = (float(j["lat"]), float(j["lon"]))
|
||
for v, veh in enumerate(vehicles):
|
||
if veh.get("home_lat") is not None and veh.get("home_lon") is not None:
|
||
coords[n_jobs + v] = (float(veh["home_lat"]), float(veh["home_lon"]))
|
||
|
||
supplied = req.get("matrix")
|
||
|
||
def travel_min(a, b):
|
||
if supplied is not None:
|
||
return supplied[a][b]
|
||
ca, cb = coords[a], coords[b]
|
||
if ca is None or cb is None:
|
||
return 0 # coordonnée inconnue → trajet neutre (ne pas pénaliser à tort)
|
||
return _haversine_km(ca, cb) * 60.0 / speed
|
||
|
||
service = [0] * N
|
||
for i, j in enumerate(jobs):
|
||
service[i] = int(j.get("service_min") or default_service)
|
||
|
||
starts = [n_jobs + v for v in range(n_veh)]
|
||
manager = pywrapcp.RoutingIndexManager(N, n_veh, starts, starts) # départ == retour == domicile
|
||
routing = pywrapcp.RoutingModel(manager)
|
||
|
||
# Coût PAR VÉHICULE = trajet + biais spécialiste (rang de la compétence chez CE tech ; 0 = principale).
|
||
def make_cost_cb(v):
|
||
veh = vehicles[v]
|
||
vskills = veh.get("skills") or []
|
||
nsk = len(vskills)
|
||
|
||
def cb(fi, ti):
|
||
f = manager.IndexToNode(fi)
|
||
t = manager.IndexToNode(ti)
|
||
base = travel_min(f, t)
|
||
if t < n_jobs:
|
||
sk = jobs[t].get("skill")
|
||
rank = vskills.index(sk) if (sk and sk in vskills) else 0
|
||
base += rank * rank_w # spécialité : garde le spécialiste sur sa compétence
|
||
base += max(0, nsk - 1) * vers_w # polyvalence pénalisée → cas simples aux moins polyvalents
|
||
base += (eff_of(veh, sk) - 1.0) * eff_bias # efficacité : rapide (<1) récompensé, lent (>1) pénalisé
|
||
return int(round(max(0.0, base)))
|
||
return cb
|
||
|
||
for v in range(n_veh):
|
||
cidx = routing.RegisterTransitCallback(make_cost_cb(v))
|
||
routing.SetArcCostEvaluatorOfVehicle(cidx, v)
|
||
|
||
# Dimension TEMPS = trajet + temps sur place (SERVICE) du nœud de départ, ce dernier MIS À L'ÉCHELLE par l'efficacité
|
||
# du véhicule pour la compétence du job (rapide = moins de temps → tient plus de jobs) → callbacks PAR VÉHICULE.
|
||
def make_time_cb(v):
|
||
veh = vehicles[v]
|
||
|
||
def cb(fi, ti):
|
||
f = manager.IndexToNode(fi)
|
||
svc = int(round(service[f] * eff_of(veh, jobs[f].get("skill")))) if f < n_jobs else 0
|
||
return int(round(travel_min(f, manager.IndexToNode(ti)) + svc))
|
||
return cb
|
||
time_cbs = [routing.RegisterTransitCallback(make_time_cb(v)) for v in range(n_veh)]
|
||
|
||
horizon = 1440
|
||
for veh in vehicles:
|
||
horizon = max(horizon, int(veh.get("shift_end_min") or 0) + int(veh.get("overtime_min") or 0))
|
||
for j in jobs:
|
||
if j.get("tw_end_min") is not None:
|
||
horizon = max(horizon, int(j["tw_end_min"]))
|
||
routing.AddDimensionWithVehicleTransits(time_cbs, horizon, horizon, False, "Time") # temps PAR VÉHICULE (efficacité) ; slack=horizon
|
||
time_dim = routing.GetDimensionOrDie("Time")
|
||
|
||
# Fenêtre de QUART par véhicule. La fin peut aller jusqu'à +overtime_min (plafond ~120%), MAIS toute heure au-delà du quart
|
||
# nominal est PÉNALISÉE (borne souple) → le solveur ÉTALE l'excédent vers les techs encore sous 100% avant de faire des heures sup.
|
||
ot_coef = int(req.get("overtime_coef") or 20)
|
||
for v, veh in enumerate(vehicles):
|
||
s0 = int(veh.get("shift_start_min") or 0)
|
||
s1 = int(veh.get("shift_end_min") or horizon)
|
||
if s1 < s0:
|
||
s1 = horizon
|
||
ot = max(0, int(veh.get("overtime_min") or 0))
|
||
time_dim.CumulVar(routing.Start(v)).SetRange(s0, s1)
|
||
time_dim.CumulVar(routing.End(v)).SetRange(s0, s1 + ot) # jusqu'à ~120 % du quart
|
||
if ot > 0:
|
||
time_dim.SetCumulVarSoftUpperBound(routing.End(v), s1, ot_coef) # au-delà du quart nominal = coûteux → équilibre
|
||
|
||
# Fenêtre horaire par job (optionnelle) + compétence (véhicules autorisés) + pénalité de drop.
|
||
# NB : SetAllowedVehiclesForIndex a un typemap Span cassé en ortools 9.15 → on contraint VehicleVar ∈ autorisés ∪ {-1}
|
||
# (-1 = job non planifié ; la disjonction le permet).
|
||
sv = routing.solver()
|
||
for i, j in enumerate(jobs):
|
||
idx = manager.NodeToIndex(i)
|
||
tws, twe = j.get("tw_start_min"), j.get("tw_end_min")
|
||
if tws is not None and twe is not None and int(twe) >= int(tws):
|
||
time_dim.CumulVar(idx).SetRange(int(tws), int(twe)) # fenêtre AM/PM (dur) : ex. AM [480,720], PM [720,960]
|
||
uw = int(j.get("urgent_weight") or 0)
|
||
if uw > 0:
|
||
# URGENCE → servir TÔT : coût = uw × heure d'arrivée (borne souple à 0) → tire le job en début de tournée
|
||
# (conditionne le début de journée du tech). Reste souple : ne casse pas la faisabilité, se combine aux fenêtres AM/PM.
|
||
time_dim.SetCumulVarSoftUpperBound(idx, 0, uw)
|
||
sk = j.get("skill")
|
||
allowed = [v for v, veh in enumerate(vehicles) if skill_ok(veh, sk)]
|
||
if len(allowed) < n_veh: # compétence restreint réellement → filtre DUR
|
||
sv.Add(sv.MemberCt(routing.VehicleVar(idx), allowed + [-1]))
|
||
routing.AddDisjunction([idx], base_pen + int(j.get("priority_boost") or 0)) # laisser tomber coûte cher → préfère assigner
|
||
|
||
params = pywrapcp.DefaultRoutingSearchParameters()
|
||
params.first_solution_strategy = routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
|
||
params.local_search_metaheuristic = routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
|
||
params.time_limit.FromSeconds(max(1, int(max_seconds)))
|
||
sol = routing.SolveWithParameters(params)
|
||
|
||
if sol is None:
|
||
return {"status": "NO_SOLUTION", "routes": [], "unassigned": [j["id"] for j in jobs] + unservable,
|
||
"total_travel_min": 0, "objective": None, "solve_ms": 0,
|
||
"message": "Aucune tournée faisable (fenêtres/quarts trop serrés)."}
|
||
|
||
routes = []
|
||
assigned = set()
|
||
total_travel = 0
|
||
for v, veh in enumerate(vehicles):
|
||
idx = routing.Start(v)
|
||
stops = []
|
||
route_travel = 0
|
||
while not routing.IsEnd(idx):
|
||
node = manager.IndexToNode(idx)
|
||
nxt = sol.Value(routing.NextVar(idx))
|
||
route_travel += int(round(travel_min(node, manager.IndexToNode(nxt))))
|
||
if node < n_jobs:
|
||
stops.append({"job_id": jobs[node]["id"],
|
||
"arrival_min": int(sol.Value(time_dim.CumulVar(idx))),
|
||
"service_min": service[node]})
|
||
assigned.add(node)
|
||
idx = nxt
|
||
if stops:
|
||
total_travel += route_travel
|
||
routes.append({"vehicle": veh["id"], "vehicle_name": veh.get("name", veh["id"]),
|
||
"stops": stops, "travel_min": int(route_travel)})
|
||
|
||
unassigned = [jobs[i]["id"] for i in range(n_jobs) if i not in assigned] + unservable
|
||
try:
|
||
solve_ms = int(routing.solver().WallTime())
|
||
except Exception:
|
||
solve_ms = None
|
||
return {
|
||
"status": "OK",
|
||
"routes": routes,
|
||
"unassigned": unassigned,
|
||
"total_travel_min": int(total_travel),
|
||
"objective": sol.ObjectiveValue(),
|
||
"solve_ms": solve_ms,
|
||
}
|