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.  The upper bound is
#: generous enough for a strong specialty brew (a 1.150 wort is real) while
#: still catching typos such as 900 or 1500.
_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.

        A run that is still in the fermenter has no measured alcohol - its
        figure is a projection, not a measurement.  Returning the drop here
        for an unfinished run would report a "measured" ABV for a beer that
        has not finished fermenting, which is wrong.
        """
        if self.is_active():
            return None
        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 name is resolved by preferring, in order:

        1. The name on the **final** reading.  That is the one the operator
           wrote when the brew was finished, so it is the name they settled
           on.  Run 4 resolves to "Nocturna Christmas beer" rather than
           "Necatron-Christmas Beer" from the first reading.
        2. The name used on the most readings, breaking ties towards the most
           recent.  This catches runs whose name changed exactly once.
        3. The name on the most recent reading, if it appears only once.

        A run whose readings carry no name at all falls back to "Run N".
        """
        named = [r for r in self.readings if r.beer_type]
        if not named:
            return f"Run {self.beer_run_id}"

        final = self.final_reading()
        if final is not None and final.beer_type:
            return final.beer_type

        counts: dict[str, int] = {}
        for reading in named:
            counts[reading.beer_type] = counts.get(reading.beer_type, 0) + 1

        return max(counts, key=lambda label: (counts[label], self._last_index_of(label)))

    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 the stored gravity as a decimal reading, e.g. ``1039`` -> ``1.039``."""
    if points is None:
        return ""
    return f"{points / 1000:.3f}"


def _fmt_point_scale(points: int | None) -> str:
    """Format the gravity in brewing "points".

    A reading of 1.039 is 39 points in the homebrew sense - the digits after
    the decimal point.  The stored value ``1039`` is the full reading, not a
    point count, so it must not be labelled "points".
    """
    if points is None:
        return ""
    return f"{points % 1000}"


def fermentation_curve(run: Run) -> list[dict]:
    """The run's gravity readings shaped for plotting.

    Each point carries the reading date, the gravity, and how far through
    the fermentation it was - the fraction of the total gravity drop that
    had happened by that reading.  That makes two runs with different
    gravities directly comparable, which is the point of the chart.
    """
    values = run.sg_values()
    if len(values) < 2:
        return []

    og = values[0]
    total_drop = og - values[-1]
    if total_drop <= 0:
        # No drop to measure progress against.
        return []

    points = []
    for reading in run.readings:
        if reading.sg_points is None:
            continue
        drop = og - reading.sg_points
        points.append({
            "date": reading.date.isoformat() if reading.date else None,
            "sg": reading.sg_points,
            "sg_decimal": _fmt_decimal(reading.sg_points),
            "point_scale": _fmt_point_scale(reading.sg_points),
            # 0.0 at the start, 1.0 at the final reading.
            "progress": round(max(0.0, min(1.0, drop / total_drop)), 4),
            "final": reading.final_reading,
            "comment": reading.comment,
        })
    return points


def attenuation_percent(run: Run) -> float | None:
    """The apparent attenuation: how much of the original gravity was
    fermented away.

    This is the number a brewer actually compares between recipes, and it is
    more informative than ABV alone because it normalises for how strong the
    starting wort was.

    Returns ``None`` when the figure would be meaningless:

    - The original gravity was not above 1.000, so there is nothing to
      attenuate.
    - The final gravity is below 1.000.  Beer cannot be less dense than
      water, so a reading below 1.000 means the reading is wrong (a
      mis-entry, or a hydrometer read at the wrong temperature).  Run 12 in
      the live data has a final gravity of 900, which would otherwise show
      as a nonsense 338% attenuation.  The run is reported separately by
      :func:`implausible_runs` rather than being quietly folded into the
      averages.
    """
    og = run.original_gravity()
    fg = run.final_gravity()
    if og is None or fg is None or og <= 1000 or fg < 1000:
        return None
    attenuation = (og - fg) / (og - 1000) * 100
    if attenuation > 100:
        # Cannot ferment away more than the sugar that was there.
        return None
    return round(attenuation, 1)


def implausible_runs(runs: Iterable[Run]) -> list[dict]:
    """The runs whose gravity readings are physically impossible.

    Beer cannot be less dense than water, so a final gravity below 1.000
    means a reading is wrong.  These runs are excluded from the averages and
    listed on the analysis page instead of being silently included.
    """
    problems = []
    for run in runs:
        og = run.original_gravity()
        fg = run.final_gravity()
        if fg is not None and fg < 1000:
            problems.append({
                "beer_run_id": run.beer_run_id,
                "og_points": og,
                "fg_points": fg,
                "problem": (
                    f"final gravity {fg / 1000:.3f} is below 1.000 — "
                    "beer cannot be less dense than water, so this reading "
                    "is probably wrong"
                ),
            })
        elif og is not None and fg is not None and og > 1000 and fg > 1000 \
                and (og - fg) / (og - 1000) * 100 > 100:
            problems.append({
                "beer_run_id": run.beer_run_id,
                "og_points": og,
                "fg_points": fg,
                "problem": "the gravity drop implies more than 100% attenuation",
            })
    return problems


def days_to_target(run: Run, target: int) -> int | None:
    """How many days the run took to reach *target* gravity.

    Used to answer "how long does this usually take?", which is the question
    a brewer actually has when planning the next batch.
    """
    for reading in run.readings:
        if reading.sg_points is not None and reading.sg_points <= target:
            first = run.first_reading()
            if first is None or first.date is None or reading.date is None:
                return None
            return days_between(first.date, reading.date)
    return None


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),
        "og_point_scale": _fmt_point_scale(og),
        "fg_point_scale": _fmt_point_scale(fg),
        "max_sg": run.max_sg(),
        "min_sg": run.min_sg(),
        "min_sg_decimal": _fmt_decimal(run.min_sg()),
        "min_sg_point_scale": _fmt_point_scale(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())
                       if not run.is_active() else None),
        "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(),
        "last_comment": run.comments()[-1] if run.comments() else "",
        "attenuation": attenuation_percent(run),
        "curve": fermentation_curve(run),
        "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
        ],
    }