piwebapiconnect 0.1.6

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())

---

Functions Provided

All service functions are asynchronous and must be awaited.

Standard Data Fetching

• get_current_value(tag_name: str): Fetch the current snapshot value of a tag.
• get_real_time_data(tag_name: str): An alias for get_current_value.
• get_recorded_values(tag_name: str, start_time: str, end_time: str): Fetch actual historical recorded values for a time range.
• get_interpolated_values(tag_name: str, start_time: str, end_time: str, interval: str): Fetch evenly spaced interpolated values (e.g., interval="1h").
• get_multiple_current_values(tag_names: List[str]): Fetch values for multiple tags simultaneously using high-efficiency StreamSets.
• get_summary_data(tag_name: str, start_time: str, end_time: str, summary_type: str = "Average"): Get Server-calculated statistics (Average, Minimum, Maximum, Total, etc.).

Live Data Streams

• get_live_data_stream(tag_name: str, interval_seconds: int = 60): Returns an asynchronous generator yielding live value updates continuously.

  async for data in service.get_live_data_stream("SINUSOID", interval_seconds=5):
      print(data)
  

Data Manipulation

• write_value(tag_name: str, value: Any, timestamp: Any = None): Write a new value to a PI tag at the specified timestamp.
• delete_value(tag_name: str, timestamp: str): Delete an exact point from a PI tag.

Metadata & Architecture Options

• search_tags(query: str): Search for PI tags based on a query pattern (e.g., *temperature*).
• get_tag_attributes(tag_name: str): Fetch the configuration and metadata associated with a PI point.
• get_asset_servers(): List available PI AF Asset Servers.
• get_organizations(asset_server: str): List AF Databases on a specified server.

Utilities

• test_connection(): Safely verify connection status without throwing exceptions.
• close(): Cleanly drop the internal async HTTP client connections.

---

Exception Handling

The library utilizes a custom PIWebAPIError class to safely report network crashes, HTTP errors, and typo issues, preventing your script from shutting down abruptly.

from piwebapiconnect.service import PIWebAPIError

try:
    data = await service.get_current_value("INVALID_TAG_NAME")
except PIWebAPIError as e:
    print(f"PI Database Error: {e}")