Automate OEE Shift Reports with Python and TimescaleDB

A plant that still generates its shift OEE report by hand — someone exporting a spreadsheet from the historian at 6am — has already lost the argument for trustworthy reporting, because the number depends on who ran the export and when. This recipe builds the scheduled path instead: a TimescaleDB continuous aggregate pre-computes per-asset state durations and counts, a Python job assembles them into a timezone-correct shift window and composes Availability × Performance × Quality once, and the result is rendered and delivered on a fixed schedule with a guarantee that re-running it for a closed shift produces the same numbers. It sits directly under OEE reporting and shift rollups, which covers the weighting rules this job implements; here the focus is the concrete five-step build.

1. Build the hourly continuous aggregate #Permalink to this section

The job must never scan raw state-interval rows at report time — on a plant logging sub-second transitions that scan is too slow to run on a schedule, and a report that takes minutes to render is a report nobody trusts enough to wait for. Pre-aggregate to an hourly grain with a TimescaleDB continuous aggregate, keyed by asset, and refresh it on a short cadence so a report generated shortly after shift-end sees complete data. The view carries every additive quantity the composition step needs — planned seconds, downtime seconds, run seconds, total and good counts, and the count-weighted ideal-time product — so no later step ever has to touch raw interval facts.

CREATE MATERIALIZED VIEW oee_hourly_asset
WITH (timescaledb.continuous) AS
SELECT
    asset_id,
    time_bucket('1 hour', interval_start) AS bucket,
    SUM(planned_seconds)                  AS planned_seconds,
    SUM(downtime_seconds)                 AS downtime_seconds,
    SUM(run_seconds)                      AS run_seconds,
    SUM(total_count)                      AS total_count,
    SUM(good_count)                       AS good_count,
    SUM(ideal_cycle_time_sec * total_count) AS ideal_time_seconds
FROM oee_interval_facts
GROUP BY asset_id, bucket
WITH NO DATA;

SELECT add_continuous_aggregate_policy('oee_hourly_asset',
    start_offset      => INTERVAL '3 hours',
    end_offset         => INTERVAL '2 minutes',
    schedule_interval  => INTERVAL '2 minutes');

The end_offset of two minutes trades a small amount of staleness for near-guaranteed completeness by the time a shift-end report fires. Sizing the bucket and tuning retention on the raw table is covered in continuous aggregates and downsampling retention; this job only depends on the view existing and refreshing on schedule.

2. Query the shift window in the correct timezone #Permalink to this section

A shift report must reflect wall-clock shift boundaries, not a UTC day. Reuse the same shift-calendar logic that classification uses — do not re-derive shift start/end independently in the reporting job, or a DST transition gets handled twice, inconsistently. Convert the local shift window to UTC once, then filter the continuous aggregate.

from dataclasses import dataclass
from datetime import datetime
from zoneinfo import ZoneInfo

import asyncpg


@dataclass(frozen=True)
class ShiftWindow:
    shift_id: str
    plant_tz: str
    start_local: datetime  # naive local wall time
    end_local: datetime

    def to_utc(self) -> tuple[datetime, datetime]:
        zone = ZoneInfo(self.plant_tz)
        start = self.start_local.replace(tzinfo=zone)
        end = self.end_local.replace(tzinfo=zone)
        return start.astimezone(ZoneInfo("UTC")), end.astimezone(ZoneInfo("UTC"))


async def fetch_shift_totals(pool: asyncpg.Pool, asset_ids: list[str], window: ShiftWindow) -> dict:
    start_utc, end_utc = window.to_utc()
    row = await pool.fetchrow(
        """
        SELECT
            SUM(planned_seconds)          AS planned_seconds,
            SUM(downtime_seconds)         AS downtime_seconds,
            SUM(run_seconds)              AS run_seconds,
            SUM(total_count)              AS total_count,
            SUM(good_count)               AS good_count,
            SUM(ideal_time_seconds)       AS ideal_time_seconds
        FROM oee_hourly_asset
        WHERE asset_id = ANY($1::text[])
          AND bucket >= $2 AND bucket < $3
        """,
        asset_ids, start_utc, end_utc,
    )
    return dict(row) if row else {}

Filtering on the converted UTC range, rather than assuming the aggregate’s hourly buckets line up with the shift start, is what keeps a 6:30am local shift start from silently clipping its first partial hour. The full DST and midnight-crossing treatment lives in shift boundary logic.

3. Compose Availability, Performance, and Quality once #Permalink to this section

The assembly step is intentionally a single function that only divides sums — it never sees a per-hour ratio, so it cannot average one by accident. Decimal end to end avoids IEEE 754 drift across a shift’s worth of accumulated seconds.

from decimal import Decimal


def compose_shift_oee(totals: dict) -> dict[str, Decimal]:
    planned = Decimal(str(totals.get("planned_seconds") or 0))
    downtime = Decimal(str(totals.get("downtime_seconds") or 0))
    run = Decimal(str(totals.get("run_seconds") or 0))
    total_count = int(totals.get("total_count") or 0)
    good_count = int(totals.get("good_count") or 0)
    ideal_time = Decimal(str(totals.get("ideal_time_seconds") or 0))

    if planned <= 0:
        return {k: Decimal("0") for k in ("availability", "performance", "quality", "oee")}

    availability = (planned - downtime) / planned
    performance = ideal_time / run if run > 0 else Decimal("0")
    quality = Decimal(good_count) / Decimal(total_count) if total_count > 0 else Decimal("0")

    def clamp(v: Decimal) -> Decimal:
        return min(max(v, Decimal("0")), Decimal("1")).quantize(Decimal("0.0001"))

    a, p, q = clamp(availability), clamp(performance), clamp(quality)
    return {"availability": a, "performance": p, "quality": q, "oee": clamp(a * p * q)}

This is the same additive-first identity used across OEE formula validation: every input to compose_shift_oee is a sum, never a pre-divided ratio, which is what keeps a multi-asset shift report from quietly becoming a mean of percentages.

4. Render HTML and CSV from the same data #Permalink to this section

Render both formats from one composed dictionary so the numbers on the emailed HTML summary and the CSV attached for the data warehouse can never diverge — a common defect is a report template that recomputes a rounded figure independently for each output format.

import csv
import io
from datetime import datetime, timezone


def render_csv(shift_id: str, oee: dict[str, Decimal]) -> str:
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(["shift_id", "availability", "performance", "quality", "oee", "generated_at"])
    writer.writerow([
        shift_id, oee["availability"], oee["performance"], oee["quality"], oee["oee"],
        datetime.now(timezone.utc).isoformat(),
    ])
    return buf.getvalue()


def render_html(shift_id: str, oee: dict[str, Decimal]) -> str:
    pct = lambda v: f"{v * 100:.2f}%"
    return f"""
    <table>
      <caption>Shift {shift_id} OEE</caption>
      <tr><th>Availability</th><td>{pct(oee['availability'])}</td></tr>
      <tr><th>Performance</th><td>{pct(oee['performance'])}</td></tr>
      <tr><th>Quality</th><td>{pct(oee['quality'])}</td></tr>
      <tr><th>OEE</th><td>{pct(oee['oee'])}</td></tr>
    </table>
    """.strip()

Keep the multiplication (v * 100) inside the presentation layer only — every value entering it is already a clamped, quantized Decimal, so display formatting can never feed back into the stored figure.

5. Schedule the job and make re-runs idempotent #Permalink to this section

A cron trigger fires shortly after each shift closes, but the job must also tolerate being re-run — manually, after a late MES correction, or because the scheduler retried a timeout. Guard against duplicate reports with a unique constraint on (shift_id, revision) and bump the revision explicitly rather than overwriting.

async def generate_and_store_report(
    pool: asyncpg.Pool, asset_ids: list[str], window: ShiftWindow, force_revision: bool = False
) -> None:
    existing = await pool.fetchval(
        "SELECT MAX(revision) FROM oee_shift_reports WHERE shift_id = $1", window.shift_id
    )
    if existing is not None and not force_revision:
        return  # already generated; re-run is a no-op unless explicitly forced

    totals = await fetch_shift_totals(pool, asset_ids, window)
    oee = compose_shift_oee(totals)
    revision = (existing or 0) + 1

    await pool.execute(
        """
        INSERT INTO oee_shift_reports
            (shift_id, revision, availability, performance, quality, oee, generated_at)
        VALUES ($1, $2, $3, $4, $5, $6, now())
        """,
        window.shift_id, revision, oee["availability"], oee["performance"],
        oee["quality"], oee["oee"],
    )

# crontab entry: fire 10 minutes after each 8-hour shift close, plant-local time
10 6,14,22 * * * /opt/oee/venv/bin/python -m oee_reports.generate --shift auto

The force_revision flag is how a late-arriving MES correction triggers a re-run without silently clobbering the original — the new row is a new revision, and downstream consumers can subscribe to “latest revision for shift X” instead of a mutable cell.

Wrap the scheduled entry point with basic observability: emit a counter for reports generated versus skipped-as-duplicate, a gauge for the age of the continuous aggregate’s last refresh at the moment the job ran, and an alert if a shift closes without a corresponding report within, say, thirty minutes. A report job that fails silently at 2am is functionally identical to one that was never scheduled — the failure only becomes visible when a supervisor notices a missing number hours later, by which point the underlying raw telemetry may already be outside its hot-storage retention window.

Gotchas & anti-patterns #Permalink to this section

• Re-deriving shift boundaries in the report job. If the reporting code computes its own DST rules independently of the classification layer’s shift boundary logic, the two will eventually disagree on a fall-back transition and the report will not match the dashboard.
• Recomputing display percentages per output format. Rendering HTML and CSV from separately rounded values, rather than one shared Decimal dictionary, produces two “correct-looking” reports that quietly disagree in the third decimal place.
• Overwriting a shipped report on re-run. Silent overwrite erases the audit trail a floor dispute needs six months later; always insert a new revision.
• Firing the cron job before the continuous aggregate has caught up. A trigger fired at exactly shift-end can race the aggregate’s end_offset window; build in a short delay (as in the crontab example) rather than querying raw facts as a workaround.
• Treating a zero-row query result as a zero OEE. An empty result from fetch_shift_totals usually means the asset list or shift window is wrong, not that the shift genuinely produced nothing — validate row counts before composing, and alert rather than publish a false zero.

Quick reference #Permalink to this section

Related #Permalink to this section

• OEE reporting and shift rollups — parent guide covering the weighted-aggregation contract this job implements.
• Rolling up OEE from line to plant with weighted aggregation — extend this shift job to line and plant grains correctly.
• Continuous aggregates and downsampling retention — sizing the hourly rollup used in step 1.
• Shift boundary logic — the timezone- and DST-safe window logic step 2 reuses.
• OEE formula validation — the additive-first identity behind step 3.