Pharma — batch manufacturing & genealogy

The problem. A finished drug lot fails a quality check. Before anything ships, the manufacturer must reconstruct its genealogy: which intermediate lots, raw material lots, and equipment produced it, and whether any deviation in the process record explains the failure. The process data has to be exact — a pH or temperature stored with floating-point drift isn't acceptable in a regulated batch record — and the lineage has to be walkable in seconds, not days.

This scenario combines exact-value time-series with upstream graph genealogy.

Set up demo data

New workspace? Run this once (Python) to create the batch genealogy — a final lot derived from an intermediate, which came from two raw-material lots, produced on a bioreactor — so the upstream walk in step 2 returns a real lineage. Safe to re-run.

import intellistream_datahub_sdk

client = intellistream_datahub_sdk.DataHubClient.from_env()

client.resources.create(
    [intellistream_datahub_sdk.Resource(external_id=x, name=x, labels=[lbl]) for x, lbl in
     [("lot_22f_final", "Lot"), ("lot_int_88", "Lot"), ("lot_raw_acme_41", "Lot"),
      ("lot_raw_acme_42", "Lot"), ("equipment_bioreactor_3", "Equipment")]],
    [intellistream_datahub_sdk.RelForm.by_external_ids("lot_22f_final", "lot_int_88", "derived_from"),
     intellistream_datahub_sdk.RelForm.by_external_ids("lot_int_88", "lot_raw_acme_41", "derived_from"),
     intellistream_datahub_sdk.RelForm.by_external_ids("lot_int_88", "lot_raw_acme_42", "derived_from"),
     intellistream_datahub_sdk.RelForm.by_external_ids("lot_22f_final", "equipment_bioreactor_3", "produced_on")])

1. Record process data exactly

A bioreactor's pH, temperature and dissolved oxygen are the batch record. Use the NUMERIC value type so values store as exact decimals, not floats.

Java

var ph = Timeseries.of("batch_22f_ph")
    .name("Batch 22F — pH")
    .setValueType("numeric");
ph.setUnit("ph");
client.timeseries().create(ph);

client.timeseries().ingest(Map.of(
    "batch_22f_ph", List.of(Datapoint.of(Instant.now(), "7.0421"))));   // exact

Python

import intellistream_datahub_sdk, pandas as pd

client.timeseries.create([intellistream_datahub_sdk.TimeSeries(
    external_id="batch_22f_ph", name="Batch 22F — pH", unit="ph", value_type="numeric")])

client.timeseries.insert_from_lists(
    timestamps=[pd.Timestamp.now(tz="UTC")], values=[7.0421], ts="batch_22f_ph")

Rust

use intellistream_datahub_sdk::timeseries::TimeSeries;
use chrono::Utc;

let mut ph = TimeSeries::new("batch_22f_ph", "Batch 22F — pH");
ph.unit = Some("ph".into());
ph.value_type = "numeric".into();
api.time_series.create_one(&ph).await?;

api.time_series
    .insert_datapoint(None, Some("batch_22f_ph".into()), Utc::now(), "7.0421".into())
    .await?;

Exact decimals

NUMERIC stores values without floating-point rounding — see the Time-series reference for the value types. For a regulated batch record, prefer it over the float types.

2. Walk the genealogy upstream

The failed lot derived_from intermediate lots, which derived_from raw material lots; each step produced_on a piece of equipment. Walk up from the failed lot and the returned sub-graph is its complete genealogy — every input and every machine that touched it.

Java

ResourceNetwork lineage = client.resources().fetchRelated("lot_22f_final", 10);

lineage.nodes().forEach(n -> {
    if (n.getExternalId().startsWith("lot_raw_"))
        System.out.println("raw material: " + n.getExternalId());
    if (n.getExternalId().startsWith("equipment_"))
        System.out.println("equipment: " + n.getExternalId());
});

Python

lineage = client.resources.fetch_related(external_id="lot_22f_final", depth=10)

for node in lineage.nodes:
    if node.external_id.startswith("lot_raw_"):
        print("raw material:", node.external_id)
    elif node.external_id.startswith("equipment_"):
        print("equipment:", node.external_id)

Rust

use intellistream_datahub_sdk::resources::RelatedResourcesForm;

let lineage = api.resources.fetch_related(
    &RelatedResourcesForm::from_external_id("lot_22f_final").with_depth(10)).await?;

for node in lineage.nodes() {
    if node.external_id.starts_with("lot_raw_") {
        println!("raw material: {}", node.external_id);
    }
}

If a raw material lot turns up in the genealogy of other failed batches too, you've found a shared root cause — the same intersection trick as alarm correlation.

3. Capture deviations as events

Every out-of-band reading in the batch record is a process_deviation event tied to the lot, so the review team reconstructs an exact, ordered deviation timeline. See Turn readings into events.

See the result

Walking up from the failed lot returns its complete genealogy:

raw material: lot_raw_acme_41
raw material: lot_raw_acme_42
equipment:    equipment_bioreactor_3

See also

• Time-series reference — exact NUMERIC values.
• Correlate alarms with the graph — the genealogy traversal pattern.
• Turn readings into events — the deviation rule.