3. Time-series data ingestion from industrial historians

Industrial historians are the system of record for process data. Before any analysis, anomaly detection, or dashboard can run, the data must be pulled out of the historian and into a Python process in a form that preserves timestamps, quality flags, and engineering units. This chapter covers the three historian families you will meet most often in operations work — OSIsoft PI, InfluxDB, and SQL-based historians — and shows how to connect to each with a Python client, extract a bounded time window, and normalize the result into a common structure.

The core challenge is not the network call. It is that each historian speaks a different data model. PI stores values as PI Points with a tag name, a timestamp, a value, and a 16-bit quality code. InfluxDB stores measurements with tags, fields, and nanosecond timestamps. SQL historians store rows in wide or narrow tables with a timestamp column and one or more value columns, often with a separate quality column. A robust ingestion layer hides these differences behind a single interface.

Choosing a client and connection pattern

For OSIsoft PI, the supported Python path is the PI Web API, accessed over HTTPS with the requests library or the community PIconnect package. PI Web API is preferred over the legacy PI SDK because it is cross-platform and does not require a Windows-only COM layer. Authentication uses either Kerberos or a bearer token. For InfluxDB 2.x, use the official influxdb-client package with a token and an organization/bucket pair. For InfluxDB 1.x, the older influxdb package with username/password still appears in brownfield plants. For SQL historians such as Wonderware InSQL, Rockwell FactoryTalk Historian, or OSIsoft PI SQL, use pyodbc or sqlalchemy with the vendor ODBC driver.

A practical pattern is to define a small adapter class per historian, all exposing the same method signature:

def read_range(tags: list[str], start: datetime, end: datetime) -> pd.DataFrame:
    """Return a long-format DataFrame with columns:
    timestamp, tag, value, quality."""

Keeping the return contract identical means downstream code — resampling, gap filling, unit conversion — never needs to know which historian produced the data.

Pulling a window from PI Web API

PI Web API exposes recorded values through the /streams/{webId}/recorded endpoint. You first resolve a tag to a WebID, then request a time range. The following example uses a bearer token and requests values in ISO 8601 with a timezone offset:

import requests, pandas as pd

BASE = "https://piwebapi.plant.local/piwebapi"
HEADERS = {"Authorization": "Bearer <token>"}

def pi_read(tag, start, end):
    # 1. resolve tag to WebID
    r = requests.get(f"{BASE}/points", headers=HEADERS,
                     params={"path": f"\\\\PISERVER\\{tag}"}, verify=False)
    web_id = r.json()["WebId"]
    # 2. read recorded values
    r = requests.get(f"{BASE}/streams/{web_id}/recorded",
                     headers=HEADERS,
                     params={"startTime": start.isoformat(),
                             "endTime": end.isoformat(),
                             "selectedFields": "Items.Timestamp,Items.Value,Items.Good"})
    items = r.json()["Items"]
    return pd.DataFrame({
        "timestamp": pd.to_datetime([i["Timestamp"] for i in items], utc=True),
        "tag": tag,
        "value": [i["Value"] for i in items],
        "quality": [i["Good"] for i in items],
    })

Two details matter. First, PI returns timestamps in UTC; convert to local time only at the presentation layer. Second, the Good flag is a boolean summary — the full 16-bit quality code is available via Items.Quality if you need to distinguish "questionable" from "substituted". For most analytics, filtering to Good == True is sufficient, but you should log how many points were dropped.

Querying InfluxDB with Flux

InfluxDB 2.x uses Flux, a functional query language. The client returns a stream of tables that you convert to a DataFrame. A typical historian query for a bounded window looks like this:

from influxdb_client import InfluxDBClient
from influxdb_client.client.flux_table import FluxTable

client = InfluxDBClient(url="http://influx.plant.local:8086",
                        token="<token>", org="operations")

query = f'''
from(bucket: "historian")
  |> range(start: {start.isoformat()}, stop: {end.isoformat()})
  |> filter(fn: (r) => r._measurement == "process" and
             r._field == "value" and
             r.tag in ["FIC-101", "TIC-205"])
  |> keep(columns: ["_time", "tag", "_value", "_field"])
'''

tables = client.query_api().query(query)
df = client.query_api().query_data_frame(query)

Flux pushes filtering to the server, which is essential when a tag has millions of points. Avoid pulling an entire measurement and filtering in pandas. Also note that InfluxDB stores timestamps in UTC nanoseconds; query_data_frame returns a _time column already parsed as datetime64[ns, UTC].

SQL historians and the wide-table problem

SQL historians often store data in a wide table where each tag is a column, or in a narrow table with a tag name column. The narrow form is easier to query generically:

SELECT ts, tag, value, quality
FROM history
WHERE tag IN (?, ?)
  AND ts >= ? AND ts < ?
ORDER BY ts;

Bind parameters rather than string-formatting timestamps — this avoids SQL injection and lets the driver use the index on ts. With pyodbc:

import pyodbc, pandas as pd

conn = pyodbc.connect("DRIVER={FactoryTalk Historian ODBC};"
                      "SERVER=hist01;DATABASE=Runtime;Trusted_Connection=yes;")
df = pd.read_sql(query, conn, params=[tag1, tag2, start, end])

Wide tables require a pivot after extraction. Pull only the columns you need; a SELECT * on a 500-column historian table will transfer far more data than the analysis uses.

Normalizing the result

Once each adapter returns a long-format frame, apply a consistent normalization step. Convert all timestamps to UTC, drop rows where quality indicates a bad value, and attach engineering units from a tag metadata table. A minimal normalization function:

def normalize(df, units):
    df = df.copy()
    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
    df = df[df["quality"].astype(bool)]
    df["unit"] = df["tag"].map(units)
    return df.sort_values(["tag", "timestamp"]).reset_index(drop=True)

After normalization, resample to a fixed interval. Industrial data is event-driven, so a 10-second resample with mean aggregation is common for continuous signals, while counters should use last minus first over the interval. Always record the resampling rule alongside the data; a value without its aggregation semantics is ambiguous.

Failure modes to watch

• Time zone drift. PI returns UTC, some SQL historians return local time. Mixing them shifts every trend by the offset. Verify by comparing a known step change against the DCS event log.
• Silent quality drops. Filtering on Good can remove 30% of points during a sensor fault. Log the drop ratio per tag and alert if it exceeds a threshold.
• Query timeouts. Pulling a month of 1-second data for 50 tags in one call will time out. Chunk the range into daily windows and concatenate.
• Unit mismatch. A tag may report in °F while the model expects °C. Carry units through the pipeline and convert explicitly.

The adapter pattern, UTC normalization, and explicit quality handling form the foundation for everything that follows. With a reliable ingestion layer, downstream resampling, gap filling, and feature engineering become straightforward pandas operations rather than historian-specific special cases.