Scheduling preventive maintenance from usage thresholds

Problem statement #

You need a Python job that reads each instrument’s usage counters and preventive-maintenance rules, decides whether runtime hours, cycle count, or elapsed time has crossed a service threshold, and generates a work order for the ones that have — idempotently, so that a re-run against unchanged counters generates nothing and a single threshold crossing generates exactly one order tied to a grant-funded asset.

This task sits under Preventive Maintenance Scheduling, part of the broader Equipment Calibration & Lab Inventory Tracking practice. The job is narrow: it turns counters and rules into work orders and ledger entries. It does not perform the service, dispatch a technician, or reconcile cost — it produces the auditable trigger that the work-order system and the Immutable Audit Report Generation layer act on.

Prerequisites #

• Python 3.10+ for modern type hints and datetime.now(timezone.utc).
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] for the idempotent insert. Hashing uses the standard-library hashlib. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • MAINT_DB_URL — the SQLAlchemy connection string for the maintenance store.
• Policy config: a version-controlled table of per-asset PM intervals (runtime hours, cycles, elapsed days) and each asset’s funding source, kept alongside your University Policy Mapping Frameworks. Current counter readings are supplied upstream by Tracking High-Frequency Instrument Usage with IoT Sensors; this page assumes the latest readings are already available.

Step-by-step implementation #

The job loads assets and rules, reads the current counters, evaluates the whichever-comes-first trigger, generates a work order under an interval-scoped idempotency key, and writes an audit hash to the maintenance ledger. The interval-scoped key is the single thing that makes the whole job safe to re-run.

Figure: the evaluator selects the first-crossed clock; the interval-scoped key makes the resulting work order write exactly once.

Step 1 — Load assets, rules, and current counters #

Begin by assembling, per asset, the three thresholds and the current readings. Keep the interval already serviced (serviced_interval) alongside the rule — it is the baseline the evaluator compares the newly computed interval against, and it is what stops a counter parked at the boundary from re-firing.

import hashlib
import logging
from dataclasses import dataclass
from datetime import datetime, timezone

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

logging.basicConfig(level=logging.INFO, format="%(asctime)s [PM] %(message)s")
logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class AssetPM:
    asset_id: str
    funding_source: str
    runtime_threshold: float | None      # hours between services
    cycle_threshold: int | None          # cycles between services
    elapsed_threshold_days: int | None   # calendar days between services
    serviced_interval: int               # highest interval already serviced
    last_service_at: datetime
    runtime_hours: float                 # current reading
    cycle_count: int                     # current reading

Step 2 — Evaluate the whichever-comes-first trigger #

For each clock, compute the interval index — how many whole thresholds the counter has passed. A service is due only when the highest index across the clocks exceeds the interval already serviced. Returning the new interval index is essential; it becomes part of the idempotency key.

@dataclass(frozen=True)
class Trigger:
    asset_id: str
    trigger_type: str
    interval_index: int


def evaluate(asset: AssetPM, now: datetime) -> Trigger | None:
    """Return the first-crossed trigger whose interval exceeds the one already
    serviced, or None. interval_index = floor(current / threshold)."""
    reached: list[Trigger] = []

    if asset.runtime_threshold:
        idx = int(asset.runtime_hours // asset.runtime_threshold)
        reached.append(Trigger(asset.asset_id, "runtime_hours", idx))
    if asset.cycle_threshold:
        idx = int(asset.cycle_count // asset.cycle_threshold)
        reached.append(Trigger(asset.asset_id, "cycle_count", idx))
    if asset.elapsed_threshold_days:
        elapsed = (now - asset.last_service_at).days
        idx = elapsed // asset.elapsed_threshold_days
        reached.append(Trigger(asset.asset_id, "elapsed_days", idx))

    # The clock that reached the furthest interval fired first in real time.
    top = max(reached, key=lambda t: t.interval_index, default=None)
    if top is None or top.interval_index <= asset.serviced_interval:
        return None  # nothing new is due — comparing against serviced baseline
    return top

Step 3 — Derive the interval-scoped idempotency key #

The key must encode the interval, not just the asset. Two services of the same instrument (its second and its third) must produce different keys, while two runs over the same second-interval crossing must produce the same key.

def work_order_id(asset_id: str, trigger_type: str, interval_index: int) -> str:
    """Deterministic: the SAME interval always yields the SAME id, so a re-run
    collapses to a no-op; a NEW interval yields a NEW id."""
    raw = f"{asset_id}:{trigger_type}:{interval_index}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]


def audit_hash(asset_id: str, interval_index: int, funding_source: str, ts: str) -> str:
    """Fingerprint the ledger entry so a stored order can be re-verified."""
    canonical = f"{asset_id}|{interval_index}|{funding_source}|{ts}"
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

Step 4 — Generate the work order and write the ledger #

Insert with ON CONFLICT DO NOTHING on the deterministic id. Only a genuinely new row appends a ledger entry, so a re-run that inserts nothing also ledgers nothing. The funding source rides along so the maintenance cost can later be allocated to the benefiting award under 2 CFR 200.436.

from sqlalchemy.dialects.postgresql import insert as pg_insert


def generate(session: Session, asset: AssetPM, trigger: Trigger) -> str | None:
    now = datetime.now(timezone.utc)
    ts = now.isoformat(timespec="seconds")
    wo_id = work_order_id(trigger.asset_id, trigger.trigger_type, trigger.interval_index)
    ah = audit_hash(asset.asset_id, trigger.interval_index, asset.funding_source, ts)

    stmt = pg_insert(WorkOrder).values(
        work_order_id=wo_id,
        asset_id=asset.asset_id,
        trigger_type=trigger.trigger_type,
        interval_index=trigger.interval_index,
        funding_source=asset.funding_source,
        status="scheduled",
        created_at=now,
    ).on_conflict_do_nothing(index_elements=["work_order_id"])

    result = session.execute(stmt)
    if not result.rowcount:
        session.rollback()
        return None  # this interval already has an order — idempotent no-op

    session.add(MaintenanceLedger(
        work_order_id=wo_id, asset_id=asset.asset_id,
        interval_index=trigger.interval_index, trigger_type=trigger.trigger_type,
        funding_source=asset.funding_source, audit_hash=ah,
        event="scheduled", recorded_at=now,
    ))
    session.commit()
    logger.info("scheduled %s for %s (interval %d)",
                wo_id, asset.asset_id, trigger.interval_index)
    return wo_id


def run_scheduler(assets: list[AssetPM], session: Session) -> dict[str, int]:
    now = datetime.now(timezone.utc)
    stats = {"generated": 0, "skipped": 0}
    for asset in assets:
        trigger = evaluate(asset, now)
        if trigger is None:
            stats["skipped"] += 1
            continue
        wo_id = generate(session, asset, trigger)
        stats["generated" if wo_id else "skipped"] += 1
    return stats

Schema and field reference #

Verification #

1. Re-run generates nothing. Run run_scheduler twice against unchanged counters. The second call must report generated=0, and SELECT count(*) FROM maintenance_ledger WHERE recorded_at >= :run_start must return 0 on the second pass.
2. One order per interval. After a crossing, confirm exactly one work order exists for that asset_id and interval_index: SELECT count(*) FROM work_orders WHERE asset_id=:a AND interval_index=:i returns 1.
3. Reproduce the audit hash. Recompute audit_hash from the stored asset, interval, funding source, and timestamp and confirm it equals the persisted value — an equal hash proves the ledger entry matches the order it describes.
4. Next interval still fires. Advance a counter past the next threshold and confirm a new, distinct work_order_id is generated while the prior interval’s order is untouched.

Troubleshooting #

Three gotchas specific to usage-threshold scheduling:

• Double-generating without an interval-scoped key. If the idempotency key is asset_id alone, the first service blocks every subsequent one, or a per-run timestamp key lets each run create a fresh order — both are wrong. The key must be asset_id + trigger_type + interval_index so one interval maps to one order and consecutive intervals each get their own. This is the single most common cause of a grant being double-charged for one service.
• Threshold hysteresis and flapping. A counter parked just above the threshold re-evaluates as “due” on every run, generating a cancel-and-regenerate loop. Compare the computed interval index against a persisted serviced_interval baseline and advance that baseline on completion, so the evaluator measures against the next interval rather than the one just crossed.
• Elapsed-vs-usage confusion. Treating the three clocks as alternatives to pick between, rather than as parallel timers where the earliest wins, silently ignores whichever axis you did not check. Always compute all three interval indices and take the maximum; a pump can be well under its runtime limit yet long past its calendar limit, and the calendar clock must still fire.

Frequently asked questions #

Related #

• Up to the parent topic: Preventive Maintenance Scheduling
• Tracking High-Frequency Instrument Usage with IoT Sensors — the counters this job reads
• Automating Calibration Reminder Emails for Lab Equipment — the sibling due-date clock for calibration
• Escalating Overdue Calibration to Lab Managers — escalation when a due service is missed