"""How the three parts work together. NN time matrix --> CSP assigns orders to vans --> A* orders the stops of each van ^ | | a route is late: the first late pair (previous | +-- stop, late stop) becomes a new "different van" <-+ rule and the CSP re-runs (at most 5 loops) If the CSP has no solution (or loops run out) we report the unserviceable orders with a reason each and try widening their windows in 15-minute steps. """ from __future__ import annotations from dataclasses import dataclass, field from .csp import AssignmentCSP, fmt_t from .search import RouteProblem, plan_van MAX_LOOPS = 5 @dataclass class PlanResult: status: str # 'ok' | 'late' | 'no_solution' routes: dict = field(default_factory=dict) assignment: dict = field(default_factory=dict) loops: list = field(default_factory=list) learned_edges: list = field(default_factory=list) csp: AssignmentCSP | None = None unserviceable: list = field(default_factory=list) # (order id, reason) widening: dict | None = None message: str = "" @property def n_late(self): return sum(r.n_late for r in self.routes.values()) def van_problem(inst, tm, van, order_ids) -> RouteProblem: return RouteProblem([inst.order(i) for i in order_ids], inst.depot, van.shift_start, tm, inst.return_to_depot, van.shift_end, van.id) def plan(inst, tm, fc: bool = True, ac3: bool = True, algo: str = "astar", max_loops: int = MAX_LOOPS, suggest: bool = True, traces: dict | None = None) -> PlanResult: learned: list = [] loops = [] last = None for loop in range(1, max_loops + 1): csp = AssignmentCSP(inst, tm, learned) res = csp.solve(fc=fc, ac3=ac3) info = {"loop": loop, "csp_status": res.status, "csp_stats": res.stats, "n_edges_nn": sum(1 for s in csp.edges.values() if s == "nn"), "n_edges_learned": sum(1 for s in csp.edges.values() if s == "learned"), "new_rules": []} loops.append(info) if res.status != "solved": out = PlanResult("no_solution", loops=loops, learned_edges=learned, csp=csp) if last is not None: # keep the best late plan we had out.routes, out.assignment = last.routes, last.assignment out.unserviceable = explain_no_solution(csp, inst, tm) if suggest: out.widening = suggest_widening(inst, tm, [u for u, _ in out.unserviceable if u is not None]) out.message = format_no_solution(out) return out routes = {} for v in inst.vans: ids = [i for i, vv in res.assignment.items() if vv == v.id] ids.sort(key=lambda i: inst.order(i).earliest) tr = [] if traces is not None else None routes[v.id] = plan_van(van_problem(inst, tm, v, ids), algo=algo, trace=tr) if traces is not None: traces[v.id] = tr last = PlanResult("late", routes, res.assignment, loops, learned, csp) late_vans = [r for r in routes.values() if not r.feasible] if not late_vans: last.status = "ok" last.message = f"all {len(inst.orders)} orders on time after {loop} loop(s)" return last for r in late_vans: k = r.first_late() if k is None: # only the return to the depot overran the shift k = len(r.sequence) - 1 if k == 0: k = 1 if len(r.sequence) > 1 else 0 if k == 0: continue pair = (r.sequence[k - 1], r.sequence[k]) learned.append(pair) info["new_rules"].append({"van": r.van, "pair": pair, "late_by": r.lateness[k]}) last.message = (f"{last.n_late} late deliveries remain after {max_loops} loops; " f"learned rules: {learned}") last.unserviceable = [(oid, f"late by {late:.1f} min on van {r.van}") for r in last.routes.values() for oid, late in zip(r.sequence, r.lateness) if late > 1e-9] if suggest: last.widening = suggest_widening(inst, tm, [u for u, _ in last.unserviceable]) return last # --------------------------------------------------------------------------- no-solution reports def explain_no_solution(csp: AssignmentCSP, inst, tm) -> list: out = [] for oid, reasons in csp.unary_reasons.items(): if not csp.domains0[oid]: out.append((oid, "window not reachable from the depot by any van: " + "; ".join(reasons))) if out: return out tot_d = sum(o.demand for o in inst.orders) tot_c = sum(v.capacity for v in inst.vans) if tot_d > tot_c: out.append((None, f"total demand {tot_d} exceeds total van capacity {tot_c}")) tot_s = sum(v.max_stops for v in inst.vans) if len(inst.orders) > tot_s: out.append((None, f"{len(inst.orders)} orders but only {tot_s} stops available")) if out: return out clique = csp.max_clique() if len(clique) > len(inst.vans): nb = csp.neighbors() drop = max(clique, key=lambda i: (len(nb[i]), i)) for oid in clique: out.append((oid, f"orders {clique} are pairwise incompatible (cannot meet both windows in " f"either order with predicted times) but there are only {len(inst.vans)} vans" + ("; dropping this one helps most" if oid == drop else ""))) return out for o in inst.orders: # which single removal makes it solvable? sub = _without(inst, o.id) learned = [tuple(e) for e, s in csp.edges.items() if s == "learned" and o.id not in e] if AssignmentCSP(sub, tm, learned).solve(fc=True, ac3=True, max_assignments=50_000).status == "solved": out.append((o.id, "removing this order makes the assignment feasible " "(capacity + different-van rules conflict)")) if not out: out.append((None, "no assignment satisfies capacity + different-van rules")) return out def _without(inst, oid): d = inst.to_dict() d["orders"] = [o for o in d["orders"] if o["id"] != oid] from .instances import Instance return Instance.from_dict(d) def suggest_widening(inst, tm, only_ids=None, step: float = 15.0, max_steps: int = 8) -> dict: only = set(only_ids) if only_ids else None for k in range(1, max_steps + 1): w = inst.widened(step * k, only) r = plan(w, tm, suggest=False) if r.status == "ok": return {"found": True, "minutes": step * k, "orders": sorted(only) if only else "all"} return {"found": False, "minutes": step * max_steps, "orders": sorted(only) if only else "all"} def format_no_solution(res: PlanResult) -> str: lines = ["NO SOLUTION: not every order can be served on time.", "Unserviceable orders / reasons:"] for oid, why in res.unserviceable: lines.append(f" - {'order ' + str(oid) if oid is not None else 'instance'}: {why}") w = res.widening if w: who = "all windows" if w["orders"] == "all" else f"the windows of orders {w['orders']}" if w["found"]: lines.append(f"Suggestion: widening {who} by {int(w['minutes'])} min (15-min steps) makes it feasible.") else: lines.append(f"Widening {who} by up to {int(w['minutes'])} min (15-min steps) does not help.") return "\n".join(lines) def describe_plan(inst, res: PlanResult) -> str: lines = [f"status: {res.status} -- {res.message}"] for v in inst.vans: r = res.routes.get(v.id) if r is None: continue load = sum(inst.order(i).demand for i in r.sequence) lines.append(f"van {v.id} (shift {fmt_t(v.shift_start)}-{fmt_t(v.shift_end)}, load {load}/{v.capacity}, " f"{len(r.sequence)} stops, back {fmt_t(r.finish)}):") for oid, a, b, late in zip(r.sequence, r.arrivals, r.begins, r.lateness): o = inst.order(oid) flag = f" LATE {late:.1f} min" if late > 1e-9 else "" lines.append(f" order {oid:2d} zone {o.zone:2d} window {fmt_t(o.earliest)}-{fmt_t(o.latest)} " f"arrive {fmt_t(a)} serve {fmt_t(b)}{flag}") return "\n".join(lines)