244 lines
7.1 KiB
Python
244 lines
7.1 KiB
Python
from datetime import date as date_type
|
|
from datetime import timedelta
|
|
from typing import Optional
|
|
|
|
from django.db.models import Avg, Max, Min
|
|
from django.http import HttpRequest
|
|
from django.shortcuts import get_object_or_404
|
|
from django.utils import timezone
|
|
from ninja import Router, Schema
|
|
from ninja.errors import HttpError
|
|
|
|
from accounts.auth import AUTH
|
|
from tracker.models import DoseLog, Medication, PulseReading, ScheduledDose
|
|
|
|
router = Router(auth=AUTH)
|
|
|
|
|
|
# --- Schemas ---------------------------------------------------------------
|
|
|
|
|
|
class ScheduledDoseOut(Schema):
|
|
id: int
|
|
medication_id: int
|
|
medication_name: str
|
|
amount: str
|
|
time_of_day: int
|
|
time_of_day_label: str
|
|
|
|
|
|
class MedicationOut(Schema):
|
|
id: int
|
|
name: str
|
|
is_active: bool
|
|
notes: str
|
|
scheduled_doses: list[ScheduledDoseOut]
|
|
|
|
|
|
class DoseStatusOut(ScheduledDoseOut):
|
|
"""A scheduled dose plus whether it was given on the day being looked at."""
|
|
|
|
taken: Optional[bool] # None = not recorded yet
|
|
notes: str
|
|
|
|
|
|
class PulseOut(Schema):
|
|
date: date_type
|
|
bpm: int
|
|
notes: str
|
|
|
|
|
|
class DayOut(Schema):
|
|
date: date_type
|
|
doses: list[DoseStatusOut]
|
|
pulse: Optional[PulseOut]
|
|
|
|
|
|
class DoseLogIn(Schema):
|
|
scheduled_dose_id: int
|
|
date: Optional[date_type] = None
|
|
taken: bool = True
|
|
notes: str = ""
|
|
|
|
|
|
class PulseIn(Schema):
|
|
bpm: int
|
|
date: Optional[date_type] = None
|
|
notes: str = ""
|
|
|
|
|
|
class PulseStatsOut(Schema):
|
|
count: int
|
|
average: Optional[float]
|
|
minimum: Optional[int]
|
|
maximum: Optional[int]
|
|
trend_bpm_per_day: Optional[float]
|
|
first_date: Optional[date_type]
|
|
last_date: Optional[date_type]
|
|
|
|
|
|
# --- Helpers ---------------------------------------------------------------
|
|
|
|
|
|
def _serialize_scheduled_dose(dose: ScheduledDose) -> dict:
|
|
return {
|
|
"id": dose.id,
|
|
"medication_id": dose.medication_id,
|
|
"medication_name": dose.medication.name,
|
|
"amount": dose.amount,
|
|
"time_of_day": dose.time_of_day,
|
|
"time_of_day_label": dose.get_time_of_day_display(),
|
|
}
|
|
|
|
|
|
def build_day(day: date_type) -> dict:
|
|
"""The checklist for one day: every active dose, with its recorded state."""
|
|
# Chronological, so the day reads morning → night and groups cleanly by time.
|
|
doses = (
|
|
ScheduledDose.objects.filter(is_active=True, medication__is_active=True)
|
|
.select_related("medication")
|
|
.order_by("time_of_day", "medication__name")
|
|
)
|
|
|
|
logs = {
|
|
log.scheduled_dose_id: log
|
|
for log in DoseLog.objects.filter(date=day, scheduled_dose__in=doses)
|
|
}
|
|
|
|
return {
|
|
"date": day,
|
|
"doses": [
|
|
{
|
|
**_serialize_scheduled_dose(dose),
|
|
"taken": logs[dose.id].taken if dose.id in logs else None,
|
|
"notes": logs[dose.id].notes if dose.id in logs else "",
|
|
}
|
|
for dose in doses
|
|
],
|
|
"pulse": PulseReading.objects.filter(date=day).first(),
|
|
}
|
|
|
|
|
|
def pulse_trend(readings: list[PulseReading]) -> Optional[float]:
|
|
"""Least-squares slope in bpm/day: ~0 means stable, positive means climbing."""
|
|
if len(readings) < 2:
|
|
return None
|
|
|
|
origin = readings[0].date
|
|
xs = [(r.date - origin).days for r in readings]
|
|
ys = [r.bpm for r in readings]
|
|
n = len(xs)
|
|
mean_x = sum(xs) / n
|
|
mean_y = sum(ys) / n
|
|
|
|
variance = sum((x - mean_x) ** 2 for x in xs)
|
|
if variance == 0:
|
|
return None
|
|
|
|
covariance = sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys))
|
|
return round(covariance / variance, 3)
|
|
|
|
|
|
# --- Medications -----------------------------------------------------------
|
|
|
|
|
|
@router.get("/medications", response=list[MedicationOut])
|
|
def list_medications(request: HttpRequest, include_inactive: bool = False):
|
|
"""The medication plan. Edit it in the Django admin."""
|
|
medications = Medication.objects.prefetch_related("scheduled_doses__medication")
|
|
if not include_inactive:
|
|
medications = medications.filter(is_active=True)
|
|
|
|
return [
|
|
{
|
|
"id": med.id,
|
|
"name": med.name,
|
|
"is_active": med.is_active,
|
|
"notes": med.notes,
|
|
"scheduled_doses": [
|
|
_serialize_scheduled_dose(dose)
|
|
for dose in med.scheduled_doses.all()
|
|
if dose.is_active or include_inactive
|
|
],
|
|
}
|
|
for med in medications
|
|
]
|
|
|
|
|
|
# --- Daily doses -----------------------------------------------------------
|
|
|
|
|
|
@router.get("/day", response=DayOut)
|
|
def get_day(request: HttpRequest, date: Optional[date_type] = None):
|
|
"""Everything for one day (defaults to today): the dose checklist and the pulse."""
|
|
return build_day(date or timezone.localdate())
|
|
|
|
|
|
@router.post("/doses", response=DayOut)
|
|
def log_dose(request: HttpRequest, payload: DoseLogIn):
|
|
"""Tick a dose off (or untick it). Idempotent per (dose, day)."""
|
|
day = payload.date or timezone.localdate()
|
|
dose = get_object_or_404(ScheduledDose, id=payload.scheduled_dose_id)
|
|
|
|
log, _ = DoseLog.objects.get_or_create(
|
|
scheduled_dose=dose, date=day, defaults={"taken": payload.taken}
|
|
)
|
|
log.taken = payload.taken
|
|
log.notes = payload.notes
|
|
log.save()
|
|
|
|
return build_day(day)
|
|
|
|
|
|
# --- Pulse -----------------------------------------------------------------
|
|
|
|
|
|
@router.get("/pulse", response=list[PulseOut])
|
|
def list_pulse(request: HttpRequest, days: int = 90):
|
|
"""Readings for the graph, oldest first."""
|
|
since = timezone.localdate() - timedelta(days=days)
|
|
return PulseReading.objects.filter(date__gte=since).order_by("date")
|
|
|
|
|
|
@router.post("/pulse", response=PulseOut)
|
|
def record_pulse(request: HttpRequest, payload: PulseIn):
|
|
"""One reading per day: posting twice for the same day overwrites it."""
|
|
if not 20 <= payload.bpm <= 400:
|
|
raise HttpError(422, "A pulse of that value is not plausible for a cat.")
|
|
|
|
day = payload.date or timezone.localdate()
|
|
reading, _ = PulseReading.objects.update_or_create(
|
|
date=day, defaults={"bpm": payload.bpm, "notes": payload.notes}
|
|
)
|
|
return reading
|
|
|
|
|
|
@router.get("/pulse/stats", response=PulseStatsOut)
|
|
def pulse_stats(request: HttpRequest, days: int = 90):
|
|
"""Is the pulse steady or drifting? `trend_bpm_per_day` is the slope.
|
|
|
|
Declared before /pulse/{date}, which would otherwise match "stats" first.
|
|
"""
|
|
since = timezone.localdate() - timedelta(days=days)
|
|
readings = list(PulseReading.objects.filter(date__gte=since).order_by("date"))
|
|
|
|
aggregates = PulseReading.objects.filter(date__gte=since).aggregate(
|
|
average=Avg("bpm"), minimum=Min("bpm"), maximum=Max("bpm")
|
|
)
|
|
|
|
return {
|
|
"count": len(readings),
|
|
"average": round(aggregates["average"], 1) if aggregates["average"] else None,
|
|
"minimum": aggregates["minimum"],
|
|
"maximum": aggregates["maximum"],
|
|
"trend_bpm_per_day": pulse_trend(readings),
|
|
"first_date": readings[0].date if readings else None,
|
|
"last_date": readings[-1].date if readings else None,
|
|
}
|
|
|
|
|
|
@router.delete("/pulse/{date}")
|
|
def delete_pulse(request: HttpRequest, date: date_type):
|
|
get_object_or_404(PulseReading, date=date).delete()
|
|
return {"detail": "Deleted."}
|