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

# At or below this, the cartridge is done and the arithmetic stops being the
# useful answer. A supply sitting at 1% that drains a tenth of a point a day
# computes to ten days; a printer at 1% is out of toner as far as anyone
# standing at it is concerned, and it is what should be ordered first. Rate is
# still reported - only the days-left figure is floored.
EMPTY_LEVEL = 5


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, currentlevel=None):
    """Everything the report needs about one supply.

    `currentlevel` is the live reading from the printer, when the caller has
    one. It is what the days-left figure is computed against, because it is
    what the report displays: taking the level from the display and the
    countdown from the last stored history point lets the two disagree, and a
    row reading "20% - 4 days left" is read as a broken report, correctly.

    Returns:
      currentlevel     live reading if given, else the latest stored one
      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': currentlevel, 'daysleft': None, 'burnrateperday': None,
        'reason': None, 'replacements': 0, 'lastreplaced': None,
        'basisdays': 0, 'points': [],
    }
    if not points:
        # A live level with no history is still worth acting on when it is low.
        if currentlevel is not None and currentlevel <= EMPTY_LEVEL:
            result['daysleft'] = 0
        else:
            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)
    stored = run[-1][1]
    level = stored if currentlevel is None else currentlevel
    result['currentlevel'] = level
    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)

    # A live level well above the stored run means it was swapped since the
    # last stored reading. The run describes the cartridge that came out.
    if currentlevel is not None and currentlevel - stored >= rise:
        result['replacements'] += 1
        result['reason'] = 'replaced recently'
        return result

    rate = burn_rate(run)

    # Empty is empty. Ordering by a rate below this level ranks a dead
    # cartridge behind a healthy one that happens to be draining faster.
    if level <= EMPTY_LEVEL:
        result['daysleft'] = 0
        result['burnrateperday'] = round(rate, 2) if rate is not None else None
        return result

    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(level / 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