217 lines
8.6 KiB
Python
217 lines
8.6 KiB
Python
from datetime import datetime, time, timedelta, timezone
|
|
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, get_sleep_counts_by_start_date
|
|
from app.schemas import DailySummaryResponse, TimetableResponse, WakeupBedtimeResponse, SummaryResponse
|
|
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"),
|
|
timezone_offset: int = Query(0),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
if to_date.hour == 0 and to_date.minute == 0:
|
|
to_date = to_date.replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
|
|
try:
|
|
day_start_time = time.fromisoformat(day_start)
|
|
day_end_time = time.fromisoformat(day_end)
|
|
except ValueError:
|
|
return {"days": []}
|
|
|
|
utc_from_date = from_date + timedelta(minutes=timezone_offset)
|
|
utc_to_date = to_date + timedelta(minutes=timezone_offset)
|
|
adj_utc_from_date = utc_from_date - timedelta(days=1)
|
|
sessions, _ = await get_sessions(session, from_date=adj_utc_from_date, to_date=utc_to_date)
|
|
summary_map = {}
|
|
|
|
for s in sessions:
|
|
if not s.end_time:
|
|
continue
|
|
local_start = s.start_time - timedelta(minutes=timezone_offset)
|
|
local_end = s.end_time - timedelta(minutes=timezone_offset)
|
|
segments = get_day_night_splits(
|
|
local_start, local_end, day_start_time, day_end_time
|
|
)
|
|
for seg in segments:
|
|
if not (from_date.date() <= seg["start"].date() <= to_date.date()):
|
|
continue
|
|
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"),
|
|
timezone_offset: int = Query(0),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
if to_date.hour == 0 and to_date.minute == 0:
|
|
to_date = to_date.replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
|
|
utc_from_date = from_date + timedelta(minutes=timezone_offset)
|
|
utc_to_date = to_date + timedelta(minutes=timezone_offset)
|
|
adj_utc_from_date = utc_from_date - timedelta(days=1)
|
|
sessions, _ = await get_sessions(session, from_date=adj_utc_from_date, to_date=utc_to_date)
|
|
slots = []
|
|
now_utc = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
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_local = datetime.combine(current_date, time(0, 0)) + timedelta(
|
|
minutes=i * 15
|
|
)
|
|
slot_end_local = slot_start_local + timedelta(minutes=15)
|
|
slot_start_str = slot_start_local.strftime("%H:%M")
|
|
slot_end_str = slot_end_local.strftime("%H:%M")
|
|
|
|
# Convert local slots to UTC for comparison with DB sessions
|
|
slot_start_utc = slot_start_local + timedelta(minutes=timezone_offset)
|
|
slot_end_utc = slot_end_local + timedelta(minutes=timezone_offset)
|
|
|
|
is_sleeping = False
|
|
for s in sessions:
|
|
effective_end_time_utc = s.end_time if s.end_time is not None else now_utc
|
|
if s.start_time <= slot_end_utc and effective_end_time_utc >= slot_start_utc:
|
|
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("/summary", response_model=SummaryResponse)
|
|
async def get_summary(
|
|
from_date: datetime,
|
|
to_date: datetime,
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
# Period Stats
|
|
period_counts = await get_sleep_counts_by_start_date(session, from_date=from_date, to_date=to_date)
|
|
total_sleeps = sum(count for _, count in period_counts)
|
|
num_days = (to_date.date() - from_date.date()).days + 1
|
|
avg_sleeps_per_day = total_sleeps / num_days if num_days > 0 else 0.0
|
|
|
|
# All-Time Stats
|
|
all_counts = await get_sleep_counts_by_start_date(session)
|
|
total_sleeps_all_time = sum(count for _, count in all_counts)
|
|
avg_sleeps_per_day_all_time = 0.0
|
|
if all_counts:
|
|
# counts are ordered by desc date (from crud.py)
|
|
# so last one is earliest.
|
|
earliest_date_str = all_counts[-1][0]
|
|
# SQLite date() returns YYYY-MM-DD string
|
|
earliest_date = datetime.strptime(earliest_date_str, "%Y-%m-%d").date()
|
|
today = datetime.now(timezone.utc).date()
|
|
days_all_time = (today - earliest_date).days + 1
|
|
avg_sleeps_per_day_all_time = total_sleeps_all_time / days_all_time if days_all_time > 0 else 0.0
|
|
|
|
return {
|
|
"total_sleeps": total_sleeps,
|
|
"avg_sleeps_per_day": avg_sleeps_per_day,
|
|
"total_sleeps_all_time": total_sleeps_all_time,
|
|
"avg_sleeps_per_day_all_time": avg_sleeps_per_day_all_time,
|
|
}
|
|
|
|
|
|
@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"),
|
|
timezone_offset: int = Query(0),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
if to_date.hour == 0 and to_date.minute == 0:
|
|
to_date = to_date.replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
|
|
try:
|
|
w_ref_time = time.fromisoformat(wakeup_ref)
|
|
b_ref_time = time.fromisoformat(bedtime_ref)
|
|
except ValueError:
|
|
return {"days": []}
|
|
|
|
utc_from_date = from_date + timedelta(minutes=timezone_offset)
|
|
utc_to_date = to_date + timedelta(minutes=timezone_offset)
|
|
adj_utc_from_date = utc_from_date - timedelta(days=1)
|
|
sessions, _ = await get_sessions(session, from_date=adj_utc_from_date, to_date=utc_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:
|
|
# Convert to local for comparison
|
|
s_start_local = s.start_time - timedelta(minutes=timezone_offset)
|
|
s_end_local = s.end_time - timedelta(minutes=timezone_offset) if s.end_time else None
|
|
|
|
if s_start_local < midnight_today and s_end_local and s_end_local >= midnight_today:
|
|
night_session = (s, s_start_local, s_end_local)
|
|
break
|
|
|
|
if night_session:
|
|
s, s_start_local, s_end_local = night_session
|
|
# Bedtime offset from b_ref_time on the day it started (local)
|
|
s_start_date = s_start_local.date()
|
|
ref_datetime_bedtime = datetime.combine(s_start_date, b_ref_time)
|
|
bedtime_min = int((s_start_local - ref_datetime_bedtime).total_seconds() / 60.0)
|
|
|
|
# Wakeup offset from w_ref_time on the current day (local)
|
|
ref_datetime_wakeup = datetime.combine(current_date, w_ref_time)
|
|
wakeup_min = int((s_end_local - 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}
|