Skip to main content

Seed a sandbox

Most guides in this section read series, resources and events that have to exist first. This page creates them, once, using the SDK's own calls, so every guide can be run end to end against a sandbox rather than read.

What needs seeding, and what does not
GuideNeeds
High-throughput ingestionNothing. It writes its own data
Query & aggregateA the engine series
Model assets as a graphNothing. It creates the graph it reads
Consume live dataA the engine series
Turn readings into eventsA the engine series, which include a hot stretch so the rule actually fires
Attach files to assetsNothing. It uploads what it then reads
Work with unitsNothing. It creates the series it converts
Correlate alarmsB the cooling system, its two sensors and their alarms

Everything below is Python (numpy + pandas + the SDK), the same shape as Generate sample data for the advanced scenarios. Java and Rust reach the same result through the calls each guide already shows.

import intellistream_datahub_sdk, numpy as np, pandas as pd

client = intellistream_datahub_sdk.DataHubClient.from_env()

def ingest(external_id, index, values, unit=None, name=None):
    client.timeseries.create([intellistream_datahub_sdk.TimeSeries(
        external_id=external_id, name=name or external_id,
        unit=unit or "value", value_type="float")])
    client.timeseries.insert_from_lists(
        timestamps=index, values=np.asarray(values), ts=external_id)
    print(f"ingested {len(values):,} points → {external_id}")

A. An engine, with the last twenty minutes running hot​

Three series for one engine: a day of readings at one a minute. The temperature ends in a climb past 110 °C, because Turn readings into events looks at the last hour and fires above that limit. Seed this and the guide has something to find.

idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=24*60, freq="1min")
n, t = len(idx), np.arange(24*60, dtype=float)

hot = np.where(t > n - 20, (t - (n - 20)) * 1.6, 0.0)          # the last twenty minutes
ingest("engine_temperature", idx,
       88 + 4*np.sin(t/90) + np.random.normal(0, 0.8, n) + hot, unit="celsius")
ingest("engine_rpm", idx,
       1500 + 60*np.sin(t/140) + np.random.normal(0, 12, n), unit="rpm")
ingest("engine_oil_pressure", idx,
       320 - 0.4*hot + np.random.normal(0, 4, n), unit="kpa")

Check that the temperature really does cross the limit the events guide uses, before running that guide and concluding it does not work:

print(f"peak {(88 + 4*np.sin(t/90) + hot).max():.1f} °C")   # > 110

B. A cooling system, two sensors and two alarms​

Correlate alarms walks outward from two alarmed sensors and looks for what they have in common. That only works if the chain exists, so this builds sensor_a and sensor_b three hops below a shared cooling_system, and raises one alarm on each. The relationship name has to match the one the guide filters on, PART_OF, or the traversal comes back empty and the guide looks broken.

nodes = [
    intellistream_datahub_sdk.Resource(external_id="cooling_system", name="Cooling system", labels=["System"]),
    intellistream_datahub_sdk.Resource(external_id="skid_1", name="Cooling skid 1", labels=["Skid"]),
    intellistream_datahub_sdk.Resource(external_id="pump_a", name="Cooling pump A", labels=["Pump"]),
    intellistream_datahub_sdk.Resource(external_id="pump_b", name="Cooling pump B", labels=["Pump"]),
    intellistream_datahub_sdk.Resource(external_id="sensor_a", name="Pump A discharge sensor", labels=["Sensor"]),
    intellistream_datahub_sdk.Resource(external_id="sensor_b", name="Pump B discharge sensor", labels=["Sensor"]),
]
edges = [
    intellistream_datahub_sdk.RelForm.by_external_ids("skid_1", "cooling_system", "PART_OF"),
    intellistream_datahub_sdk.RelForm.by_external_ids("pump_a", "skid_1", "PART_OF"),
    intellistream_datahub_sdk.RelForm.by_external_ids("pump_b", "skid_1", "PART_OF"),
    intellistream_datahub_sdk.RelForm.by_external_ids("sensor_a", "pump_a", "PART_OF"),
    intellistream_datahub_sdk.RelForm.by_external_ids("sensor_b", "pump_b", "PART_OF"),
]

graph = client.resources.create(nodes, edges)
print(len(graph.nodes), "nodes,", len(graph.relations), "edges")

Two alarms, seconds apart, one on each sensor. They look like two incidents and are one.

now = pd.Timestamp.now(tz="UTC")
client.events.create([
    intellistream_datahub_sdk.Event(external_id=f"alarm_sensor_a_{int(now.timestamp())}",
                      type="alarm", status="open", event_time=now,
                      metadata={"resource": "sensor_a", "reason": "high discharge temperature"}),
    intellistream_datahub_sdk.Event(external_id=f"alarm_sensor_b_{int(now.timestamp())}",
                      type="alarm", status="open", event_time=now + pd.Timedelta(seconds=8),
                      metadata={"resource": "sensor_b", "reason": "high discharge temperature"}),
])
Attach the alarm to the resource properly

The metadata above is enough for the correlation walk, which starts from the sensors. In production an alarm should carry relatedResources so the link is a real edge rather than a string somebody has to parse. Events reference →

Check the sandbox is populated​

points = client.timeseries.retrieve_datapoints(intellistream_datahub_sdk.RetrieveFilter(
    ts="engine_temperature",
    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=1),
    end=pd.Timestamp.now(tz="UTC")))[0].get_datapoints()

alarms = list(client.events.filter(intellistream_datahub_sdk.EventFilter(
    basic_filter=intellistream_datahub_sdk.BasicEventFilter(type="alarm"), limit=10)))

print(f"{len(points)} readings in the last hour, {len(alarms)} alarms")
# 60 readings in the last hour, 2 alarms

If both numbers are non-zero, every guide on this site has something to run against.

See also​