from datetime import date, datetime, time, timedelta from typing import Optional, List from fastapi import APIRouter, HTTPException from pydantic import BaseModel import api_database router = APIRouter(prefix="/dashboard/{station_id}", tags=["dashboard", "local"]) def validate_daterange(start_date: date, end_date: date): days_requested = (end_date - start_date).days if days_requested < 0: raise HTTPException(status_code=400, detail="Requested date-range is negative") class StationDashboard(BaseModel): id: Optional[int] commonName: Optional[str] lat: Optional[float] lon: Optional[float] maxEndDate: Optional[date] maxStartDate: Optional[date] @router.get("/", response_model=StationDashboard) def get_general_dashboard(station_id: int): return api_database.get_dashboard(station_id)[0] class StationDashboardTopStationsEntry(BaseModel): stationName: str stationId: int stationLat: float stationLon: float number: int avgDuration: int @router.get("/to", response_model=List[StationDashboardTopStationsEntry]) def get_to_dashboard_for_station(station_id: int, start_date: date, end_date: date): validate_daterange(start_date, end_date) return api_database.get_dashboard_to(station_id, start_date, end_date) @router.get("/from", response_model=List[StationDashboardTopStationsEntry]) def get_from_dashboard_for_station(station_id: int, start_date: date, end_date: date): validate_daterange(start_date, end_date) return api_database.get_dashboard_from(station_id, start_date, end_date) class StationDashboardDurationGroup(BaseModel): number: int minutesGroup: str @router.get("/duration", response_model=List[StationDashboardDurationGroup]) def get_duration_dashboard_for_station(station_id: int, start_date: date, end_date: date): validate_daterange(start_date, end_date) db_data = api_database.get_dashboard_duration(station_id, start_date, end_date) ret_val = [] for group in ['0-5', '5-15', '15-30', '30-45', '45+']: curr_minute_group = list(filter(lambda x: x['minutesGroup'] == group, db_data)) if curr_minute_group: item = curr_minute_group.pop() ret_val.append({ 'number': item['number'], 'minutesGroup': item['minutesGroup'] }) else: ret_val.append({'number': 0, 'minutesGroup': group}) return ret_val class StationDashboardTimeGroup(BaseModel): timeFrame: str number: int avgDuration: int @router.get("/time", response_model=List[StationDashboardTimeGroup]) def get_time_dashboard_for_station(station_id: int, start_date: date, end_date: date): validate_daterange(start_date, end_date) db_data = api_database.get_dashboard_time(station_id, start_date, end_date) ret_val = [] init_date = datetime.combine(date.today(), time(0, 0)) for i in range(144): curr_interval = (init_date + timedelta(minutes=10 * i)).strftime("%H:%M") search_interval = list(filter(lambda x: x['timeFrame'] == curr_interval, db_data)) if search_interval: ret_val.append(search_interval.pop()) else: ret_val.append({'timeFrame': curr_interval, 'number': 0, 'avgDuration': 0}) return ret_val