Reading Values

POINT_WEB_ID = "your-web-id"

response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/value",
    params={
        "selectedFields": "Timestamp;Value;Good;UnitsAbbreviation",
    },
)
data = response.json()

print(f"Value:     {data['Value']}")
print(f"Timestamp: {data['Timestamp']}")
print(f"Good:      {data['Good']}")
print(f"Units:     {data.get('UnitsAbbreviation', 'N/A')}")

response = session.get(f"{BASE_URL}/streams/{WEB_ID}/value")
data = response.json()

value = data["Value"]

# Check if the value is a digital state (dict) or numeric (int/float)
if isinstance(value, dict):
    # Digital state: {"Name": "Active", "Value": 1}
    state_name = value.get("Name", "Unknown")
    state_value = value.get("Value", None)
    print(f"Digital state: {state_name} (code: {state_value})")
elif isinstance(value, (int, float)):
    print(f"Numeric value: {value}")
elif isinstance(value, str):
    # String point (rare, but possible)
    print(f"String value: {value}")
else:
    print(f"Unexpected value type: {type(value)} = {value}")

# Helper function for safe value extraction
def extract_numeric_value(pi_value):
    """Extract a numeric value from a PI Web API value response.

    Returns None for digital states and bad quality values.
    """
    value = pi_value.get("Value")
    if not pi_value.get("Good", True):
        return None
    if isinstance(value, dict):
        return None  # Digital state
    if isinstance(value, (int, float)):
        return float(value)
    return None

response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/recorded",
    params={
        "startTime": "*-1d",       # 1 day ago
        "endTime": "*",            # Now
        "maxCount": 1000,          # Limit results
        "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
    },
)

items = response.json()["Items"]
print(f"Retrieved {len(items)} recorded values")

# Warning: if len(items) == maxCount, data may be truncated
if len(items) == 1000:
    print("Results may be truncated! Use pagination for complete data.")

# Use Interpolated boundary for accurate time-window calculations
response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/recorded",
    params={
        "startTime": "2026-03-15T08:00:00Z",
        "endTime": "2026-03-15T16:00:00Z",
        "boundaryType": "Interpolated",
        "maxCount": 10000,
    },
)
# Now the first value is interpolated AT 08:00:00 and the last AT 16:00:00

# Only return values above 50
response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/recorded",
    params={
        "startTime": "*-1d",
        "endTime": "*",
        "filterExpression": "'.' > 50",   # '.' refers to the current value
        "maxCount": 10000,
    },
)

# Only return good quality values
response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/recorded",
    params={
        "startTime": "*-1d",
        "endTime": "*",
        "filterExpression": "IsGood('.')",
        "maxCount": 10000,
    },
)

response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/interpolated",
    params={
        "startTime": "*-1d",
        "endTime": "*",
        "interval": "1h",   # One value per hour = 24 values for 1 day
        "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
    },
)

items = response.json()["Items"]
for item in items:
    print(f"{item['Timestamp']}: {item['Value']}")

# Read aligned interpolated values for multiple points
params = {
    "startTime": "t",        # Today at midnight
    "endTime": "*",          # Now
    "interval": "5m",        # 5-minute intervals
}

temp_items = session.get(
    f"{BASE_URL}/streams/{TEMP_WEB_ID}/interpolated", params=params
).json()["Items"]

pressure_items = session.get(
    f"{BASE_URL}/streams/{PRESSURE_WEB_ID}/interpolated", params=params
).json()["Items"]

# Because we used the same params, timestamps are aligned
# and we can zip them together
for temp, pressure in zip(temp_items, pressure_items):
    # temp["Timestamp"] == pressure["Timestamp"]
    print(f"{temp['Timestamp']}: T={temp['Value']:.1f} P={pressure['Value']:.2f}")

response = session.get(
    f"{BASE_URL}/streams/{POINT_WEB_ID}/summary",
    params={
        "startTime": "*-1d",
        "endTime": "*",
        "summaryType": "Average,Minimum,Maximum,StdDev,Count,PercentGood,Range",
        "calculationBasis": "TimeWeighted",
    },
)

summaries = response.json()["Items"]
for summary in summaries:
    stype = summary["Type"]
    value = summary["Value"]["Value"]
    good = summary["Value"]["Good"]
    print(f"{stype}: {value} (Good: {good})")

import pandas as pd

def pi_recorded_to_dataframe(
    session, base_url, web_id, start="*-7d", end="*",
    max_count=10000, column_name="Value"
):
    """Load PI recorded values into a pandas DataFrame.

    Handles:
    - Timezone-aware timestamps (UTC)
    - Digital state filtering (non-numeric values become NaN)
    - Quality flag filtering (bad values become NaN)
    - Proper DatetimeIndex for time-series analysis
    """
    response = session.get(
        f"{base_url}/streams/{web_id}/recorded",
        params={
            "startTime": start,
            "endTime": end,
            "maxCount": max_count,
            "selectedFields": "Items.Timestamp;Items.Value;Items.Good",
        },
    )
    response.raise_for_status()
    items = response.json().get("Items", [])

    if not items:
        return pd.DataFrame(columns=[column_name])

    # Build DataFrame
    rows = []
    for item in items:
        value = item["Value"]
        good = item.get("Good", True)

        # Handle digital states (value is a dict like {"Name": "Active", "Value": 1})
        if isinstance(value, dict):
            numeric_value = None  # or value.get("Value") if you want the integer code
        elif isinstance(value, (int, float)):
            numeric_value = float(value) if good else None
        else:
            numeric_value = None

        rows.append({
            "Timestamp": item["Timestamp"],
            column_name: numeric_value,
        })

    df = pd.DataFrame(rows)
    df["Timestamp"] = pd.to_datetime(df["Timestamp"], utc=True)
    df = df.set_index("Timestamp").sort_index()

    return df


# Usage
df = pi_recorded_to_dataframe(session, BASE_URL, POINT_WEB_ID)

print(f"Shape: {df.shape}")
print(f"Time range: {df.index.min()} to {df.index.max()}")
print(f"NaN count: {df.isna().sum().iloc[0]} (bad quality or digital states)")
print(f"\nStatistics:")
print(df.describe())

# Multiple points into one aligned DataFrame
points = {
    "Temperature": TEMP_WEB_ID,
    "Pressure": PRESSURE_WEB_ID,
    "Flow": FLOW_WEB_ID,
}

frames = {}
for name, wid in points.items():
    frames[name] = pi_recorded_to_dataframe(
        session, BASE_URL, wid, column_name=name
    )[name]

combined = pd.concat(frames, axis=1)
print(combined.head())