beer_data/src/flask/brewing.py

"""Pure brewing logic for the beer_data app.

Everything in this module is a pure function over plain data (no Flask, no Mongo,
no pandas) so the maths can be unit-tested without a database.

Gravity convention
------------------
The Mongo collection stores specific gravity as integer *points*: the digits after
the decimal point of the usual 1.0xx reading.  ``1039`` means 1.039.  This matches
how the readings were entered and keeps the stored data unchanged.

Alcohol by volume
-----------------
Standard homebrew formula (Charlie Papazian / "The Joy of Homebrewing"):

    ABV% = (OG - FG) * 131.25

With gravities expressed in points the difference is directly in points, so:

    ABV% = points_drop * 0.13125

The legacy application used ``(max - min) / 7.36 + 0.5`` which is not a recognised
formula and overstates the result (it also returns 0.5% for a run with no drop at
all).  :func:`legacy_abv_percent` keeps the old maths available so the history page
can show the difference between the two, but new code uses :func:`abv_percent`.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Iterable, Sequence

__all__ = [
    "ABV_FACTOR",
    "ASSUMED_ORIGINAL_GRAVITY",
    "Reading",
    "Run",
    "abv_percent",
    "assumed_og_abv_percent",
    "days_between",
    "final_reading",
    "first_reading",
    "gravity_drop_points",
    "is_active",
    "is_plausible_sg",
    "last_reading",
    "legacy_abv_percent",
    "parse_date",
    "parse_sg_points",
    "run_summary",
    "sg_points_to_decimal",
]

#: Conversion factor for points-drop -> percent alcohol (131.25 / 1000).
ABV_FACTOR = 0.13125

#: Original gravity assumed for an in-progress run when no OG was recorded.
#: The legacy app hard-coded 1012; kept as the default so existing numbers line up.
ASSUMED_ORIGINAL_GRAVITY = 1012

#: Plausible range for a hydrometer reading, in points.
_MIN_PLAUSIBLE_SG = 990
_MAX_PLAUSIBLE_SG = 1200


# --------------------------------------------------------------------------- #
# Parsing helpers
# --------------------------------------------------------------------------- #

def parse_sg_points(raw: object) -> int | None:
    """Return the gravity in points for *raw*, or ``None`` if it is not usable.

    Accepts ints, floats and strings.  ``"1039"``, ``1039`` and ``1039.0`` all
    become ``1039``.  Values like ``"1.039"`` are also accepted and converted, and
    anything unparseable yields ``None`` rather than raising.
    """
    if raw is None:
        return None
    if isinstance(raw, bool):  # bool is an int subclass; never a gravity
        return None
    if isinstance(raw, (int, float)):
        value = float(raw)
    else:
        text = str(raw).strip()
        if not text:
            return None
        try:
            value = float(text)
        except ValueError:
            return None

    if value > 100:  # already stored as points (1039)
        return int(round(value))
    if value >= 0.9:  # a decimal reading such as 1.039
        return int(round(value * 1000))
    return None


def sg_points_to_decimal(points: int | None) -> float | None:
    """Convert points (``1039``) to the conventional decimal reading (``1.039``)."""
    if points is None:
        return None
    return round(points / 1000.0, 4)


def parse_date(raw: object) -> date | None:
    """Parse a ``YYYY-MM-DD`` date, tolerating the odd alternative layout."""
    if raw is None:
        return None
    if isinstance(raw, datetime):
        return raw.date()
    if isinstance(raw, date):
        return raw
    text = str(raw).strip()
    if not text:
        return None
    for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%Y/%m/%d"):
        try:
            return datetime.strptime(text, fmt).date()
        except ValueError:
            continue
    return None


def _as_bool(value: object) -> bool:
    """Interpret the stored ``final_reading`` flag.

    The collection stores it as the *string* ``"True"``/``"False"`` for historic
    rows and as a real boolean for newer ones, so both are accepted.
    """
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    return str(value).strip().lower() in {"true", "1", "yes", "y", "t"}


# --------------------------------------------------------------------------- #
# Domain objects
# --------------------------------------------------------------------------- #

@dataclass(frozen=True)
class Reading:
    """A single gravity reading taken on a given day."""

    date: date | None
    sg_points: int | None
    final_reading: bool = False
    comment: str = ""
    beer_type: str = ""
    _id: str = ""

    @property
    def sg_decimal(self) -> float | None:
        return sg_points_to_decimal(self.sg_points)

    @property
    def is_plausible(self) -> bool:
        return is_plausible_sg(self.sg_points)

    @classmethod
    def from_doc(cls, doc: dict) -> "Reading":
        """Build a :class:`Reading` from a raw Mongo document."""
        return cls(
            date=parse_date(doc.get("date")),
            sg_points=parse_sg_points(doc.get("sg")),
            final_reading=_as_bool(doc.get("final_reading")),
            comment=(doc.get("comment") or "").strip(),
            beer_type=(doc.get("beer_type") or "").strip(),
            _id=str(doc.get("_id") or ""),
        )


@dataclass
class Run:
    """All readings that belong to one brew run."""

    beer_run_id: str
    readings: list[Reading] = field(default_factory=list)

    @classmethod
    def from_docs(cls, beer_run_id: str, docs: Iterable[dict]) -> "Run":
        readings = [Reading.from_doc(d) for d in docs]
        readings.sort(key=lambda r: (r.date or date.min, r.sg_points or 0))
        return cls(beer_run_id=str(beer_run_id), readings=readings)

    # -- aggregate accessors ------------------------------------------------ #

    def sg_values(self) -> list[int]:
        """Gravity points of every *usable* reading, in chronological order."""
        return [r.sg_points for r in self.readings if r.sg_points is not None]

    def original_gravity(self) -> int | None:
        """OG = the first usable reading of the run."""
        values = self.sg_values()
        return values[0] if values else None

    def final_gravity(self) -> int | None:
        """FG = the last usable reading of the run (if the run is finished)."""
        values = self.sg_values()
        return values[-1] if values else None

    def max_sg(self) -> int | None:
        values = self.sg_values()
        return max(values) if values else None

    def min_sg(self) -> int | None:
        values = self.sg_values()
        return min(values) if values else None

    def is_active(self) -> bool:
        return is_active(self.readings)

    def final_reading(self) -> Reading | None:
        return final_reading(self.readings)

    def first_reading(self) -> Reading | None:
        return first_reading(self.readings)

    def last_reading(self) -> Reading | None:
        return last_reading(self.readings)

    def gravity_drop_points(self) -> int | None:
        og, fg = self.original_gravity(), self.final_gravity()
        if og is None or fg is None:
            return None
        return og - fg

    def abv_percent(self) -> float | None:
        """Standard ABV for a finished run."""
        drop = self.gravity_drop_points()
        return abv_percent(drop) if drop is not None else None

    def projected_abv_percent(self, assumed_og: int = ASSUMED_ORIGINAL_GRAVITY,
                              og_override: int | None = None) -> float | None:
        # `assumed_og` is retained for API compatibility but is deliberately
        # NOT used as a fallback — see the docstring.
        """ABV estimate for an *unfinished* run, or ``None`` if unknowable.

        The original gravity is taken from, in order of preference:

        1. ``og_override`` — the OG recorded against the run by the user.
        2. The run's first reading, when there are two or more readings.
           The first reading of a run IS its original gravity — that is what
           a reading taken on brew day is for.
        With a single reading and no recorded OG there is NO reliable
        estimate: the reading may have been taken on brew day or mid-
        fermentation.  The legacy app assumed 1012 anyway, which produced
        *negative* alcohol whenever the reading was above 1012 — a perfectly
        normal OG.  Rather than show a misleading figure, ``None`` is
        returned and the UI shows a dash until an OG is known.  Record the OG
        against the run and the estimate appears.
        """
        values = self.sg_values()
        if not values:
            return None

        if og_override is not None:
            og = og_override
        elif len(values) >= 2:
            og = values[0]
        else:
            # A single reading and no recorded OG: there is no reliable
            # original gravity to measure against, so there is no estimate.
            return None

        fg = self.min_sg()
        if fg is None:
            return None

        drop = og - fg
        if drop <= 0:
            # Gravity has risen during fermentation — an anomaly, not alcohol.
            return None

        return abv_percent(drop)

    def days_in_fermenter(self, today: date | None = None) -> int | None:
        first = self.first_reading()
        if first is None or first.date is None:
            return None
        final = self.final_reading()
        end = final.date if (final is not None and final.date is not None) else (today or date.today())
        if end is None:
            return None
        return days_between(first.date, end)

    def display_name(self) -> str:
        """Best available human label for the run.

        The stored ``beer_type`` drifts between readings of the same run, so the
        most frequently used non-empty value wins; ties break towards the most
        recent reading.  Run 4 would resolve to "Nocturna the Christmas beer"
        rather than "Necatron-Christmas Beer".
        """
        counts: dict[str, int] = {}
        for reading in self.readings:
            label = reading.beer_type
            if label:
                counts[label] = counts.get(label, 0) + 1
        if not counts:
            return f"Run {self.beer_run_id}"
        best = max(counts, key=lambda label: (counts[label], self._last_index_of(label)))
        return best

    def _last_index_of(self, label: str) -> int:
        for index in range(len(self.readings) - 1, -1, -1):
            if self.readings[index].beer_type == label:
                return index
        return -1

    def comments(self) -> list[str]:
        """Every non-empty comment, oldest first."""
        return [r.comment for r in self.readings if r.comment]

    def outlier_readings(self) -> list[Reading]:
        """Readings whose gravity is outside the plausible hydrometer range."""
        return [r for r in self.readings if r.sg_points is not None and not r.is_plausible]


# --------------------------------------------------------------------------- #
# Module level helpers (usable without a Run instance)
# --------------------------------------------------------------------------- #

def abv_percent(points_drop: int | float | None) -> float | None:
    """Standard ABW/ABV from a gravity drop expressed in points."""
    if points_drop is None:
        return None
    return round(float(points_drop) * ABV_FACTOR, 2)


def legacy_abv_percent(max_points: int | float | None,
                        min_points: int | float | None) -> float | None:
    """The legacy app's formula, kept for the legacy-vs-standard comparison.

    The old code computed ``((max - min) / 7.36) + 0.5``.  It is not a recognised
    brewing formula and returns 0.5% even when nothing has fermented.
    """
    if max_points is None or min_points is None:
        return None
    return round(((float(max_points) - float(min_points)) / 7.36) + 0.5, 2)


def assumed_og_abv_percent(current_points: int | None,
                           assumed_og: int = ASSUMED_ORIGINAL_GRAVITY) -> float | None:
    """ABV for an in-progress run measured against an assumed original gravity."""
    if current_points is None:
        return None
    return abv_percent(assumed_og - current_points)


def is_plausible_sg(points: int | None) -> bool:
    """True when a gravity reading sits inside a believable hydrometer range."""
    if points is None:
        return False
    return _MIN_PLAUSIBLE_SG <= points <= _MAX_PLAUSIBLE_SG


def is_active(readings: Sequence[Reading]) -> bool:
    """A run is active while it has no reading marked as the final one."""
    return not any(r.final_reading for r in readings)


def final_reading(readings: Sequence[Reading]) -> Reading | None:
    """The reading flagged as final, or ``None``."""
    for reading in readings:
        if reading.final_reading:
            return reading
    return None


def first_reading(readings: Sequence[Reading]) -> Reading | None:
    """The chronologically first reading."""
    dated = [r for r in readings if r.date is not None]
    if dated:
        return min(dated, key=lambda r: r.date)  # type: ignore[arg-type,return-value]
    return readings[0] if readings else None


def last_reading(readings: Sequence[Reading]) -> Reading | None:
    """The chronologically last reading."""
    dated = [r for r in readings if r.date is not None]
    if dated:
        return max(dated, key=lambda r: r.date)  # type: ignore[arg-type,return-value]
    return readings[-1] if readings else None


def days_between(start: date, end: date) -> int:
    return (end - start).days


def _fmt_decimal(points: int | None) -> str:
    """Format gravity points as a decimal reading, e.g. ``1039`` -> ``1.039``."""
    if points is None:
        return ""
    return f"{points / 1000:.3f}"


def run_summary(run: Run, today: date | None = None,
                og_override: int | None = None) -> dict:
    """A flat dict describing a run, ready for a template or a JSON response."""
    final = run.final_reading()
    first = run.first_reading()
    og = run.original_gravity()
    fg = run.final_gravity()
    return {
        "beer_run_id": run.beer_run_id,
        "name": run.display_name(),
        "active": run.is_active(),
        "og_points": og,
        "fg_points": fg,
        "og_decimal": sg_points_to_decimal(og),
        "fg_decimal": sg_points_to_decimal(fg),
        # Pre-formatted for the templates so no arithmetic happens in Jinja.
        "og_decimal_str": _fmt_decimal(og),
        "fg_decimal_str": _fmt_decimal(fg),
        "max_sg": run.max_sg(),
        "min_sg": run.min_sg(),
        "min_sg_decimal": _fmt_decimal(run.min_sg()),
        "drop_points": run.gravity_drop_points(),
        "abv": run.abv_percent(),
        "projected_abv": run.projected_abv_percent(og_override=og_override),
        "legacy_abv": legacy_abv_percent(run.max_sg(), run.min_sg()),
        "reading_count": len(run.readings),
        "first_date": first.date.isoformat() if (first and first.date) else None,
        "final_date": final.date.isoformat() if (final and final.date) else None,
        "days_in_fermenter": run.days_in_fermenter(today),
        "comments": run.comments(),
        "outliers": [r.sg_points for r in run.outlier_readings()],
        "readings": [
            {
                "date": r.date.isoformat() if r.date else None,
                "sg": r.sg_points,
                "sg_decimal": r.sg_decimal,
                "final_reading": r.final_reading,
                "comment": r.comment,
                "beer_type": r.beer_type,
            }
            for r in run.readings
        ],
    }