OSHA Laboratory Standard Automation

The OSHA Laboratory Standard — 29 CFR 1910.1450 — is written for people, not for machines: it obliges a university to keep a Chemical Hygiene Plan, to determine employee exposure, to impose extra controls on particularly hazardous substances, and to keep those obligations current as bottles move on and off the shelf. Most institutions discharge that duty with a binder and an annual walk-through, which means the inventory of record and the compliance state of record drift apart the moment a graduate student opens a new can of diethyl ether. This guide closes that gap by turning the standard into executable checks that run against a live chemical inventory, and it is one of the compliance layers anchored to the parent guide on Hazardous Material & Chemical Inventory Compliance. Each container row is evaluated against a policy matrix derived from the CHP; the evaluator flags over-threshold quantities, missing engineering or administrative controls, expired peroxide-forming reagents, and any container that has no linked safety data sheet, then routes the offending rows to a quarantine list an environmental-health-and-safety officer works through.

The audience is concrete: the chemical hygiene officer who signs the CHP, the lab manager who owns the shelf, the compliance analyst who answers to an OSHA inspector, and the Python developer who has to keep the two in sync. This guide shares its container schema with Safety Data Sheet Management, hands disposal-bound containers off to EPA RCRA Hazardous Waste Tracking, and reads its physical container counts from the same asset store used by Equipment Calibration & Lab Inventory Tracking. The step-by-step Python that implements a single evaluation pass lives in the companion how-to on automating OSHA chemical hygiene inventory checks.

Problem framing #

The Laboratory Standard is deceptively broad, and the failure it invites is silent. A CHP written in 2019 names benzene as a select carcinogen requiring a designated area; three years later benzene shows up on a shelf in a room the CHP never designated, and nobody notices because no automated check ever compares the two. A one-liter bottle of tetrahydrofuran is opened, its peroxide clock starts, and twelve months later it is an explosion risk that the annual inspection missed by six weeks. A newly received reagent has a container barcode but no linked safety data sheet, so the hazard communication chain required under 29 CFR 1910.1200 is broken from the first day. None of these are exotic; they are the ordinary entropy of a working lab, and a binder cannot keep up with them.

The design principle behind this layer is that compliance is a derived property of the inventory, recomputed on every run, never a document someone remembers to update. Three contracts hold the line:

• Determinism. The same inventory and the same CHP matrix produce the same set of flags every time. Containers are processed in a stable, hash-sorted order so the output never depends on database row order.
• Idempotency. Each evaluation writes to a compliance ledger keyed on container plus content fingerprint; re-running the evaluator over an unchanged inventory adds no duplicate ledger rows and raises no duplicate flags.
• Flag, don’t fix. The evaluator never edits inventory quantities, never relocates a container, and never manufactures a missing SDS. It records a flag with a reason code and routes the container to a quarantine list an EHS officer resolves — because remediation under the Laboratory Standard is a human decision with a signature attached.

Policy constraints #

Compliance is the boundary this layer enforces, not a report it produces afterward. Every check below traces to a specific subsection of the Laboratory Standard or the Hazard Communication Standard it depends on, and the policy matrix that drives the evaluator is itself a version-controlled artifact reviewed under the same University Policy Mapping Frameworks that govern grant policy.

Operational boundary. The CHP dictates which substances are particularly hazardous, what controls each hazard class requires, and how long a peroxide-forming container may sit before it must be tested or discarded. The evaluator supplies the mechanics: it joins each container to its matrix row, compares quantities to thresholds, checks control presence, computes expiry against the received or opened date, and confirms SDS linkage. It must never quietly downgrade a hazard class to make a container pass, and no evaluation run may be skipped to clear a shelf before an inspection. Peroxide-former expiry windows — commonly three months for opened Class I formers, six months for Class II, and twelve months for stabilized Class III — are policy inputs, not constants baked into code, so an institution can tighten them without a redeploy.

Data schema & field mapping #

The evaluator reads a single canonical container record. Source inventories arrive with wildly different column names — a barcode system calls it BOTTLE_ID, a legacy spreadsheet calls it Chem # — so before evaluation those fields are mapped onto the schema below, whose constraints encode the regulatory rules from the previous section. The CAS number is the join key into the CHP matrix; it must be normalized and checksum-validated, because a mistyped CAS number silently detaches a container from its hazard policy.

The flag field is the only evaluator-owned column; everything above it maps from the source inventory or is derived from the CHP matrix. A container may accrue more than one flag in a single pass — an expired peroxide former with no linked SDS trips both expired and no_sds — so the evaluator emits a list of reasons per container rather than a single verdict, and the ledger stores each reason as its own row so an auditor can count unresolved findings by category.

Implementation #

The evaluation layer has three composable parts: a CHP matrix loader that proves the policy is well-formed before any container is checked, a deterministic evaluator that partitions the inventory into compliant and flagged containers, and an idempotent write path that appends findings to the compliance ledger keyed on container plus content hash. The Python below is the reference for a full pass; the companion how-to breaks the same logic into annotated steps.

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone

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

# Peroxide-former expiry windows in days, keyed by peroxide class.
# These are POLICY INPUTS sourced from the CHP, not constants — an
# institution may tighten them without touching evaluator code.
PEROXIDE_WINDOWS_DAYS: dict[str, int] = {"I": 90, "II": 180, "III": 365}


@dataclass(frozen=True)
class ChpRule:
    """One row of the Chemical Hygiene Plan policy matrix, keyed by CAS."""
    cas_number: str
    hazard_class: str
    phs: bool
    threshold_qty: float          # normalized to grams or millilitres
    threshold_unit: str
    control_required: str | None  # e.g. "designated_area", "fume_hood"
    peroxide_class: str | None    # "I" | "II" | "III" | None


@dataclass
class Finding:
    container_id: str
    cas_number: str
    reasons: list[str] = field(default_factory=list)


class ChpEvaluator:
    """Deterministic, idempotent evaluator for OSHA 29 CFR 1910.1450 checks."""

    def __init__(self, matrix: dict[str, ChpRule], matrix_version: str) -> None:
        # Pre-flight: a container with no matrix row cannot be evaluated, so an
        # empty or malformed matrix must fail loudly at startup, not silently
        # pass every container.
        if not matrix:
            raise ValueError("CHP policy matrix is empty; refusing to evaluate.")
        self.matrix = matrix
        self.matrix_version = matrix_version

    @staticmethod
    def _content_hash(container: dict) -> str:
        """Deterministic SHA-256 over canonical container content."""
        canonical = json.dumps(container, sort_keys=True, separators=(",", ":"), default=str)
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def _evaluate_one(self, c: dict, today: date) -> Finding:
        finding = Finding(container_id=c["container_id"], cas_number=c["cas_number"])
        rule = self.matrix.get(c["cas_number"])

        # 1910.1450(e): an unmatched hazard class means the CHP does not cover
        # this chemical — a finding in its own right.
        if rule is None:
            finding.reasons.append("unmapped_hazard_class")
            return finding

        # 1910.1450(d): exposure determination via quantity threshold.
        if c["quantity"] > rule.threshold_qty:
            finding.reasons.append("over_threshold")

        # 1910.1450 App A: particularly hazardous substances need the declared
        # control physically present.
        if rule.control_required and c.get("control_present") != rule.control_required:
            finding.reasons.append("control_missing")

        # Peroxide-former expiry: 3/6/12-month window by peroxide class.
        if rule.peroxide_class:
            window = timedelta(days=PEROXIDE_WINDOWS_DAYS[rule.peroxide_class])
            expiry = c.get("expiry_date")
            if expiry is None or today > expiry or today > (c["received_date"] + window):
                finding.reasons.append("expired")

        # 1910.1200: hazard communication requires a linked SDS.
        if not c.get("sds_id"):
            finding.reasons.append("no_sds")

        return finding

    def evaluate_batch(
        self, containers: list[dict], today: date | None = None
    ) -> tuple[list[Finding], list[Finding]]:
        """Return (compliant, flagged) findings in deterministic order."""
        today = today or datetime.now(timezone.utc).date()
        compliant: list[Finding] = []
        flagged: list[Finding] = []

        # Sort by content hash so output order never depends on inventory order.
        for container in sorted(containers, key=self._content_hash):
            finding = self._evaluate_one(container, today)
            (flagged if finding.reasons else compliant).append(finding)

        logger.info("evaluated %d containers: %d compliant, %d flagged",
                    len(containers), len(compliant), len(flagged))
        return compliant, flagged

The evaluator is stateless and pure: it reads containers and a matrix, returns findings, and mutates nothing. That is what makes a second run over the same inventory reproduce the first exactly. The write path below turns findings into an append-only ledger, and the idempotency guarantee lives in the database, not in application memory.

from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session


def commit_findings(
    findings: list[Finding],
    matrix_version: str,
    session: Session,
    quarantine,  # callable: (Finding) -> None
) -> dict[str, int]:
    """Append findings to the compliance ledger idempotently; route flagged
    containers to quarantine. Safe to re-run against an unchanged inventory."""
    stats = {"recorded": 0, "skipped": 0, "quarantined": 0}

    for finding in findings:
        # One ledger row per (container, reason) so unresolved findings can be
        # counted by category. An empty reasons list records a single
        # "compliant" row for the container.
        reasons = finding.reasons or ["compliant"]
        for reason in reasons:
            audit_hash = hashlib.sha256(
                f"{finding.container_id}:{reason}:{matrix_version}".encode("utf-8")
            ).hexdigest()
            stmt = pg_insert(ComplianceLedger).values(
                container_id=finding.container_id,
                cas_number=finding.cas_number,
                reason=reason,
                matrix_version=matrix_version,
                audit_hash=audit_hash,
                recorded_at=datetime.now(timezone.utc),
            )
            # The audit_hash is the natural key: an identical (container, reason,
            # matrix_version) never inserts twice, so re-runs are no-ops.
            stmt = stmt.on_conflict_do_nothing(index_elements=["audit_hash"])
            result = session.execute(stmt)
            if result.rowcount:
                stats["recorded"] += 1
            else:
                stats["skipped"] += 1

        if finding.reasons:
            stats["quarantined"] += 1
            quarantine(finding)

    session.commit()
    return stats

The on_conflict_do_nothing guarded by the deterministic audit_hash is what makes the ledger idempotent at the database layer: re-evaluating an unchanged shelf inserts nothing, a newly expired container inserts exactly one expired row, and a resolved finding simply stops being emitted on the next pass. Because each finding carries the matrix_version it was judged under, an inspector can prove which CHP a container was measured against — essential when the plan is revised mid-year and a substance is newly designated as particularly hazardous.

Integration points #

The evaluator never writes to the inventory of record; it reads a snapshot and emits findings that adjacent systems consume by key. Each boundary has an explicit contract:

• Lab inventory / LIMS. Container counts and locations come from the same asset store described in Equipment Calibration & Lab Inventory Tracking, and threshold policy is coordinated with Inventory Threshold Tuning so a reorder rule and a hazard threshold never contradict each other. The evaluator reads by container_id and writes nothing back.
• Safety data sheets. The sds_id on each container resolves through Safety Data Sheet Management; a no_sds flag is a direct signal to that subsystem that a sheet is missing or unlinked.
• Waste disposal. A container flagged expired — a peroxide former past its window — is a candidate for disposal, so its container_id is handed to EPA RCRA Hazardous Waste Tracking, where its accumulation clock begins.

An example finding published to the quarantine list for a flagged container:

{
  "container_id": "LAB07-THF-00412",
  "cas_number": "109-99-9",
  "chemical_name": "Tetrahydrofuran",
  "matrix_version": "2026.2",
  "reasons": ["expired", "no_sds"],
  "detail": {
    "peroxide_class": "I",
    "received_date": "2025-10-01",
    "window_days": 90,
    "evaluated_on": "2026-07-15"
  }
}

Verification & audit #

Every finding appends to an immutable ComplianceLedger (container id, CAS, reason, matrix version, audit hash, timestamp). That ledger is the artifact an EHS officer reconstructs an inspection from, and it shares its append-only, hash-addressed design with Immutable Audit Report Generation so a chemical finding and a financial finding are provable the same way.

To confirm a run was correct:

1. Coverage parity. The count of distinct container_id values in the ledger for a run must equal the number of containers read. A missing container is a silent drop — a defect, not an accepted state.
2. Reproduce the partition. Re-run the evaluator against the same snapshot and the same matrix_version; recorded must be 0 on the second pass and every audit_hash must already exist. A non-zero second pass means the evaluation is non-deterministic.
3. Quarantine reconciliation. Every flagged container must carry at least one reason row; the count of unresolved over_threshold, control_missing, expired, and no_sds rows is a reportable compliance metric an officer drives to zero.

from sqlalchemy import select, func


def verify_run(session: Session, matrix_version: str) -> dict[str, int]:
    rows = session.execute(
        select(ComplianceLedger.reason, func.count())
        .where(ComplianceLedger.matrix_version == matrix_version)
        .group_by(ComplianceLedger.reason)
    ).all()
    return {reason: count for reason, count in rows}

Because the ledger is append-only and every row is keyed by a deterministic audit hash, an inspector can pin any finding back to the exact container, CHP version, and moment it was raised. Retain the ledger for the duration OSHA requires for exposure-related records, and apply cryptographic checksums so a resolved finding can never be quietly deleted to clean up a report.

Failure modes & recovery #

When the evaluator meets a dirty inventory or a stale CHP, findings are isolated without halting the pass. Every recovery is idempotent-safe: re-running it cannot duplicate a ledger row.

Role boundaries. The chemical hygiene officer owns the CHP matrix and approves version promotions; they do not modify evaluator code. The Python developer owns idempotency, ordering, and the write path; they do not set hazard thresholds. The lab manager owns data accuracy at the shelf — correct CAS numbers, attested controls, linked sheets — and resolves quarantined containers before the next pass. No one clears a flag by editing the ledger; a flag clears only when the underlying container is fixed and re-evaluated.

Frequently asked questions #

Related #

• Parent guide: Hazardous Material & Chemical Inventory Compliance
• Child how-to: Automating OSHA Chemical Hygiene Inventory Checks
• EPA RCRA Hazardous Waste Tracking — where expired containers begin their disposal clock
• Safety Data Sheet Management — the source of the sds_id linkage each container needs
• Equipment Calibration & Lab Inventory Tracking — the asset store the evaluator reads container counts from