"""Consumable life tracking: page counters, replacement baselines, dropped pages. Everything here hangs off one hard-won fact: CUPS cannot be trusted to say what the printer actually did. Its page log only sees jobs it printed itself, and it counts pages the printer silently discarded - it logged "total 50" for a job that produced 3 sheets. Only the printer's own @PJL INFO PAGECOUNT is authoritative, so replacement baselines are kept on that scale and the gap between the two counters is what exposes dropped pages. """ from __future__ import annotations import json import logging from pathlib import Path import re from python_pkg.brother_printer.constants import ( CONSUMABLE_STATE_DIR, CUPS_PAGE_LOG_PATH, DRUM_RATED_PAGES, GREEN, RESET, TONER_RATED_PAGES, _out, ) from python_pkg.brother_printer.data_classes import PageCountEstimate logger = logging.getLogger(__name__) CUPS_PAGE_LOG = Path(CUPS_PAGE_LOG_PATH) CONSUMABLE_STATE_FILE = Path.home() / CONSUMABLE_STATE_DIR / "state.json" # state.json schema versions. Version 1 (implicit, no "schema" key) recorded # replacement baselines as CUPS page-log counts; version 2 records them on the # printer's own lifetime counter. See _migrate_state_to_printer_scale. STATE_SCHEMA_CUPS_SCALE = 1 STATE_SCHEMA_PRINTER_SCALE = 2 # ── Consumable state management ────────────────────────────────────── def _get_cups_total_pages() -> int: """Parse CUPS page_log to get total pages printed (deduplicated by job).""" if not CUPS_PAGE_LOG.exists(): return 0 try: text = CUPS_PAGE_LOG.read_text(encoding="utf-8", errors="replace") except OSError: return 0 jobs: dict[str, int] = {} for line in text.splitlines(): match = re.search(r"\s(\d+)\s+\[.*?\]\s+total\s+(\d+)", line) if match: job_id = match.group(1) pages = int(match.group(2)) jobs[job_id] = max(jobs.get(job_id, 0), pages) return sum(jobs.values()) def _load_consumable_state() -> dict[str, int]: """Load consumable replacement state from disk.""" defaults: dict[str, int] = { "toner_replaced_at": 0, "drum_replaced_at": 0, "schema": STATE_SCHEMA_CUPS_SCALE, # Snapshot of both counters at the last successful printer read, used to # spot dropped pages and to place a CUPS-log figure on the printer's # scale when the printer cannot be reached. "last_printer_count": 0, "last_cups_total": 0, } if not CONSUMABLE_STATE_FILE.exists(): return defaults try: data = json.loads( CONSUMABLE_STATE_FILE.read_text(encoding="utf-8"), ) return { "toner_replaced_at": int(data.get("toner_replaced_at", 0)), "drum_replaced_at": int(data.get("drum_replaced_at", 0)), "schema": int(data.get("schema", STATE_SCHEMA_CUPS_SCALE)), "last_printer_count": int(data.get("last_printer_count", 0)), "last_cups_total": int(data.get("last_cups_total", 0)), } except OSError, json.JSONDecodeError, ValueError, TypeError: return defaults def _migrate_state_to_printer_scale( state: dict[str, int], printer_total: int, ) -> dict[str, int]: """Rebase replacement baselines from the CUPS page log onto the printer's counter. Baselines written before this migration counted pages the CUPS log had seen, which undercounts the printer's own lifetime counter by however many pages were printed without CUPS logging them. Shift each baseline by the gap measured now, so "pages since replacement" - and therefore the reported percentages - stay put across the switch. A zero baseline means "never replaced" rather than "replaced at page zero", so it is left alone: on the printer's scale it already means "as old as the printer", which is the truthful reading. Args: state: Loaded state, possibly still on the CUPS page-log scale. printer_total: Lifetime page count from the printer's own counter. Returns: State on the printer's scale, migrated at most once. """ if state.get("schema", STATE_SCHEMA_CUPS_SCALE) >= STATE_SCHEMA_PRINTER_SCALE: return state offset = printer_total - _get_cups_total_pages() migrated = dict(state) migrated["schema"] = STATE_SCHEMA_PRINTER_SCALE if offset > 0: for key in ("toner_replaced_at", "drum_replaced_at"): if state[key] > 0: migrated[key] = state[key] + offset _save_consumable_state(migrated) return migrated def _save_consumable_state(state: dict[str, int]) -> None: """Persist consumable replacement state to disk.""" CONSUMABLE_STATE_FILE.parent.mkdir(parents=True, exist_ok=True) CONSUMABLE_STATE_FILE.write_text( json.dumps(state, indent=2) + "\n", encoding="utf-8", ) def reset_consumable(name: str, printer_total: int = 0) -> None: """Record current page count as replacement point for a consumable. Args: name: Consumable to reset, "toner" or "drum". printer_total: Lifetime count from the printer's own counter. Falls back to the CUPS page log when zero, i.e. the printer was unreachable. """ total = printer_total if printer_total > 0 else _get_cups_total_pages() state = _load_consumable_state() key = f"{name}_replaced_at" state[key] = total # Baselines written here are already on the printer's scale, so mark the # state migrated to stop the rebase shifting them a second time. if printer_total > 0: state["schema"] = STATE_SCHEMA_PRINTER_SCALE _save_consumable_state(state) _out(f"{GREEN}✓ {name.capitalize()} counter reset at page count {total}.{RESET}") _out(f" State saved to {CONSUMABLE_STATE_FILE}") def _cups_total_on_printer_scale(state: dict[str, int]) -> int: """Express the CUPS page-log count on the printer's counter scale. Replacement baselines are stored against the printer's lifetime counter, so comparing a raw CUPS figure against them subtracts two different scales. The CUPS log runs behind the printer (it only sees jobs it printed itself), so the raw number reads too low - low enough to go negative and clamp to zero, which reported a spent cartridge as 100% full. Shift by the gap measured at the last successful printer read. It is an approximation, which is why callers flag the estimate as approximate. Args: state: Loaded consumable state holding the last counter snapshot. Returns: The CUPS total shifted onto the printer's scale, or the raw total when no snapshot exists to shift it by. """ cups_total = _get_cups_total_pages() if cups_total <= 0: return 0 offset = state["last_printer_count"] - state["last_cups_total"] if state["last_printer_count"] <= 0 or offset <= 0: return cups_total return cups_total + offset def estimate_consumable_life(printer_total: int = 0) -> PageCountEstimate: """Estimate toner/drum life from pages printed since the last replacement. Args: printer_total: Lifetime count from @PJL INFO PAGECOUNT. When zero the printer could not be asked, so the CUPS page log stands in and the estimate is flagged approximate. Returns: The estimate; total_pages is zero when no counter could be read at all. """ approximate = printer_total <= 0 state = _load_consumable_state() total = _cups_total_on_printer_scale(state) if approximate else printer_total if total <= 0: return PageCountEstimate() if not approximate: state = _migrate_state_to_printer_scale(state, total) toner_pages = max(0, total - state["toner_replaced_at"]) drum_pages = max(0, total - state["drum_replaced_at"]) toner_pct = max(0, 100 - (toner_pages * 100 // TONER_RATED_PAGES)) drum_pct = max(0, 100 - (drum_pages * 100 // DRUM_RATED_PAGES)) return PageCountEstimate( total_pages=total, toner_pages=toner_pages, drum_pages=drum_pages, toner_pct_remaining=toner_pct, drum_pct_remaining=drum_pct, approximate=approximate, toner_exhausted=toner_pages >= TONER_RATED_PAGES, toner_low=toner_pages >= TONER_RATED_PAGES * 80 // 100, drum_near_end=drum_pages >= DRUM_RATED_PAGES * 90 // 100, )