Writing SPC Events Back to MES and Historians

A ticket tells the quality team an excursion happened; a write-back tells the shop floor. This guide builds the adapter that persists a detected out-of-control event — the offending value, the control limits in force, and the disposition — back into the manufacturing-execution system and the process historian, so the systems the line already watches carry the current quality state. It is the sibling of pushing alerts to Jira and ServiceNow under the routing area, implementing the same upsert contract but against a historian tag or an MES quality table rather than a ticketing REST API.

Prerequisites

  • Python 3.10+ with the AlertEvent and Violation definitions from the routing area importable.
  • A connection to the target system injected in — an OPC UA client session to the historian, or a database connection to the MES quality schema — built and authenticated in your composition root, following the patterns in connecting Python to MES and SCADA systems.
  • Write access to a quality-event table or a set of historian tags reserved for SPC state, with a unique constraint available on the event key.
  • The calculation_hash (or equivalent dataset-version identifier) that produced the control limits, so each written event is traceable back to the exact calculation.
  • A defined disposition vocabulary agreed with quality (for example OPEN, CONTAINED, DISPOSITIONED) that the write-back can set and later update.

Step 1 — Define the quality-event record

Decide exactly what the shop-floor system must hold. The record carries enough to reconstruct the signal and its state without reaching back to any other system: the identity, the point and its limits, the disposition, and the traceability hash. Build it in one place so every write path is consistent.

from __future__ import annotations

import time
from typing import Any

# Disposition lifecycle the MES/historian record moves through.
VALID_DISPOSITIONS = {"OPEN", "CONTAINED", "DISPOSITIONED", "FALSE_ALARM"}


def build_quality_record(
    event: "AlertEvent",
    calculation_hash: str,
    disposition: str = "OPEN",
) -> dict[str, Any]:
    """Assemble the row/tag payload persisted to the MES or historian."""
    if disposition not in VALID_DISPOSITIONS:
        raise ValueError(f"Unknown disposition {disposition!r}; expected {VALID_DISPOSITIONS}")
    v = event.violation
    return {
        "spc_key": event.key,            # unique idempotency key
        "process_id": event.process_id,
        "rule_id": event.rule_id,
        "alarm_code": event.alarm_code,
        "severity": event.severity.name,
        "point_value": float(v.value),
        "center": float(v.center),
        "ucl": float(v.ucl),
        "lcl": float(v.lcl),
        "signal_epoch": float(v.timestamp),
        "calculation_hash": calculation_hash,  # traceability to the limit calc
        "disposition": disposition,
        "written_epoch": time.time(),
    }

Storing the limits alongside the value is deliberate: an auditor reading the historian months later must see the point and the limits it violated, not just a flag, and the calculation_hash ties those limits to the exact dataset and code that produced them.

Step 2 — Choose the write target

The same record persists two ways depending on the backend, and the injected connection hides the difference from the adapter:

  • Historian via OPC UA. Write the event to a small set of structured tags reserved for SPC state on the process node — for example a string tag holding the alarm code and a numeric tag holding the point value — using the OPC UA client's write service. Historians timestamp and archive each tag write, giving you the time series for free.
  • MES quality table via SQL. Insert (or upsert) a row into the MES quality-event table. This is the better fit when the disposition lifecycle matters, because a relational row is easy to update in place as the state moves from OPEN to DISPOSITIONED.

Prefer the MES table when you need the full disposition lifecycle and traceability joins; prefer historian tags when you mainly need the event to appear on operator trends alongside the process value. Many plants do both, writing a tag for visibility and a row for the record of disposition.

Step 3 — Implement the transactional, idempotent write

The write-back must be atomic and idempotent: a retried delivery from the router must not double-insert, and a partial write must not leave a row without its limits. Wrap the insert-or-update in a transaction and key it on spc_key, so the first write creates the row and any re-fire updates it in place.

class MesWriteBackAdapter:
    """Persist SPC events to an MES quality table with a transactional upsert.

    The database connection is injected; the adapter is agnostic to the exact
    driver and can be tested against a fake connection with no database.
    """

    def __init__(self, conn: Any, name: str = "mes", table: str = "spc_quality_event") -> None:
        self.conn = conn
        self.name = name
        self.table = table

    def upsert(self, event: "AlertEvent", calculation_hash: str = "") -> str:
        """Insert or update the quality-event row for event.key; return the key."""
        record = build_quality_record(event, calculation_hash)
        cols = list(record.keys())
        placeholders = ", ".join(["?"] * len(cols))
        col_list = ", ".join(cols)
        # Update every column except the immutable key on conflict, so a re-fire
        # refreshes the row instead of raising or duplicating it.
        updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "spc_key")

        sql = (
            f"INSERT INTO {self.table} ({col_list}) VALUES ({placeholders}) "
            f"ON CONFLICT(spc_key) DO UPDATE SET {updates}"
        )

        try:
            with self.conn:  # transaction: commit on success, roll back on error
                self.conn.execute(sql, tuple(record.values()))
        except Exception as exc:  # noqa: BLE001 - surface to the router's retry logic
            raise RuntimeError(f"MES write-back failed for {event.key}: {exc}") from exc
        return event.key

The with self.conn: block is the transaction boundary — either the row is fully written or nothing is, so a failure mid-write never leaves a value without its limits. The ON CONFLICT clause is the idempotency mechanism: the unique constraint on spc_key guarantees one row per event no matter how many times the router retries.

Step 4 — Update the disposition as the reaction closes

The write-back is not one-and-done. As the out-of-control action plan runs, the disposition moves, and the same row is updated so the shop-floor state stays current. Because the row is keyed on spc_key, closing the loop is a targeted update.

def set_disposition(conn: Any, table: str, spc_key: str, disposition: str) -> None:
    """Advance the quality-event disposition, e.g. OPEN -> DISPOSITIONED."""
    if disposition not in VALID_DISPOSITIONS:
        raise ValueError(f"Unknown disposition {disposition!r}")
    with conn:
        cur = conn.execute(
            f"UPDATE {table} SET disposition = ?, written_epoch = ? WHERE spc_key = ?",
            (disposition, time.time(), spc_key),
        )
        if cur.rowcount == 0:
            raise LookupError(f"No quality-event row for key {spc_key}")

Writing the disposition back is what closes the audit loop: the MES now holds not just that an excursion happened but how it was resolved, tied by calculation_hash to the limits and by spc_key to the ticket the ticketing adapter filed.

Verification

Use an in-memory SQLite connection as the fake so the test exercises the real transactional and conflict logic without a live MES.

import sqlite3


def make_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    conn.execute(
        "CREATE TABLE spc_quality_event ("
        "spc_key TEXT PRIMARY KEY, process_id TEXT, rule_id TEXT, alarm_code TEXT, "
        "severity TEXT, point_value REAL, center REAL, ucl REAL, lcl REAL, "
        "signal_epoch REAL, calculation_hash TEXT, disposition TEXT, written_epoch REAL)"
    )
    return conn


def test_idempotent_writeback_and_disposition():
    conn = make_conn()
    adapter = MesWriteBackAdapter(conn)
    event = make_event(key="k-42", process_id="LINE7-BORE-DIA")  # test helper

    for _ in range(3):                       # three re-fires of one excursion
        adapter.upsert(event, calculation_hash="hash-abc")

    rows = conn.execute("SELECT COUNT(*) FROM spc_quality_event").fetchone()[0]
    assert rows == 1                         # one row, not three

    set_disposition(conn, "spc_quality_event", "k-42", "DISPOSITIONED")
    state = conn.execute(
        "SELECT disposition, calculation_hash FROM spc_quality_event WHERE spc_key=?",
        ("k-42",),
    ).fetchone()
    assert state == ("DISPOSITIONED", "hash-abc")
    print("verified single row with closed disposition:", state)

Expected output is a single row whose disposition is DISPOSITIONED and whose calculation_hash is preserved. If the count comes back as three, the unique constraint or the ON CONFLICT clause is missing; if the disposition update raises LookupError, the key written does not match the key queried.

Root-Cause Table

Symptom Cause Fix
Duplicate rows for one excursion No unique constraint on the event key, or a plain INSERT without conflict handling Add a unique key on spc_key and use insert-or-update on conflict
Rows written without limits after a crash Non-transactional multi-statement write Wrap the write in a single transaction so it is all-or-nothing
Historian shows the flag but not the value Only the alarm code tag was written Write the point value and limits as structured tags too, so trends show the context
Cannot reconstruct which limits fired calculation_hash omitted from the record Persist the calculation hash with every event to tie it to the exact limit calculation
Disposition update hits no row Key mismatch between write-back and update, or the row was never created Verify the same spc_key is used throughout; the write-back must precede any disposition change
Write-back stalls the pipeline on an MES outage Errors swallowed instead of surfaced Raise on failure so the router's retry and dead-letter logic owns recovery

For the router abstraction and delivery guarantees that call this write-back, return to routing SPC alerts to ticketing and MES systems.