Of course! Here is a comprehensive guide on using OSIsoft PI Python, covering the official library, common use cases, best practices, and a complete example.

Overview: The Official PI Python SDK

OSIsoft provides an official Python Software Development Kit (SDK) called pisdk. This is the recommended and most powerful way to interact with the PI System from Python.

• Library Name: pisdk (for the classic PI SDK) and pisyscfg (for PI System configuration).
• Installation: You install it using pip.
• Connection: It connects to the PI Data Archive (the main server) using standard PI protocols.
• Features: You can read data, write data, search for tags, get current values, perform calculations, and more.

Installation

First, you need to install the pisdk library. Open your terminal or command prompt and run:

pip install pisdk

Connecting to the PI Data Archive

Before you can do anything, you need to establish a connection to your PI Data Archive. The primary object for this is the PISDK.PIServer object.

import pisdk
# --- Connection Details ---
# Replace with your PI Data Archive server name or IP address
server_name = "YOUR_PI_SERVER"
# Optional: Provide a username and password if security is enabled
# username = "your_username"
# password = "your_password"
try:
    # Connect to the server
    # If no username/password is provided, it will use Windows authentication
    pi_server = pisdk.PIServer(server_name, username, password)
    # Optional: Test the connection
    if pi_server.Connected:
        print(f"Successfully connected to PI Server: {pi_server.Name}")
    else:
        print("Failed to connect to PI Server.")
except Exception as e:
    print(f"An error occurred during connection: {e}")
# You should always disconnect when you're done
# pi_server.Disconnect()

Key Concepts and Objects

Once connected, you'll work with a few key objects:

• PIServer: Represents the connection to the PI Data Archive. You already have this.
• PIPoint: Represents a single tag (or "point") in the PI System. This is the object you'll use to read/write data for a specific tag.
• PIPoints: A collection of PIPoint objects. Useful for getting multiple points at once.
• PIValue: Represents a single data value (with its timestamp, quality, etc.).
• PIValues: A collection of PIValue objects, representing a time range of data for a tag.

Common Operations

a. Finding a Tag (PIPoint)

You need to get a PIPoint object before you can read or write data. You can do this by searching for the tag's name.

# Assuming 'pi_server' is your connected server object
point_name = "SINUSOID" # A common example tag in PI servers
try:
    # Get the PIPoint object
    my_point = pi_server.PIPoints(point_name)
    # You can also search for points
    # found_points = pi_server.PIPoints.FindTag("[cdtT]sinusoid") # Wildcard search
    # if found_points:
    #     my_point = found_points[0]
    print(f"Found PI Point: {my_point.Name}")
except Exception as e:
    print(f"Error finding point '{point_name}': {e}")

b. Reading Current Value

The simplest read operation is to get the current value of a tag.

if my_point:
    try:
        current_value = my_point.Data.Snapshot
        print(f"Current value of '{point_name}': {current_value.Value} at {current_value.Timestamp}")
    except Exception as e:
        print(f"Error reading current value: {e}")

c. Reading Historical Data (Time Range)

This is the most common operation. You read data over a specified time range.

from datetime import datetime, timedelta
if my_point:
    try:
        # Define the time range
        end_time = datetime.now()
        start_time = end_time - timedelta(days=1) # Last 24 hours
        # Get the values
        # The 'return_raw_values' parameter returns PIValue objects
        values = my_point.RecordedValues(start_time, end_time, "", return_raw_values=True)
        print(f"\nReading historical data for '{point_name}' from {start_time} to {end_time}:")
        # Iterate through the returned values
        for value in values:
            # value.Value is the data value
            # value.Timestamp is the timestamp
            # value.StatusCode is the data quality (0 = Good)
            print(f"  Timestamp: {value.Timestamp}, Value: {value.Value}, Quality: {value.StatusCode}")
    except Exception as e:
        print(f"Error reading historical data: {e}")

d. Writing Data

You can write a single value or a list of values to a tag.

if my_point:
    try:
        # --- Option 1: Write a single value ---
        # You need a timestamp for the value
        write_timestamp = datetime.now()
        write_value = 123.45
        # Create a PIValue object to write
        value_to_write = pisdk.PIValue()
        value_to_write.Value = write_value
        value_to_write.Timestamp = write_timestamp
        # Write the value (this overwrites existing data at that exact timestamp)
        my_point.Data.UpdateValue(value_to_write, pisdk.UpdateOption.ReplaceOrAdd)
        print(f"\nWrote single value {write_value} to '{point_name}' at {write_timestamp}")
        # --- Option 2: Write a list of values ---
        # This is more efficient for bulk writes
        values_to_write = []
        for i in range(5):
            ts = write_timestamp - timedelta(minutes=i*5)
            val = 100 + i * 5
            pv = pisdk.PIValue()
            pv.Value = val
            pv.Timestamp = ts
            values_to_write.append(pv)
        my_point.Data.UpdateValues(values_to_write, pisdk.UpdateOption.ReplaceOrAdd)
        print(f"Wrote {len(values_to_write)} values to '{point_name}'.")
    except Exception as e:
        print(f"Error writing data: {e}")

Complete Example Script

Here is a single, runnable script that combines connection, reading, and writing.

import pisdk
from datetime import datetime, timedelta
# --- Configuration ---
PI_SERVER = "YOUR_PI_SERVER" # e.g., "PISRV01"
TAG_NAME = "SINUSOID"       # A tag that exists on your server
USERNAME = None             # Set to None for Windows Auth, or your username
PASSWORD = None             # Set to None for Windows Auth, or your password
def main():
    pi_server = None
    try:
        # 1. Connect to the PI Server
        print(f"Connecting to PI Server: {PI_SERVER}...")
        pi_server = pisdk.PIServer(PI_SERVER, USERNAME, PASSWORD)
        if not pi_server.Connected:
            print("Connection failed. Exiting.")
            return
        print("Connection successful.")
        # 2. Get the PIPoint object
        print(f"Finding PI Point: {TAG_NAME}...")
        my_point = pi_server.PIPoints(TAG_NAME)
        print(f"Found PI Point: {my_point.Name}")
        # 3. Read the current value
        print("\n--- Reading Current Value ---")
        current_value = my_point.Data.Snapshot
        print(f"Current Value: {current_value.Value} at {current_value.Timestamp}")
        # 4. Read historical data (last 24 hours)
        print("\n--- Reading Historical Data (Last 24 Hours) ---")
        end_time = datetime.now()
        start_time = end_time - timedelta(days=1)
        values = my_point.RecordedValues(start_time, end_time, "", return_raw_values=True)
        print(f"Found {len(values)} values in the time range.")
        # Print the first 5 values
        for i, value in enumerate(values[:5]):
            print(f"  {i+1}. Timestamp: {value.Timestamp}, Value: {value.Value}")
        if len(values) > 5:
            print(f"  ... and {len(values) - 5} more values.")
        # 5. Write a new value
        print("\n--- Writing a New Value ---")
        new_timestamp = datetime.now()
        new_value = 999.99
        value_to_write = pisdk.PIValue()
        value_to_write.Value = new_value
        value_to_write.Timestamp = new_timestamp
        my_point.Data.UpdateValue(value_to_write, pisdk.UpdateOption.ReplaceOrAdd)
        print(f"Successfully wrote value {new_value} at {new_timestamp}")
        # 6. Verify the write by reading the current value again
        print("\n--- Verifying the Write ---")
        updated_value = my_point.Data.Snapshot
        print(f"Updated Current Value: {updated_value.Value} at {updated_value.Timestamp}")
    except Exception as e:
        print(f"\nAn error occurred: {e}")
    finally:
        # 7. Disconnect from the server
        if pi_server and pi_server.Connected:
            print("\nDisconnecting from PI Server.")
            pi_server.Disconnect()
if __name__ == "__main__":
    main()

Advanced Topics & Best Practices

• Performance (Bulk Operations): Always prefer reading and writing in bulk (RecordedValues, UpdateValues) over looping and doing single operations. This drastically reduces network latency.
• Error Handling: The PI SDK can raise various exceptions (e.g., pisdk.PIException). Always wrap your PI SDK calls in try...except blocks to handle connection issues, tag not found errors, and bad data quality gracefully.
• Asynchronous Operations: For high-performance applications, consider using the PI SDK's asynchronous methods (e.g., RecordedValuesAsync) to avoid blocking your main thread. This is more advanced but can significantly improve throughput.
• PI AF SDK: If you are working with PI Asset Framework (AF), OSIsoft provides a separate but similar SDK called piafsdk. It allows you to browse and interact with the hierarchical AF database (elements, attributes, templates). The concepts are very similar to the PI SDK.
• Connection Pooling: If your application needs to connect frequently, consider maintaining a persistent connection object instead of connecting and disconnecting for every small task.

Alternatives to the Official SDK

While the pisdk is the official choice, other libraries exist:

• PI Web API: This is a modern, RESTful web API for the PI System. You can interact with it from Python using the requests library. It's excellent for web applications, microservices, and situations where you can't install a client-side SDK.
• PI OLEDB Provider: This is an older database-like interface. You could connect to it from Python using libraries like pyodbc, but it's generally less convenient and powerful than the native Python SDK.