Automating Calibration Reminder Emails for Lab Equipment

Problem statement #

You need a Python job that scans a synchronized equipment inventory, emails calibration reminders to the asset owner and compliance officer for every instrument approaching its due date, and — critically — can be re-run daily without ever sending a duplicate, so that a missed calibration window never becomes an NIH or OSHA audit finding.

This task sits under Calibration Due Date Routing, part of the broader Equipment Calibration & Lab Inventory Tracking practice. The dispatcher is intentionally narrow: it reads due dates, classifies urgency, delivers mail, escalates failures, and fingerprints every attempt. It does not own the canonical inventory or resolve tenant ownership — that routing logic belongs to the parent cluster — and it does not adjudicate whether an instrument is in or out of compliance; it surfaces the deadline and records the dispatch for a human to act on.

Prerequisites #

Before deploying the reminder job, confirm the following environment and policy configuration:

• Python 3.10+ (the code uses modern type hints and datetime.timezone). No third-party packages are required — smtplib, sqlite3, and hashlib are all standard library, which keeps campus IT review and the deployment footprint minimal.
• A synchronized inventory store. This example uses SQLite for portability, but the same schema and idempotency key work against PostgreSQL behind SQLAlchemy. Asset ownership and recipient resolution must already be populated by the Calibration Due Date Routing layer so the dispatcher never hard-codes institutional addresses.
• Environment variables for SMTP credentials (never commit them, consistent with Security Boundary Configuration):
  • SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASS — campus relay connection details.
• A monitored fallback inbox. When the primary relay rejects or times out, escalations route here. It must be watched by compliance staff, not an automated ticket bot, so a human acknowledges the missed delivery. Persistent relay outages should divert through your Fallback Routing Protocols.

Policy and compliance context #

The job is architected so the regulatory boundary is satisfied by design, not bolted on:

• NIH & NSF grant compliance. Federal sponsors require documented maintenance histories for shared instrumentation. Each reminder is traceable to a specific grant-funded asset, and delivery logs must survive an institutional audit — that is what the append-only dispatch_log and its audit hash provide.
• OSHA Laboratory Standard (29 CFR 1910.1450). Safety-critical equipment (fume hoods, centrifuges, gas monitors) must remain calibrated. The urgency tier guarantees that a breached threshold escalates rather than sitting silently in a queue.
• EPA environmental monitoring. Instruments tracking emissions or chemical storage face strict calibration intervals; the cryptographic fingerprint satisfies record-keeping requirements for regulatory inspection.

Reminders must never expose personally identifiable information in transit, must use least-privilege database access, and must keep an immutable audit trail for every dispatch attempt — enforced below through state tracking and SHA-256 hashing.

Step-by-step implementation #

The flow the dispatcher enforces: query assets inside the lookback window, skip any already notified this cycle (the idempotency guard), compute an urgency tier, attempt delivery, escalate failures to the fallback recipients, and fingerprint every outcome into an append-only log. The composite unique key on (asset_id, cycle_date, recipient_hash) is what makes a re-run safe.

Figure: the UNIQUE dispatch-log constraint makes re-running the cycle safe — already-notified assets are skipped, failures escalate.

Step 1 — Declare the schema with the idempotency constraint #

The dispatch_log table is the audit ledger. Its UNIQUE(asset_id, cycle_date, recipient_hash) constraint is the single guarantee that powers idempotency: a second insert for the same asset, cycle, and recipient set is silently ignored rather than producing a duplicate email.

import os
import json
import hashlib
import logging
import smtplib
import sqlite3
from datetime import datetime, timedelta, timezone
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Dict, List

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.FileHandler("calibration_audit.log"), logging.StreamHandler()],
)


def init_db(db_path: str) -> None:
    """Create the inventory and append-only dispatch ledger if absent."""
    with sqlite3.connect(db_path) as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS equipment (
                asset_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                calibration_due DATE NOT NULL,
                owner_email TEXT NOT NULL,
                compliance_email TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS dispatch_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                asset_id TEXT NOT NULL,
                cycle_date DATE NOT NULL,
                recipient_hash TEXT NOT NULL,
                status TEXT NOT NULL,
                audit_hash TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(asset_id, cycle_date, recipient_hash)  -- idempotency key
            );
            """
        )
        conn.commit()

Step 2 — Classify urgency and fingerprint the payload #

Urgency tiering drives both the subject line and the escalation decision. The audit hash is computed from a canonical, sorted JSON dump so the same dispatch always produces the same fingerprint — the property a compliance officer relies on to prove a record was not altered after the fact.

def calculate_urgency(due_date: str) -> str:
    """CRITICAL <= 7 days, WARNING <= 30, else INFO (OSHA-driven thresholds)."""
    days_remaining = (datetime.strptime(due_date, "%Y-%m-%d") - datetime.now()).days
    if days_remaining <= 7:
        return "CRITICAL"
    if days_remaining <= 30:
        return "WARNING"
    return "INFO"


def generate_audit_hash(payload: dict) -> str:
    """Deterministic SHA-256 fingerprint for non-repudiation."""
    canonical = json.dumps(payload, sort_keys=True).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


def recipient_fingerprint(recipients: List[str]) -> str:
    """Stable hash of the recipient set so order never breaks the idempotency key."""
    return hashlib.sha256("|".join(sorted(recipients)).encode("utf-8")).hexdigest()

Step 3 — Guard against duplicates and record every attempt #

Before sending, the dispatcher checks the ledger; after sending, it writes the outcome with INSERT OR IGNORE so even a race between two concurrent runs cannot double-send. Note PII never lands in the log — only the recipient hash is stored.

def already_dispatched(db_path: str, asset_id: str, cycle_date: str, recipients: List[str]) -> bool:
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            "SELECT 1 FROM dispatch_log WHERE asset_id=? AND cycle_date=? AND recipient_hash=?",
            (asset_id, cycle_date, recipient_fingerprint(recipients)),
        ).fetchone()
        return row is not None


def log_dispatch(db_path: str, asset_id: str, cycle_date: str,
                 recipients: List[str], status: str, audit_hash: str) -> None:
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            """INSERT OR IGNORE INTO dispatch_log
               (asset_id, cycle_date, recipient_hash, status, audit_hash)
               VALUES (?, ?, ?, ?, ?)""",
            (asset_id, cycle_date, recipient_fingerprint(recipients), status, audit_hash),
        )
        conn.commit()

Step 4 — Deliver, escalate, and run the cycle #

Primary delivery is attempted against the campus relay. On any SMTPException the payload escalates to the monitored fallback inbox and is logged as FALLBACK_ESCALATED — the failure is never swallowed.

def send_email(smtp_config: Dict, to_addrs: List[str], subject: str, body: str) -> bool:
    msg = MIMEMultipart("alternative")
    msg["Subject"], msg["From"], msg["To"] = subject, smtp_config["sender"], ", ".join(to_addrs)
    msg.attach(MIMEText(body, "plain"))
    try:
        with smtplib.SMTP(smtp_config["host"], smtp_config["port"], timeout=20) as server:
            if smtp_config.get("use_tls", True):
                server.starttls()
            server.login(smtp_config["user"], smtp_config["password"])
            server.sendmail(smtp_config["sender"], to_addrs, msg.as_string())
        return True
    except smtplib.SMTPException as exc:
        logging.error("SMTP delivery failed: %s", exc)
        return False


def run_dispatch_cycle(db_path: str, smtp_config: Dict, fallback_email: str,
                       lookback_days: int = 45) -> None:
    threshold = (datetime.now() + timedelta(days=lookback_days)).strftime("%Y-%m-%d")
    cycle_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    with sqlite3.connect(db_path) as conn:
        assets = conn.execute(
            """SELECT asset_id, name, calibration_due, owner_email, compliance_email
               FROM equipment WHERE calibration_due <= ?""",
            (threshold,),
        ).fetchall()

    for asset_id, name, due_date, owner, compliance in assets:
        recipients = [owner, compliance]
        if already_dispatched(db_path, asset_id, cycle_date, recipients):
            logging.info("Skipping %s: already dispatched for cycle %s", asset_id, cycle_date)
            continue

        urgency = calculate_urgency(due_date)
        subject = f"[{urgency}] Calibration Reminder: {name} (Due: {due_date})"
        body = (
            f"Asset: {name} ({asset_id})\n"
            f"Calibration Due: {due_date}\nUrgency: {urgency}\n"
            "Please schedule service to maintain compliance.\n"
        )
        audit_hash = generate_audit_hash(
            {"asset_id": asset_id, "due_date": due_date, "urgency": urgency, "recipients": recipients}
        )

        if send_email(smtp_config, recipients, subject, body):
            log_dispatch(db_path, asset_id, cycle_date, recipients, "DELIVERED", audit_hash)
            logging.info("Dispatched %s | audit=%s", asset_id, audit_hash)
        else:
            fallback = [fallback_email, compliance]
            send_email(smtp_config, fallback, f"[FALLBACK] {subject}", body)
            log_dispatch(db_path, asset_id, cycle_date, fallback, "FALLBACK_ESCALATED", audit_hash)
            logging.warning("Escalated %s to fallback | audit=%s", asset_id, audit_hash)


if __name__ == "__main__":
    config = {
        "host": os.getenv("SMTP_HOST", "smtp.university.edu"),
        "port": int(os.getenv("SMTP_PORT", "587")),
        "use_tls": True,
        "user": os.getenv("SMTP_USER"),
        "password": os.getenv("SMTP_PASS"),
        "sender": "lab-compliance@university.edu",
    }
    init_db("lab_inventory.db")
    run_dispatch_cycle("lab_inventory.db", config, "compliance-escalation@university.edu")

Schedule this with cron or a campus task runner. Because the idempotency key is scoped to cycle_date, running it daily, weekly, or on demand re-issues a reminder at most once per calendar day per recipient set.

Schema and field reference #

The two tables the dispatcher relies on. Widen them in your version-controlled inventory config rather than in code.

Verification #

Confirm a run behaved correctly before trusting it in production:

1. Dry-run idempotency: seed one due asset, run the cycle twice. Exactly one dispatch_log row must exist for that (asset_id, cycle_date, recipient_hash), and the second run must log Skipping ... already dispatched.
2. Reproduce the hash: re-run generate_audit_hash on the same payload dict and confirm it matches SELECT audit_hash FROM dispatch_log WHERE asset_id = ?. An equal hash proves the record was not altered.
3. Force the fallback path: point SMTP_HOST at an unreachable port and confirm the row lands as FALLBACK_ESCALATED and the fallback inbox receives the [FALLBACK] subject.
4. Tier boundaries: set calibration_due to today+7, today+30, and today+31 and confirm CRITICAL, WARNING, and INFO respectively.

Troubleshooting #

Three gotchas specific to this dispatcher:

• Duplicate emails after a schema change. If you altered dispatch_log and dropped the composite UNIQUE constraint, INSERT OR IGNORE silently degrades to a plain insert and re-runs double-send. Recreate the table with the UNIQUE(asset_id, cycle_date, recipient_hash) constraint, then verify with the dry-run above.
• cycle_date drift across timezones. Mixing a naive datetime.now() for the cycle with a UTC value elsewhere can roll the date over midnight and reset the idempotency guard, re-sending everything. Pin cycle_date to a single timezone (UTC above) for every run.
• SMTP 550 rejection. The campus relay enforces SPF/DKIM alignment and your sender domain does not match institutional DNS. Align the sender to the authorized domain with IT, and during a sustained relay outage divert through your Fallback Routing Protocols rather than letting the queue saturate.

Frequently asked questions #

Related #

• Up to the parent topic: Calibration Due Date Routing
• Tracking High-Frequency Instrument Usage with IoT Sensors
• Setting Dynamic Reorder Thresholds for Lab Consumables
• Mapping Equipment Locations Across Multi-Building Campuses
• Fallback Routing Protocols