Streaming OSIsoft PI Tags into a Compliance Pipeline

An OSIsoft PI System — now branded AVEVA PI — is the historian of record at most mid-size and large water utilities, and the concrete task on this page is pulling tag data out of it through the PI Web API and landing each sample in a validated compliance pipeline with a defensible quality flag attached. This is a reference implementation inside the parent Historian Integration Patterns for Compliance Pipelines section, written for the automation engineers who operate the extraction service and the environmental staff who answer for the numbers it produces. The design principle throughout is separation of duties: the PI archive is authoritative for what a sensor measured and when, but it is not the place where EPA thresholds are evaluated — that judgment happens later, once readings have been normalized, quality-graded, and handed onward to the Violation Detection & Rule Engine Logic. The extractor described here reads over a strictly read-only PI identity, distinguishes recorded from interpolated values, resolves each tag to a durable WebID, and pages through long time ranges without losing a sample or duplicating one.

Prerequisites & Environment Setup

The client targets Python 3.10+ and httpx, which gives you connection pooling, HTTP/2, timeouts, and clean Basic authentication without the surprises of hand-rolled requests retry loops. Schema enforcement uses pydantic v2, and the normalized records are the same shape consumed one step downstream when writing validated readings to InfluxDB for compliance.

Two things must exist before any code runs, and neither ships in a package. First, a read-only PI identity: a PI mapping or explicit account that has read access to the point database and the streams you need and write access to nothing. The extractor should never be able to modify the archive, and the PI Web API honors that boundary only if the identity itself is scoped that way. Second, network reachability to the PI Web API host, which normally lives on a separate services tier from the raw control network; route to it through the same read-only egress discipline described in Security Boundary Design, and pin the server’s TLS certificate rather than disabling verification.

python3 -m venv .venv && source .venv/bin/activate
pip install "httpx[http2]==0.27.*" "pydantic==2.7.*"
# Optional: pandas for downstream framing before the compliance windows
pip install "pandas==2.2.*"

Confirm connectivity and your identity in one call before writing any extraction logic. The PI Web API root returns a small set of links, and a 401 here means the credential is wrong, while a 403 means the account authenticated but lacks read rights on the resource you asked for.

curl -s -u 'DOMAIN\pi_readonly' https://pi.example.gov/piwebapi/ | head -c 400

Step-by-Step Implementation

Step 1 — Resolve a tag to a stable WebID

Every PI point has a human name such as PLANT1.EFF.TURBIDITY and an opaque, encoded WebID that the PI Web API uses to address its stream. Names can be renamed by SCADA engineers; WebIDs are stable for the life of the point, so resolve the name once and cache the WebID for the run. The lookup is a single request to the points controller using the point’s full path in PI’s backslash notation.

import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Iterator, Optional
import httpx

logger = logging.getLogger("pi_extractor")


class PIWebAPIClient:
    """Thin, read-only PI Web API client over a pooled httpx session."""

    def __init__(self, base_url: str, username: str, password: str,
                 verify: str | bool = True, timeout: float = 30.0):
        self._client = httpx.Client(
            base_url=base_url.rstrip("/"),
            auth=httpx.BasicAuth(username, password),
            verify=verify,          # path to the CA bundle in production
            http2=True,
            timeout=httpx.Timeout(timeout),
            headers={"Accept": "application/json"},
        )

    def resolve_webid(self, server: str, tag: str) -> str:
        """Look up a PI point WebID from its \\\\server\\tag path."""
        path = f"\\\\{server}\\{tag}"
        resp = self._client.get("/points", params={
            "path": path,
            "selectedFields": "WebId;Name;PointType",
        })
        resp.raise_for_status()
        body = resp.json()
        web_id = body.get("WebId")
        if not web_id:
            raise KeyError(f"PI point not found: {path}")
        logger.info("Resolved %s -> %s (%s)", tag, web_id, body.get("PointType"))
        return web_id

    def close(self) -> None:
        self._client.close()

The selectedFields query parameter is not cosmetic — PI Web API responses are large by default, carrying self-referential links for every resource, and trimming them to the fields you consume cuts payload size and latency materially across a bulk backfill. Cache the resolved WebID in memory for the duration of the run, but do not persist it across deployments as if it were configuration: a point that is deleted and recreated in PI receives a fresh WebID, and a stale cached value will silently return 404 long after the tag name still works. Resolving by name at startup keeps the extractor correct through those archive edits.

Step 2 — Page through recorded values

For compliance you almost always want recorded values: the actual archived samples with their original timestamps and status, not resampled points. The recorded controller returns items in ascending time order and caps each response at maxCount. There is no server cursor, so paging is done by time: request a window, and if the response comes back full, advance the window start to just past the last timestamp received and request again. The generator below walks an arbitrary range in bounded pages and yields raw items.

    def iter_recorded(self, web_id: str, start: datetime, end: datetime,
                      page_size: int = 1000) -> Iterator[dict[str, Any]]:
        """Yield recorded PI values from start to end, paging by timestamp."""
        cursor = start.astimezone(timezone.utc)
        end = end.astimezone(timezone.utc)
        while cursor < end:
            resp = self._client.get(
                f"/streams/{web_id}/recorded",
                params={
                    "startTime": cursor.isoformat(),
                    "endTime": end.isoformat(),
                    "maxCount": page_size,
                    "boundaryType": "Inside",
                    "selectedFields": (
                        "Items.Timestamp;Items.Value;Items.Good;"
                        "Items.Questionable;Items.Substituted;"
                        "Items.UnitsAbbreviation"
                    ),
                },
            )
            resp.raise_for_status()
            items = resp.json().get("Items", [])
            if not items:
                return
            for item in items:
                yield item
            if len(items) < page_size:
                return  # last (partial) page reached
            last_ts = _parse_ts(items[-1]["Timestamp"])
            # advance one microsecond past the last sample to avoid a duplicate
            cursor = last_ts + timedelta(microseconds=1)

boundaryType=Inside keeps the window half-open so the sample sitting exactly on a page boundary is never emitted twice — the paired guard is the one-microsecond nudge to the cursor. Between those two rules, a multi-day backfill streams cleanly with no gaps and no repeats.

Step 3 — Normalize into a typed reading with a quality flag

A raw PI item is not yet a compliance record. Its Value field is polymorphic: for an analog point it is a number, but when the source is a digital state — I/O Timeout, Bad, Shutdown, Comm Fail — Value is a nested object with IsSystem: true. Treat those as non-numeric and grade them BAD; never coerce a system state into a float. The three boolean status fields map to the quality ladder: anything not Good is questionable at best, and a substituted value carries an operator edit that compliance reviewers must be able to see.

from enum import Enum
from pydantic import BaseModel, ConfigDict


class Quality(str, Enum):
    GOOD = "GOOD"
    QUESTIONABLE = "QUESTIONABLE"
    SUBSTITUTED = "SUBSTITUTED"
    BAD = "BAD"


class PIReading(BaseModel):
    model_config = ConfigDict(frozen=True)
    tag: str
    timestamp: datetime
    value: Optional[float]
    units: Optional[str]
    quality: Quality
    recorded: bool = True


def _parse_ts(raw: str) -> datetime:
    return datetime.fromisoformat(raw.replace("Z", "+00:00")).astimezone(timezone.utc)


def normalize(tag: str, item: dict[str, Any]) -> PIReading:
    raw_value = item.get("Value")
    ts = _parse_ts(item["Timestamp"])

    # A digital/system state arrives as a dict, not a scalar.
    if isinstance(raw_value, dict) and raw_value.get("IsSystem"):
        logger.warning("System state on %s at %s: %s", tag, ts, raw_value.get("Name"))
        return PIReading(tag=tag, timestamp=ts, value=None,
                         units=item.get("UnitsAbbreviation"), quality=Quality.BAD)

    if not item.get("Good", False):
        quality = Quality.BAD
    elif item.get("Substituted", False):
        quality = Quality.SUBSTITUTED
    elif item.get("Questionable", False):
        quality = Quality.QUESTIONABLE
    else:
        quality = Quality.GOOD

    value = float(raw_value) if isinstance(raw_value, (int, float)) else None
    if value is None and quality == Quality.GOOD:
        quality = Quality.BAD  # numeric expected but absent

    return PIReading(tag=tag, timestamp=ts, value=value,
                     units=item.get("UnitsAbbreviation"), quality=quality)

The frozen=True model config makes each reading immutable once built, which matters when the same object is fanned out to several consumers — the rule engine cannot accidentally mutate what the historian said. Wiring the three steps together, a full extraction run is compact.

def extract(client: PIWebAPIClient, server: str, tag: str,
            start: datetime, end: datetime) -> Iterator[PIReading]:
    web_id = client.resolve_webid(server, tag)
    for item in client.iter_recorded(web_id, start, end):
        yield normalize(tag, item)

The paging-and-normalize loop is easier to reason about as a diagram than as prose: resolve once, then loop the recorded endpoint until a short page signals the end.

PI Web API recorded-value paging and normalization flow Resolve the tag WebID once, then repeatedly GET a page of recorded values. If the page is full the extractor advances the start time past the last timestamp and loops back for the next page; when a short page arrives it normalizes the items into typed PIReading records and passes them into the compliance pipeline. Resolve tag WebID GET …/recorded one page Normalize → PIReading Page full? no yes · advance startTime into the compliance pipeline

Step 4 — Know when interpolation is legitimate

The recorded controller gives you archived truth. The sibling interpolated controller resamples the archive onto an even grid — for example one point per hour — by drawing a straight line between the two recorded samples that bracket each grid time. For a value between recorded points (t0,v0)(t_0, v_0) and (t1,v1)(t_1, v_1), PI returns

v(t)=v0+(v1−v0)⋅t−t0t1−t0v(t) = v_0 + (v_1 - v_0)\cdot\frac{t - t_0}{t_1 - t_0}

Interpolated points are convenient for dashboards and for evenly spaced feeds, but they are synthetic: the value at that timestamp was never measured. The distinction has regulatory teeth. A running annual average or a maximum-contaminant-level check that quietly folds in interpolated fill can understate a genuine excursion, because the straight-line estimate always sits between the two real samples and can never exceed either of them. Grade every interpolated reading with recorded=False so downstream consumers can exclude it from any calculation that must rest on measured data, and reconcile the mixed cadence of recorded and interpolated feeds with the time-series alignment strategies module before it reaches a compliance window. A practical rule that serves most utilities well: pull recorded for anything that feeds a reportable calculation, and reserve interpolated for visualization and for gap-aware feeds that already carry their own provenance. Chlorine and turbidity streams that originate on the OPC UA data extraction side of the plant frequently land in PI as well, so a utility often blends both sources on the same tag axis.

Configuration Reference

The parameters below are the ones that vary by site and by tag. Treat the server name, tag paths, and page size as versioned configuration rather than inline literals.

Parameter Endpoint / field Type Default Meaning
base_url /piwebapi str — Root of the PI Web API service
path GET /points str — \\server\tag point path to resolve
web_id GET /streams/{webId}/recorded str — Stable stream identifier from the lookup
startTime / endTime recorded / interpolated ISO 8601 — UTC window bounds for the query
maxCount recorded int 1000 Page size cap per request
boundaryType recorded str Inside Half-open window; excludes the boundary sample
interval interpolated duration — Resample step, e.g. 1h, 15m
selectedFields any controller str — Semicolon list trimming the response payload
Good item status bool — false maps to quality BAD
Questionable item status bool — true maps to quality QUESTIONABLE
Substituted item status bool — true maps to quality SUBSTITUTED

Verification & Testing

Normalization is the part of this pipeline most worth pinning with tests, because its inputs are irregular and its output is what compliance trusts. The suite below covers a clean analog sample, a system-state item, and a substituted value — the three cases that most often slip through.

def test_good_analog_reading():
    item = {"Timestamp": "2026-07-01T00:00:00Z", "Value": 0.42,
            "Good": True, "Questionable": False, "Substituted": False,
            "UnitsAbbreviation": "NTU"}
    r = normalize("PLANT1.EFF.TURBIDITY", item)
    assert r.value == 0.42
    assert r.quality is Quality.GOOD
    assert r.timestamp.tzinfo is timezone.utc


def test_system_state_is_bad_and_nonnumeric():
    item = {"Timestamp": "2026-07-01T01:00:00Z",
            "Value": {"Name": "I/O Timeout", "IsSystem": True},
            "Good": False}
    r = normalize("PLANT1.EFF.TURBIDITY", item)
    assert r.value is None
    assert r.quality is Quality.BAD


def test_substituted_value_is_flagged():
    item = {"Timestamp": "2026-07-01T02:00:00Z", "Value": 1.1,
            "Good": True, "Substituted": True}
    r = normalize("PLANT1.EFF.TURBIDITY", item)
    assert r.quality is Quality.SUBSTITUTED
    assert r.value == 1.1

Acceptance criteria before promoting the extractor to production:

Troubleshooting & Gotchas

  • Backfill loops forever or stalls on the last page. The exit conditions are a short page or an empty page; if you advance the cursor without the len(items) < page_size check, a range whose final page is exactly full will loop. Keep both the microsecond nudge and the short-page return, and the paging terminates cleanly.
  • Duplicate samples at page boundaries. Without boundaryType=Inside plus the one-microsecond cursor advance, the sample on the seam is emitted by both the page that ends on it and the page that starts on it. The two rules together make the window half-open.
  • Value is sometimes a dict and crashes float(). That is a PI system digital state, not a number. Test isinstance(raw_value, dict) and route it to BAD before any numeric conversion, as normalize does.
  • A 403 only on certain tags. The account authenticated but lacks read access to those specific points or their parent element. This is a PI security-mapping problem, not a code bug — grant read on the points to the read-only identity rather than widening the account.
  • Timestamps drift by whole seconds after conversion. PI returns ISO 8601 with a trailing Z; parse it as UTC-aware and keep it UTC through the whole path. Never strip the offset to a naive datetime, or the writing validated readings to InfluxDB for compliance step will misplace points across the compliance window boundary.

Related pages