Upload app.py

app.py

@@ -0,0 +1,322 @@
+from datetime import datetime # type: ignore
+import sys # type: ignore
+from pathlib import Path
+
+# Add parent directory to path to allow imports from api/
+sys.path.insert(0, str(Path(__file__).parent.parent))
+# Add src directory to path to allow imports from same directory
+sys.path.insert(0, str(Path(__file__).parent))
+
+from api.bus_cache import AsyncBusCache # type: ignore
+from api.utils import hms_to_seconds, get_service_day_start_ts, translate_occupancy # type: ignore
+from db_manager import init_db # type: ignore
+from dotenv import load_dotenv # type: ignore
+from fastapi import FastAPI, HTTPException # type: ignore
+from fastapi.middleware.cors import CORSMiddleware # type: ignore
+
+load_dotenv()
+
+ttc_cache = AsyncBusCache(ttl=20)
+
+# Initialize database connection globally
+db = init_db()
+
+app = FastAPI(title="WheresMyBus v2.0 API")
+
+# Setup CORS for your React frontend
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"], # In production, use your actual React URL
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def health_check():
+    """Simple health check endpoint"""
+    return "backend is running"
+
+@app.get("/api/vehicles")
+async def get_vehicles():
+    data = await ttc_cache.get_data()
+    vehicles = data.get("vehicles", [])
+    return {
+        "status": "success",
+        "count": len(vehicles),
+        "vehicles": vehicles
+    }
+
+@app.get("/api/routes")
+async def get_all_routes():
+    """
+    Returns a complete list of TTC routes with their display names and colors.
+    """
+    try:
+        # Run the query against DuckDB
+        # We handle missing colors by providing defaults (TTC Red: #FF0000)
+        query = """
+            SELECT 
+                route_id, 
+                route_short_name, 
+                route_long_name, 
+                COALESCE(route_color, 'FF0000') as route_color,
+                COALESCE(route_text_color, 'FFFFFF') as route_text_color
+            FROM routes 
+            ORDER BY 
+                CASE 
+                    WHEN CAST(route_short_name AS VARCHAR) ~ '^[0-9]+$' THEN CAST(route_short_name AS INTEGER) 
+                    ELSE 999 
+                END, 
+                route_short_name;
+        """
+        
+        results = db.execute(query).fetchall()
+        
+        # Convert to a clean list of dictionaries
+        route_list = [
+            {
+                "id": r[0],
+                "number": r[1],
+                "name": r[2],
+                "color": f"#{r[3]}",
+                "text_color": f"#{r[4]}"
+            }
+            for r in results
+        ]
+        
+        return {
+            "status": "success",
+            "count": len(route_list),
+            "routes": route_list
+        }
+        
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+@app.get("/api/routes/{route_id}")
+async def get_route_view(route_id: str):
+    data = await ttc_cache.get_data()
+    all_buses = data.get("vehicles", [])
+    route_buses = [v for v in all_buses if v['route'] == route_id]
+    
+    if not route_buses:
+        return {"route": route_id, "vehicles": []}
+
+    # IMPORTANT: Cast Trip IDs to strings to ensure they match the DB
+    trip_ids = [str(v['trip_id']) for v in route_buses]
+    placeholders = ','.join(['?'] * len(trip_ids))
+    
+    # We use CAST(? AS VARCHAR) to force DuckDB to match strings to strings
+    query = f"""
+        SELECT 
+            CAST(st.trip_id AS VARCHAR), 
+            CAST(st.stop_id AS VARCHAR), 
+            st.arrival_time as scheduled_time, 
+            t.trip_headsign
+        FROM stop_times st
+        JOIN trips t ON CAST(st.trip_id AS VARCHAR) = CAST(t.trip_id AS VARCHAR)
+        WHERE CAST(st.trip_id AS VARCHAR) IN ({placeholders})
+    """
+    db_rows = db.execute(query, trip_ids).fetchall()
+    
+    # Check if we got ANYTHING back from the DB
+    if not db_rows:
+        print(f"DEBUG: No matches in DB for Trip IDs: {trip_ids[:3]}")
+
+    schedule_map = {(r[0], r[1]): r[2] for r in db_rows}
+    name_map = {r[0]: r[3] for r in db_rows}
+
+    service_day_ts = get_service_day_start_ts()
+    enriched = []
+
+    for bus in route_buses:
+        # Default delay is 0 if no prediction exists
+        raw_delay_mins = 0
+        
+        pred_time = bus.get('predicted_time')
+        stop_id = bus.get('next_stop_id')
+        
+        if pred_time and stop_id:
+            sched_hms = schedule_map.get((str(bus['trip_id']), str(stop_id)))
+            if sched_hms:
+                # DELAY = SCHEDULED - PREDICTED (negative = late, positive = early)
+                plan_ts = service_day_ts + hms_to_seconds(sched_hms)
+                raw_delay_mins = round((plan_ts - pred_time) / 60)
+
+        enriched.append({
+            "number": bus['id'],
+            "name": name_map.get(str(bus['trip_id']), "Not in Schedule"), # This is the destination
+            "location": {"lat": bus['lat'], "lon": bus['lon']},
+            "delay_mins": raw_delay_mins, # Actual integer: 5 = 5m late, -2 = 2m early
+            "fullness": translate_occupancy(bus['occupancy'])
+        })
+
+    return {
+        "route": route_id,
+        "count": len(enriched),
+        "vehicles": enriched
+    }
+
+@app.get("/api/vehicles/{vehicle_id}")
+async def get_vehicle_view(vehicle_id: str):
+    # 1. Pull latest from cache
+    data = await ttc_cache.get_data()
+    vehicles = data.get("vehicles", [])
+    
+    # 2. Find this specific bus in the list
+    bus = next((v for v in vehicles if str(v['id']) == vehicle_id), None)
+    
+    if not bus:
+        raise HTTPException(status_code=404, detail="Vehicle not active or not found")
+
+    trip_id = str(bus['trip_id'])
+    next_stop_id = bus.get('next_stop_id')
+    predicted_time = bus.get('predicted_time')
+
+    # 3. Handshake with Database (Cast to VARCHAR to avoid type errors)
+    # We get the destination name and the specific scheduled arrival time
+    destination = "Not in Schedule"
+    delay_mins = 0
+    
+    if next_stop_id:
+        query = """
+            SELECT 
+                t.trip_headsign, 
+                st.arrival_time as scheduled_time
+            FROM trips t
+            JOIN stop_times st ON CAST(t.trip_id AS VARCHAR) = CAST(st.trip_id AS VARCHAR)
+            WHERE CAST(t.trip_id AS VARCHAR) = ? 
+              AND CAST(st.stop_id AS VARCHAR) = ?
+            LIMIT 1
+        """
+        row = db.execute(query, [trip_id, str(next_stop_id)]).fetchone()
+        
+        if row:
+            destination = row[0]
+            scheduled_hms = row[1]
+            
+            # DELAY = SCHEDULED - PREDICTED (negative = late, positive = early)
+            if predicted_time:
+                service_day_ts = get_service_day_start_ts()
+                plan_ts = service_day_ts + hms_to_seconds(scheduled_hms)
+                delay_mins = round((plan_ts - predicted_time) / 60)
+    else:
+        # If no next_stop_id, try to get destination from trip_id only
+        query = """
+            SELECT trip_headsign
+            FROM trips
+            WHERE CAST(trip_id AS VARCHAR) = ?
+            LIMIT 1
+        """
+        row = db.execute(query, [trip_id]).fetchone()
+        if row:
+            destination = row[0]
+
+    return {
+        "vehicle_number": vehicle_id,
+        "route_id": bus['route'],
+        "name": destination,
+        "location": {
+            "lat": bus['lat'],
+            "lon": bus['lon']
+        },
+        "delay_mins": delay_mins,
+        "fullness": translate_occupancy(bus['occupancy']),
+        "trip_id": trip_id
+    }
+
+@app.get("/api/stop/{stop_code}")
+async def get_stop_view(stop_code: str):
+    # 1. Translate Pole Number to Database ID
+    stop_info = db.execute("SELECT stop_id, stop_name FROM stops WHERE CAST(stop_code AS VARCHAR) = ? LIMIT 1", [str(stop_code)]).fetchone()
+    if not stop_info:
+        return {"error": "Stop code not found"}
+    
+    target_id = str(stop_info[0])
+    stop_name = stop_info[1]
+
+    # 2. Get the Cache structure (dict with vehicles, predictions, alerts)
+    cached_data = await ttc_cache.get_data()
+    vehicles_list = cached_data.get("vehicles", [])
+    predictions = cached_data.get("predictions", {})
+    
+    # Build vehicles map for quick lookup
+    vehicles = {str(v['trip_id']): v for v in vehicles_list}
+
+    now = datetime.now().timestamp()
+    two_hours_out = now + 7200
+    arrivals = []
+
+    # 3. Search the FULL itineraries for our target_id
+    for trip_id, itinerary in predictions.items():
+        if target_id in itinerary:
+            pred_time = itinerary[target_id]
+            
+            # Only include if the bus hasn't passed the stop yet and is within 2 hours
+            if now <= pred_time <= two_hours_out:
+                
+                # 4. Handshake with DB for destination and schedule
+                query = """
+                    SELECT t.trip_headsign, st.arrival_time as scheduled_time, r.route_short_name
+                    FROM trips t
+                    JOIN stop_times st ON CAST(t.trip_id AS VARCHAR) = CAST(st.trip_id AS VARCHAR)
+                    JOIN routes r ON t.route_id = r.route_id
+                    WHERE CAST(t.trip_id AS VARCHAR) = ? AND CAST(st.stop_id AS VARCHAR) = ?
+                    LIMIT 1
+                """
+                row = db.execute(query, [trip_id, target_id]).fetchone()
+
+                if row:
+                    # Find the actual bus for fullness (if it's on the road)
+                    bus = vehicles.get(trip_id)
+                    
+                    # DELAY = SCHEDULED - PREDICTED (negative = late, positive = early)
+                    plan_ts = get_service_day_start_ts() + hms_to_seconds(row[1])
+                    
+                    arrivals.append({
+                        "route": row[2],
+                        "destination": row[0],
+                        "eta_mins": round((pred_time - now) / 60),
+                        "delay_mins": round((plan_ts - pred_time) / 60),
+                        "fullness": translate_occupancy(bus['occupancy']) if bus else "Unknown",
+                        "vehicle_id": bus['id'] if bus else "In Transit"
+                    })
+
+    arrivals.sort(key=lambda x: x['eta_mins'])
+    return {"stop_name": stop_name, "stop_code": stop_code, "arrivals": arrivals}
+
+@app.get("/api/alerts")
+async def get_all_alerts():
+    """
+    Returns every active service alert for the entire TTC network.
+    """
+    data = await ttc_cache.get_data()
+    return {
+        "timestamp": datetime.now().timestamp(),
+        "count": len(data["alerts"]),
+        "alerts": data["alerts"]
+    }
+
+@app.get("/api/alerts/{route_id}")
+async def get_alerts_for_route(route_id: str):
+    data = await ttc_cache.get_data()
+    alerts = data.get("alerts", {})
+    route_alerts = alerts.get(route_id, [])
+    
+    if not route_alerts:
+        return {
+            "route_id": route_id,
+            "count": 0,
+            "alerts": "No alerts"
+        }
+    
+    return {
+        "route_id": route_id,
+        "count": len(route_alerts),
+        "alerts": route_alerts
+    }
+
+if __name__ == "__main__":
+    import uvicorn # type: ignore
+    # Start the server
+    uvicorn.run(app, host="0.0.0.0", port=7860)