Frank Schwenk
119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import aiosqlite
|
|
|
|
from server.config_loader import MedsConfig, Slot, load_meds_config, slot_datetime, today_str
|
|
from server.db import utc_now_iso
|
|
|
|
|
|
async def get_log_for_day(db: aiosqlite.Connection, day: str) -> dict[str, dict[str, Any]]:
|
|
cur = await db.execute(
|
|
"SELECT slot_id, status, logged_at, source FROM intake_log WHERE day = ? ORDER BY logged_at DESC",
|
|
(day,),
|
|
)
|
|
rows = await cur.fetchall()
|
|
await cur.close()
|
|
result: dict[str, dict[str, Any]] = {}
|
|
for slot_id, status, logged_at, source in rows:
|
|
if slot_id not in result:
|
|
result[slot_id] = {"status": status, "logged_at": logged_at, "source": source}
|
|
return result
|
|
|
|
|
|
async def get_snooze_for_day(db: aiosqlite.Connection, day: str) -> dict[str, str]:
|
|
cur = await db.execute("SELECT slot_id, snooze_until FROM snooze WHERE day = ?", (day,))
|
|
rows = await cur.fetchall()
|
|
await cur.close()
|
|
return {slot_id: snooze_until for slot_id, snooze_until in rows}
|
|
|
|
|
|
def compute_status(
|
|
slot: Slot,
|
|
now: datetime,
|
|
tz: ZoneInfo,
|
|
log_entry: dict[str, Any] | None,
|
|
snooze_until: str | None,
|
|
) -> str:
|
|
if log_entry:
|
|
return log_entry["status"]
|
|
|
|
slot_dt = slot_datetime(now, slot, tz)
|
|
window_end = slot_dt + timedelta(minutes=slot.reminder_window_minutes)
|
|
|
|
if snooze_until:
|
|
try:
|
|
snooze_dt = datetime.fromisoformat(snooze_until)
|
|
if now < snooze_dt:
|
|
return "snoozed"
|
|
except ValueError:
|
|
pass
|
|
|
|
if now < slot_dt:
|
|
return "upcoming"
|
|
|
|
if now <= window_end:
|
|
return "pending"
|
|
|
|
return "overdue"
|
|
|
|
|
|
async def build_today(db: aiosqlite.Connection, config: MedsConfig | None = None) -> dict[str, Any]:
|
|
config = config or load_meds_config()
|
|
tz = config.timezone
|
|
now = datetime.now(tz)
|
|
day = today_str(tz)
|
|
logs = await get_log_for_day(db, day)
|
|
snoozes = await get_snooze_for_day(db, day)
|
|
|
|
slots_out: list[dict[str, Any]] = []
|
|
for slot in config.slots:
|
|
log_entry = logs.get(slot.id)
|
|
snooze_until = snoozes.get(slot.id)
|
|
status = compute_status(slot, now, tz, log_entry, snooze_until)
|
|
slot_dt = slot_datetime(now, slot, tz)
|
|
window_end = slot_dt + timedelta(minutes=slot.reminder_window_minutes)
|
|
slots_out.append(
|
|
{
|
|
"id": slot.id,
|
|
"time": slot.time.strftime("%H:%M"),
|
|
"label": slot.label,
|
|
"meds": [{"name": m.name, "dose": m.dose} for m in slot.meds],
|
|
"status": status,
|
|
"logged_at": log_entry["logged_at"] if log_entry else None,
|
|
"window_end": window_end.isoformat(),
|
|
"snooze_until": snooze_until,
|
|
}
|
|
)
|
|
|
|
return {"day": day, "timezone": str(tz), "now": now.isoformat(), "slots": slots_out}
|
|
|
|
|
|
async def mark_missed_for_overdue(db: aiosqlite.Connection, config: MedsConfig | None = None) -> list[str]:
|
|
from server.db import new_id
|
|
|
|
config = config or load_meds_config()
|
|
tz = config.timezone
|
|
now = datetime.now(tz)
|
|
day = today_str(tz)
|
|
logs = await get_log_for_day(db, day)
|
|
marked: list[str] = []
|
|
|
|
for slot in config.slots:
|
|
if slot.id in logs:
|
|
continue
|
|
status = compute_status(slot, now, tz, None, None)
|
|
if status == "overdue":
|
|
await db.execute(
|
|
"INSERT INTO intake_log (id, slot_id, day, status, logged_at, source) VALUES (?, ?, ?, ?, ?, ?)",
|
|
(new_id(), slot.id, day, "missed", utc_now_iso(), "system"),
|
|
)
|
|
marked.append(slot.id)
|
|
|
|
if marked:
|
|
await db.commit()
|
|
return marked
|