Spaces:
Sleeping
Sleeping
File size: 2,608 Bytes
57ed4c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Loop-break week summary for Home banners and agent context."""
from __future__ import annotations
import re
from datetime import date, datetime, timedelta, timezone
from typing import Any
from fastapi import APIRouter, Depends, Query, Request
from app.deps import require_login
from app.models import ApiEnvelope, ok
from app.store_daily import indoors_streak
router = APIRouter(prefix="/api/loop", tags=["loop"])
_LABEL_ONLY = re.compile(r"\blabel(?:ing|ling)?\b", re.IGNORECASE)
def loop_summary_for(
*,
entry_store: Any,
daily_store: Any,
as_of: date,
days: int = 7,
) -> dict[str, Any]:
"""Aggregate indoors / proof / label-only / movement / pending for a window."""
start = as_of - timedelta(days=max(1, days) - 1)
rows = [r for r in daily_store._load() if start <= r.date <= as_of]
entries = []
for entry in entry_store._load():
day = entry.ts.date() if hasattr(entry.ts, "date") else entry.ts
if start <= day <= as_of:
entries.append(entry)
streak = indoors_streak(daily_store._load(), as_of=as_of)
proof_days = sum(1 for r in rows if r.proof_brick_done)
proof_rate = (proof_days / len(rows)) if rows else 0.0
movement_sum = sum(int(r.movement_minutes or 0) for r in rows)
high_i = [e for e in entries if (e.intensity or 0) >= 7]
label_fails = [
e
for e in high_i
if _LABEL_ONLY.search(e.remedy or "")
and e.result.value in ("failed", "partial", "pending")
]
label_only_fail_rate = (len(label_fails) / len(high_i)) if high_i else 0.0
pending_count = sum(1 for e in entry_store._load() if e.result.value == "pending")
return {
"days": days,
"as_of": as_of.isoformat(),
"indoors_streak": streak,
"proof_rate": round(proof_rate, 4),
"proof_days": proof_days,
"days_with_daily": len(rows),
"label_only_fail_rate": round(label_only_fail_rate, 4),
"label_only_fail_n": len(label_fails),
"high_intensity_n": len(high_i),
"movement_minutes_sum": movement_sum,
"pending_count": pending_count,
}
@router.get("/summary", response_model=ApiEnvelope)
def get_loop_summary(
request: Request,
days: int = Query(default=7, ge=1, le=30),
_user: str = Depends(require_login),
) -> dict[str, Any]:
as_of = datetime.now(timezone.utc).date()
return ok(
loop_summary_for(
entry_store=request.app.state.entry_store,
daily_store=request.app.state.daily_store,
as_of=as_of,
days=days,
)
)
|