Example Gallery

import requests

session = requests.Session()
session.auth = ("DOMAIN\\username", "password")
session.verify = "/path/to/ca-bundle.pem"  # Or False for dev only

BASE_URL = "https://myserver/piwebapi"

# Verify connection
resp = session.get(f"{BASE_URL}/")
resp.raise_for_status()
info = resp.json()
print(f"Connected to {info['ProductTitle']} {info['ProductVersion']}")

import requests
from requests_kerberos import HTTPKerberosAuth, REQUIRED

session = requests.Session()
session.auth = HTTPKerberosAuth(mutual_authentication=REQUIRED)
session.verify = "/path/to/ca-bundle.pem"

resp = session.get(f"{BASE_URL}/")
resp.raise_for_status()
print(f"Authenticated as: {resp.json().get('Links', {}).get('Self')}")

const BASE_URL = "https://myserver/piwebapi";
const headers = {
  "Authorization": "Basic " + btoa("DOMAIN\\username:password"),
  "Accept": "application/json",
};

const resp = await fetch(`${BASE_URL}/`, { headers });
const info = await resp.json();
console.log(`Connected to ${info.ProductTitle} ${info.ProductVersion}`);

using System.Net.Http;
using System.Net.Http.Headers;
using System.Text;

var handler = new HttpClientHandler();
// For self-signed certs in dev only:
// handler.ServerCertificateCustomValidationCallback = (_, _, _, _) => true;

var client = new HttpClient(handler);
var authBytes = Encoding.ASCII.GetBytes("DOMAIN\\username:password");
client.DefaultRequestHeaders.Authorization =
    new AuthenticationHeaderValue("Basic", Convert.ToBase64String(authBytes));

var resp = await client.GetAsync("https://myserver/piwebapi/");
resp.EnsureSuccessStatusCode();
var json = await resp.Content.ReadAsStringAsync();
Console.WriteLine(json);

# Find a PI point by its full path (most reliable method)
path = "\\\\MY-SERVER\\sinusoid"
resp = session.get(
    f"{BASE_URL}/points",
    params={
        "path": path,
        "selectedFields": "WebId;Name;PointType;EngineeringUnits",
    },
)
resp.raise_for_status()
point = resp.json()
print(f"Name: {point['Name']}")
print(f"WebID: {point['WebId']}")
print(f"Type: {point['PointType']}")
print(f"Units: {point.get('EngineeringUnits', 'N/A')}")

# First, get the data server WebID
ds_resp = session.get(
    f"{BASE_URL}/dataservers",
    params={"selectedFields": "Items.WebId;Items.Name"},
)
DATA_SERVER_WEB_ID = ds_resp.json()["Items"][0]["WebId"]

# Search for PI points by name pattern
resp = session.get(
    f"{BASE_URL}/dataservers/{DATA_SERVER_WEB_ID}/points",
    params={
        "nameFilter": "Temperature*",
        "maxCount": 20,
        "selectedFields": "Items.WebId;Items.Name;Items.PointType",
    },
)
resp.raise_for_status()
points = resp.json()["Items"]
print(f"Found {len(points)} points:")
for p in points:
    print(f"  {p['Name']}: {p['WebId'][:20]}...")

const path = encodeURIComponent("\\\\MY-SERVER\\sinusoid");
const resp = await fetch(
  `${BASE_URL}/points?path=${path}&selectedFields=WebId;Name;PointType`,
  { headers }
);
const point = await resp.json();
console.log(`${point.Name}: ${point.WebId}`);

# Read current (snapshot) value
resp = session.get(
    f"{BASE_URL}/streams/{web_id}/value",
    params={"selectedFields": "Timestamp;Value;Good"},
)
resp.raise_for_status()
data = resp.json()
print(f"Value: {data['Value']} at {data['Timestamp']}")
print(f"Quality: {'Good' if data.get('Good', True) else 'Bad'}")

# Expected response:
# {
#   "Timestamp": "2026-03-15T14:30:00Z",
#   "Value": 72.34,
#   "Good": true
# }

# Read recorded values over last 24 hours
resp = session.get(
    f"{BASE_URL}/streams/{web_id}/recorded",
    params={
        "startTime": "*-24h",
        "endTime": "*",
        "maxCount": 10000,
        "boundaryType": "Inside",
        "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
    },
)
resp.raise_for_status()
items = resp.json()["Items"]

# Check for truncation
if len(items) == 10000:
    print("WARNING: Data may be truncated. Increase maxCount or narrow time range.")

# Filter and display
good_values = [i for i in items if i.get("Good", True)]
print(f"Got {len(items)} values ({len(good_values)} good quality)")
for item in good_values[:5]:
    print(f"  {item['Timestamp']}: {item['Value']}")

# Read interpolated values at regular 1-hour intervals
resp = session.get(
    f"{BASE_URL}/streams/{web_id}/interpolated",
    params={
        "startTime": "*-8h",
        "endTime": "*",
        "interval": "1h",
        "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
    },
)
resp.raise_for_status()
items = resp.json()["Items"]

# Interpolated values are evenly spaced and much faster than recorded
print(f"Got {len(items)} interpolated values:")
for item in items:
    print(f"  {item['Timestamp']}: {item['Value']:.2f}")

const resp = await fetch(
  `${BASE_URL}/streams/${webId}/value?selectedFields=Timestamp;Value;Good`,
  { headers }
);
const data = await resp.json();
console.log(`Value: ${data.Value} at ${data.Timestamp}`);

var url = $"{baseUrl}/streams/{webId}/value?selectedFields=Timestamp;Value;Good";
var resp = await client.GetAsync(url);
resp.EnsureSuccessStatusCode();

var json = await resp.Content.ReadFromJsonAsync<JsonElement>();
var value = json.GetProperty("Value").GetDouble();
var timestamp = json.GetProperty("Timestamp").GetString();
Console.WriteLine($"Value: {value} at {timestamp}");

# Write a single value to the current timestamp
resp = session.post(
    f"{BASE_URL}/streams/{web_id}/value",
    json={"Value": 72.5, "Timestamp": "*"},
    params={"updateOption": "Replace"},
)
print(f"Status: {resp.status_code}")  # 202 = accepted

# Expected: HTTP 202 Accepted (write buffered)
# or HTTP 204 No Content (write committed immediately)

# Write multiple historical values
# NOTE: Use a list of value objects (not wrapped in {"Items": [...]})
values = [
    {"Value": 70.1, "Timestamp": "2026-03-15T10:00:00Z"},
    {"Value": 71.3, "Timestamp": "2026-03-15T10:05:00Z"},
    {"Value": 72.0, "Timestamp": "2026-03-15T10:10:00Z"},
    {"Value": 73.5, "Timestamp": "2026-03-15T10:15:00Z"},
]
resp = session.post(
    f"{BASE_URL}/streams/{web_id}/recorded",
    json=values,
    params={"updateOption": "Replace"},
)
print(f"Status: {resp.status_code}")

# Check for partial errors
if resp.status_code not in (200, 202, 204):
    print(f"Errors: {resp.json().get('Errors', resp.text[:200])}")

const resp = await fetch(`${BASE_URL}/streams/${webId}/value?updateOption=Replace`, {
  method: "POST",
  headers: { ...headers, "Content-Type": "application/json" },
  body: JSON.stringify({ Value: 72.5, Timestamp: new Date().toISOString() }),
});
console.log(`Write status: ${resp.status}`);

# Read current values for multiple points in ONE request
batch = {
    "Temp": {
        "Method": "GET",
        "Resource": f"{BASE_URL}/streams/{temp_wid}/value?selectedFields=Timestamp;Value;Good",
    },
    "Pressure": {
        "Method": "GET",
        "Resource": f"{BASE_URL}/streams/{pressure_wid}/value?selectedFields=Timestamp;Value;Good",
    },
    "Flow": {
        "Method": "GET",
        "Resource": f"{BASE_URL}/streams/{flow_wid}/value?selectedFields=Timestamp;Value;Good",
    },
}

resp = session.post(f"{BASE_URL}/batch", json=batch)
resp.raise_for_status()

# IMPORTANT: Check each sub-request status individually
for name, result in resp.json().items():
    if result["Status"] == 200:
        content = result["Content"]
        quality = "Good" if content.get("Good", True) else "Bad"
        print(f"{name}: {content['Value']} ({quality})")
    else:
        print(f"{name}: ERROR {result['Status']}")

# Look up a point AND read its value in one batch request
# The second request depends on the first (uses ParentIds)
batch = {
    "lookup": {
        "Method": "GET",
        "Resource": f"{BASE_URL}/points?path=\\\\MY-SERVER\\sinusoid&selectedFields=WebId",
    },
    "read": {
        "Method": "GET",
        "Resource": "{0}/streams/{1}/value?selectedFields=Timestamp;Value",
        "Parameters": [BASE_URL, "$.lookup.Content.WebId"],
        "ParentIds": ["lookup"],
    },
}

resp = session.post(f"{BASE_URL}/batch", json=batch)
resp.raise_for_status()
results = resp.json()

if results["read"]["Status"] == 200:
    val = results["read"]["Content"]
    print(f"Value: {val['Value']} at {val['Timestamp']}")
else:
    print(f"Error: {results['read']['Status']}")

import pandas as pd

# Fetch recorded values
resp = session.get(
    f"{BASE_URL}/streams/{web_id}/recorded",
    params={
        "startTime": "*-7d",
        "endTime": "*",
        "maxCount": 50000,
        "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
    },
)
resp.raise_for_status()
items = resp.json()["Items"]

# Convert to DataFrame
df = pd.DataFrame(items)
df["Timestamp"] = pd.to_datetime(df["Timestamp"], utc=True)
df = df.set_index("Timestamp").sort_index()

# Handle digital state values (come as dicts, not numbers)
def clean_value(v):
    if isinstance(v, dict):
        return v.get("Name", str(v))
    return v

df["Value"] = df["Value"].apply(clean_value)

# Filter to good quality only
df_good = df[df["Good"] == True][["Value"]]

# Convert to numeric (digital states will become NaN)
df_good["Value"] = pd.to_numeric(df_good["Value"], errors="coerce")

# Analysis
print(df_good["Value"].describe())
print(f"\nHourly averages (last 24h):")
print(df_good["Value"].last("24h").resample("1h").mean())

import pandas as pd

def read_interpolated_multi(session, base_url, points, start, end, interval):
    """Read interpolated values for multiple points and align into one DataFrame.

    Args:
        points: list of (name, web_id) tuples
        interval: e.g., "5m", "1h", "1d"
    """
    frames = {}
    for name, wid in points:
        resp = session.get(
            f"{base_url}/streams/{wid}/interpolated",
            params={
                "startTime": start,
                "endTime": end,
                "interval": interval,
                "selectedFields": "Items.Timestamp;Items.Value",
            },
        )
        resp.raise_for_status()
        items = resp.json()["Items"]
        s = pd.DataFrame(items)
        s["Timestamp"] = pd.to_datetime(s["Timestamp"], utc=True)
        frames[name] = s.set_index("Timestamp")["Value"]

    return pd.DataFrame(frames)


# Usage
points = [
    ("Temperature", temp_wid),
    ("Pressure", pressure_wid),
    ("Flow", flow_wid),
]
df = read_interpolated_multi(session, BASE_URL, points, "*-24h", "*", "5m")
print(df.describe())
print(f"\nCorrelation matrix:\n{df.corr()}")