ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Forecast when a printer runs out, and count what it has been through

Browse Source 
The toner report says what is empty now. It could not say what to order, and
nothing recorded how fast anything drains - every level read was cached for
five minutes and then discarded.

Zabbix has been keeping the history all along; we simply never asked. One
history.get gives both answers, because a cartridge only goes DOWN while it is
in use: a rise is a replacement. Count the rises and you have how many
cartridges a printer has been through; fit a slope to the readings SINCE the
last rise and you have days-to-empty. Fitting across a replacement averages a
spent cartridge with a fresh one and describes neither.

Sorted by days left, which is the point. A cartridge at 60% dropping 5% a day
needs ordering before one sitting at 8% that has not moved in months, and a
level-sorted list ranks those backwards.

It refuses to guess. Too few readings, a level that has not moved enough - many
printers report in 10% steps and sit on a plateau for a fortnight - or a recent
replacement each produce no estimate and say which. Those printers are listed
separately rather than sorted in as 0 or as 999, since a printer without an
estimate is neither urgent nor safe. Estimates show what they rest on, because
"9 days from 21 days of readings" and "9 days from 2 readings" are not the same
claim.

A separate report card, not an extension of the toner report: that one is an
exceptions list a tech acts on today, this is an ordering view read monthly,
and the history query is heavier than the live read it would have slowed down.

The analysis is pure arithmetic over a list of readings, so the 14 tests cover
the noise wobble, the plateau, the swap, junk rows and division by zero without
needing Zabbix. Zabbix being unreachable is reported as such rather than
rendering an empty table that reads as "nothing is due".
This commit is contained in:
cproudlock
2026-08-12 11:45:40 -04:00
parent 2fce81f33f
commit d109314123
8 changed files with 627 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
plugins/printers/api/asset_routes.py
View File

@@ -1482,3 +1482,102 @@ def dashboard_supplies():
rows.sort(key=lambda r: (not r['iscritical'], r['printername']))
return success_response(rows)
# =============================================================================
# Supply forecast
#
# The toner report answers "what is empty now". This answers "what will be, and
# what have we been getting through" - a purchasing question, on a different
# cadence, off data Zabbix has been keeping all along.
# =============================================================================
@printers_asset_bp.route('/supplies/forecast', methods=['GET'])
@jwt_required(optional=True)
def supplies_forecast():
"""Days-to-empty and replacement counts per printer.
?days=90 how far back to read (Zabbix retention is the real ceiling)
Printers sort by their soonest supply. Anything without an honest estimate
is returned separately with the reason, rather than sorted as though it
were fine or dropped as though it did not exist.
"""
from ..services.supply_history import analyse, soonest
try:
days = max(1, min(365, int(request.args.get('days', 90))))
except (TypeError, ValueError):
days = 90
service = ZabbixService()
if not service.isconfigured or not service.isreachable:
return success_response({
'printers': [], 'unestimated': [], 'days': days,
'available': False,
'reason': 'Zabbix is not configured or not reachable',
})
rows = (
db.session.query(Printer, Asset, Communication, Vendor)
.join(Asset, Asset.assetid == Printer.assetid)
.join(Communication, Communication.assetid == Asset.assetid)
.outerjoin(Vendor, Vendor.vendorid == Printer.vendorid)
.filter(Asset.isactive == True,
Communication.ipaddress.isnot(None),
Communication.ipaddress != '')
.all()
)
seen = set()
estimated, unestimated = [], []
for printer, asset, comm, vendor in rows:
if printer.printerid in seen:
continue
seen.add(printer.printerid)
supplies = service.getsuppliesbyip_cached(comm.ipaddress)
if not supplies:
continue
itemids = [s['itemid'] for s in supplies if s.get('itemid')]
history = service.gethistory(itemids, days=days)
analysed = []
for supply in supplies:
points = history.get(str(supply.get('itemid')), [])
detail = analyse(points)
detail['name'] = supply.get('name')
detail['color'] = supply.get('color')
# Trust the live read for the level; history can lag a poll behind.
detail['currentlevel'] = supply.get('level', detail['currentlevel'])
analysed.append(detail)
entry = {
'printerid': printer.printerid,
'printername': asset.name or printer.hostname or '',
'assetnumber': asset.assetnumber or '',
'ipaddress': comm.ipaddress,
'vendor': vendor.vendor if vendor else None,
'supplies': analysed,
'daysleft': soonest(analysed),
'replacements': sum(s['replacements'] for s in analysed),
}
(estimated if entry['daysleft'] is not None else unestimated).append(entry)
# Soonest first: the point of the report is what to order next.
estimated.sort(key=lambda p: p['daysleft'])
unestimated.sort(key=lambda p: p['printername'])
return success_response({
'printers': estimated,
'unestimated': unestimated,
'days': days,
'available': True,
'summary': {
'estimated': len(estimated),
'unestimated': len(unestimated),
'replacements': sum(p['replacements'] for p in estimated + unestimated),
'duewithin30': sum(1 for p in estimated if p['daysleft'] <= 30),
},
})

6
plugins/printers/frontend/routes.js
View File

@@ -8,6 +8,12 @@ export default [
component: () => import('./views/TonerReport.vue'),
meta: { plugin: 'printers' }
},
{
path: 'reports/toner-forecast',
name: 'toner-forecast',
component: () => import('./views/TonerForecast.vue'),
meta: { plugin: 'printers' }
},
{
path: 'settings/printertypes',
name: 'printer-types',

182
plugins/printers/frontend/views/TonerForecast.vue Normal file
View File

@@ -0,0 +1,182 @@
<template>
<div>
<div class="page-header">
<h1>Toner Forecast</h1>
<div class="actions">
<select v-model.number="days" class="form-control" @change="load">
<option :value="30">Last 30 days</option>
<option :value="90">Last 90 days</option>
<option :value="180">Last 180 days</option>
</select>
<router-link to="/reports/toner" class="btn btn-secondary">Toner Report</router-link>
</div>
</div>
<div v-if="loading" class="loading">Reading history...</div>
<!-- Zabbix off is not the same as nothing running out. Say which. -->
<div v-else-if="!available" class="card empty-state">
<p>{{ reason || 'Supply history is unavailable.' }}</p>
<p class="muted">
Levels and history both come from Zabbix. With it unreachable there is
nothing to forecast from - this is not a report of "nothing is due".
</p>
</div>
<template v-else>
<div class="summary-row">
<div class="summary-card">
<span class="summary-number">{{ summary.duewithin30 }}</span>
<span class="summary-label">due within 30 days</span>
</div>
<div class="summary-card">
<span class="summary-number">{{ summary.estimated }}</span>
<span class="summary-label">printers with an estimate</span>
</div>
<div class="summary-card">
<span class="summary-number">{{ summary.replacements }}</span>
<span class="summary-label">cartridges changed in {{ days }} days</span>
</div>
</div>
<div class="card">
<div class="table-container">
<table>
<thead>
<tr>
<th>Printer</th>
<th>Runs out in</th>
<th>Supply</th>
<th>Level</th>
<th>Burn rate</th>
<th>Changed</th>
<th>Based on</th>
</tr>
</thead>
<tbody>
<!-- Sorted by days left, not by level: a cartridge at 60% falling
fast is ordered before one sitting at 8% that never moves. -->
<template v-for="p in printers" :key="p.printerid">
<tr v-for="(s, index) in p.supplies" :key="p.printerid + s.name"
:class="{ 'row-group-start': index === 0 }">
<td v-if="index === 0" :rowspan="p.supplies.length">
<router-link :to="`/printers/${p.printerid}`">
{{ p.printername || p.assetnumber }}
</router-link>
<div class="muted small">{{ p.ipaddress }}</div>
</td>
<td v-if="index === 0" :rowspan="p.supplies.length">
<span class="days" :class="urgency(p.daysleft)">
{{ p.daysleft }} days
</span>
</td>
<td>{{ s.name }}</td>
<td>{{ s.currentlevel != null ? Math.round(s.currentlevel) + '%' : '-' }}</td>
<td>{{ s.burnrateperday != null ? s.burnrateperday + '%/day' : '-' }}</td>
<td>{{ s.replacements || 0 }}</td>
<td class="muted small">
{{ s.reason ? s.reason : s.basisdays + ' days of readings' }}
</td>
</tr>
</template>
<tr v-if="!printers.length">
<td colspan="7" class="muted" style="text-align:center;">
Nothing can be estimated yet from {{ days }} days of history.
</td>
</tr>
</tbody>
</table>
</div>
</div>
<!-- Kept separate rather than sorted in as 0 or as 999: a printer with no
estimate is neither urgent nor safe, and the reason is the useful part. -->
<div v-if="unestimated.length" class="card">
<h3 class="section-title">No estimate yet ({{ unestimated.length }})</h3>
<div class="table-container">
<table>
<thead>
<tr><th>Printer</th><th>Supply</th><th>Level</th><th>Why</th></tr>
</thead>
<tbody>
<template v-for="p in unestimated" :key="p.printerid">
<tr v-for="s in p.supplies" :key="p.printerid + s.name">
<td>
<router-link :to="`/printers/${p.printerid}`">
{{ p.printername || p.assetnumber }}
</router-link>
</td>
<td>{{ s.name }}</td>
<td>{{ s.currentlevel != null ? Math.round(s.currentlevel) + '%' : '-' }}</td>
<td class="muted">{{ s.reason || '-' }}</td>
</tr>
</template>
</tbody>
</table>
</div>
</div>
</template>
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { printersApi } from '@/api'
import { apiError } from '@/utils/apiError'
const loading = ref(true)
const available = ref(false)
const reason = ref('')
const days = ref(90)
const printers = ref([])
const unestimated = ref([])
const summary = ref({ duewithin30: 0, estimated: 0, replacements: 0 })
function urgency(daysleft) {
if (daysleft <= 7) return 'critical'
if (daysleft <= 30) return 'low'
return 'ok'
}
async function load() {
loading.value = true
try {
const response = await printersApi.supplyForecast(days.value)
const data = response.data.data
available.value = data.available
reason.value = data.reason || ''
printers.value = data.printers || []
unestimated.value = data.unestimated || []
summary.value = data.summary || summary.value
} catch (err) {
available.value = false
reason.value = apiError(err, 'Failed to read supply history')
} finally {
loading.value = false
}
}
onMounted(load)
</script>
<style scoped>
.actions { display: flex; gap: 0.5rem; align-items: center; }
.actions .form-control { width: auto; }
.summary-row { display: flex; gap: 1rem; margin-bottom: 1.25rem; flex-wrap: wrap; }
.summary-card {
background: var(--bg-card-solid);
border: 1px solid var(--border);
border-radius: 0.5rem;
padding: 1rem 1.5rem;
min-width: 12rem;
}
.summary-number { display: block; font-size: 1.8rem; font-weight: 700; }
.summary-label { color: var(--text-light); font-size: 0.85rem; }
.days { font-weight: 700; }
.days.critical { color: var(--danger); }
.days.low { color: var(--warning); }
.small { font-size: 0.8rem; }
.row-group-start td { border-top: 2px solid var(--border); }
.empty-state { padding: 2rem; text-align: center; }
.section-title { padding: 0.75rem 1rem 0; }
</style>

11
plugins/printers/plugin.py
View File

@@ -358,6 +358,17 @@ class PrintersPlugin(BasePlugin):
'category': 'printers',
'route': '/reports/toner',
},
{
# Separate from the toner report on purpose: that one is an
# exceptions list a tech acts on today, this is an ordering
# view read monthly, and it runs a heavier history query.
'id': 'tonerforecast',
'name': 'Toner Forecast',
'description': 'Estimated days until each printer runs out, '
'and how many cartridges it has been through',
'category': 'printers',
'route': '/reports/toner-forecast',
},
]
def get_permissions(self) -> List:

3
plugins/printers/services/__init__.py
View File

@@ -9,6 +9,7 @@ from .supply_parts import (
alerttier,
)
from .supply_alerts import check_supplies
from .supply_history import analyse, soonest
from .seed_supplies import seedsupplies
__all__ = [
@@ -19,5 +20,7 @@ __all__ = [
'lookupsupplies',
'alerttier',
'check_supplies',
'analyse',
'soonest',
'seedsupplies',
]

162
plugins/printers/services/supply_history.py Normal file
View File

@@ -0,0 +1,162 @@
"""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

34
plugins/printers/services/zabbix_service.py
View File

@@ -19,6 +19,7 @@ Configuration (database Setting overrides env var):
"""
import logging
import time
from typing import Dict, List, Optional
import requests
@@ -219,6 +220,39 @@ class ZabbixService:
})
return supplies
def gethistory(self, itemids, days=90, limit=5000):
"""Raw level history for supply items: {itemid: [(clock, value), ...]}.
Zabbix keeps this already - we simply never asked for it. history=3 is
the unsigned-integer table, which is where a percent-remaining item
lands; a site that types the item as float would need history=0, so a
miss falls back rather than erroring.
Returns {} when Zabbix is off or unreachable, so callers degrade to
"no estimate" instead of failing.
"""
if not itemids:
return {}
timefrom = int(time.time()) - days * 86400
collected = {}
for historytype in (3, 0):
rows = self._apicall("history.get", {
"output": "extend",
"history": historytype,
"itemids": list(itemids),
"time_from": timefrom,
"sortfield": "clock",
"sortorder": "ASC",
"limit": limit,
}) or []
for row in rows:
collected.setdefault(str(row.get("itemid")), []).append(
(row.get("clock"), row.get("value")))
# Items live in one table or the other; stop once something answered.
if collected:
break
return collected
def getpingstatus(self, ip: str) -> str:
"""ICMP ping state for a printer: '1' up, '0' down, '-1' unknown."""
hostid = self.gethostidbyip(ip)

130
tests/test_plugins/test_supply_history.py Normal file
View File

@@ -0,0 +1,130 @@
"""Tests for toner burn-rate and replacement counting.
The analysis is pure arithmetic over a list of readings, so it is tested that
way - no Zabbix, no fixtures. What matters is that it refuses to guess: the
cases where an estimate would be dishonest are the ones most likely to reach a
purchasing decision.
"""
from datetime import datetime, timedelta, timezone
from plugins.printers.services.supply_history import (
analyse, burn_rate, current_run, find_replacements, normalise, soonest,
)
START = datetime(2026, 6, 1, tzinfo=timezone.utc)
def series(levels, hours=24):
"""[(clock, value)] one reading per `hours`, as Zabbix returns them."""
return [(str(int((START + timedelta(hours=i * hours)).timestamp())), str(v))
for i, v in enumerate(levels)]
def test_counts_one_replacement_per_upward_step():
points = normalise(series([80, 60, 40, 20, 100, 80, 60, 95, 70]))
assert len(find_replacements(points)) == 2
def test_ignores_a_wobble_that_is_not_a_replacement():
"""SNMP rounding and a gauge settling both nudge a reading upward."""
points = normalise(series([60, 58, 61, 57, 55]))
assert find_replacements(points) == []
def test_estimate_uses_only_the_current_cartridge():
"""Fitting across a swap averages a spent cartridge with a fresh one."""
points = normalise(series([90, 60, 30, 100, 90, 80, 70]))
run = current_run(points)
assert [level for _, level in run] == [100, 90, 80, 70]
def test_days_left_from_a_steady_drain():
# 10 points a day apart, 5% a day, ending at 55%
result = analyse(series([100, 95, 90, 85, 80, 75, 70, 65, 60, 55]))
assert result['burnrateperday'] == 5.0
assert result['daysleft'] == 11 # 55 / 5
assert result['reason'] is None
assert result['basisdays'] == 9.0
def test_no_estimate_from_two_readings():
"""Two points through a coarse gauge can prove any rate at all."""
result = analyse(series([100, 90]))
assert result['daysleft'] is None
assert result['reason'] == 'not enough history yet'
def test_no_estimate_while_the_gauge_has_not_moved():
"""A printer reporting in 10% steps sits on a plateau for a fortnight."""
result = analyse(series([70, 70, 70, 70, 70, 70]))
assert result['daysleft'] is None
assert result['reason'] == 'level has not moved enough to estimate'
def test_a_recent_replacement_says_so_rather_than_guessing():
result = analyse(series([40, 30, 20, 10, 100, 98]))
assert result['daysleft'] is None
assert result['reason'] == 'replaced recently'
assert result['replacements'] == 1
assert result['lastreplaced'] is not None
def test_empty_history_is_reported_not_crashed():
result = analyse([])
assert result['reason'] == 'no history'
assert result['currentlevel'] is None
assert result['replacements'] == 0
def test_junk_rows_are_dropped_not_fatal():
points = normalise([('notaclock', '50'), ('1780000000', 'n/a'),
('1780000000', '50')])
assert len(points) == 1
def test_days_left_never_goes_negative():
result = analyse(series([20, 15, 10, 5, 0]))
assert result['daysleft'] == 0
def test_printer_sorts_by_its_soonest_supply():
"""A colour MFP is as urgent as its most pressing cartridge."""
supplies = [{'daysleft': 40}, {'daysleft': 6}, {'daysleft': None}]
assert soonest(supplies) == 6
def test_soonest_is_none_when_nothing_can_be_estimated():
assert soonest([{'daysleft': None}, {'daysleft': None}]) is None
def test_burn_rate_needs_elapsed_time():
"""Several readings in the same second is not a rate."""
clock = str(int(START.timestamp()))
points = normalise([(clock, '90'), (clock, '80'), (clock, '70'), (clock, '60')])
assert burn_rate(points) is None
def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db):
"""A dashboard must be told the data is missing, not shown an empty list
that reads as 'nothing runs out soon'."""
response = client.get('/api/printers/supplies/forecast')
assert response.status_code == 200
body = response.get_json()['data']
assert body['available'] is False
assert 'not configured' in body['reason']
assert body['printers'] == []