"""The analysis pages: what is active, and how the history has gone.""" from __future__ import annotations from collections import defaultdict from datetime import date from flask import Blueprint, redirect, render_template, request, session, url_for from brewing import Run, is_plausible_sg, run_summary bp = Blueprint("analysis", __name__) def repo(): """The request-scoped repository (one Mongo client per request).""" from flask import current_app return current_app.extensions["beer_repo"]() def login_required(fn): """Redirect to the login page when there is no session.""" from functools import wraps @wraps(fn) def wrapper(*args, **kwargs): if not session.get("logged_in"): return redirect(url_for("auth.login", next=request.path)) return fn(*args, **kwargs) return wrapper def _fmt_days(value: float) -> str: """Format a day count: a whole number stays whole, a fraction keeps one decimal place.""" if float(value).is_integer(): return str(int(value)) return f"{value:.1f}" def _decorate(run: Run, metadata: dict) -> dict: """Attach the stored metadata to a run summary.""" meta = metadata.get(run.beer_run_id, {}) summary = run_summary(run, og_override=meta.get("og_points")) summary["canonical_name"] = meta.get("name") or run.display_name() summary["tap_number"] = meta.get("tap_number") summary["notes"] = meta.get("notes", "") return summary @bp.route("/active") @login_required def active(): """The active beers page: what is in the fermenter right now.""" runs = repo().runs() metadata = repo().all_run_metadata() today = date.today() cards = [] for run in runs: if not run.is_active(): continue summary = _decorate(run, metadata) summary["days_in_fermenter"] = run.days_in_fermenter(today) cards.append(summary) # Most recently started first — that is what is fermenting right now. cards.sort(key=lambda s: (s["first_date"] or "0000-00-00"), reverse=True) ages = [c["days_in_fermenter"] for c in cards if c["days_in_fermenter"] is not None] return render_template( "active.html", cards=cards, today=today, totals={ "count": len(cards), "avg_days": _fmt_days(sum(ages) / len(ages)) if ages else 0, "longest_days": max(ages) if ages else None, }, ) @bp.route("/history") @login_required def history(): """Historical brews and how they performed.""" runs = repo().runs() metadata = repo().all_run_metadata() summaries = [_decorate(run, metadata) for run in runs] completed = [s for s in summaries if not s["active"]] # --- ABV comparison: standard formula vs the legacy one ---------------- # abv_rows = [] for s in completed: if s["abv"] is None: continue abv_rows.append({ "beer_run_id": s["beer_run_id"], "name": s["canonical_name"], "final_date": s["final_date"], "abv": s["abv"], "legacy_abv": s["legacy_abv"], "delta": round((s["legacy_abv"] or 0) - s["abv"], 2), }) abv_rows.sort(key=lambda r: (r["final_date"] or "0000-00-00")) # --- brews per month --------------------------------------------------- # per_month: dict[str, int] = defaultdict(int) for s in summaries: stamp = s["final_date"] or s["first_date"] if stamp: per_month[stamp[:7]] += 1 months = sorted(per_month.items()) # --- outliers ---------------------------------------------------------- # outliers = [] for s in summaries: for reading in s["readings"]: sg = reading["sg"] if sg is not None and not is_plausible_sg(sg): outliers.append({ "beer_run_id": s["beer_run_id"], "name": s["canonical_name"], "date": reading["date"], "sg": sg, "comment": reading["comment"], }) avg_abv = round(sum(r["abv"] for r in abv_rows) / len(abv_rows), 2) if abv_rows else 0 best = max(abv_rows, key=lambda r: r["abv"]) if abv_rows else None worst = min(abv_rows, key=lambda r: r["abv"]) if abv_rows else None return render_template( "history.html", summaries=summaries, completed=completed, abv_rows=abv_rows, months=months, outliers=outliers, stats={ "total_runs": len(summaries), "completed": len(completed), "active": len(summaries) - len(completed), "avg_abv": avg_abv, "best": best, "worst": worst, }, )