69 lines
3.0 KiB
Python
69 lines
3.0 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone
|
|
from typing import AsyncGenerator
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from app.database import get_session
|
|
from app.crud import get_sessions, get_all_time_stats
|
|
from sqlmodel import select, func
|
|
from app.models import SleepSession
|
|
|
|
async def get_session_context() -> AsyncGenerator[AsyncSession, None]:
|
|
async for session in get_session():
|
|
yield session
|
|
|
|
async def verify_stats():
|
|
# Use the async generator correctly
|
|
gen = get_session_context()
|
|
session: AsyncSession = await gen.__anext__()
|
|
|
|
try:
|
|
import datetime as dt
|
|
to_date = datetime.now(timezone.utc).replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
from_date = (to_date - dt.timedelta(days=6)).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
print(f"Verifying range: {from_date} to {to_date}")
|
|
|
|
# 1. Verify 'in period' stats via get_sessions (as used in router)
|
|
sessions, _ = await get_sessions(session, from_date=from_date, to_date=to_date)
|
|
total_sleeps = len(sessions)
|
|
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
|
|
|
|
print(f"In-period: Total={total_sleeps}, Days={num_days}, Avg={avg_sleeps_per_day}")
|
|
|
|
# 2. Verify 'all time' stats via get_all_time_stats
|
|
total_sleeps_all_time, avg_sleeps_per_day_all_time = await get_all_time_stats(session)
|
|
print(f"All-time: Total={total_sleeps_all_time}, Avg={avg_sleeps_per_day_all_time}")
|
|
|
|
# 3. Direct DB check for raw session count
|
|
statement = select(func.count()).select_from(SleepSession)
|
|
raw_total_result = await session.exec(statement)
|
|
db_total = raw_total_result.one()
|
|
print(f"DB Raw Total: {db_total}")
|
|
|
|
# 4. Direct DB check for first session to see if it matches
|
|
first_stmt = select(SleepSession.start_time).order_by(SleepSession.start_time.asc()).limit(1)
|
|
first_result = await session.exec(first_stmt)
|
|
first_time = first_result.one()
|
|
print(f"First Session in DB: {first_time}")
|
|
|
|
if total_sleeps > 0 and total_sleeps_all_time == total_sleeps:
|
|
print("INFO: All sessions are within the period.")
|
|
elif total_sleeps < total_sleeps_all_time:
|
|
print(f"INFO: Found {total_sleeps_all_time - total_sleeps} sessions outside the period.")
|
|
|
|
if avg_sleeps_per_day > avg_sleeps_per_day_all_time and total_sleeps_all_time > 0:
|
|
print("!!! ALERT: In-period average is HIGHER than All-time average. This contradicts the user's observation if all sessions are in period.")
|
|
else:
|
|
print("SUCCESS: Average calculation seems logically consistent (in-period <= all-time).")
|
|
|
|
finally:
|
|
# Cleanup the generator
|
|
try:
|
|
await gen.__anext__()
|
|
except StopAsyncIteration:
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(verify_stats())
|