151 lines
5.4 KiB
Python
151 lines
5.4 KiB
Python
from datetime import datetime, time, timedelta
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from app.database import get_session
|
|
from app.crud import get_sessions
|
|
from app.schemas import DailySummaryResponse, TimetableResponse, WakeupBedtimeResponse
|
|
from app.utils.time_utils import get_day_night_splits
|
|
|
|
router = APIRouter(prefix="/api/stats", tags=["stats"])
|
|
|
|
|
|
@router.get("/daily-summary", response_model=DailySummaryResponse)
|
|
async def get_daily_summary(
|
|
from_date: datetime,
|
|
to_date: datetime,
|
|
day_start: str = Query("07:00"),
|
|
day_end: str = Query("20:00"),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
day_start_time = time.fromisoformat(day_start)
|
|
day_end_time = time.fromisoformat(day_end)
|
|
except ValueError:
|
|
return {"days": []}
|
|
|
|
sessions, _ = await get_sessions(session, from_date=from_date, to_date=to_date)
|
|
summary_map = {}
|
|
|
|
for s in sessions:
|
|
if not s.end_time:
|
|
continue
|
|
segments = get_day_night_splits(
|
|
s.start_time, s.end_time, day_start_time, day_end_time
|
|
)
|
|
for seg in segments:
|
|
date_str = seg["start"].strftime("%Y-%m-%d")
|
|
if date_str not in summary_map:
|
|
summary_map[date_str] = {"day": 0.0, "night": 0.0}
|
|
duration = (seg["end"] - seg["start"]).total_seconds() / 60.0
|
|
if seg["type"] == "day":
|
|
summary_map[date_str]["day"] += duration
|
|
else:
|
|
summary_map[date_str]["night"] += duration
|
|
|
|
days = []
|
|
for date_str in sorted(summary_map.keys()):
|
|
data = summary_map[date_str]
|
|
days.append(
|
|
{
|
|
"date": date_str,
|
|
"total_minutes": int(round(data["day"] + data["night"])),
|
|
"day_minutes": int(round(data["day"])),
|
|
"night_minutes": int(round(data["night"])),
|
|
}
|
|
)
|
|
return {"days": days}
|
|
|
|
|
|
@router.get("/timetable", response_model=TimetableResponse)
|
|
async def get_timetable(
|
|
from_date: datetime,
|
|
to_date: datetime,
|
|
day_start: str = Query("07:00"),
|
|
day_end: str = Query("20:00"),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
sessions, _ = await get_sessions(session, from_date=from_date, to_date=to_date)
|
|
slots = []
|
|
current_date = from_date.date()
|
|
while current_date <= to_date.date():
|
|
date_str = current_date.strftime("%Y-%m-%d")
|
|
for i in range(96):
|
|
slot_start_time = datetime.combine(current_date, time(0, 0)) + timedelta(
|
|
minutes=i * 15
|
|
)
|
|
slot_end_time = slot_start_time + timedelta(minutes=15)
|
|
slot_start_str = slot_start_time.strftime("%H:%M")
|
|
slot_end_str = slot_end_time.strftime("%H:%M")
|
|
is_sleeping = False
|
|
for s in sessions:
|
|
if s.start_time <= slot_end_time and (
|
|
s.end_time is None or s.end_time >= slot_start_time
|
|
):
|
|
is_sleeping = True
|
|
break
|
|
slots.append(
|
|
{
|
|
"date": date_str,
|
|
"slot_start": slot_start_str,
|
|
"slot_end": slot_end_str,
|
|
"sleeping": is_sleeping,
|
|
}
|
|
)
|
|
current_date += timedelta(days=1)
|
|
return {"slots": slots}
|
|
|
|
|
|
@router.get("/wake-sleep-times", response_model=WakeupBedtimeResponse)
|
|
async def get_wake_sleep_times(
|
|
from_date: datetime,
|
|
to_date: datetime,
|
|
wakeup_ref: str = Query("04:00"),
|
|
bedtime_ref: str = Query("19:00"),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
w_ref_time = time.fromisoformat(wakeup_ref)
|
|
b_ref_time = time.fromisoformat(bedtime_ref)
|
|
except ValueError:
|
|
return {"days": []}
|
|
|
|
# Fetch sessions starting from one day before to catch night sleep ending on from_date
|
|
adj_from_date = from_date - timedelta(days=1)
|
|
sessions, _ = await get_sessions(session, from_date=adj_from_date, to_date=to_date)
|
|
days_data = []
|
|
current_date = from_date.date()
|
|
while current_date <= to_date.date():
|
|
date_str = current_date.strftime("%Y-%m-%d")
|
|
wakeup_min = None
|
|
bedtime_min = None
|
|
|
|
midnight_today = datetime.combine(current_date, time(0, 0))
|
|
|
|
night_session = None
|
|
for s in sessions:
|
|
# A session that runs over midnight into current_date
|
|
if s.start_time < midnight_today and s.end_time and s.end_time >= midnight_today:
|
|
night_session = s
|
|
break
|
|
|
|
if night_session:
|
|
# Bedtime offset from b_ref_time on the day it started
|
|
s_start_date = night_session.start_time.date()
|
|
ref_datetime_bedtime = datetime.combine(s_start_date, b_ref_time)
|
|
bedtime_min = int((night_session.start_time - ref_datetime_bedtime).total_seconds() / 60.0)
|
|
|
|
# Wakeup offset from w_ref_time on the current day
|
|
ref_datetime_wakeup = datetime.combine(current_date, w_ref_time)
|
|
wakeup_min = int((night_session.end_time - ref_datetime_wakeup).total_seconds() / 60.0)
|
|
|
|
days_data.append(
|
|
{
|
|
"date": date_str,
|
|
"wakeup_minutes": wakeup_min,
|
|
"bedtime_minutes": bedtime_min,
|
|
}
|
|
)
|
|
current_date += timedelta(days=1)
|
|
return {"days": days_data}
|