208 lines
7.1 KiB
Python
208 lines
7.1 KiB
Python
import calendar
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from models import EggCollection, FlockHistory, FeedPurchase
|
|
from schemas import DashboardStats, BudgetStats, MonthlySummary
|
|
|
|
router = APIRouter(prefix="/api/stats", tags=["stats"])
|
|
|
|
|
|
def _avg_per_hen_30d(db: Session, start_30d: date) -> float | None:
|
|
"""
|
|
For each collection in the last 30 days, look up the flock size that was
|
|
in effect on that date using a correlated subquery, then average eggs/hen
|
|
across those days. This gives an accurate result even when flock size changed.
|
|
"""
|
|
flock_at_date = (
|
|
select(FlockHistory.chicken_count)
|
|
.where(FlockHistory.date <= EggCollection.date)
|
|
.order_by(FlockHistory.date.desc())
|
|
.limit(1)
|
|
.correlate(EggCollection)
|
|
.scalar_subquery()
|
|
)
|
|
|
|
rows = db.execute(
|
|
select(EggCollection.eggs, flock_at_date.label('flock_count'))
|
|
.where(EggCollection.date >= start_30d)
|
|
).all()
|
|
|
|
valid = [(r.eggs, r.flock_count) for r in rows if r.flock_count]
|
|
if not valid:
|
|
return None
|
|
return round(sum(e / f for e, f in valid) / len(valid), 3)
|
|
|
|
|
|
def _current_flock(db: Session) -> int | None:
|
|
row = db.scalars(
|
|
select(FlockHistory).order_by(FlockHistory.date.desc()).limit(1)
|
|
).first()
|
|
return row.chicken_count if row else None
|
|
|
|
|
|
def _total_eggs(db: Session, start: date | None = None, end: date | None = None) -> int:
|
|
q = select(func.coalesce(func.sum(EggCollection.eggs), 0))
|
|
if start:
|
|
q = q.where(EggCollection.date >= start)
|
|
if end:
|
|
q = q.where(EggCollection.date <= end)
|
|
return db.scalar(q)
|
|
|
|
|
|
def _total_feed_cost(db: Session, start: date | None = None, end: date | None = None):
|
|
q = select(
|
|
func.coalesce(func.sum(FeedPurchase.bags * FeedPurchase.price_per_bag), 0)
|
|
)
|
|
if start:
|
|
q = q.where(FeedPurchase.date >= start)
|
|
if end:
|
|
q = q.where(FeedPurchase.date <= end)
|
|
return db.scalar(q)
|
|
|
|
|
|
@router.get("/dashboard", response_model=DashboardStats)
|
|
def dashboard_stats(db: Session = Depends(get_db)):
|
|
today = date.today()
|
|
start_30d = today - timedelta(days=30)
|
|
start_7d = today - timedelta(days=7)
|
|
|
|
total_alltime = _total_eggs(db)
|
|
total_30d = _total_eggs(db, start=start_30d)
|
|
total_7d = _total_eggs(db, start=start_7d)
|
|
flock = _current_flock(db)
|
|
|
|
# Count how many distinct days have a collection logged
|
|
days_tracked = db.scalar(
|
|
select(func.count(func.distinct(EggCollection.date)))
|
|
)
|
|
|
|
# Average eggs per day over the last 30 days (only counting days with data)
|
|
days_with_data_30d = db.scalar(
|
|
select(func.count(func.distinct(EggCollection.date)))
|
|
.where(EggCollection.date >= start_30d)
|
|
)
|
|
|
|
avg_per_day = round(total_30d / days_with_data_30d, 2) if days_with_data_30d else None
|
|
avg_per_hen = _avg_per_hen_30d(db, start_30d)
|
|
|
|
return DashboardStats(
|
|
current_flock=flock,
|
|
total_eggs_alltime=total_alltime,
|
|
total_eggs_30d=total_30d,
|
|
total_eggs_7d=total_7d,
|
|
avg_eggs_per_day_30d=avg_per_day,
|
|
avg_eggs_per_hen_day_30d=avg_per_hen,
|
|
days_tracked=days_tracked,
|
|
)
|
|
|
|
|
|
@router.get("/monthly", response_model=list[MonthlySummary])
|
|
def monthly_stats(db: Session = Depends(get_db)):
|
|
MONTH_NAMES = ['Jan','Feb','Mar','Apr','May','Jun','Jul','Aug','Sep','Oct','Nov','Dec']
|
|
|
|
# Monthly egg totals
|
|
egg_rows = db.execute(
|
|
select(
|
|
func.year(EggCollection.date).label('year'),
|
|
func.month(EggCollection.date).label('month'),
|
|
func.sum(EggCollection.eggs).label('total_eggs'),
|
|
func.count(EggCollection.date).label('days_logged'),
|
|
)
|
|
.group_by(func.year(EggCollection.date), func.month(EggCollection.date))
|
|
.order_by(func.year(EggCollection.date).desc(), func.month(EggCollection.date).desc())
|
|
).all()
|
|
|
|
if not egg_rows:
|
|
return []
|
|
|
|
# Monthly feed costs
|
|
feed_rows = db.execute(
|
|
select(
|
|
func.year(FeedPurchase.date).label('year'),
|
|
func.month(FeedPurchase.date).label('month'),
|
|
func.sum(FeedPurchase.bags * FeedPurchase.price_per_bag).label('feed_cost'),
|
|
)
|
|
.group_by(func.year(FeedPurchase.date), func.month(FeedPurchase.date))
|
|
).all()
|
|
|
|
feed_map = {(r.year, r.month): r.feed_cost for r in feed_rows}
|
|
|
|
results = []
|
|
for row in egg_rows:
|
|
y, m = int(row.year), int(row.month)
|
|
last_day = calendar.monthrange(y, m)[1]
|
|
month_end = date(y, m, last_day)
|
|
|
|
# Flock size in effect at the end of this month
|
|
flock_row = db.scalars(
|
|
select(FlockHistory)
|
|
.where(FlockHistory.date <= month_end)
|
|
.order_by(FlockHistory.date.desc())
|
|
.limit(1)
|
|
).first()
|
|
flock = flock_row.chicken_count if flock_row else None
|
|
|
|
total_eggs = int(row.total_eggs)
|
|
days_logged = int(row.days_logged)
|
|
avg_per_day = round(total_eggs / days_logged, 2) if days_logged else None
|
|
avg_per_hen = round(avg_per_day / flock, 3) if (avg_per_day and flock) else None
|
|
|
|
raw_feed_cost = feed_map.get((y, m))
|
|
feed_cost = round(Decimal(str(raw_feed_cost)), 2) if raw_feed_cost else None
|
|
cpe = round(Decimal(str(raw_feed_cost)) / Decimal(total_eggs), 4) if (raw_feed_cost and total_eggs) else None
|
|
cpd = round(cpe * 12, 4) if cpe else None
|
|
|
|
results.append(MonthlySummary(
|
|
year=y,
|
|
month=m,
|
|
month_label=f"{MONTH_NAMES[m - 1]} {y}",
|
|
total_eggs=total_eggs,
|
|
days_logged=days_logged,
|
|
avg_eggs_per_day=avg_per_day,
|
|
flock_at_month_end=flock,
|
|
avg_eggs_per_hen_per_day=avg_per_hen,
|
|
feed_cost=feed_cost,
|
|
cost_per_egg=cpe,
|
|
cost_per_dozen=cpd,
|
|
))
|
|
|
|
return results
|
|
|
|
|
|
@router.get("/budget", response_model=BudgetStats)
|
|
def budget_stats(db: Session = Depends(get_db)):
|
|
today = date.today()
|
|
start_30d = today - timedelta(days=30)
|
|
|
|
total_cost = _total_feed_cost(db)
|
|
total_cost_30d = _total_feed_cost(db, start=start_30d)
|
|
total_eggs = _total_eggs(db)
|
|
total_eggs_30d = _total_eggs(db, start=start_30d)
|
|
|
|
def cost_per_egg(cost, eggs):
|
|
if not eggs or not cost:
|
|
return None
|
|
return round(Decimal(str(cost)) / Decimal(eggs), 4)
|
|
|
|
def cost_per_dozen(cpe):
|
|
return round(cpe * 12, 4) if cpe else None
|
|
|
|
cpe = cost_per_egg(total_cost, total_eggs)
|
|
cpe_30d = cost_per_egg(total_cost_30d, total_eggs_30d)
|
|
|
|
return BudgetStats(
|
|
total_feed_cost=round(Decimal(str(total_cost)), 2) if total_cost else None,
|
|
total_feed_cost_30d=round(Decimal(str(total_cost_30d)), 2) if total_cost_30d else None,
|
|
total_eggs_alltime=total_eggs,
|
|
total_eggs_30d=total_eggs_30d,
|
|
cost_per_egg=cpe,
|
|
cost_per_dozen=cost_per_dozen(cpe),
|
|
cost_per_egg_30d=cpe_30d,
|
|
cost_per_dozen_30d=cost_per_dozen(cpe_30d),
|
|
)
|