Skip to main content

Manufacturing — production quality & OEE

The problem. A factory floor is dozens of stations, each emitting a cycle time, a unit count, and the occasional scrap event — thousands of readings a minute. Plant managers don't want the raw firehose; they want Overall Equipment Effectiveness (availability × performance × quality), rolled up per station, line and shift, plus an alert when a station's scrap rate spikes.

This scenario centres on ingesting at volume and aggregating the result into the hourly and shift-level numbers a production dashboard actually shows.

Set up demo data​

New workspace? Run this once (Python) to create the station's series with an 8-hour shift of readings, so the OEE roll-up has data. Safe to re-run.

import datahub_sdk, numpy as np, pandas as pd

client = datahub_sdk.DataHubClient.from_env()

for s, u in [("station_07_cycle_time_s", "s"), ("station_07_scrap", "count")]:
    client.timeseries.create([datahub_sdk.TimeSeries(external_id=s, name=s, unit=u, value_type="float")])
idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=8 * 60, freq="1min")
client.timeseries.insert_from_lists(timestamps=idx, values=11 + np.random.normal(0, 0.6, len(idx)), ts="station_07_cycle_time_s")
client.timeseries.insert_from_lists(timestamps=idx, values=np.random.poisson(0.3, len(idx)).astype(float), ts="station_07_scrap")

1. Model the line and stream its output​

A plant contains lines, a line contains stations. Each station's signals are series: station_07_cycle_time_s, station_07_units, station_07_scrap.

var cycleTime = Timeseries.of("station_07_cycle_time_s").name("Station 07 cycle time");
cycleTime.setUnit("s");
var scrap = Timeseries.of("station_07_scrap").name("Station 07 scrap count");
scrap.setUnit("count");
client.timeseries().create(cycleTime, scrap);

// readings arrive continuously across every station — hand them off in bulk
client.timeseries().ingest(Map.of(
    "station_07_cycle_time_s", cycleTimes,   // List<Datapoint>
    "station_07_scrap", scrapCounts));

See high-throughput ingestion for sustaining floor-wide volume.

2. Roll up to hourly OEE inputs​

The dashboard needs per-hour averages and totals, not raw cycles. Ask for the aggregates and granularity and the platform returns one value per bucket — the performance inputs to OEE. (There's no count aggregate; a per-hour count comes from sum-ming a per-event count series.) Full mechanics in Query & aggregate.

var filter = new RetrieveFilter();
filter.setExternalId("station_07_cycle_time_s");
filter.setStart(ZonedDateTime.now().minusHours(8));   // one shift
filter.setEnd(ZonedDateTime.now());
filter.setAggregates(List.of("avg", "sum"));
filter.setGranularity("1h");

var request = new DataRetriever<RetrieveFilter>();
request.setItems(List.of(filter));

client.timeseries().retrieve(request).getItems().get(0).getDatapoints()
    .forEach(p -> recordHourlyPerformance(p.getTimestamp(), p.getValue()));

3. Flag a quality excursion​

When a station's hourly scrap rate jumps past its limit, record a scrap_excursion event so the quality team sees a ranked list of stations needing attention.

EventModel excursion = new EventModel();
excursion.setExternalId("scrap_excursion_st07_" + System.currentTimeMillis());
excursion.setType("scrap_excursion");
excursion.setStatus("open");
excursion.setMetadata(Map.of("station", "station_07", "scrap_rate_pct", "4.8", "limit_pct", "2.0"));
excursion.setEventTime(ZonedDateTime.now());
client.events().create(List.of(excursion));

See the result​

Chart the hourly average cycle time — the line the production dashboard watches:

import matplotlib.pyplot as plt

rf = datahub_sdk.RetrieveFilter(ts="station_07_cycle_time_s",
    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=8), end=pd.Timestamp.now(tz="UTC"),
    aggregates=["avg"], granularity="1h")
vals = [dp.average for dp in client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()]
plt.plot(vals); plt.title("Station 07 hourly avg cycle time (s)"); plt.show()

See also​