Environmental — air-quality monitoring

The problem. An air-quality network mixes sensor types: particulates in micrograms per cubic metre, nitrogen dioxide in parts per billion, carbon monoxide in parts per million. If a series doesn't carry its unit, 42 is meaningless — is that a healthy CO reading or a hazardous one? Getting units right, consistently, across a heterogeneous fleet is what makes the data comparable and the dashboards trustworthy.

This is the scenario where units are the hero: pick the right unit for each pollutant and tag every series with it.

Set up demo data

New workspace? Run this once (Python) to create the station's pollutant series (each with its unit) and a day of readings — so the daily-average roll-up has data. The unit catalogue browsed in step 1 is built-in reference data. Safe to re-run.

import intellistream_datahub_sdk, numpy as np, pandas as pd

client = intellistream_datahub_sdk.DataHubClient.from_env()

for s, u in [("station_kirkeveien_pm25", "ug_m3"), ("station_kirkeveien_no2", "ppb"),
             ("station_kirkeveien_co", "ppm")]:
    client.timeseries.create([intellistream_datahub_sdk.TimeSeries(external_id=s, name=s, unit=u)])
idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=24, freq="1h")
client.timeseries.insert_from_lists(timestamps=idx, values=np.random.uniform(5, 45, 24), ts="station_kirkeveien_pm25")

1. Find the right unit

Browse the unit catalogue to get the canonical external id for each pollutant's unit. See the Units reference.

Java

client.units().list().getItems().forEach(u ->
    System.out.println(u.getExternalId() + "  " + u.getName() + "  (" + u.getSymbol() + ")"));
// e.g. ug_m3  micrograms per cubic metre  (µg/m³)

Python

for u in client.units.list():
    print(u.external_id, u.name, u.symbol)

ugm3 = client.units.by_external_ids("ug_m3")     # confirm the one you'll use

Rust

for u in api.units.list().await?.get_items() {
    println!("{}  {}  ({})", u.external_id, u.name, u.symbol);
}
let ugm3 = api.units.by_external_id("ug_m3").await?;

2. Tag each series with its unit

Create one series per pollutant per station, each carrying its unit's external id — so every value is self-describing and a chart can label its axis correctly.

Java

var pm25 = Timeseries.of("station_kirkeveien_pm25").name("Kirkeveien PM2.5");
pm25.setUnit("ug_m3");
var no2 = Timeseries.of("station_kirkeveien_no2").name("Kirkeveien NO2");
no2.setUnit("ppb");
var co = Timeseries.of("station_kirkeveien_co").name("Kirkeveien CO");
co.setUnit("ppm");
client.timeseries().create(pm25, no2, co);

Python

import intellistream_datahub_sdk

client.timeseries.create([
    intellistream_datahub_sdk.TimeSeries(external_id="station_kirkeveien_pm25", name="Kirkeveien PM2.5", unit="ug_m3"),
    intellistream_datahub_sdk.TimeSeries(external_id="station_kirkeveien_no2", name="Kirkeveien NO2", unit="ppb"),
    intellistream_datahub_sdk.TimeSeries(external_id="station_kirkeveien_co", name="Kirkeveien CO", unit="ppm")])

Rust

use intellistream_datahub_sdk::timeseries::TimeSeries;

for (id, name, unit) in [
    ("station_kirkeveien_pm25", "Kirkeveien PM2.5", "ug_m3"),
    ("station_kirkeveien_no2", "Kirkeveien NO2", "ppb"),
    ("station_kirkeveien_co", "Kirkeveien CO", "ppm"),
] {
    let mut ts = TimeSeries::new(id, name);
    ts.unit = Some(unit.into());
    api.time_series.create_one(&ts).await?;
}

3. Daily averages and exceedances

Roll each pollutant up to daily averages for the public index, and raise an exceedance event when a station crosses a regulatory limit — the metadata carrying the pollutant, its unit, and the threshold so the record is unambiguous. See Query & aggregate.

See the result

Chart the day's PM2.5 — now self-describing, because the series carries its ug_m3 unit:

import matplotlib.pyplot as plt

rf = intellistream_datahub_sdk.RetrieveFilter(ts="station_kirkeveien_pm25",
    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(days=1), 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("Kirkeveien PM2.5 (µg/m³)"); plt.show()

See also

• Units reference — the unit catalogue and lookups.
• Work with units of measure — tagging series with units.
• Query & aggregate — daily-average index.