piwebapiconnect 0.1.8

A professional library for PI Web API integration

piwebapiconnect

A professional Python library for OSIsoft PI Web API integration. Built cleanly with httpx and asyncio to be purely asynchronous and incredibly robust.

Installation

pip install piwebapiconnect

Setup & Connecting

1. Using a .env file (Recommended)

Create a .env file in your root folder:

PI_WEB_API_URL=https://your-pi-server/piwebapi
PI_WEB_API_USERNAME=domain\user
PI_WEB_API_PASSWORD=your-password
PI_SERVER_NAME=PISERVER01
PI_WEB_API_VERIFY_SSL=False  # Set to True in production

from piwebapiconnect import get_service
service = get_service()

2. Passing credentials directly

import asyncio
from piwebapiconnect import get_service

async def main():
    service = get_service(
        url="https://your-pi-server/piwebapi",
        username="domain\\user",
        password="your-password",
        server_name="PISERVER01",
        verify_ssl=False
    )

    print(await service.test_connection())
    await service.close()

if __name__ == "__main__":
    asyncio.run(main())

---

Detailed Example Usage

All methods in this library are asynchronous and should be used within an async context.

Initialization

import asyncio
from piwebapiconnect import get_service

async def main():
    # Initialize service with demo credentials
    service = get_service(
        url="https://demo-pi-server/piwebapi",
        username="domain\\user",
        password="password123",
        server_name="PISERVER01",
        verify_ssl=False
    )
    
    # ... use service methods here ...

    await service.close()

if __name__ == "__main__":
    asyncio.run(main())

1. Fetch Current Data

Retrieve the latest value for a single tag.

# Both methods are identical
val = await service.get_current_value("SINUSOID")
val = await service.get_real_time_data("SINUSOID")

print(f"Value: {val['value']} at {val['timestamp']}")

2. Fetch Multiple Current Values

Retrieve values for multiple tags in a single efficient batch request.

tags = ["SINUSOID", "CDT158", "BA:LEVEL.1"]
results = await service.get_multiple_current_values(tags)
for tag, data in results.items():
    print(f"{tag}: {data['value']}")

3. Historical Data (Recorded & Interpolated)

# Actual stored points
history = await service.get_recorded_values("SINUSOID", "*-24h", "*")

# Interpolated points at specific intervals
interpolated = await service.get_interpolated_values("SINUSOID", "*-24h", "*", interval="1h")

4. Summary Statistics

Get calculated values like Average, Min, or Max directly from the PI Server.

summary = await service.get_summary_data("SINUSOID", "*-24h", "*", summary_type="Average")
print(f"Average: {summary['value']}")

5. Live Data Stream

Yields continuous updates for one or more tags.

# Stream for multiple tags
async for batch in service.get_live_data_stream(["SINUSOID", "CDT158"], interval_seconds=5):
    for update in batch:
        print(f"Update - {update['tag']}: {update['value']}")

# Stream for a single tag (yields Dict instead of List)
async for update in service.get_live_data_stream("SINUSOID", interval_seconds=5):
    print(f"Single Update: {update['value']}")

6. Data Manipulation (Write & Delete)

# Write a value
await service.write_value("TEST_TAG", 123.4)

# Delete a value at a specific timestamp
await service.delete_value("TEST_TAG", "2023-10-01T12:00:00Z")

7. Metadata & Hierarchy Browsing

# Search for tags
tags = await service.search_tags("*pressure*")

# Get tag configuration/attributes
attrs = await service.get_tag_attributes("SINUSOID")

# Browse Asset Servers and Databases
servers = await service.get_asset_servers()
databases = await service.get_organizations("AF_SERVER_DEMO")

---

Error Handling

The library provides a PIWebAPIError for handling API-specific issues gracefully.

from piwebapiconnect.service import PIWebAPIError

try:
    data = await service.get_current_value("INVALID_TAG")
except PIWebAPIError as e:
    print(f"Caught PI Error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

---

License

This project is licensed under the MIT License.