| """ |
| Advanced Risk Engine - Explains crowd predictions with reasoning and recommendations |
| """ |
|
|
| from typing import Dict, List |
|
|
| RISK_CONFIG = { |
| "Low": { |
| "score": 15, |
| "color": "#10b981", |
| "bg": "#d1fae5", |
| "emoji": "🟢", |
| "badge": "LOW RISK", |
| "advice": "Comfortable travel expected. Enjoy your journey!", |
| }, |
| "Medium": { |
| "score": 50, |
| "color": "#f59e0b", |
| "bg": "#fef3c7", |
| "emoji": "🟡", |
| "badge": "MODERATE", |
| "advice": "Moderate crowd. Book window seat in advance if possible.", |
| }, |
| "Extreme": { |
| "score": 90, |
| "color": "#ef4444", |
| "bg": "#fee2e2", |
| "emoji": "🔴", |
| "badge": "EXTREME", |
| "advice": "Very crowded! Consider alternate timing or route.", |
| }, |
| } |
|
|
| HOUR_LABELS = { |
| 5: "Early Morning", 6: "Early Morning", |
| 7: "Morning Rush", 8: "Peak Morning Rush", 9: "Morning Rush", |
| 10: "Mid Morning", 11: "Late Morning", |
| 12: "Noon", 13: "Afternoon", 14: "Afternoon", 15: "Mid Afternoon", 16: "Late Afternoon", |
| 17: "Evening Rush", 18: "Peak Evening Rush", 19: "Evening Rush", |
| 20: "Late Evening", 21: "Night", 22: "Late Night", |
| } |
|
|
| WESTERN_STATIONS_ORDER = [ |
| "Churchgate", "Marine Lines", "Charni Road", "Grant Road", |
| "Mumbai Central", "Mahalaxmi", "Lower Parel", "Dadar", |
| "Bandra", "Andheri", "Borivali", "Virar" |
| ] |
|
|
| CENTRAL_STATIONS_ORDER = [ |
| "CSMT", "Masjid", "Sandhurst Road", "Byculla", |
| "Dadar", "Kurla", "Ghatkopar", "Thane", "Kalyan" |
| ] |
|
|
| HARBOUR_STATIONS_ORDER = [ |
| "CSMT", "Masjid", "Dockyard Road", "Wadala Road", |
| "Kurla", "Vashi", "Nerul", "Panvel" |
| ] |
|
|
|
|
| def get_risk_info(crowd_level: str) -> Dict: |
| return RISK_CONFIG.get(crowd_level, RISK_CONFIG["Low"]) |
|
|
|
|
| def build_reasoning( |
| line: str, station: str, destination: str, |
| direction: str, hour: int, day_of_week: str, |
| crowd_level: str |
| ) -> List[str]: |
| reasons = [] |
| is_weekday = day_of_week not in ["Saturday", "Sunday"] |
| is_morning_peak = 7 <= hour <= 9 |
| is_evening_peak = 17 <= hour <= 19 |
| is_southbound = direction == "Southbound" |
| hour_label = HOUR_LABELS.get(hour, "") |
|
|
| |
| if is_morning_peak: |
| reasons.append(f"⏰ {hour_label} ({hour}:00) — peak office hours create high demand") |
| elif is_evening_peak: |
| reasons.append(f"⏰ {hour_label} ({hour}:00) — peak return hours, trains are packed") |
| elif hour < 6: |
| reasons.append(f"⏰ Early morning ({hour}:00) — very few commuters at this hour") |
| elif hour > 21: |
| reasons.append(f"⏰ Late night ({hour}:00) — minimal traffic") |
| else: |
| reasons.append(f"⏰ {hour_label} ({hour}:00) — off-peak period") |
|
|
| |
| if is_weekday: |
| reasons.append(f"📅 {day_of_week} is a working day — 20-30% more passengers than weekends") |
| else: |
| reasons.append(f"📅 {day_of_week} is a weekend — lighter traffic overall") |
|
|
| |
| if is_morning_peak and is_southbound and is_weekday: |
| reasons.append("🚆 Southbound during morning rush — commuters flooding into the city") |
| elif is_morning_peak and not is_southbound and is_weekday: |
| reasons.append("🚆 Northbound during morning rush — outbound direction, relatively lighter") |
| elif is_evening_peak and not is_southbound and is_weekday: |
| reasons.append("🚆 Northbound during evening rush — commuters heading home to suburbs") |
| elif is_evening_peak and is_southbound and is_weekday: |
| reasons.append("🚆 Southbound during evening — reverse commute, lighter than northbound") |
|
|
| |
| if line == "Western": |
| reasons.append("🛤️ Western Line — highest ridership line in Mumbai (approx. 4M+ daily)") |
| elif line == "Central": |
| reasons.append("🛤️ Central Line — second busiest, serves eastern suburbs & Thane district") |
| elif line == "Harbour": |
| reasons.append("🛤️ Harbour Line — connects Navi Mumbai, generally less crowded") |
|
|
| return reasons |
|
|
|
|
| def get_best_hours( |
| model, le_station, le_dest, |
| line: str, station: str, destination: str, direction: str, day: str, |
| predict_fn |
| ) -> List[Dict]: |
| results = [] |
| for h in range(5, 23): |
| pred = predict_fn(model, le_station, le_dest, line, station, destination, direction, h, day) |
| results.append({ |
| "hour": h, |
| "label": f"{h:02d}:00", |
| "crowd_level": pred["crowd_level"], |
| "risk_score": RISK_CONFIG[pred["crowd_level"]]["score"], |
| "confidence": pred["confidence"] |
| }) |
| return results |
|
|
|
|
| def get_travel_tip(crowd_level: str, direction: str, hour: int) -> str: |
| tips = { |
| "Low": [ |
| "Perfect time to travel! You'll likely get a seat.", |
| "Great window! Trains are comfortable right now.", |
| "Off-peak bliss — smooth, uncrowded journey ahead.", |
| ], |
| "Medium": [ |
| "Manageable crowd — you may have to stand for a while.", |
| "Arrive 5 mins early to board comfortably.", |
| "It's busy but not terrible. Avoid the first bogie.", |
| ], |
| "Extreme": [ |
| "Packed trains! If flexible, shift by 30–60 minutes.", |
| "Consider the Ladies' Special or next train.", |
| "Carry water, maintain patience — it's Mumbai rush!", |
| ] |
| } |
| import random |
| opts = tips.get(crowd_level, tips["Medium"]) |
| return opts[hour % len(opts)] |
|
|
