Building an OCAP Decision Table in Python

Modeling an out-of-control action plan as data — rather than as branching if statements buried in an alert handler — is what makes the reaction reviewable, testable, and auditable. This how-to builds the table as a keyed structure, wraps it in a resolver that never fails on an unknown signal, proves that every rule code the detector can emit is covered, and serializes the whole plan to CSV and YAML so a quality engineer can review it without reading Python. It is the data-modeling companion to the broader out-of-control action plans (OCAP) guide, and it assumes the signals arrive as codes from out-of-control rule detection.

The reason to keep the plan as data is separation of concerns: the policy (which signal gets which reaction) changes far more often than the mechanism (how a signal is resolved). When the plan is data, a change to policy is a change to a row that an auditor can diff, not a code change that needs a developer and a deploy.

Prerequisites

Confirm these before building the table:

Python 3.10+ (pip install pyyaml for the YAML export; csv and dataclasses are standard library)
The exhaustive list of rule codes your detector can emit, e.g. Western Electric WE1–WE4 and Nelson NELSON1–NELSON8
The list of monitored characteristics whose reactions differ (a small set — most share a generic reaction)
Agreement from quality on the escalation tiers (operator, line lead, quality engineer) and which signals hold product
A place to persist the serialized table under version control, so every change to the plan is diffable and dated

Step 1 — Model one OCAP row as a typed record

Start with the atomic unit: one row that separates containment from correction and names an owner and tier. Freezing the dataclass makes each row hashable and prevents accidental mutation after the table is built.

from __future__ import annotations

from dataclasses import dataclass
from enum import IntEnum
from typing import Optional


class Severity(IntEnum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass(frozen=True)
class OcapRow:
    """One codified reaction. characteristic=None means 'any characteristic'."""
    rule_code: str
    characteristic: Optional[str]
    containment: str          # stop the bleeding
    correction: str           # fix this occurrence
    owner_role: str
    escalation_tier: int      # 1=operator, 2=line lead, 3=quality engineer
    hold_product: bool
    severity: Severity

Verify in isolation: construct one row and assert it is hashable (hash(row) does not raise) and immutable (assigning to row.owner_role raises FrozenInstanceError). Both properties are what let the row act as a safe, comparable table entry.

Step 2 — Assemble the table as a keyed dictionary

Model the plan as a dict keyed on (rule_code, characteristic). A None characteristic is the generic reaction for that code; an explicit characteristic overrides it. Building the dict from a flat list of rows keeps the source readable and lets you catch duplicate keys immediately.

def build_table(rows: list[OcapRow]) -> dict[tuple[str, Optional[str]], OcapRow]:
    """Index rows by (rule_code, characteristic); reject duplicate keys."""
    table: dict[tuple[str, Optional[str]], OcapRow] = {}
    for row in rows:
        key = (row.rule_code, row.characteristic)
        if key in table:
            raise ValueError(f"Duplicate OCAP row for key {key!r}")
        table[key] = row
    return table


ROWS = [
    OcapRow("WE1", "bore_diameter",
            "Stop machine; quarantine parts since last good subgroup",
            "Verify tool offset; re-qualify with first-article check",
            "line_lead", 2, True, Severity.CRITICAL),
    OcapRow("WE1", None,
            "Quarantine parts since last good subgroup",
            "Confirm setup and re-center the process",
            "operator", 1, True, Severity.HIGH),
    OcapRow("NELSON2", None,
            "Flag the run; increase inspection frequency",
            "Investigate mean shift; adjust after root cause confirmed",
            "line_lead", 2, False, Severity.MEDIUM),
]

TABLE = build_table(ROWS)

Verify: passing two rows with the same (rule_code, characteristic) must raise ValueError. A silent overwrite would let one reaction shadow another with no warning — exactly the kind of latent policy bug the table format is meant to eliminate.

Step 3 — Write the resolver with a deterministic fallback chain

The resolver turns a signal into a row through a fixed precedence: exact match, then the code's generic row, then a safe default. It must never raise on an unknown code — an unhandled signal on the shop floor is worse than a conservative over-reaction.

import logging

logger = logging.getLogger("ocap.resolver")

DEFAULT_ROW = OcapRow(
    rule_code="*", characteristic=None,
    containment="Quarantine suspect product; hold the line for review",
    correction="Investigate before resuming; no automatic adjustment",
    owner_role="quality_engineer", escalation_tier=3,
    hold_product=True, severity=Severity.HIGH,
)


def resolve(table: dict[tuple[str, Optional[str]], OcapRow],
            rule_code: str, characteristic: str) -> tuple[OcapRow, str]:
    """Resolve a signal to (row, match_kind). Never raises on an unknown code.

    Precedence: exact (code, characteristic) -> (code, None) -> default.
    """
    if not rule_code or not characteristic:
        raise ValueError("rule_code and characteristic must both be non-empty.")

    exact = table.get((rule_code, characteristic))
    if exact is not None:
        return exact, "exact"

    generic = table.get((rule_code, None))
    if generic is not None:
        return generic, "characteristic_default"

    logger.warning("No OCAP row for (%s, %s); using default.", rule_code, characteristic)
    return DEFAULT_ROW, "fallback"

Verify: resolve(TABLE, "WE1", "bore_diameter") returns the exact row; resolve(TABLE, "WE1", "surface_finish") returns the generic WE1 row with match kind characteristic_default; and resolve(TABLE, "NELSON7", "torque") returns DEFAULT_ROW with match kind fallback. The three outcomes exercise all three precedence tiers.

Step 4 — Prove full coverage against the detector's code list

Coverage is the property that turns the table from hopeful into trustworthy: every code the detector can emit must resolve to a real row, not silently drop to the default. Assert it against the detector's authoritative code list so a newly added rule fails the test until someone writes its reaction.

DETECTOR_CODES = ["WE1", "WE2", "WE3", "WE4",
                  "NELSON1", "NELSON2", "NELSON3", "NELSON4"]


def uncovered_codes(table: dict[tuple[str, Optional[str]], OcapRow],
                    detector_codes: list[str]) -> list[str]:
    """Return detector codes that have no explicit row (would hit the default)."""
    covered = {code for (code, _char) in table}
    return sorted(c for c in detector_codes if c not in covered)


def assert_full_coverage(table, detector_codes) -> None:
    missing = uncovered_codes(table, detector_codes)
    if missing:
        raise AssertionError(
            f"OCAP table missing explicit rows for: {missing}. "
            "Every detector rule code needs a defined reaction (IATF 16949)."
        )

Verify: with the sample TABLE, uncovered_codes returns the codes that lack a row (WE2, WE3, WE4, NELSON1, NELSON3, NELSON4), and assert_full_coverage raises listing them. In your real project the assertion should pass; a non-empty list is a to-do list of reactions still to be written and agreed with quality.

Step 5 — Serialize the table to CSV and YAML for auditors

The plan has to be reviewable by people who do not read Python. Export it to CSV for a spreadsheet review and to YAML for a human-diffable, version-controlled source of truth. Both must round-trip: what you write must re-load into the same rows.

import csv
import io
from dataclasses import asdict, fields

import yaml


def rows_to_dicts(rows: list[OcapRow]) -> list[dict]:
    out = []
    for r in rows:
        d = asdict(r)
        d["severity"] = int(r.severity)   # store the numeric class, not the enum repr
        out.append(d)
    return out


def to_csv(rows: list[OcapRow]) -> str:
    header = [f.name for f in fields(OcapRow)]
    buf = io.StringIO()
    writer = csv.DictWriter(buf, fieldnames=header)
    writer.writeheader()
    writer.writerows(rows_to_dicts(rows))
    return buf.getvalue()


def to_yaml(rows: list[OcapRow]) -> str:
    return yaml.safe_dump(rows_to_dicts(rows), sort_keys=False)

Verify: assert the CSV has one header line plus one line per row, and that yaml.safe_load(to_yaml(ROWS)) returns a list whose length equals len(ROWS) with the severity restored as an integer. A round-trip that preserves row count and severity proves the export is a faithful record, not a lossy view.

Verification

Run the full chain on the sample table and assert the three properties that matter — deterministic resolution, correct fallback, and a faithful export:

def test_ocap_table() -> None:
    table = build_table(ROWS)

    # Deterministic: same key resolves identically every time.
    first, kind = resolve(table, "WE1", "bore_diameter")
    for _ in range(1000):
        again, kind_again = resolve(table, "WE1", "bore_diameter")
        assert again == first and kind_again == kind
    assert kind == "exact" and first.severity == Severity.CRITICAL

    # Generic and fallback tiers.
    _, generic_kind = resolve(table, "WE1", "surface_finish")
    assert generic_kind == "characteristic_default"
    default_row, fallback_kind = resolve(table, "NELSON7", "torque")
    assert fallback_kind == "fallback" and default_row.hold_product is True

    # Serialization round-trips without losing rows or severity.
    loaded = yaml.safe_load(to_yaml(ROWS))
    assert len(loaded) == len(ROWS)
    assert loaded[0]["severity"] == int(ROWS[0].severity)

    print("OCAP table OK:", len(table), "rows,", "fallback holds product")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    test_ocap_table()

Expected output resembles OCAP table OK: 3 rows, fallback holds product, with a warning logged for the NELSON7 fallback. If the determinism loop ever fails, state or clock has leaked into the resolver; if the fallback row does not hold product, the default is failing open and must be corrected before deployment.

Root-cause table

Symptom	Cause	Fix
Two reactions for the same signal, one silently wins	Duplicate `(rule_code, characteristic)` key overwritten in the dict	Build via `build_table`, which raises on duplicate keys (Step 2)
A new detector rule produces no reaction	Code added upstream but no row written; it drops to the default	Run `assert_full_coverage` in CI against the detector's code list (Step 4)
Auditor cannot review the plan	Policy encoded as `if` branches in code, not as data	Serialize to CSV and YAML and version-control the export (Step 5)
Severity reads as `Severity.HIGH` in the CSV	Enum written via its repr instead of its value	Cast to `int(severity)` in `rows_to_dicts` before export (Step 5)
Resolver raises on an unmapped code	Fallback tier missing or a bare dict lookup used	Route through `resolve` with the exact, generic, default chain (Step 3)

Escalation tiers and reaction time for OCAP — turn the tier field into reaction-time budgets and promotion logic
Out-of-control rule detection (Nelson / Western Electric) — the source of the rule codes this table is keyed on

Up one level: Out-of-Control Action Plans (OCAP).