"""Single-file Gradio web app for the Field Adjusters Routing Optimizer. Run: .venv/bin/python app.py then open http://127.0.0.1:7860 Workflow: upload claims.csv / adjusters.csv (same schemas as data/, see README) or generate a synthetic instance, pick a solver time limit, and Solve. Results: fleet summary, interactive Folium map, per-adjuster schedule, dropped-claim reschedule queue, and a CSV download of the assignments. Phase 2 (LLM claim intake / schedule Q&A) plugs in behind this same solve pipeline. """ from __future__ import annotations import html import os import shutil import tempfile import csv import folium import gradio as gr import pandas as pd import assistant import config import data_gen import distance import solver from data_gen import min_to_hhmm # Latest solved instance, shared with the AI assistant tabs. resolver # re-runs the same backend + toggles as the last Solve so what-if # hypotheticals are apples-to-apples with the schedule on screen. LAST = {"claims": None, "adjusters": None, "sol": None, "mode": "global", "resolver": None, "backend_label": "ortools", "lunch_break": False, "balance": False, "time_limit": 10, "matrix_builder": None, "distance_source": "haversine"} # Folium's Marker icons accept a fixed color vocabulary; keep hex colors # (for lines/circles) and icon color names aligned per adjuster. HEX_COLORS = ["#d62728", "#1f77b4", "#2ca02c", "#9467bd", "#ff7f0e", "#8c564b", "#e377c2", "#17becf", "#bcbd22", "#7f7f7f"] ICON_COLORS = ["red", "blue", "green", "purple", "orange", "darkred", "pink", "lightblue", "beige", "gray"] PRIORITY_RADIUS = {config.PRIORITY_MUST_TODAY: 11, config.PRIORITY_HIGH: 8, config.PRIORITY_NORMAL: 6} # --------------------------------------------------------------------------- # Data loading # --------------------------------------------------------------------------- DEMO_DIR = os.path.join(config.DATA_DIR, "demo") def _win_str(c) -> str: """All availability windows of a claim as display text.""" return data_gen.win_str(c) def demo_scenarios() -> list[str]: if not os.path.isdir(DEMO_DIR): return [] return sorted(d for d in os.listdir(DEMO_DIR) if os.path.isdir(os.path.join(DEMO_DIR, d))) def _load_instance(claims_file, adjusters_file, n_claims, n_adjusters, seed, demo_choice=None): """Return (claims, adjusters, source_description). Priority: uploaded CSVs win when both are provided; otherwise a selected demo scenario is loaded from data/demo/; otherwise a synthetic instance is generated. Files are staged into a temp dir so the existing data_gen loaders can be reused unchanged. """ use_demo = demo_choice and demo_choice != "(none)" with tempfile.TemporaryDirectory() as tmp: if claims_file and adjusters_file: shutil.copy(claims_file, os.path.join(tmp, "claims.csv")) shutil.copy(adjusters_file, os.path.join(tmp, "adjusters.csv")) src = "uploaded CSVs" elif claims_file or adjusters_file: raise gr.Error("Please upload BOTH claims.csv and adjusters.csv " "(or neither, to use a demo scenario or a " "synthetic instance).") elif use_demo: d = os.path.join(DEMO_DIR, demo_choice) if not os.path.isdir(d): raise gr.Error(f"Demo scenario {demo_choice!r} not found - " "run make_demo_data.py to rebuild data/demo.") for fname in ("claims.csv", "adjusters.csv"): shutil.copy(os.path.join(d, fname), os.path.join(tmp, fname)) src = f"demo: {demo_choice}" else: data_gen.generate(n_claims=int(n_claims), n_adjusters=int(n_adjusters), seed=int(seed), data_dir=tmp) src = (f"synthetic instance (seed {int(seed)})") try: claims = data_gen.load_claims(tmp) adjusters = data_gen.load_adjusters(tmp) except (KeyError, ValueError) as e: raise gr.Error(f"Could not parse CSVs: {e}. Expected the " "schemas described in the README.") if not claims or not adjusters: raise gr.Error("Empty claims or adjusters file.") return claims, adjusters, src # --------------------------------------------------------------------------- # Folium map # --------------------------------------------------------------------------- def build_map(sol, claims) -> str: lats = [c.lat for c in claims] + [r.adjuster.home_lat for r in sol.routes] lons = [c.lon for c in claims] + [r.adjuster.home_lon for r in sol.routes] m = folium.Map(location=[sum(lats) / len(lats), sum(lons) / len(lons)], tiles="OpenStreetMap") for v, route in enumerate(sol.routes): hexc = HEX_COLORS[v % len(HEX_COLORS)] iconc = ICON_COLORS[v % len(ICON_COLORS)] adj = route.adjuster label = f"{adj.adjuster_id} {adj.name}" folium.Marker( [adj.home_lat, adj.home_lon], icon=folium.Icon(color=iconc, icon="home", prefix="fa"), tooltip=f"{label} - home", popup=folium.Popup( f"{label}
skills: {', '.join(adj.skills)}
" f"shift {min_to_hhmm(adj.shift_start)}-" f"{min_to_hhmm(adj.shift_end)}
" f"{len(route.stops)} claims, {route.total_miles:.0f} mi", max_width=260), ).add_to(m) if route.stops: points = ([[adj.home_lat, adj.home_lon]] + [[s.claim.lat, s.claim.lon] for s in route.stops] + [[adj.home_lat, adj.home_lon]]) folium.PolyLine(points, color=hexc, weight=3, opacity=0.85, tooltip=f"{label} route " f"({route.total_miles:.0f} mi)").add_to(m) for seq, s in enumerate(route.stops, start=1): c = s.claim folium.CircleMarker( [c.lat, c.lon], radius=PRIORITY_RADIUS[c.priority], color=hexc, fill=True, fill_color=hexc, fill_opacity=0.9, tooltip=f"#{seq} {c.claim_id} ({c.peril}) - {label}", popup=folium.Popup( f"{c.claim_id} [{config.PRIORITY_LABEL[c.priority]}]" f"
peril: {c.peril}
" f"window(s) {_win_str(c)}
" f"on-site {min_to_hhmm(s.arrival_min)}-" f"{min_to_hhmm(s.departure_min)}
" f"adjuster: {label} (stop #{seq})", max_width=260), ).add_to(m) for c in sol.dropped: folium.Marker( [c.lat, c.lon], icon=folium.Icon(color="lightgray", icon="remove"), tooltip=f"DROPPED {c.claim_id} ({c.peril})", popup=folium.Popup( f"{c.claim_id} - dropped, reschedule
" f"[{config.PRIORITY_LABEL[c.priority]}] peril: {c.peril}
" f"window(s) {_win_str(c)}, " f"{c.service_minutes} min on-site", max_width=260), ).add_to(m) m.fit_bounds([[min(lats), min(lons)], [max(lats), max(lons)]], padding=(20, 20)) # Embed via srcdoc iframe: reliable sizing inside Gradio. raw = m.get_root().render() return (f'') # --------------------------------------------------------------------------- # Result tables # --------------------------------------------------------------------------- def schedule_table(sol) -> pd.DataFrame: rows = [] for route in sol.routes: adj = route.adjuster for seq, s in enumerate(route.stops, start=1): c = s.claim rows.append({ "Adjuster": f"{adj.adjuster_id} {adj.name}", "Stop": seq, "Claim": c.claim_id, "Peril": c.peril, "Priority": config.PRIORITY_LABEL[c.priority], "Window": f"{min_to_hhmm(c.window_start)}-" f"{min_to_hhmm(c.window_end)}", "Drive (mi)": round(s.travel_miles_from_prev, 1), "Arrive": min_to_hhmm(s.arrival_min), "Depart": min_to_hhmm(s.departure_min), }) return pd.DataFrame(rows) def dropped_table(sol) -> pd.DataFrame: rows = [] for c in sorted(sol.dropped, key=lambda c: c.priority): rows.append({ "Claim": c.claim_id, "Peril": c.peril, "Priority": config.PRIORITY_LABEL[c.priority], "Window": _win_str(c), "On-site (min)": c.service_minutes, "Drop penalty": config.DROP_PENALTY[c.priority], "Note": ("NO ELIGIBLE ADJUSTER (skill/territory)" if c in sol.unservable else "reschedule"), }) return pd.DataFrame(rows) def assignments_csv(sol) -> str: df = schedule_table(sol) path = os.path.join(tempfile.mkdtemp(prefix="routing_"), "assignments.csv") df.to_csv(path, index=False) return path # --------------------------------------------------------------------------- # Solve callback # --------------------------------------------------------------------------- def working_dataset_files(claims, adjusters): """Export the CURRENT working dataset as re-uploadable CSVs: every applied scenario (hires, hours, priorities), intake addition, and age column is reflected. This is how changes get back to the user - a web app can never modify the files on their machine.""" pending = [c for c in ADDED_SINCE_SOLVE if c.claim_id not in {x.claim_id for x in claims}] d = tempfile.mkdtemp(prefix="working_") cpath = data_gen.write_claims_csv(list(claims) + pending, os.path.join(d, "claims_updated.csv")) apath = data_gen.write_adjusters_csv( adjusters, os.path.join(d, "adjusters_updated.csv")) return cpath, apath def render_solution(sol, claims, src): """Build the result-panel outputs from a solved instance.""" served = sum(len(r.stops) for r in sol.routes) # Multi-part labels (scenario | solver | proof | notes) read badly # crammed into one line - render each part as its own bullet. if " | " in src: head = "### Results\n" + "\n".join( f"- {part}" for part in src.split(" | ")) + "\n\n" else: head = f"### Results - {src}\n" summary = head + ( f"| Served | Fleet miles | Driving time | On-site time | Objective |\n" f"|---|---|---|---|---|\n" f"| **{served} / {len(claims)}** claims " f"| {sol.total_miles:.1f} mi " f"| {sol.total_travel_min // 60}h {sol.total_travel_min % 60}m " f"| {sum(r.total_service_min for r in sol.routes) // 60}h " f"{sum(r.total_service_min for r in sol.routes) % 60}m " f"| {sol.objective} |" ) if sol.dropped_must_today: ids = ", ".join(c.claim_id for c in sol.dropped_must_today) banner = (f'
VIOLATION: ' f'{len(sol.dropped_must_today)} MUST-TODAY claim(s) could ' f'not be scheduled: {ids}. Add adjusters, extend shifts, ' f'or relax lower-priority work.
') else: banner = ('
All MUST-TODAY claims are ' 'scheduled.
') cpath, apath = working_dataset_files(claims, LAST["adjusters"]) return (summary, banner, build_map(sol, claims), schedule_table(sol), dropped_table(sol), assignments_csv(sol), cpath, apath) def dispatch_backend(solver_choice, adjusters, claims, miles, travel_min, time_limit_s, lunch_break=False, balance=False): """The Solve button's backend dispatch, shared with the AI assistant's what-if re-solves so hypotheticals run on the same engine and toggles as the schedule on screen. Returns (sol, extras) where extras carries backend-specific proof info for the banner.""" if solver_choice == "pyvrp": import pyvrp_solver return pyvrp_solver.solve(adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s)), {} if solver_choice == "hybrid": import pyvrp_solver return pyvrp_solver.solve_hybrid( adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), lunch_break=bool(lunch_break), balance=bool(balance)), {} if solver_choice == "milp (exact)": import milp_solver sol, exact = milp_solver.solve_as_backend( adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), engine="auto", lunch_break=bool(lunch_break), balance=bool(balance)) return sol, {"exact": exact} if solver_choice == "cpsat (exact, license-free)": import cpsat_solver sol, cpsat_info = cpsat_solver.solve( adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), lunch_break=bool(lunch_break), balance=bool(balance)) return sol, {"cpsat": cpsat_info} if solver_choice in ("sequence (pre-assigned, exact)", "sequence-milp (pre-assigned, MILP proof)"): import sequencer seq_method = ("milp" if solver_choice.startswith("sequence-milp") else "enumeration") sol, seq_info = sequencer.solve_sequenced( adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), method=seq_method, lunch_break=bool(lunch_break), balance=bool(balance)) return sol, {"seq_info": seq_info, "seq_method": seq_method} if solver_choice == "setpart (route pool + MILP)": import setpartition sol, sp_info = setpartition.solve_setpartition( adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), engine="auto", lunch_break=bool(lunch_break), balance=bool(balance)) return sol, {"sp_info": sp_info} if solver_choice == "ortools": return solver.solve(adjusters, claims, miles, travel_min, time_limit_s=int(time_limit_s), lunch_break=bool(lunch_break), balance=bool(balance)), {} # Strict on purpose: a silent ortools fallback would mask a renamed # dropdown label and quietly run every what-if on the wrong engine. raise ValueError(f"unknown solver backend: {solver_choice!r}") def make_matrix_builder(distance_source): """Rebuild travel matrices the same way the last Solve did, so what-if re-solves price hypothetical rosters on the same distance model (google matrices are disk-cached, so unchanged location pairs do not re-bill).""" def _build(adjusters, claims): if distance_source == "google": import google_distance try: return google_distance.build_matrices(adjusters, claims) except google_distance.GoogleMatrixError as e: raise RuntimeError(f"Google Routes API: {e}") return distance.build_matrices(adjusters, claims) return _build def make_resolver(solver_choice, lunch_break, balance): """Bind the Solve button's backend choice and toggles into a callable the assistant re-runs for what-if scenarios.""" def _resolve(adjusters, claims, miles, travel_min, time_limit_s): sol, _ = dispatch_backend(solver_choice, adjusters, claims, miles, travel_min, time_limit_s, lunch_break, balance) return sol return _resolve def run_solve(claims_file, adjusters_file, n_claims, n_adjusters, seed, time_limit, distance_source, solver_choice="ortools", lunch_break=False, balance=False, demo_choice="(none)", google_key=""): claims, adjusters, src = _load_instance( claims_file, adjusters_file, n_claims, n_adjusters, seed, demo_choice=demo_choice) # A key pasted in the UI takes effect for this process only (also # enables address geocoding on the AI intake tab). if google_key and google_key.strip(): os.environ["GOOGLE_MAPS_API_KEY"] = google_key.strip() if distance_source == "google": import google_distance try: miles, travel_min = google_distance.build_matrices( adjusters, claims) except google_distance.GoogleMatrixError as e: raise gr.Error(f"Google Routes API: {e}") else: miles, travel_min = distance.build_matrices(adjusters, claims) if solver_choice in ("sequence (pre-assigned, exact)", "sequence-milp (pre-assigned, MILP proof)") \ and not any(c.assigned_to for c in claims): raise gr.Error( "The sequence backends need pre-assigned claims: add an " "assigned_to column to claims.csv (adjuster ids), or " "pick the 'pre-assigned' demo scenario.") _mw_ok = ("ortools", "cpsat (exact, license-free)", "milp (exact)") if (any(c.extra_windows for c in claims) and solver_choice not in _mw_ok): raise gr.Error( "This day contains split-availability claims (multiple " "time windows). Backends supporting them so far: ortools, " "cpsat, and milp (exact) - pick one of those.") sol, extras = dispatch_backend(solver_choice, adjusters, claims, miles, travel_min, time_limit, lunch_break, balance) if "exact" in extras: exact = extras["exact"] proof_tag = (f"PROVEN OPTIMAL [{exact.engine}]" if exact.status == "Optimal" else f"exact solve not proven in time [{exact.engine}]") elif "cpsat" in extras: ci = extras["cpsat"] proof_tag = (f"PROVEN OPTIMAL [cp-sat, {ci['wall_s']}s]" if ci["proven"] else f"best found in time, bound {ci.get('bound', '?')} " f"[cp-sat]") elif "seq_info" in extras: seq_info = extras["seq_info"] proof_tag = ("every adjuster's route PROVEN OPTIMAL" if seq_info["proven_optimal"] else "sequencing search hit its time guard") if extras["seq_method"] == "milp": proof_tag += ( f" + MILP certificate [{seq_info['engine']}]" if seq_info["milp_all_certified"] else " (MILP certification incomplete)") if seq_info["drop_reasons"]: why = "; ".join(f"{cid}: {r}" for cid, r in sorted(seq_info["drop_reasons"].items())) proof_tag += f" | for rescheduling - {why}" elif "sp_info" in extras and sol is not None: proof_tag = (f"MILP selected the best of " f"{extras['sp_info']['pool_columns']} candidate " f"routes [{extras['sp_info']['engine']}]") if sol is None: if "cpsat" in extras: raise gr.Error(f"CP-SAT ended with status " f"{extras['cpsat']['status']} - allow more " f"time, or check that claim windows overlap " f"adjuster shifts.") raise gr.Error("No solution found - check that claim windows " "overlap adjuster shifts.") LAST.update(claims=claims, adjusters=adjusters, sol=sol, mode=("sequence" if solver_choice.startswith( "sequence") else "global"), resolver=make_resolver(solver_choice, bool(lunch_break), bool(balance)), backend_label=solver_choice, lunch_break=bool(lunch_break), balance=bool(balance), time_limit=int(time_limit), matrix_builder=make_matrix_builder(distance_source), distance_source=distance_source) ADDED_SINCE_SOLVE.clear() label = src + f" | solver: {solver_choice}" if solver_choice in ("milp (exact)", "cpsat (exact, license-free)", "setpart (route pool + MILP)", "sequence (pre-assigned, exact)", "sequence-milp (pre-assigned, MILP proof)"): label += f" | {proof_tag}" _toggle_backends = ("ortools", "milp (exact)", "cpsat (exact, license-free)", "hybrid", "setpart (route pool + MILP)", "sequence (pre-assigned, exact)", "sequence-milp (pre-assigned, MILP proof)") if lunch_break and solver_choice in _toggle_backends: label += " | lunch break on" if balance and solver_choice in _toggle_backends: label += " | workload balance on" return render_solution(sol, claims, label) # --------------------------------------------------------------------------- # Phase 2: AI claim intake + schedule assistant # --------------------------------------------------------------------------- def run_extract(fnol_text): if not fnol_text or not fnol_text.strip(): raise gr.Error("Paste FNOL text (an email or call notes) first.") try: result = assistant.extract_claims(fnol_text) except assistant.AssistantError as e: raise gr.Error(str(e)) if not result.claims: return (pd.DataFrame(), [], "No claims found in that text.") rows = [{ "Policyholder": c.policyholder_name or "-", "Address": c.address or "-", "Peril": c.peril, "Priority": config.PRIORITY_LABEL[c.priority], "Window": f"{c.window_start}-{c.window_end}", "On-site (min)": c.service_minutes, "Coordinates": (f"{c.lat}, {c.lon}" if c.lat is not None else "needs geocoding"), "Notes": c.notes, } for c in result.claims] note = (f"Extracted {len(result.claims)} claim(s). Review, then add " "them to data/claims.csv and re-solve. Claims without " "coordinates are placed at the region center until geocoded " "(production: wire in a geocoding API).") return pd.DataFrame(rows), [c.model_dump() for c in result.claims], note ADDED_SINCE_SOLVE: list = [] # intake claims awaiting a solve/repair def add_extracted_claims(pending): if not pending: raise gr.Error("Nothing to add - extract a claim first.") claims_path = os.path.join(config.DATA_DIR, "claims.csv") # The working dataset (and its Download-tab export) is the canonical # record; the server-side data/claims.csv append is best-effort for # local directory-based workflows. known_ids = [c.claim_id for c in (LAST["claims"] or [])] + \ [c.claim_id for c in ADDED_SINCE_SOLVE] if os.path.exists(claims_path): known_ids += [c.claim_id for c in data_gen.load_claims()] nums = [int(i.split("-")[1]) for i in known_ids if "-" in i and i.split("-")[1].isdigit()] next_num = 1 + max(nums, default=0) added, geocoded, new_rows = [], [], [] for i, c in enumerate(pending): lat, lon = c["lat"], c["lon"] if lat is None and c.get("address"): try: import geocode lat, lon = geocode.geocode(c["address"]) geocoded.append(c["address"]) except Exception: pass # no key / lookup failed: placeholder below if lat is None: lat, lon = config.REGION_CENTER claim_id = f"CLM-{next_num + i:03d}" added.append(claim_id) new_rows.append([claim_id, lat, lon, c["peril"], c["priority"], c["window_start"], c["window_end"], c["service_minutes"]]) ADDED_SINCE_SOLVE.append(data_gen.Claim( claim_id=claim_id, lat=float(lat), lon=float(lon), peril=c["peril"], priority=int(c["priority"]), window_start=data_gen.hhmm_to_min(c["window_start"]), window_end=data_gen.hhmm_to_min(c["window_end"]), service_minutes=int(c["service_minutes"]))) if os.path.exists(claims_path): # best-effort local-dir append with open(claims_path, "a", newline="") as f: csv.writer(f).writerows(new_rows) note = f"Added {', '.join(added)} to the working dataset." if geocoded: note += f" Geocoded {len(geocoded)} address(es)." return (note + " Solve to schedule them for a fresh day, or use " "In-day repair mid-day. Download-tab exports include them.", []) def chat_turn(user_msg, history, assist_state): """Streaming chat: yields the reply as Claude generates it.""" if not user_msg or not user_msg.strip(): yield "", history, assist_state return if LAST["sol"] is None: history = history + [ {"role": "user", "content": user_msg}, {"role": "assistant", "content": "Solve a schedule first (Solve button), then ask " "me about it."}] yield "", history, assist_state return if assist_state is None: assist_state = assistant.ScheduleAssistant() assist_state.set_context( LAST["claims"], LAST["adjusters"], LAST["sol"], resolver=LAST["resolver"], backend_label=LAST["backend_label"], toggles={"lunch_break": LAST["lunch_break"], "balance": LAST["balance"]}, default_time_limit=LAST["time_limit"], mode=LAST["mode"], matrix_builder=LAST["matrix_builder"], distance_label=LAST["distance_source"]) base = history + [{"role": "user", "content": user_msg}] yield "", base + [{"role": "assistant", "content": "..."}], assist_state reply = "" for partial in assist_state.ask_stream(user_msg): reply = partial yield ("", base + [{"role": "assistant", "content": reply}], assist_state) def apply_scenario(assist_state, confirm): """Promote the assistant's last what-if to the working schedule - an explicit, human-confirmed action.""" if not confirm: raise gr.Error("Tick the confirmation box first - this replaces " "the working schedule.") if assist_state is None or assist_state.last_what_if is None: raise gr.Error("No what-if scenario yet - ask the assistant a " "'what if...' question first.") if LAST["sol"] is None: raise gr.Error("Solve a schedule first.") scen = assist_state.last_what_if import copy from data_gen import hhmm_to_min adjusters = copy.deepcopy( [a for a in LAST["adjusters"] if a.adjuster_id not in set(scen["exclude_adjuster_ids"])]) for ch in scen.get("shift_changes", []): for a in adjusters: if a.adjuster_id == ch["adjuster_id"]: if ch.get("new_shift_start"): a.shift_start = hhmm_to_min(ch["new_shift_start"]) if ch.get("new_shift_end"): a.shift_end = hhmm_to_min(ch["new_shift_end"]) for spec in scen.get("add_adjusters", []): adjusters.append(data_gen.Adjuster(**spec)) claims = copy.deepcopy(LAST["claims"]) for c in claims: if c.claim_id in set(scen["must_today_claim_ids"]): c.priority = config.PRIORITY_MUST_TODAY for ch in scen.get("priority_changes", []): for c in claims: if c.claim_id == ch["claim_id"]: c.priority = int(ch["new_priority"]) build = LAST["matrix_builder"] or distance.build_matrices try: miles, travel_min = build(adjusters, claims) except RuntimeError as e: raise gr.Error(str(e)) resolver = LAST["resolver"] or make_resolver("ortools", False, False) sol = resolver(adjusters, claims, miles, travel_min, LAST["time_limit"] or 10) if sol is None: raise gr.Error("Scenario has no feasible solution.") LAST.update(claims=claims, adjusters=adjusters, sol=sol) shifts = ", ".join( f"{ch['adjuster_id']} -> " f"{ch.get('new_shift_start') or ''}" f"{'-' if ch.get('new_shift_start') and ch.get('new_shift_end') else ''}" f"{ch.get('new_shift_end') or ''}" for ch in scen.get("shift_changes", [])) or "none" hires = ", ".join(f"{s['adjuster_id']} ({'/'.join(s['skills'])})" for s in scen.get("add_adjusters", [])) or "none" prios = ", ".join( f"{ch['claim_id']} -> {config.PRIORITY_LABEL[int(ch['new_priority'])]}" for ch in scen.get("priority_changes", [])) or "none" label = (f"APPLIED SCENARIO - excluded: " f"{scen['exclude_adjuster_ids'] or 'none'}; escalated: " f"{scen['must_today_claim_ids'] or 'none'}; shifts: {shifts}; " f"added: {hires}; priorities: {prios}") return render_solution(sol, claims, label) def run_repair(now_hhmm, repair_limit): """Re-plan the rest of the day from the current time, keeping every completed and in-progress stop exactly as it happened.""" import repair as repair_mod if LAST["sol"] is None: raise gr.Error("Solve a schedule first - repair needs a baseline " "day to work from.") try: now_min = data_gen.hhmm_to_min(str(now_hhmm).strip()) assert 0 <= now_min < 24 * 60 except Exception: raise gr.Error("Enter the current time as HH:MM, e.g. 13:00.") known = {c.claim_id for c in LAST["claims"]} new_claims = [c for c in ADDED_SINCE_SOLVE if c.claim_id not in known] all_claims = LAST["claims"] + new_claims sequenced = LAST.get("mode") == "sequence" combined, state = repair_mod.solve_repair( LAST["adjusters"], all_claims, LAST["sol"], now_min, time_limit_s=int(repair_limit), respect_assignments=sequenced) if combined is None: raise gr.Error("Repair found no feasible plan - check the time.") if sequenced: # Repair may place brand-new intake claims (no assigned_to) on # any qualified adjuster; stamp that placement as the claim's # assignment so later what-ifs and Apply keep it binding instead # of dropping the claim as unassigned. by_id = {c.claim_id: c for c in all_claims} for r in combined.routes: for s in r.stops: if not s.claim.assigned_to: s.claim.assigned_to = r.adjuster.adjuster_id c = by_id.get(s.claim.claim_id) if c is not None and not c.assigned_to: c.assigned_to = r.adjuster.adjuster_id LAST.update(claims=all_claims, sol=combined) ADDED_SINCE_SOLVE.clear() frozen = sum(len(v) for v in state.frozen_stops.values()) busy = ", ".join(f"{k} (finishing {v.claim.claim_id})" for k, v in state.in_progress.items()) or "none" status = (f"**Repaired from {now_hhmm}.** Kept {frozen} completed/" f"in-progress stops exactly as they happened. Mid-" f"inspection: {busy}. New claims included: " f"{', '.join(c.claim_id for c in new_claims) or 'none'}. " + ("Upstream assignments stay binding (pre-assigned " "mode); new claims may go to any qualified " "adjuster. " if sequenced else "") + f"The tabs below now show the full day: frozen morning " f"+ re-optimized afternoon.") outs = run_repair_render(combined, all_claims, now_hhmm) return (status, *outs) def run_repair_render(sol, claims, now_hhmm): return render_solution(sol, claims, f"IN-DAY REPAIR from {now_hhmm}") def make_tomorrow_files(): """Multi-day rollover: served claims out, dropped claims carried forward one day older (so SLA escalation lifts them tomorrow), and the current roster as-is for the user to curate.""" if LAST["sol"] is None: raise gr.Error("Solve a schedule first.") import copy carried = copy.deepcopy(LAST["sol"].dropped) for c in carried: c.age_days += 1 pending = [c for c in ADDED_SINCE_SOLVE if c.claim_id not in {x.claim_id for x in LAST["claims"]}] d = tempfile.mkdtemp(prefix="tomorrow_") cpath = data_gen.write_claims_csv( carried + pending, os.path.join(d, "claims_tomorrow.csv")) apath = data_gen.write_adjusters_csv( LAST["adjusters"], os.path.join(d, "adjusters_tomorrow.csv")) temp_ids = [a.adjuster_id for a in LAST["adjusters"] if a.adjuster_id.startswith("TEMP-")] note = (f"**Tomorrow's starting files.** {len(carried)} unserved " f"claim(s) carried forward with age_days + 1" + (f", plus {len(pending)} not-yet-scheduled intake claim(s)" if pending else "") + ". The roster is today's working roster" + (f" - review temporary hire(s) {', '.join(temp_ids)} and " f"delete their row(s) if they were a one-day arrangement" if temp_ids else "") + ". Edit in Excel as needed, then upload both files " "tomorrow morning and Solve.") return note, cpath, apath def make_briefings(assist_state): if LAST["sol"] is None: raise gr.Error("Solve a schedule first.") state = assist_state or assistant.ScheduleAssistant() state.set_context(LAST["claims"], LAST["adjusters"], LAST["sol"]) try: md = assistant.generate_briefings(state) except assistant.AssistantError as e: raise gr.Error(str(e)) path = os.path.join(tempfile.mkdtemp(prefix="briefing_"), "morning_briefings.md") with open(path, "w") as f: f.write(md) return md, path # --------------------------------------------------------------------------- # Call Planner (prototype): guided appointment calling for pre-assigned # days. The adjuster keeps making the calls; after every logged call the # sequence solver replans the rest of the day around what was agreed. # --------------------------------------------------------------------------- def _cp_solve(adjuster_id, booked, deferred, offered=None): """Replan one adjuster's day. booked: {claim_id: minute} appointments (window pinned); deferred: claim ids moved to tomorrow; offered: optional {claim_id: (lo, hi)} availability from the current call.""" import copy adj = next(a for a in LAST["adjusters"] if a.adjuster_id == adjuster_id) claims = [copy.deepcopy(c) for c in LAST["claims"] if c.assigned_to == adjuster_id and c.claim_id not in deferred] for c in claims: t = booked.get(c.claim_id) if t is not None: c.window_start, c.window_end = t, t + c.service_minutes elif offered and c.claim_id in offered: lo, hi = offered[c.claim_id] c.window_start = max(c.window_start, lo) c.window_end = min(c.window_end, hi) if c.window_start > c.window_end: c.window_start, c.window_end = lo, hi build = LAST["matrix_builder"] or distance.build_matrices miles, travel_min = build([adj], claims) import sequencer as _seq sol, info = _seq.solve_sequenced([adj], claims, miles, travel_min, time_limit_s=5) return adj, sol def _cp_view(booked, deferred, sol): rows, to_call = [], [] order = 1 for r in sol.routes: for s in r.stops: cid = s.claim.claim_id if cid in booked: rows.append([order, cid, min_to_hhmm(s.arrival_min), "BOOKED"]) else: rows.append([order, cid, f"propose {min_to_hhmm(s.arrival_min)} " f"(they're free " f"{min_to_hhmm(s.claim.window_start)}-" f"{min_to_hhmm(s.claim.window_end)})", "to call"]) to_call.append(cid) order += 1 for c in sol.dropped: rows.append(["-", c.claim_id, "-", "won't fit today - reschedule"]) if c.claim_id not in booked: to_call.append(c.claim_id) for cid in sorted(deferred): rows.append(["-", cid, "-", "moved to tomorrow (agreed)"]) return rows, to_call def mc_load_appointments(): """Today's planned visits, as three pickers of claim ids.""" if LAST["sol"] is None: raise gr.Error("Solve a day first - the sweep applies this " "morning's answers to a real plan.") ids = sorted(s.claim.claim_id for r in LAST["sol"].routes for s in r.stops) if not ids: raise gr.Error("The last solve served no claims.") up = gr.update(choices=ids, value=[]) return up, up, up def mc_apply(noanswer, cancel, resched, days): """Apply the morning's call outcomes: withhold what nobody confirmed, re-solve the confirmed day, rebook what was withheld.""" import confirmations as cf if LAST["sol"] is None: raise gr.Error("Solve a day first.") claims, adjusters = LAST["claims"], LAST["adjusters"] build = LAST["matrix_builder"] or distance.build_matrices miles, travel_min = build(adjusters, claims) planned = {s.claim.claim_id for r in LAST["sol"].routes for s in r.stops} outcomes = {c.claim_id: cf.CONFIRMED for c in claims} for cid in (noanswer or []): outcomes[cid] = cf.NO_ANSWER for cid in (cancel or []): outcomes[cid] = cf.CANCEL for cid in (resched or []): outcomes[cid] = cf.RESCHEDULE if not (noanswer or cancel or resched): raise gr.Error("Log at least one non-confirmation - otherwise " "the morning changed nothing.") d = cf.morning_sweep(adjusters, claims, miles, travel_min, outcomes, time_limit_s=int(LAST["time_limit"] or 10), lunch_break=LAST["lunch_break"], balance=LAST["balance"]) rebooked = {} if d.held: pending = [c for c in claims if c.claim_id not in planned or c.claim_id in set(d.held_ids)] try: rebooked, _over = cf.rebook(adjusters, pending, miles, claims, days=int(days), time_limit_s=20) except Exception: rebooked = {} n_now = sum(len(r.stops) for r in d.dispatched.routes) n_blind = sum(len(r.stops) for r in d.baseline.routes) lines = [ "### This morning's dispatch", f"- **{n_now} confirmed visits** dispatched, " f"{d.dispatched_travel_min} drive min " f"(the blind plan: {n_blind} visits, " f"{d.baseline_travel_min} drive min)", ] if d.held: lines.append(f"- **{len(d.held)} visits withheld** - " f"**{d.exposure_min} drive min** were riding on " f"doors nobody confirmed") if d.rebooked_today: lines.append(f"- {len(d.rebooked_today)} policyholders moved " f"to a new time **today**: " f"{', '.join(sorted(d.rebooked_today))}") if d.backfilled: lines.append(f"- freed capacity absorbed " f"**{', '.join(c.claim_id for c in d.backfilled)}" f"** - would otherwise have rolled to another day") if d.dispatched.dropped_must_today: lines.append("- :warning: a MUST-TODAY claim still cannot be " "routed - human decision needed") lines.append("\n*Exposure is minutes at risk, not guaranteed " "savings: an unanswered policyholder might still have " "been home. What the number states precisely is how " "much driving was unconfirmed.*") rows = [] for c in d.held: day = rebooked.get(c.claim_id) when = ("beyond the horizon - escalate" if day is None else "tomorrow" if day == 1 else f"in {day} days") rows.append([c.claim_id, d.hold_reasons.get(c.claim_id, "unconfirmed"), when]) return "\n".join(lines), rows or [["-", "nothing withheld", "-"]] def cp_load_adjusters(): ids = sorted({c.assigned_to for c in (LAST["claims"] or []) if c.assigned_to}) if not ids: raise gr.Error( "The Call Planner needs a pre-assigned day: pick the " "'pre-assigned' demo scenario (or upload claims.csv with " "an assigned_to column), click Solve, then come back.") return gr.update(choices=ids, value=ids[0]) def cp_start(adjuster_id): if not adjuster_id: raise gr.Error("Click 'Load adjusters' and pick one first.") state = {"adjuster": adjuster_id, "booked": {}, "deferred": [], "msgs": [], "call": None} _, sol = _cp_solve(adjuster_id, {}, set()) rows, to_call = _cp_view({}, set(), sol) _sms(state, "sys", f"{_now_hm()} - confirmation texts sent to " f"all policyholders") for r in sol.routes: for s in r.stops: cid = s.claim.claim_id _sms(state, "out", f"Good morning! John Smith from your insurance " f"company about claim {cid}. Confirming my " f"inspection visit today between {_slot_of(cid)}. " f"Reply YES to confirm, R to reschedule.", who=f"John to {cid}") status = (f"**Calling session for {adjuster_id}.** Call in the " f"order below and propose the suggested time. After " f"each call, log what happened - the plan replans " f"itself around every answer.") return (status, rows, gr.update(choices=to_call, value=to_call[0] if to_call else None), state, _msgs_html(state), _call_html(state)) def _cp_slot_label(slot): return f"{min_to_hhmm(slot[0])}-{min_to_hhmm(slot[1])}" def cp_slot_menu(state, claim_id): """The feasibility-filtered slot menu: for each standard company slot, trial-solve the day with this claim pinned into that slot and report whether it can be offered - so the adjuster reads a menu to the policyholder instead of negotiating freeform.""" if not state or not state.get("adjuster"): raise gr.Error("Start a calling session first.") if not claim_id: raise gr.Error("Pick which claim you're discussing.") adjuster_id = state["adjuster"] booked = {c: t for c, t in state["booked"].items() if c != claim_id} deferred = set(state["deferred"]) lines = [f"**Slots you can offer {claim_id}:**"] any_fit = False for slot in config.CALL_SLOTS: _, trial = _cp_solve(adjuster_id, booked, deferred, offered={claim_id: slot}) stop = next((s for r in trial.routes for s in r.stops if s.claim.claim_id == claim_id), None) if stop is not None: any_fit = True lines.append(f"- {_cp_slot_label(slot)}: **yes** - would " f"arrive about {min_to_hhmm(stop.arrival_min)}") else: lines.append(f"- {_cp_slot_label(slot)}: not possible " f"today") if not any_fit: lines.append("**No slot fits today** - offer tomorrow " "(log 'Defer to tomorrow').") else: lines.append("When they pick, log the call with 'Booked into " "a standard slot'.") spoken = [ln.replace("**", "").replace("- ", "") for ln in lines[1:-1]] state["call"] = {"cid": claim_id, "lines": [ ("Policyholder", "I'm sorry - I can't make my window " "anymore."), ("John", "No problem at all. Give me one second to check " "the rest of my day..."), ("John", "Here is what I can honestly offer you today: " + "; ".join(spoken) + "."), ]} return " \n".join(lines), state, _call_html(state) def _now_hm(): import datetime return datetime.datetime.now().strftime("%H:%M") _PHONE_CSS = """ """ def _phone_html(title, body_html, badge="simulated demo thread - not " "a live SMS service"): return (_PHONE_CSS + f'
' f'
{title}' f'
{body_html}
' f'
{badge}
' f'
') def _msgs_html(state): if not state or not state.get("msgs"): return _phone_html("Messages", '
Start a calling ' 'session to see the thread.
') parts = [] for m in state["msgs"]: if m["kind"] == "sys": parts.append(f'
{m["text"]}
') else: parts.append(f'
{m["who"]} - ' f'{m["time"]}
' f'
' f'{m["text"]}
') return _phone_html("Messages", "".join(parts)) def _call_html(state): call = (state or {}).get("call") if not call: return _phone_html("Call", '
The transcript of the ' 'latest logged call appears here.
') import urllib.parse lines = [] speech = [] for who, text in call["lines"]: kind = "out" if who == "John" else "in" lines.append(f'
{who}
' f'
{text}
') speech.append({"who": who, "text": text}) payload = urllib.parse.quote(str(speech).replace("'", '"')) play = ('
') return _phone_html(f'Call - {call["cid"]}', "".join(lines) + play, badge="simulated call captions; voice uses " "your browser's built-in speech") def _slot_of(cid): c = next((c for c in (LAST["claims"] or []) if c.claim_id == cid), None) if c is None: return "your window" return f"{min_to_hhmm(c.window_start)}-{min_to_hhmm(c.window_end)}" def _sms(state, kind, text, who=None): state.setdefault("msgs", []).append( {"kind": kind, "text": text, "time": _now_hm(), "who": who or ("John" if kind == "out" else "Policyholder")}) def cp_log(state, claim_id, outcome, time_text, slot_label=None): if not state or not state.get("adjuster"): raise gr.Error("Start a calling session first.") if not claim_id: raise gr.Error("Pick which claim you just called.") adjuster_id = state["adjuster"] booked = dict(state["booked"]) deferred = set(state["deferred"]) _, before = _cp_solve(adjuster_id, booked, deferred) drops_before = {c.claim_id for c in before.dropped} note = "" if outcome == "Booked at the proposed time": stop = next((s for r in before.routes for s in r.stops if s.claim.claim_id == claim_id), None) if stop is None: raise gr.Error(f"{claim_id} has no proposed time right " f"now - it doesn't fit today. Defer it, or " f"book a specific time.") booked[claim_id] = stop.arrival_min elif outcome == "Booked at this time": try: booked[claim_id] = data_gen.hhmm_to_min(time_text.strip()) except Exception: raise gr.Error("Enter the agreed time as HH:MM, e.g. 14:30") elif outcome == "Booked into a standard slot": if not slot_label: raise gr.Error("Pick which slot they chose.") slot = next((s for s in config.CALL_SLOTS if _cp_slot_label(s) == slot_label), None) if slot is None: raise gr.Error(f"Unknown slot {slot_label!r}.") booked.pop(claim_id, None) _, trial = _cp_solve(adjuster_id, booked, deferred, offered={claim_id: slot}) stop = next((s for r in trial.routes for s in r.stops if s.claim.claim_id == claim_id), None) if stop is None: raise gr.Error(f"The {slot_label} slot no longer fits - " f"use 'Which slots can I offer?' for the " f"current menu.") booked[claim_id] = stop.arrival_min note = (f"Planner picked {min_to_hhmm(stop.arrival_min)} " f"inside the {slot_label} slot. ") elif outcome == "They offered a window - planner picks": try: lo_s, hi_s = time_text.strip().split("-") lo, hi = (data_gen.hhmm_to_min(lo_s.strip()), data_gen.hhmm_to_min(hi_s.strip())) except Exception: raise gr.Error("Enter their window as HH:MM-HH:MM, " "e.g. 14:00-16:00") _, trial = _cp_solve(adjuster_id, booked, deferred, offered={claim_id: (lo, hi)}) stop = next((s for r in trial.routes for s in r.stops if s.claim.claim_id == claim_id), None) if stop is None: raise gr.Error(f"Even inside {time_text} the visit can't " f"fit today's plan - suggest tomorrow " f"instead, or ask for another window.") booked[claim_id] = stop.arrival_min note = (f"Planner picked {min_to_hhmm(stop.arrival_min)} " f"inside their {time_text}. ") elif outcome == "Defer to tomorrow": deferred.add(claim_id) booked.pop(claim_id, None) else: # No answer - retry later note = f"{claim_id} parked - it stays on the list to retry. " _, after = _cp_solve(adjuster_id, booked, deferred) drops_after = {c.claim_id for c in after.dropped} hurt = sorted(drops_after - drops_before - deferred) warn = "" if hurt: warn = (f" \n**Careful:** that answer means " f"{', '.join(hurt)} no longer fits today - consider " f"calling them next to rearrange, or defer.") if claim_id in booked: note += (f"{claim_id} booked at " f"{min_to_hhmm(booked[claim_id])}.") elif claim_id in deferred: note += f"{claim_id} moved to tomorrow." # phone-screen storytelling (simulated thread + call captions) msgs = list((state or {}).get("msgs", [])) call = (state or {}).get("call") tmp = {"msgs": msgs} t_booked = (min_to_hhmm(booked[claim_id]) if claim_id in booked else None) if outcome == "Booked at the proposed time": _sms(tmp, "in", "YES", who=claim_id) _sms(tmp, "out", f"Great - you're all set for " f"{_slot_of(claim_id)}. I'll text about 30 " f"minutes before I arrive.", who=f"John to {claim_id}") elif outcome == "Booked into a standard slot": _sms(tmp, "in", "R - something came up, can we find " "another time?", who=claim_id) call = {"cid": claim_id, "lines": [ ("Policyholder", "I can't make my window anymore - " "what else can you do?"), ("John", "One second while I check the rest of my " "day..."), ("John", f"I can offer {slot_label} - I'd be at your " f"door around {t_booked}. Does that work?"), ("Policyholder", "Yes, that works."), ("John", "Booked. You'll get my on-my-way text about " "30 minutes out.")]} _sms(tmp, "out", f"Confirmed: today {slot_label}, arriving " f"about {t_booked}. - John", who=f"John to {claim_id}") elif outcome == "Booked at this time": call = {"cid": claim_id, "lines": [ ("Policyholder", f"Could you come at {t_booked} " f"instead?"), ("John", f"Let me check... yes, {t_booked} works on my " f"end. Booked."), ("Policyholder", "Thank you!")]} _sms(tmp, "out", f"Confirmed: today at {t_booked}. - John", who=f"John to {claim_id}") elif outcome == "They offered a window - planner picks": call = {"cid": claim_id, "lines": [ ("Policyholder", f"I'm only free {time_text.strip()} " f"today."), ("John", "One moment... I can make that work - I'd " f"arrive about {t_booked}."), ("Policyholder", "Perfect, see you then.")]} _sms(tmp, "out", f"Confirmed: arriving about {t_booked} " f"(within {time_text.strip()}). - John", who=f"John to {claim_id}") elif outcome == "Defer to tomorrow": call = {"cid": claim_id, "lines": [ ("Policyholder", "I'm sorry, today won't work at all."), ("John", "I don't want to promise a time I can't keep - " "let's set tomorrow instead. You'll be one of " "my first stops."), ("Policyholder", "Thank you for understanding.")]} _sms(tmp, "out", "Rescheduled to tomorrow - you'll get my " "confirmation text tonight. - John", who=f"John to {claim_id}") else: _sms(tmp, "sys", f"call to {claim_id}: no answer - parked, " f"will retry") if hurt: _sms(tmp, "sys", f"planner: {', '.join(hurt)} affected - " f"see warning") msgs = tmp["msgs"] rows, to_call = _cp_view(booked, deferred, after) state = {"adjuster": adjuster_id, "booked": booked, "deferred": sorted(deferred), "msgs": msgs, "call": call} booked_n = len(booked) left = len([c for c in to_call if c not in booked]) status = (f"**{note}**{warn} \nBooked {booked_n} - " f"{left} still to call.") return (status, rows, gr.update(choices=[c for c in to_call if c not in booked], value=next((c for c in to_call if c not in booked), None)), state, _msgs_html(state), _call_html(state)) # --------------------------------------------------------------------------- # Booking horizon (25-day rolling schedule) # --------------------------------------------------------------------------- def run_horizon(claims_file, adjusters_file, n_claims, n_adjusters, seed, demo_choice, hz_days, hz_day_secs): import horizon as horizon_mod claims, adjusters, src = _load_instance( claims_file, adjusters_file, n_claims, n_adjusters, seed, demo_choice=demo_choice) miles, travel_min = distance.build_matrices(adjusters, claims) plan = horizon_mod.plan_horizon( adjusters, claims, miles, travel_min, horizon=int(hz_days), day_time_limit_s=int(hz_day_secs)) rows = [r for r in horizon_mod.summary_rows(plan) if r[1] or r[4]] total = sum(r[2] for r in horizon_mod.summary_rows(plan)) lines = [f"### Booking book - next {int(hz_days)} days", f"- {src}: {len(claims)} claims in the backlog", f"- day assignment " f"{'**PROVEN OPTIMAL**' if plan.assignment_proven else f'within **{plan.assignment_gap:.3%}** of optimal (certified bound)'}" f" (exact CP-SAT), then each day routed and repaired " f"forward", f"- **{total} of {len(claims)} claims scheduled** inside " f"the horizon"] for cid, planned, actual in plan.must_alerts: when = (f"only on day {actual}" if actual is not None else "on no day inside the horizon") lines.append(f"- :warning: **MUST-TODAY {cid}**: assigned day " f"{planned}, physically routable {when} - " f"needs human attention") if plan.overflow: lines.append(f"- :warning: **overflow beyond the horizon**: " f"{', '.join(c.claim_id for c in plan.overflow)}") if plan.unservable: lines.append(f"- no eligible adjuster: " f"{', '.join(c.claim_id for c in plan.unservable)}") lines.append("\n*Day 1 is an operational plan; later days are a " "capacity-checked booking book, re-solved each " "morning as reality arrives.*") book = [["day", "claim", "adjuster", "arrival"]] for dp in plan.days: if dp.solution is None: continue for r in dp.solution.routes: for s in r.stops: book.append([dp.day, s.claim.claim_id, r.adjuster.adjuster_id, min_to_hhmm(s.arrival_min)]) import csv as _csv path = os.path.join(tempfile.mkdtemp(prefix="horizon_"), "booking_book.csv") with open(path, "w", newline="") as f: _csv.writer(f).writerows(book) return " \n".join(lines), rows, path # --------------------------------------------------------------------------- # UI # --------------------------------------------------------------------------- APP_CSS = """ .gradio-container {max-width: 1320px !important; margin: 0 auto !important;} #solve-btn {height: 56px; font-size: 1.15em; letter-spacing: 0.02em;} #panel-head h3 {margin: 0.1em 0 0.4em 0;} footer {display: none !important;} """ with gr.Blocks(title="Field Adjusters Routing Optimizer") as demo: gr.Markdown("# Field Adjusters Routing Optimizer\n" "Assign claims to field adjusters and sequence each " "adjuster's day to minimize driving - respecting " "policyholder windows, specialties, territories, shifts, " "and inspection durations.") with gr.Group(): with gr.Row(equal_height=False): with gr.Column(scale=1, min_width=300): gr.Markdown("### 1  Data", elem_id="panel-head") with gr.Row(): claims_file = gr.File(label="claims.csv", type="filepath", file_types=[".csv"], height=110) adjusters_file = gr.File(label="adjusters.csv", type="filepath", file_types=[".csv"], height=110) demo_choice = gr.Dropdown( ["(none)"] + demo_scenarios(), value="(none)", label="Or pick a demo scenario", info="quiet-day: all served | typical-day: normal " "trade-offs | storm-surge: overload + red " "banner | territories: radius limits | " "aged-backlog: SLA escalation | pre-assigned: " "assigned_to column for the sequence solver | " "split-windows: two availability windows per " "claim (ortools/cpsat/milp) | cat-backlog: " "a 160-claim storm month | metro-month, " "cat-major, storm-week: production-scale " "backlogs for the Booking horizon tab | " "johns-day: built for Morning " "confirmations. Uploads always take " "priority.") with gr.Accordion("Synthetic instance settings", open=False): n_claims = gr.Slider(5, 500, value=25, step=1, label="Claims", info="on 100+ claim days: " "'ortools' for quick " "solves, 'hybrid' with " "600 s+ for the best " "plan") n_adjusters = gr.Slider(2, 50, value=5, step=1, label="Adjusters") seed = gr.Number(value=42, precision=0, label="Random seed") with gr.Column(scale=1, min_width=300): gr.Markdown("### 2  Solver", elem_id="panel-head") time_limit = gr.Slider(5, 900, value=10, step=5, label="Solver time limit (s)", info="long budgets suit " "night-before planning on " "big days; the CLI is " "uncapped for batch runs") solver_choice = gr.Dropdown( ["ortools", "pyvrp", "hybrid", "milp (exact)", "cpsat (exact, license-free)", "setpart (route pool + MILP)", "sequence (pre-assigned, exact)", "sequence-milp (pre-assigned, MILP proof)"], value="ortools", label="Solver backend", info=( "\u2022 **ortools** - the default: best for quick " "interactive solves\n" "\u2022 **pyvrp** - alternative heuristic engine " "(iterated local search)\n" "\u2022 **hybrid** - OR-Tools start + PyVRP " "improve: best quality on mid-size days " "(25-80 claims) with 30s+ budgets, and on " "500-claim days with 600s+; at shorter " "big-day budgets plain ortools measured " "strongest\n" "\u2022 **milp (exact)** - mathematical proof of " "optimality via Gurobi/HiGHS; small days " "(~15-25 claims) only\n" "\u2022 **cpsat** - proof of optimality via " "CP-SAT (no big-M, no license limits): " "fastest prover on referee-size days\n" "\u2022 **setpart** - several heuristic runs feed " "a route pool, a small MILP picks the provably " "best combination: never worse than the best " "single run\n" "\u2022 **sequence** - claims arrive pre-assigned " "(assigned_to column); each adjuster's stop " "order is solved exactly, misfits are dropped " "and reported for rescheduling\n" "\u2022 **sequence-milp** - sequence, plus a " "per-adjuster MILP certificate (Gurobi or " "HiGHS) confirming each optimum independently")) lunch_break = gr.Checkbox( label="Lunch break (30 min, 11:30-13:30)", info="works with every backend except pyvrp " "(greyed out when pyvrp is selected)") balance = gr.Checkbox( label="Balance workload", info="works with every backend except pyvrp " "(greyed out when pyvrp is selected)") with gr.Column(scale=1, min_width=300): gr.Markdown("### 3  Distances & run", elem_id="panel-head") distance_source = gr.Radio( ["haversine", "google"], value="haversine", label="Distance model", info="haversine: free estimate. google: real " "driving routes (billed per matrix element)") google_key = gr.Textbox( label="Google Maps API key", type="password", placeholder="paste key here when using google", info="Needed only for the google distance model " "(and address geocoding on AI intake). Used " "in this session only - never saved to disk. " "Alternatively set GOOGLE_MAPS_API_KEY before " "launching.") solve_btn = gr.Button("Solve", variant="primary", elem_id="solve-btn") with gr.Row(): with gr.Column(scale=3): summary_md = gr.Markdown() with gr.Column(scale=2): banner_html = gr.HTML() with gr.Tabs(): with gr.Tab("Map"): map_html = gr.HTML() with gr.Tab("Schedule"): schedule_out = gr.Dataframe(interactive=False) with gr.Tab("Dropped claims"): dropped_out = gr.Dataframe(interactive=False) with gr.Tab("Download"): gr.Markdown( "**Your files are the record - download before closing.** " "The app can never modify the CSVs on your computer; " "instead, every applied change (hires, hours, " "priorities, intake claims) lands in the *updated* " "exports below. On the hosted demo, server storage is " "wiped on restart - these downloads are the " "persistence.") with gr.Row(): csv_out = gr.File(label="assignments.csv (dispatch)") updated_claims_out = gr.File( label="claims_updated.csv (working dataset)") updated_adjusters_out = gr.File( label="adjusters_updated.csv (working roster)") gr.Markdown("---") gr.Markdown( "**Prepare tomorrow's files** - served claims removed, " "unserved claims carried forward one day older (their " "penalties grow automatically), today's roster as-is " "for you to curate. Upload both tomorrow morning and " "Solve.") tomorrow_btn = gr.Button("Prepare tomorrow's files", elem_id="tomorrow-btn") tomorrow_note = gr.Markdown() with gr.Row(): tomorrow_claims_out = gr.File(label="claims_tomorrow.csv") tomorrow_adjusters_out = gr.File( label="adjusters_tomorrow.csv") with gr.Tab("Morning confirmations"): gr.Markdown( "**Never drive to an appointment nobody confirmed.** " "Solve first, then log this morning's call results: " "anything not confirmed is withdrawn from today, the " "confirmed day is re-solved (freed capacity is offered " "to claims that would otherwise have rolled), and every " "withheld visit is rebooked on the first day that " "actually has room. The minutes reported are *exposure* " "- driving that was riding on an unconfirmed door.") with gr.Row(): mc_noanswer = gr.Dropdown( [], multiselect=True, label="No answer", info="texted and called, nothing back") mc_cancel = gr.Dropdown( [], multiselect=True, label="Cannot do today", info="policyholder declined today outright") mc_resched = gr.Dropdown( [], multiselect=True, label="Wants another time", info="slot menu is trial-solved for each") with gr.Row(): mc_load = gr.Button("Load today's appointments") mc_days = gr.Slider(5, 25, value=25, step=1, label="Rebooking horizon (days)") mc_btn = gr.Button("Apply confirmations & re-dispatch", variant="primary") mc_status = gr.Markdown() mc_table = gr.Dataframe( headers=["claim", "why held", "comes back"], interactive=False, label="Withheld today - and when each returns") with gr.Tab("AI claim intake"): gr.Markdown( "Paste a First Notice of Loss email or call notes; Claude " "extracts structured claim records (peril, priority, " "availability window, inspection time). Requires " "`ANTHROPIC_API_KEY` in the environment.") fnol_box = gr.Textbox( label="FNOL text", lines=9, placeholder="e.g. 'Policyholder Maria Gonzalez called - " "kitchen fire last night at 4413 Almeda Rd, " "family staying with relatives, needs someone " "out today. She's only reachable before noon.'") extract_btn = gr.Button("Extract claims", variant="primary", elem_id="extract-btn") intake_df = gr.Dataframe(interactive=False, label="Extracted claims") intake_note = gr.Markdown() pending_state = gr.State([]) add_btn = gr.Button("Add to claims.csv") add_status = gr.Markdown() with gr.Tab("In-day repair"): gr.Markdown( "**A new claim arrived mid-day? Re-plan the rest of the " "day without rewriting the past.** Repair reads the " "current schedule and the clock: stops already finished " "stay exactly as they happened, anyone mid-inspection " "finishes it, and every adjuster's remaining afternoon " "is re-optimized from where they actually are - " "including any claims just added on the AI claim intake " "tab, anything dropped this morning, and all the usual " "rules (windows, skills, territories, home by shift " "end). Assumes adjusters followed the plan so far.") with gr.Row(): now_box = gr.Textbox(label="Current time (HH:MM)", value="13:00", max_lines=1, scale=1) repair_limit = gr.Slider( 5, 30, value=10, step=5, scale=2, label="Repair time limit (s)") repair_btn = gr.Button("Repair rest of day", variant="primary", scale=1, elem_id="repair-btn") repair_status = gr.Markdown() with gr.Tab("AI assistant"): gr.Markdown( "Ask about the solved schedule ('Why was CLM-007 " "dropped?') or run what-ifs ('What if ADJ-01 is out " "sick?'). Solve first. Requires `ANTHROPIC_API_KEY`.") chatbot = gr.Chatbot(height=430, label="Dispatch assistant") with gr.Row(): chat_box = gr.Textbox( label="Message", scale=4, lines=1, placeholder="Why was CLM-007 dropped?") chat_btn = gr.Button("Send", variant="primary", scale=1, elem_id="chat-btn") assist_state = gr.State(None) gr.Markdown("---") gr.Markdown("**Apply the last what-if** - replaces the " "working schedule with the assistant's most " "recent hypothetical scenario (adjusters " "excluded / claims escalated).") with gr.Row(): apply_confirm = gr.Checkbox( label="I understand this replaces the current " "schedule") apply_btn = gr.Button("Apply scenario", elem_id="apply-btn") with gr.Tab("AI briefings"): gr.Markdown( "One click: Claude writes each adjuster a plain-language " "morning briefing from the solved schedule. Requires " "`ANTHROPIC_API_KEY`.") brief_btn = gr.Button("Generate morning briefings", variant="primary", elem_id="brief-btn") brief_md = gr.Markdown() brief_file = gr.File(label="morning_briefings.md") with gr.Tab("Call planner"): gr.Markdown( "**Guided appointment calling** (prototype). For days " "where each adjuster phones their own policyholders: " "the planner suggests the calling order and a time to " "propose on each call, then replans the rest of the " "day after every answer - and warns before a promise " "would wreck the schedule. When a policyholder can't " "make their window, click 'Which slots can I offer?' " "to get the feasibility-checked menu of standard " "company slots to read to them. Needs a pre-assigned " "day (the 'pre-assigned' demo scenario): Solve first, " "then start a session here.") with gr.Row(): cp_load_btn = gr.Button("Load adjusters") cp_adj = gr.Dropdown(label="Adjuster", choices=[]) cp_start_btn = gr.Button("Start calling session", variant="primary") with gr.Row(): with gr.Column(scale=3): cp_status = gr.Markdown() cp_script = gr.Dataframe( headers=["call order", "claim", "suggested proposal", "status"], interactive=False, label="Calling script (live)") with gr.Row(): cp_claim = gr.Dropdown(label="Claim just called", choices=[]) cp_outcome = gr.Radio( ["Booked at the proposed time", "Booked into a standard slot", "Booked at this time", "They offered a window - planner picks", "Defer to tomorrow", "No answer - retry later"], label="What happened on the call?", value="Booked at the proposed time") with gr.Column(): cp_slot = gr.Dropdown( [_cp_slot_label(s) for s in config.CALL_SLOTS], label="Standard slot they chose", info="for 'Booked into a standard slot'") cp_time = gr.Textbox( label="Time (HH:MM) or window " "(HH:MM-HH:MM)", placeholder="only for 'this time' / " "'window'") with gr.Row(): cp_menu_btn = gr.Button("Which slots can I " "offer?") cp_log_btn = gr.Button("Log this call", variant="primary") with gr.Column(scale=1, min_width=340): with gr.Tabs(): with gr.Tab("Messages"): cp_phone_msgs = gr.HTML(_msgs_html(None)) with gr.Tab("Call"): cp_phone_call = gr.HTML(_call_html(None)) cp_state = gr.State(None) with gr.Tab("Booking horizon"): gr.Markdown( "**Schedule the whole backlog across the next N days.** " "An exact day-assignment model spreads every claim over " "the horizon (urgent first, capacity respected), each " "day is routed by the daily engine, and anything a " "day's exact routing can't fit rolls forward " "automatically. Uses the same data sources as Solve " "(uploads / demo / synthetic).") with gr.Row(): hz_days = gr.Slider(5, 25, value=25, step=1, label="Horizon (days)") hz_day_secs = gr.Slider(3, 15, value=5, step=1, label="Routing seconds per day") hz_btn = gr.Button("Plan the horizon", variant="primary") hz_status = gr.Markdown() hz_table = gr.Dataframe( headers=["day", "booked", "served", "drive min", "rolls fwd", "claims rolling forward"], interactive=False, label="The booking book, day by day") hz_csv = gr.File(label="booking_book.csv") def sync_toggle_availability(choice): # pyvrp is the one backend with no lunch/balance mechanism; # grey out (and clear) the toggles so they can't be ticked. if choice == "pyvrp": off = gr.update(interactive=False, value=False) return off, off on = gr.update(interactive=True) return on, on solver_choice.change(sync_toggle_availability, inputs=[solver_choice], outputs=[lunch_break, balance]) extract_btn.click(run_extract, inputs=[fnol_box], outputs=[intake_df, pending_state, intake_note]) add_btn.click(add_extracted_claims, inputs=[pending_state], outputs=[add_status, pending_state]) chat_btn.click(chat_turn, inputs=[chat_box, chatbot, assist_state], outputs=[chat_box, chatbot, assist_state]) chat_box.submit(chat_turn, inputs=[chat_box, chatbot, assist_state], outputs=[chat_box, chatbot, assist_state]) solve_btn.click( run_solve, inputs=[claims_file, adjusters_file, n_claims, n_adjusters, seed, time_limit, distance_source, solver_choice, lunch_break, balance, demo_choice, google_key], outputs=[summary_md, banner_html, map_html, schedule_out, dropped_out, csv_out, updated_claims_out, updated_adjusters_out], ) apply_btn.click( apply_scenario, inputs=[assist_state, apply_confirm], outputs=[summary_md, banner_html, map_html, schedule_out, dropped_out, csv_out, updated_claims_out, updated_adjusters_out], ) tomorrow_btn.click( make_tomorrow_files, outputs=[tomorrow_note, tomorrow_claims_out, tomorrow_adjusters_out], ) brief_btn.click(make_briefings, inputs=[assist_state], outputs=[brief_md, brief_file]) cp_load_btn.click(cp_load_adjusters, outputs=[cp_adj]) cp_start_btn.click(cp_start, inputs=[cp_adj], outputs=[cp_status, cp_script, cp_claim, cp_state, cp_phone_msgs, cp_phone_call]) cp_menu_btn.click(cp_slot_menu, inputs=[cp_state, cp_claim], outputs=[cp_status, cp_state, cp_phone_call]) hz_btn.click(run_horizon, inputs=[claims_file, adjusters_file, n_claims, n_adjusters, seed, demo_choice, hz_days, hz_day_secs], outputs=[hz_status, hz_table, hz_csv]) mc_load.click(mc_load_appointments, outputs=[mc_noanswer, mc_cancel, mc_resched]) mc_btn.click(mc_apply, inputs=[mc_noanswer, mc_cancel, mc_resched, mc_days], outputs=[mc_status, mc_table]) cp_log_btn.click(cp_log, inputs=[cp_state, cp_claim, cp_outcome, cp_time, cp_slot], outputs=[cp_status, cp_script, cp_claim, cp_state, cp_phone_msgs, cp_phone_call]) repair_btn.click( run_repair, inputs=[now_box, repair_limit], outputs=[repair_status, summary_md, banner_html, map_html, schedule_out, dropped_out, csv_out, updated_claims_out, updated_adjusters_out], ) if __name__ == "__main__": # Cloud-aware launch. Local laptop: binds 127.0.0.1:7860 as before. # Hugging Face Spaces (SPACE_ID set) and containers (PORT set, e.g. # Cloud Run) bind 0.0.0.0 on the platform's port. Optional demo # login: set DEMO_USERNAME + DEMO_PASSWORD in the environment (on # Spaces: Settings -> Variables and secrets). on_cloud = bool(os.environ.get("SPACE_ID") or os.environ.get("PORT")) auth = None if os.environ.get("DEMO_USERNAME") and os.environ.get("DEMO_PASSWORD"): auth = (os.environ["DEMO_USERNAME"], os.environ["DEMO_PASSWORD"]) demo.launch( server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0" if on_cloud else "127.0.0.1"), server_port=int(os.environ.get("PORT") or os.environ.get("GRADIO_SERVER_PORT") or 7860), # Gradio 6's Node SSR proxy can fail to signal readiness on HF # Spaces, leaving the Space stuck on "Restarting" although the # app works. SSR only speeds up first paint - keep it off. ssr_mode=False, auth=auth, theme=gr.themes.Soft(primary_hue="emerald", neutral_hue="slate"), css=APP_CSS)