shopdb-flask/plugins/printers/services/supply_history.py

"""Read a supply's level history and say what it means.

Two questions come out of the same series of readings:

  * how fast is it draining, and when does it hit empty
  * how many times has it been replaced

Both hang on one observation: a cartridge only ever goes DOWN while it is in
use, so a rise is a replacement. Everything here is built on locating those
rises and treating each stretch between them as one cartridge's life.

The maths is deliberately kept away from Zabbix so it can be tested against a
list of numbers. The service layer fetches; this decides.
"""

from datetime import datetime, timezone

# A reading can wobble by a point without anything happening - SNMP rounding,
# a gauge settling after a power cycle. A rise has to clear this to count as a
# new cartridge rather than noise.
REPLACEMENT_RISE = 10

# Below this many readings a slope is arithmetic, not evidence. Two points
# through a coarse gauge can "prove" any rate at all.
MIN_POINTS_FOR_ESTIMATE = 4

# Many printers report in 10% steps, so a fortnight can pass on one plateau.
# Without a minimum observed drop the slope reads as zero and the forecast
# says "never", which is worse than saying nothing.
MIN_DROP_FOR_ESTIMATE = 2


def _asfloat(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def normalise(points):
    """[(clock, value)] -> sorted [(datetime, float)], junk dropped.

    Zabbix returns clock as a unix string and value as a string; a history
    table can also carry the odd unparseable row.
    """
    out = []
    for clock, value in points:
        seconds = _asfloat(clock)
        level = _asfloat(value)
        if seconds is None or level is None:
            continue
        out.append((datetime.fromtimestamp(seconds, tz=timezone.utc), level))
    out.sort(key=lambda p: p[0])
    return out


def find_replacements(points, rise=REPLACEMENT_RISE):
    """Timestamps where the level jumped up - one per cartridge change.

    Returns [] for a series that only falls. A rise smaller than `rise` is
    treated as noise, not a replacement.
    """
    replacements = []
    for (_, previous), (when, current) in zip(points, points[1:]):
        if current - previous >= rise:
            replacements.append(when)
    return replacements


def current_run(points, rise=REPLACEMENT_RISE):
    """The readings since the last replacement.

    Fitting across a replacement averages a spent cartridge with a fresh one
    and produces a slope that describes neither.
    """
    if not points:
        return []
    start = 0
    for index in range(1, len(points)):
        if points[index][1] - points[index - 1][1] >= rise:
            start = index
    return points[start:]


def burn_rate(points):
    """Percent consumed per day over these readings, or None.

    None means "no honest estimate": too few readings, no elapsed time, or a
    drop too small to distinguish from a gauge that has not moved yet.
    """
    if len(points) < MIN_POINTS_FOR_ESTIMATE:
        return None
    first_when, first_level = points[0]
    last_when, last_level = points[-1]
    days = (last_when - first_when).total_seconds() / 86400
    if days <= 0:
        return None
    drop = first_level - last_level
    if drop < MIN_DROP_FOR_ESTIMATE:
        return None
    return drop / days


def analyse(points, rise=REPLACEMENT_RISE):
    """Everything the report needs about one supply.

    Returns:
      currentlevel     latest reading, or None
      daysleft         at the current rate, or None when there is no estimate
      burnrateperday   percent per day, or None
      reason           why there is no estimate - shown rather than hidden, so
                       a missing forecast is explained instead of looking broken
      replacements     how many times it has been changed in this window
      lastreplaced     when, or None
      basisdays        span of readings the estimate rests on
      points           the run since the last replacement, for the chart
    """
    points = normalise(points)
    result = {
        'currentlevel': None, 'daysleft': None, 'burnrateperday': None,
        'reason': None, 'replacements': 0, 'lastreplaced': None,
        'basisdays': 0, 'points': [],
    }
    if not points:
        result['reason'] = 'no history'
        return result

    replacements = find_replacements(points, rise=rise)
    result['replacements'] = len(replacements)
    result['lastreplaced'] = replacements[-1].isoformat() if replacements else None

    run = current_run(points, rise=rise)
    result['currentlevel'] = run[-1][1]
    result['points'] = [(when.isoformat(), level) for when, level in run]
    if len(run) >= 2:
        result['basisdays'] = round(
            (run[-1][0] - run[0][0]).total_seconds() / 86400, 1)

    rate = burn_rate(run)
    if rate is None:
        # Say which of the three it is; "no estimate" alone invites a bug report.
        if len(run) < MIN_POINTS_FOR_ESTIMATE:
            result['reason'] = ('replaced recently' if replacements
                                else 'not enough history yet')
        else:
            result['reason'] = 'level has not moved enough to estimate'
        return result

    result['burnrateperday'] = round(rate, 2)
    result['daysleft'] = max(0, int(run[-1][1] / rate))
    return result


def soonest(supplies):
    """Days-left of the supply that runs out first, or None if none estimate.

    The report sorts printers by this: a printer is as urgent as its most
    pressing cartridge, and listing it once per supply would scatter it down
    the page.
    """
    days = [s['daysleft'] for s in supplies if s.get('daysleft') is not None]
    return min(days) if days else None