OEE Pipeline Generator

You generate a complete, production-grade Overall Equipment Effectiveness measurement pipeline for a manufacturing line. The output is real code (Python or Node + Grafana/Streamlit), not pseudocode or chat advice.

OEE = Availability × Performance × Quality. World-class is ~85%. Most plants run 40-60%. The gap is almost always invisible because nobody is measuring correctly. Your job is to make it measurable in a few hours of work.

============================================================ === PRE-FLIGHT === ============================================================

Before generating any code, gather and verify:

• [ ] Data source identified. What is the source of truth for production counts and machine state?
• MQTT broker (most common in Industry 4.0 deployments — Sparkplug B preferred)
• OPC UA endpoint (typical for greenfield brownfield with newer PLCs)
• Modbus TCP/RTU (legacy PLCs — Allen-Bradley, Siemens S7, Mitsubishi)
• Direct PLC tag read (Rockwell pylogix, Siemens snap7, Beckhoff ADS)
• CSV / SQL export from existing MES (Ignition, Wonderware, Rockwell FactoryTalk)
• Manual entry (last resort — generate web form scaffold)
• [ ] Cycle time / planned production rate known. OEE Performance = (Total Count × Ideal Cycle Time) / Run Time. Without ideal cycle time, you cannot compute Performance. If unknown, prompt the user or assume the published rate from the OEM datasheet.
• [ ] Shift schedule defined. Planned production time vs scheduled downtime (lunch, meetings, planned PM) must be distinguishable.
• [ ] Reject/scrap counting available. Quality = Good Count / Total Count. If only Total is available, generate a stub that flags this as a known limitation.
• [ ] Output language chosen. Default Python (best ecosystem for industrial data — pandas, numpy_financial, pymodbus, asyncua, paho-mqtt). Use Node if the user has a Node-first stack.

Recovery if anything is missing:

• Generate a stub data source adapter with mock data so the rest of the pipeline can be tested in isolation.
• Mark the stub clearly: # TODO: replace with real <protocol> connection — see references/connectors.md.
• Do NOT halt the entire skill — output a working scaffold with a known gap, not nothing.

============================================================ === PHASE 1: SCAFFOLD THE PROJECT === ============================================================

Generate this layout in the current working directory (or a subfolder if the user specifies one):

oee/
├── README.md                    # Quickstart, formulas, deployment notes
├── pyproject.toml               # or package.json if Node
├── docker-compose.yml           # Local stack: broker + Grafana + InfluxDB
├── src/
│   ├── ingest/
│   │   ├── mqtt_source.py       # paho-mqtt + Sparkplug B decoder
│   │   ├── opcua_source.py      # asyncua subscription
│   │   ├── modbus_source.py     # pymodbus TCP/RTU
│   │   └── mock_source.py       # for testing without hardware
│   ├── oee/
│   │   ├── calc.py              # core A×P×Q formulas
│   │   ├── six_big_losses.py    # downtime categorizer
│   │   └── metrics.py           # MTBF, MTTR, TEEP, NEE
│   ├── storage/
│   │   └── influx_writer.py     # InfluxDB or TimescaleDB writer
│   └── dashboard/
│       └── grafana_dashboard.json
└── tests/
    └── test_oee_calc.py         # validates formulas against published examples

VALIDATION: Every directory exists. Every file has at least a docstring and one runnable function. Tests pass against the published OEE example: A=0.80, P=0.85, Q=0.95 → OEE=0.6460.

FALLBACK: If the user's existing project structure is incompatible (Django app, Flask, etc.), embed the OEE module as a single package without forcing the folder layout.

============================================================ === PHASE 2: OEE CALCULATION ENGINE === ============================================================

Generate src/oee/calc.py with the exact formulas:

# Availability = Run Time / Planned Production Time
# Planned Production Time = Shift Length - Planned Stops
# Run Time = Planned Production Time - Unplanned Stops

# Performance = (Ideal Cycle Time × Total Count) / Run Time
# Equivalent: Performance = (Total Count / Run Time) / Ideal Run Rate

# Quality = Good Count / Total Count

# OEE = Availability × Performance × Quality
# TEEP = OEE × Utilization (Utilization = Planned Production Time / All Time)
# NEE  = Availability_uptime × Performance × Quality (excludes planned downtime)

Edge cases the engine MUST handle:

• Microstops (< 5 min) → classified as Performance loss, NOT Availability loss (industry convention).
• Speed loss → Performance < 1.0 even with no stops; surface as separate metric.
• Startup/changeover rejects → categorized separately in Quality (Six Big Losses #5).
• Performance > 100% → clamp to 1.0 AND emit warning (ideal cycle time may be set too conservatively).
• Zero planned production time → return None, do NOT divide-by-zero.

Cite the formulas inline using a reference comment so future maintainers can audit:

# Source: ISO 22400-2:2014 Key Performance Indicators for Manufacturing Operations Management

VALIDATION: Unit tests in tests/test_oee_calc.py cover: known-good values, divide-by-zero, P>1.0 clamp, partial shift, all-scrap output. All tests must pass.

FALLBACK: If a formula edge case is ambiguous (e.g., how to handle a machine running during a planned stop), default to the ISO 22400-2 interpretation and add a TODO comment explaining the assumption.

============================================================ === PHASE 3: SIX BIG LOSSES CATEGORIZER === ============================================================

Generate src/oee/six_big_losses.py that classifies downtime events into the canonical six categories:

The categorizer takes a stream of state events (timestamp, state, duration, fault_code, product) and emits classified loss events. Output schema must be JSON-serializable so it flows into the dashboard without ETL.

VALIDATION: Run the categorizer against a synthetic 8-hour shift in tests; verify that the six categories sum to the total loss within rounding.

FALLBACK: If fault codes are not available, classify all unplanned stops > 5 min as Breakdowns and emit a "fault_code_missing" warning so the user knows the granularity gap.

============================================================ === PHASE 4: DATA INGESTION ADAPTER === ============================================================

Generate the ingestion adapter matching the user's data source. Each adapter MUST:

1. Externalize credentials via environment variables — never hardcode.
2. Use TLS / mutual auth where the protocol supports it (Sparkplug B + TLS, OPC UA with X.509).
3. Handle reconnect with exponential backoff (start 1s, cap 60s).
4. Emit a canonical event schema regardless of source protocol:

   {
     "ts": "ISO8601",
     "asset_id": "line-1",
     "metric": "good_count",
     "value": 1247,
     "quality": "GOOD"
   }

MQTT Sparkplug B specifics:

• Subscribe to spBv1.0/{group}/DDATA/{edge_node}/{device}.
• Decode protobuf payload (sparkplug_b.proto — include a copy in vendor/).
• Handle NBIRTH/DBIRTH to learn the metric schema before processing DDATA.

OPC UA specifics:

• Use asyncua (Python) — modern, asyncio-native, maintained.
• Subscribe to changes (not poll). Set publishing interval to match the desired sample rate.
• Browse the address space at startup, persist the discovered NodeIds to config/opcua_nodes.yaml.

Modbus specifics:

• Use pymodbus. Register maps live in config/modbus_map.yaml.
• Convert raw 16-bit registers to engineering units via the configured scale/offset.

VALIDATION: Adapter connects to a local broker (run via docker-compose) and emits at least 10 sample events in python -m src.ingest.<adapter> --test mode.

FALLBACK: If the protocol-specific library is unavailable in the user's environment, generate mock_source.py that replays a recorded CSV — pipeline still demonstrable end-to-end.

============================================================ === PHASE 5: DASHBOARD === ============================================================

Generate src/dashboard/grafana_dashboard.json (preferred) OR a Streamlit app at src/dashboard/app.py.

The dashboard MUST include:

1. Big-number OEE for the current shift (large, color-coded: red < 60%, yellow 60-85%, green > 85%).
2. A / P / Q breakdown (three stacked single-stats — shows WHERE the loss is).
3. Six Big Losses Pareto chart (descending bar — shows WHY).
4. Hourly OEE trend (last 24h line chart).
5. Top 10 downtime events (table — duration, reason, asset).
6. MTBF / MTTR / TEEP trio (secondary metrics).

Grafana panel queries must use the canonical event schema from Phase 4 so the dashboard works without per-deployment query editing.

VALIDATION: Import the JSON into a fresh Grafana instance (Grafana ≥ 10) and confirm all 6 panels render without errors against the mock data source.

FALLBACK: If Grafana setup is too heavy for the user, generate Streamlit (single streamlit run app.py).

============================================================ === PHASE 6: README & DEPLOYMENT NOTES === ============================================================

Write README.md with:

• One-paragraph summary of what the pipeline does.
• Quickstart: docker-compose up && python -m src.ingest.mock_source && open http://localhost:3000.
• Formula reference table (so non-engineers can audit calculations).
• How to swap the mock source for the real PLC/MQTT/OPC-UA source.
• Industry context: "Deploy OEE first as foundation. Layer predictive maintenance as a 12-24 month follow-on. ROI for OEE alone is typically 9-18 months." [cite: 2026 TeepTrak deployment guide]
• Known limitations (if any from PRE-FLIGHT gaps).

VALIDATION: README quickstart actually works on a fresh machine.

============================================================ === SELF-REVIEW === ============================================================

Score each dimension 1–5:

• Complete: Does the pipeline ingest real or simulated data, compute OEE correctly per ISO 22400-2, categorize the Six Big Losses, and render a usable dashboard?
• Robust: Does the calc handle microstops, divide-by-zero, P>1.0 clamp, partial shifts? Does the ingestion reconnect on failure?
• Clean: Is the code formatted (black/ruff for Python), typed (mypy/pyright clean), and free of unused stubs?
• Industry-credible: Would a controls engineer / continuous improvement engineer recognize the terminology and formulas as correct? (This is the killer dimension — wrong formulas destroy trust instantly.)

If any < 4:

• Identify the specific gap. Most common gap: Performance formula uses average cycle time instead of ideal cycle time → silently inflates OEE. Fix and re-test.
• If a gap is genuinely unfixable in this run (e.g., the user can't provide ideal cycle time), note it in the README under "Known Limitations" with a clear remediation path.

============================================================ === LEARNINGS CAPTURE === ============================================================

After completing, append to ~/.claude/skills/oee-pipeline/LEARNINGS.md:

<YYYY-MM-DD> — <site/line/protocol context, e.g., "CNC mill, Sparkplug B">

• What worked: <specific approach — formula reference, dashboard layout, adapter pattern>
• What was awkward: <step that needed retry — e.g., "user couldn't provide ideal cycle time, defaulted to OEM spec">
• Suggested patch: <concrete improvement — e.g., "add pre-flight question for whether MES data already exists">
• Verdict: [Smooth / Minor friction / Major friction]

============================================================ === STRICT RULES === ============================================================

• Never invent formula variants. Use ISO 22400-2 conventions or cite an explicit alternative.
• Never hardcode credentials or broker URLs. Externalize via env vars.
• Never claim a number you didn't calculate from real or mock data. The dashboard MUST be backed by the calc engine, not by static placeholders.
• Never skip the Six Big Losses categorizer. Total OEE without the loss breakdown is the most common failure mode — it tells management things are bad without telling engineers what to fix.
• If the user has an existing OEE platform, do NOT regenerate the pipeline — recommend a connector instead.