""" Batch Efficiency Analyser ========================= Pure-function service that analyses a delivery batch and returns: - fleet_summary : aggregate metrics + load-balance scores - rider_timelines : per-rider start/finish/pace/utilisation - substitution_opportunities : ranked, scored transfer plans - top_recommendation : best action with confidence, root-cause & risk factors No DB access here. The route handler owns fetching; this layer is fully testable with any list of dicts. Expected delivery dict keys (all optional except userid): userid : int rider id pickupcustomer : str e.g. "Daily Grubs Bhuvaneshwari" assigntime : str "YYYY-MM-DD HH:MM:SS" pickuptime : str "YYYY-MM-DD HH:MM:SS" or None deliverytime : str "YYYY-MM-DD HH:MM:SS" or None dlat / droplat : float delivery lat (either key accepted) dlon / droplon : float delivery lon (either key accepted) deliveryid : int order id (for transfer manifests) """ from __future__ import annotations import math import statistics from collections import defaultdict from datetime import datetime, timedelta from typing import Any # --------------------------------------------------------------------------- # Defaults # --------------------------------------------------------------------------- DEFAULT_KITCHEN_COORDS: dict[str, tuple[float, float]] = { "vidhya": (11.01633, 77.01478), "jayanthi": (11.03887, 76.93008), "nandhini": (11.04324, 77.00068), "bhuvaneshwari": (11.00352, 76.95455), "selvarani": (10.99274, 77.00535), } DEFAULT_KITCHEN_FRAGMENTS: list[str] = [ "bhuvaneshwari", "jayanthi", "nandhini", "vidhya", "selvarani" ] DEFAULT_ROAD_KMH: float = 13.0 DEFAULT_IDLE_THRESHOLD_MIN: float = 30.0 DEFAULT_MAX_TRANSFER: int = 4 # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _hav(la1: float, lo1: float, la2: float, lo2: float) -> float: R = 6371.0 la1, lo1, la2, lo2 = map(math.radians, [la1, lo1, la2, lo2]) a = (math.sin((la2 - la1) / 2) ** 2 + math.cos(la1) * math.cos(la2) * math.sin((lo2 - lo1) / 2) ** 2) return R * 2 * math.asin(math.sqrt(max(0.0, min(1.0, a)))) def _travel_min(km: float, kmh: float = DEFAULT_ROAD_KMH) -> float: return km / kmh * 60.0 def _parse_ts(s: Any) -> datetime | None: if not s: return None s = str(s).strip() for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"): try: return datetime.strptime(s, fmt) except ValueError: continue return None def _fmt(dt: datetime | None) -> str | None: return dt.strftime("%H:%M:%S") if dt else None def _get_coord(order: dict) -> tuple[float, float] | None: lat = order.get("dlat") or order.get("droplat") or order.get("deliverylat") lon = order.get("dlon") or order.get("droplon") or order.get("deliverylong") try: return float(lat), float(lon) except (TypeError, ValueError): return None def _detect_kitchen( pickupcustomer: str | None, fragments: list[str], ) -> str | None: kl = (pickupcustomer or "").lower() for frag in fragments: if frag in kl: return frag return None def _stdev(values: list[float]) -> float: """Population stdev; returns 0 for fewer than 2 values.""" if len(values) < 2: return 0.0 try: return statistics.stdev(values) except Exception: return 0.0 def _score_candidate( o: dict, arrive_at_kitchen: datetime, k_coord: tuple[float, float], road_kmh: float, ) -> float: """ Score a candidate order for transfer to the idle rider. Higher = better candidate. Two components: time_gain : minutes saved vs the original delivery time. Positive means idle rider genuinely arrives earlier. geo_penalty: haversine distance from kitchen to the order drop point. Penalises far orders that inflate the idle rider's extra km. Orders with negative time_gain (idle rider would be slower) still get a score, allowing the caller to filter them out with a feasibility check. """ d_ts = _parse_ts(o.get("deliverytime")) coord = _get_coord(o) if not d_ts or not coord: return -9999.0 dist_km = _hav(k_coord[0], k_coord[1], coord[0], coord[1]) est_deliver = arrive_at_kitchen + timedelta(minutes=_travel_min(dist_km, road_kmh)) time_gain_min = (d_ts - est_deliver).total_seconds() / 60.0 # Weight: time gain matters more than geography (70/30 split). # Penalise 2 min per km of extra distance so nearby clusters float up. return time_gain_min * 0.7 - dist_km * 2.0 * 0.3 # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def analyse_batch( deliveries: list[dict], rider_names: dict[int, str] | None = None, kitchen_coords: dict[str, tuple[float, float]] | None = None, kitchen_fragments: list[str] | None = None, road_kmh: float = DEFAULT_ROAD_KMH, idle_threshold_min: float = DEFAULT_IDLE_THRESHOLD_MIN, max_transfer: int = DEFAULT_MAX_TRANSFER, ) -> dict: """ Analyse a delivery batch and return substitution opportunities. Parameters ---------- deliveries : list of order dicts (see module docstring) rider_names : optional {userid: name} override kitchen_coords : optional kitchen pickup coordinates override kitchen_fragments : optional kitchen detection strings override road_kmh : estimated loaded-rider road speed idle_threshold_min: minimum idle window to flag as opportunity max_transfer : maximum orders to suggest transferring to one rider """ if kitchen_coords is None: kitchen_coords = DEFAULT_KITCHEN_COORDS if kitchen_fragments is None: kitchen_fragments = DEFAULT_KITCHEN_FRAGMENTS if rider_names is None: rider_names = {} # ------------------------------------------------------------------ # 1. Group deliveries by rider # ------------------------------------------------------------------ by_rider: dict[int, list[dict]] = defaultdict(list) for o in deliveries: uid = o.get("userid") if uid is not None: by_rider[int(uid)].append(o) if not by_rider: return { "fleet_summary": {}, "rider_timelines": [], "substitution_opportunities": [], "top_recommendation": None, "error": "No deliveries with valid userid found.", } # ------------------------------------------------------------------ # 2. Per-rider timeline # ------------------------------------------------------------------ timelines: list[dict] = [] for uid, orders in by_rider.items(): # Detect primary kitchen by majority vote kitchen_votes: dict[str, int] = defaultdict(int) for o in orders: k = _detect_kitchen(o.get("pickupcustomer"), kitchen_fragments) if k: kitchen_votes[k] += 1 primary_kitchen = ( max(kitchen_votes, key=kitchen_votes.get) if kitchen_votes else None ) kitchen_confidence = ( round(kitchen_votes[primary_kitchen] / len(orders), 2) if primary_kitchen else 0.0 ) # Timestamps finish_ts = None start_ts = None last_coord: tuple[float, float] | None = None completed_orders = 0 for o in orders: a = _parse_ts(o.get("assigntime")) d = _parse_ts(o.get("deliverytime")) if a and (start_ts is None or a < start_ts): start_ts = a if d: completed_orders += 1 if finish_ts is None or d > finish_ts: finish_ts = d coord = _get_coord(o) if coord: last_coord = coord # Active duration and pace active_minutes: float | None = None pace_orders_per_hour: float | None = None if start_ts and finish_ts and finish_ts > start_ts: active_minutes = round((finish_ts - start_ts).total_seconds() / 60, 1) pace_orders_per_hour = round(completed_orders / (active_minutes / 60), 1) if active_minutes else None timelines.append({ "userid": uid, "name": rider_names.get(uid, f"Rider {uid}"), "kitchen": primary_kitchen, "kitchen_confidence": kitchen_confidence, "order_count": len(orders), "completed_orders": completed_orders, "pending_orders": len(orders) - completed_orders, "started_at": _fmt(start_ts), "finished_at": _fmt(finish_ts), "active_minutes": active_minutes, "pace_orders_per_hour": pace_orders_per_hour, "_finish_dt": finish_ts, "_start_dt": start_ts, "last_position": ( {"lat": round(last_coord[0], 6), "lon": round(last_coord[1], 6)} if last_coord else None ), "_last_coord": last_coord, }) # Sort by finish time timelines.sort(key=lambda t: t["_finish_dt"] or datetime.min) # ------------------------------------------------------------------ # 3. Fleet summary # ------------------------------------------------------------------ valid_start = [t["_start_dt"] for t in timelines if t["_start_dt"]] valid_finish = [t["_finish_dt"] for t in timelines if t["_finish_dt"]] fleet_start = min(valid_start) if valid_start else None fleet_done = max(valid_finish) if valid_finish else None # Load balance: stdev of order counts and finish-time spread order_counts = [t["order_count"] for t in timelines] finish_offsets_min = ( [(f - fleet_start).total_seconds() / 60 for f in valid_finish] if fleet_start and valid_finish else [] ) finish_spread_min = ( round((max(valid_finish) - min(valid_finish)).total_seconds() / 60) if len(valid_finish) >= 2 else 0 ) load_balance_stdev = round(_stdev([float(c) for c in order_counts]), 2) finish_time_stdev = round(_stdev(finish_offsets_min), 1) # Utilisation: how much of the batch window each rider was actively delivering batch_duration_min = ( round((fleet_done - fleet_start).total_seconds() / 60) if fleet_start and fleet_done else None ) avg_active_minutes = ( round( sum(t["active_minutes"] for t in timelines if t["active_minutes"]) / max(1, sum(1 for t in timelines if t["active_minutes"])), 1, ) if any(t["active_minutes"] for t in timelines) else None ) avg_utilisation_pct = ( round(avg_active_minutes / batch_duration_min * 100, 1) if avg_active_minutes and batch_duration_min else None ) fleet_summary = { "total_orders": len(deliveries), "total_riders": len(by_rider), "fleet_start": _fmt(fleet_start), "fleet_done": _fmt(fleet_done), "total_duration_minutes": batch_duration_min, "orders_per_rider_avg": round(len(deliveries) / len(by_rider), 1), "load_balance_stdev": load_balance_stdev, "finish_time_spread_minutes": finish_spread_min, "finish_time_stdev_minutes": finish_time_stdev, "avg_utilisation_pct": avg_utilisation_pct, "avg_active_minutes": avg_active_minutes, } if not fleet_done: return { "fleet_summary": fleet_summary, "rider_timelines": _clean_timelines(timelines, fleet_done), "substitution_opportunities": [], "top_recommendation": None, "error": "No completed deliveries found.", } # Mark each rider's idle status and free window for t in timelines: fd = t["_finish_dt"] if fd: idle_min = (fleet_done - fd).total_seconds() / 60.0 t["idle_minutes"] = round(idle_min) t["free_window_minutes"] = round(idle_min) # time available for substitution t["status"] = "idle" if idle_min >= idle_threshold_min else "active" else: t["idle_minutes"] = 0 t["free_window_minutes"] = 0 t["status"] = "unknown" # ------------------------------------------------------------------ # 4. Substitution opportunities # ------------------------------------------------------------------ opportunities: list[dict] = [] idle_riders = [t for t in timelines if t["status"] == "idle" and t["_last_coord"]] for idle in idle_riders: idle_uid = idle["userid"] idle_finish = idle["_finish_dt"] idle_coord = idle["_last_coord"] free_window_min = idle["free_window_minutes"] if idle["kitchen"] is None: continue # can't determine origin kitchen for target_kitchen, k_coord in kitchen_coords.items(): if target_kitchen == idle["kitchen"]: continue # skip own kitchen # Travel from idle rider's last drop to the target kitchen travel_km = _hav(idle_coord[0], idle_coord[1], k_coord[0], k_coord[1]) travel_min = _travel_min(travel_km, road_kmh) # Skip if idle rider can't even reach the kitchen before fleet is done if travel_min >= free_window_min: continue arrive_at_kitchen = idle_finish + timedelta(minutes=travel_min) # ---------------------------------------------------------- # Candidate selection: score every order from this kitchen # that was delivered after the idle rider could arrive. # Score = time_gain (70%) + geo proximity (30%). # This prefers orders where idle rider is genuinely faster # AND that are close to the kitchen (less detour). # ---------------------------------------------------------- candidate_orders: list[tuple[int, dict, float]] = [] # (rid, order, score) for rid, r_orders in by_rider.items(): if rid == idle_uid: continue r_kitchen = next( (t["kitchen"] for t in timelines if t["userid"] == rid), None ) if r_kitchen != target_kitchen: continue for o in r_orders: d_ts = _parse_ts(o.get("deliverytime")) if not d_ts or d_ts <= arrive_at_kitchen: continue score = _score_candidate(o, arrive_at_kitchen, k_coord, road_kmh) candidate_orders.append((rid, o, score)) if not candidate_orders: continue # Sort best candidates first (highest score = most time gained, closest) candidate_orders.sort(key=lambda x: x[2], reverse=True) take_pool = candidate_orders[:max_transfer] # Greedy nearest-neighbour route from the kitchen through selected orders unvisited = list(range(len(take_pool))) curr_nn = k_coord greedy_take: list[tuple[int, dict]] = [] while unvisited: ni = min( unvisited, key=lambda i: ( _hav(curr_nn[0], curr_nn[1], *c) if (c := _get_coord(take_pool[i][1])) else 999.0 ), ) unvisited.remove(ni) greedy_take.append((take_pool[ni][0], take_pool[ni][1])) coord = _get_coord(take_pool[ni][1]) if coord: curr_nn = coord # Simulate idle rider executing the greedy route curr_pos = k_coord est_time = arrive_at_kitchen transfer_manifests: list[dict] = [] total_delivery_leg_min = 0.0 for orig_rid, o in greedy_take: coord = _get_coord(o) if not coord: continue d_km = _hav(curr_pos[0], curr_pos[1], coord[0], coord[1]) d_min = _travel_min(d_km, road_kmh) total_delivery_leg_min += d_min est_deliver = est_time + timedelta(minutes=d_min) orig_deliver = _parse_ts(o.get("deliverytime")) improvement = ( round((orig_deliver - est_deliver).total_seconds() / 60) if orig_deliver and est_deliver else None ) is_feasible = improvement is not None and improvement > 0 transfer_manifests.append({ "deliveryid": o.get("deliveryid"), "from_rider_id": orig_rid, "from_rider_name": rider_names.get(orig_rid, f"Rider {orig_rid}"), "original_delivery_time": _fmt(orig_deliver), "estimated_delivery_time": _fmt(est_deliver), "improvement_minutes": improvement, "is_feasible": is_feasible, "location": {"lat": round(coord[0], 6), "lon": round(coord[1], 6)}, }) curr_pos = coord est_time = est_deliver # Only keep orders where idle rider is actually faster feasible_manifests = [m for m in transfer_manifests if m["is_feasible"]] if not feasible_manifests: continue idle_new_finish = est_time # Check idle rider can complete within their free window total_obligation_min = travel_min + total_delivery_leg_min if total_obligation_min > free_window_min: # Idle rider would finish after fleet, extending rather than helping continue # New fleet done after the transfer taken_order_objs = {id(o) for (_, o) in greedy_take} new_finish_by_rider: dict[int, datetime | None] = {} for rid, r_orders in by_rider.items(): if rid == idle_uid: new_finish_by_rider[rid] = idle_new_finish continue remaining = [o for o in r_orders if id(o) not in taken_order_objs] finishes = [_parse_ts(o.get("deliverytime")) for o in remaining] valid = [f for f in finishes if f] new_finish_by_rider[rid] = max(valid) if valid else None new_fleet_done = max( (v for v in new_finish_by_rider.values() if v), default=fleet_done, ) fleet_improvement = round( (fleet_done - new_fleet_done).total_seconds() / 60 ) # Most relieved rider orig_last_by_rider = { rid: max( (f for f in [_parse_ts(o.get("deliverytime")) for o in r_o] if f), default=None, ) for rid, r_o in by_rider.items() } most_impacted_rid = max( (rid for rid in {r for r, _ in greedy_take}), key=lambda r: (orig_last_by_rider.get(r) or datetime.min), default=None, ) orig_overloaded_finish = orig_last_by_rider.get(most_impacted_rid) new_overloaded_finish = new_finish_by_rider.get(most_impacted_rid) time_saved = ( round((orig_overloaded_finish - new_overloaded_finish).total_seconds() / 60) if orig_overloaded_finish and new_overloaded_finish else 0 ) # Total extra km for idle rider (idle→kitchen + all delivery legs) total_extra_km = travel_km + sum( _hav( (k_coord if i == 0 else (_get_coord(greedy_take[i-1][1]) or k_coord))[0], (k_coord if i == 0 else (_get_coord(greedy_take[i-1][1]) or k_coord))[1], *(_get_coord(o) or k_coord), ) for i, (_, o) in enumerate(greedy_take) if _get_coord(o) ) # ---------------------------------------------------------- # Confidence score (0-100) # Measures how comfortable this transfer is given real constraints. # # Component 1 – Slack ratio: how much free time the idle rider # has beyond the time they'll spend doing the transfer. # (free_window - total_obligation) / free_window → 0..1 # # Component 2 – Feasibility ratio: what fraction of the # transferred orders actually deliver earlier than original. # feasible_count / total_transferred → 0..1 # # Component 3 – Fleet gain ratio: minutes saved as a fraction # of total batch duration. Capped at 20 min improvement for # full score so small batches don't produce inflated scores. # ---------------------------------------------------------- slack_ratio = max(0.0, (free_window_min - total_obligation_min) / free_window_min) feasibility_ratio = len(feasible_manifests) / max(1, len(transfer_manifests)) fleet_gain_ratio = min(1.0, fleet_improvement / 20.0) if fleet_improvement > 0 else 0.0 confidence_score = round( (slack_ratio * 0.4 + feasibility_ratio * 0.35 + fleet_gain_ratio * 0.25) * 100 ) efficiency_ratio = ( round(fleet_improvement / max(0.1, total_extra_km), 2) if total_extra_km > 0 else 0.0 ) if efficiency_ratio >= 5: efficiency_rating = "high" elif efficiency_ratio >= 2: efficiency_rating = "medium" else: efficiency_rating = "low" opportunities.append({ "idle_rider": { "userid": idle_uid, "name": idle["name"], "primary_kitchen": idle["kitchen"], "order_count": idle["order_count"], "finished_at": _fmt(idle_finish), "idle_minutes": idle["idle_minutes"], "free_window_minutes": free_window_min, "last_position": idle["last_position"], }, "target_kitchen": target_kitchen, "travel_to_kitchen_km": round(travel_km, 1), "travel_to_kitchen_minutes": round(travel_min), "arrive_at_kitchen": _fmt(arrive_at_kitchen), "orders_to_transfer": transfer_manifests, "total_orders_transferred": len(feasible_manifests), "feasible_orders_count": len(feasible_manifests), "most_relieved_rider": { "userid": most_impacted_rid, "name": rider_names.get(most_impacted_rid, f"Rider {most_impacted_rid}"), "original_finish": _fmt(orig_overloaded_finish), "new_finish": _fmt(new_overloaded_finish), "time_saved_minutes": time_saved, }, "extra_km_for_idle_rider": round(total_extra_km, 1), "total_obligation_minutes": round(total_obligation_min), "idle_rider_new_finish": _fmt(idle_new_finish), "original_fleet_done": _fmt(fleet_done), "new_fleet_done": _fmt(new_fleet_done), "fleet_improvement_minutes": fleet_improvement, "confidence_score": confidence_score, "efficiency_ratio": efficiency_ratio, "efficiency_rating": efficiency_rating, }) # Keep only net-positive opportunities opportunities = [o for o in opportunities if o["fleet_improvement_minutes"] > 0] # Sort: confidence first (overall quality), then fleet improvement, then rider time saved opportunities.sort( key=lambda x: ( -x["confidence_score"], -x["fleet_improvement_minutes"], -x["most_relieved_rider"]["time_saved_minutes"], ) ) # ------------------------------------------------------------------ # 5. Top recommendation # ------------------------------------------------------------------ top_recommendation = _build_recommendation( opportunities, idle_threshold_min, fleet_done, timelines, fleet_summary ) return { "fleet_summary": fleet_summary, "rider_timelines": _clean_timelines(timelines, fleet_done), "substitution_opportunities": opportunities, "top_recommendation": top_recommendation, } # --------------------------------------------------------------------------- # Helpers for clean output # --------------------------------------------------------------------------- def _clean_timelines( timelines: list[dict], fleet_done: datetime | None, ) -> list[dict]: out = [] for t in timelines: out.append({ "userid": t["userid"], "name": t["name"], "kitchen": t["kitchen"], "kitchen_confidence": t.get("kitchen_confidence", 0.0), "order_count": t["order_count"], "completed_orders": t.get("completed_orders", t["order_count"]), "pending_orders": t.get("pending_orders", 0), "started_at": t["started_at"], "finished_at": t["finished_at"], "active_minutes": t.get("active_minutes"), "pace_orders_per_hour": t.get("pace_orders_per_hour"), "idle_minutes": t.get("idle_minutes", 0), "free_window_minutes": t.get("free_window_minutes", 0), "status": t.get("status", "unknown"), "last_position": t.get("last_position"), }) return out def _build_recommendation( opportunities: list[dict], idle_threshold: float, fleet_done: datetime | None, timelines: list[dict], fleet_summary: dict, ) -> dict | None: if not opportunities: # Diagnose WHY there are no opportunities even if some riders were idle idle_count = sum(1 for t in timelines if t.get("status") == "idle") if idle_count == 0: reason = "All riders finished within the idle threshold window — batch was well balanced." else: reason = ( f"{idle_count} rider(s) finished early but no feasible substitution found: " "either travel time exceeds the idle window, or all candidate orders " "would be delivered later by the idle rider than the original." ) return { "action": "none", "reason": reason, "fleet_balance_assessment": _balance_assessment(fleet_summary), } best = opportunities[0] idle = best["idle_rider"] target = best["target_kitchen"] relieved = best["most_relieved_rider"] primary_kitchen = idle["primary_kitchen"] or "unknown" confidence = best["confidence_score"] # Root cause: why was this rider idle? root_cause = _diagnose_root_cause(idle, timelines, fleet_summary) # Risk factors risk_factors = _identify_risks(best, idle_threshold) description = ( f"{idle['name']} ({primary_kitchen}) finished all {idle['order_count']} orders " f"at {idle['finished_at']} — {idle['idle_minutes']} min before the fleet finished. " f"Assigning {best['feasible_orders_count']} {target} orders: " f"travel {best['travel_to_kitchen_km']} km ({best['travel_to_kitchen_minutes']} min), " f"arrive at {target} kitchen at {best['arrive_at_kitchen']}. " f"Relieves {relieved['name']} by {relieved['time_saved_minutes']} min " f"({relieved['original_finish']} → {relieved['new_finish']}). " f"Fleet finishes {best['fleet_improvement_minutes']} min earlier " f"({best['original_fleet_done']} → {best['new_fleet_done']}). " f"Confidence: {confidence}/100." ) # Dynamic thresholds derived from the actual batch data idle_rider_loads = [t["order_count"] for t in timelines if t.get("kitchen") == primary_kitchen] target_rider_loads = [t["order_count"] for t in timelines if t.get("kitchen") == target] activate_idle_threshold = max(6, idle.get("order_count", 6) + 2) activate_target_threshold = max(8, round(sum(target_rider_loads) / max(1, len(target_rider_loads)) * 1.1)) activate_rule = { "condition": "AND", "rules": [ { "field": f"{primary_kitchen}_order_count", "operator": "<=", "value": activate_idle_threshold, "reason": ( f"{idle['name']} had {idle['order_count']} orders today and was idle " f"{idle['idle_minutes']} min. Dual-kitchen kicks in when their load " f"stays at or below {activate_idle_threshold}." ), }, { "field": f"{target}_order_count", "operator": ">=", "value": activate_target_threshold, "reason": ( f"{target.capitalize()} had enough orders today to justify the detour " f"({sum(target_rider_loads)} total across {len(target_rider_loads)} rider(s)). " f"Activate when {target} load is ≥ {activate_target_threshold}." ), }, ], } return { "action": "dual_kitchen_assignment", "idle_rider_id": idle["userid"], "idle_rider_name": idle["name"], "primary_kitchen": primary_kitchen, "second_kitchen": target, "second_kitchen_dispatch_after": best["arrive_at_kitchen"], "description": description, "fleet_improvement_minutes": best["fleet_improvement_minutes"], "confidence_score": confidence, "efficiency_rating": best["efficiency_rating"], "root_cause": root_cause, "risk_factors": risk_factors, "activate_when": activate_rule, "fleet_balance_assessment": _balance_assessment(fleet_summary), "api_hint": { "endpoint": "/api/v1/optimize", "note": ( f"In the next batch, pre-assign the last " f"{best['feasible_orders_count']} {target} orders to " f"rider {idle['userid']} ({idle['name']}) with a " f"dispatch-after time of {best['arrive_at_kitchen']}." ), }, } def _diagnose_root_cause( idle: dict, timelines: list[dict], fleet_summary: dict, ) -> str: """ Explain WHY this rider finished early by comparing their load against the fleet average and their kitchen's order volume. """ avg_orders = fleet_summary.get("orders_per_rider_avg", 0) rider_orders = idle["order_count"] kitchen = idle["primary_kitchen"] or "their kitchen" name = idle["name"] if avg_orders > 0 and rider_orders < avg_orders * 0.7: shortfall = round(avg_orders - rider_orders, 1) return ( f"{name} received {rider_orders} orders vs fleet average of {avg_orders:.1f} " f"(−{shortfall:.1f}). {kitchen.capitalize()} kitchen generated fewer orders " f"than the fleet needed to keep this rider fully utilised. " f"This is a systematic under-loading of the {kitchen} kitchen in this batch." ) elif rider_orders <= 4: return ( f"{name} had only {rider_orders} orders — a very light load regardless of fleet average. " f"Likely a short-demand window at {kitchen} kitchen. " f"Dual-kitchen assignment is especially effective when primary kitchen load is ≤ 4 orders." ) else: spread = fleet_summary.get("finish_time_spread_minutes", 0) return ( f"{name} is simply faster than peers — finished {idle['idle_minutes']} min ahead " f"despite a normal load of {rider_orders} orders. " f"Fleet finish-time spread is {spread} min, indicating uneven workload distribution." ) def _identify_risks(best: dict, idle_threshold: float) -> list[str]: """ Enumerate operational risks for the recommended substitution. """ risks: list[str] = [] travel_min = best["travel_to_kitchen_minutes"] obligation = best["total_obligation_minutes"] free_window = best["idle_rider"]["free_window_minutes"] confidence = best["confidence_score"] extra_km = best["extra_km_for_idle_rider"] slack_min = free_window - obligation if slack_min < 10: risks.append( f"Tight schedule: only {slack_min} min of slack between idle rider's " f"estimated finish and fleet completion. Any delay (traffic, kitchen wait) " f"would eliminate the benefit." ) if travel_min > 15: risks.append( f"Long commute to target kitchen ({travel_min} min). " f"Kitchen departure time must be precise — a late start erodes time savings." ) if extra_km > 8: risks.append( f"Extra {extra_km} km for the idle rider adds fuel cost and rider fatigue. " f"Verify this is worthwhile if fleet improvement is marginal." ) if confidence < 50: risks.append( f"Low confidence ({confidence}/100): limited slack or few feasible transfers. " f"Consider this as a contingency plan rather than a guaranteed improvement." ) if best.get("feasible_orders_count", 0) < best.get("total_orders_transferred", 1): risks.append( "Not all proposed transfers save time — some orders are included to fill " "the idle rider's route but don't improve individual delivery times." ) if not risks: risks.append("No significant risks identified. Transfer looks operationally sound.") return risks def _balance_assessment(fleet_summary: dict) -> str: """Short human-readable verdict on batch balance quality.""" spread = fleet_summary.get("finish_time_spread_minutes", 0) stdev = fleet_summary.get("load_balance_stdev", 0) util = fleet_summary.get("avg_utilisation_pct") if spread <= 10 and stdev <= 1: verdict = "Excellent — riders finished close together with balanced loads." elif spread <= 20 and stdev <= 2: verdict = "Good — minor imbalance, acceptable for this fleet size." elif spread <= 35: verdict = f"Moderate imbalance — {spread} min spread between earliest and latest finish." else: verdict = ( f"High imbalance — {spread} min spread. Some riders sat idle while others overran. " f"Pre-planning dual-kitchen assignments is strongly recommended." ) if util is not None: verdict += f" Average rider utilisation: {util}% of batch window." return verdict