98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
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) or {}
|
|
|
|
|
|
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
|