first version
This commit is contained in:
+243
@@ -0,0 +1,243 @@
|
||||
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."}
|
||||
Reference in New Issue
Block a user