piafsdkconnect 0.1.1

A professional library for PI AF SDK integration

piafsdkconnect

A professional Python library for PI AF SDK integration (.NET based).

Installation

pip install piafsdkconnect

Usage

Using .env file

Create a .env file in your project root with the following variables:

PI_SERVER_NAME=YourPIServerName
PI_SDK_PATH=C:\Program Files (x86)\PIPC\AF\PublicAssemblies\4.0  # Optional, defaults to this path
PI_USERNAME=YourUsername  # Optional, for PI User/Windows Auth
PI_PASSWORD=YourPassword  # Optional

from piafsdkconnect import get_service
service = get_service()

Complete Example Usage

Below are examples of how to use each available method in the library.

Initialization

from piafsdkconnect import get_service

# Initialize service with demo credentials
service = get_service(
    server_name="PI_SERVER_DEMO",
    username="domain\\user",
    password="password123",
    sdk_path=r"C:\Program Files (x86)\PIPC\AF\PublicAssemblies\4.0"
)

1. Fetch Current Value

Get the latest snapshot value for a single PI tag.

val = service.get_current_value("SINUSOID")
print(f"Tag: {val['tag']}, Value: {val['value']}, Time: {val['timestamp']}")

2. Fetch Multiple Current Values

Efficiently retrieve values for multiple tags at once.

tags = ["SINUSOID", "CDT158", "BA:LEVEL.1"]
values = service.get_multiple_current_values(tags)
for v in values:
    print(v)

3. Fetch Historical Recorded Values

Get actual stored data points from the PI Archive for a specific time range.

history = service.get_recorded_values("SINUSOID", "*-24h", "*")
for point in history:
    print(f"Time: {point['timestamp']}, Value: {point['value']}")

4. Search for Tags

Find tags on the PI Server using wildcard patterns.

tags = service.search_tags("*water*")
print(f"Found tags: {tags}")

5. Live Data Stream

Yields a continuous stream of data for a list of tags at a specified frequency.

print("Starting live stream (3 iterations)...")
stream = service.get_live_data_stream(["SINUSOID", "CDT158"], frequency=2.5)
for i, data_batch in enumerate(stream):
    print(f"Batch {i+1}: {data_batch}")
    if i >= 2: break

6. Write New Values

Send data to a PI tag at the current time or a specific timestamp.

# Write current value
service.write_value("TEST_TAG", 45.6)

# Write with specific timestamp
service.write_value("TEST_TAG", 99.9, timestamp="2023-10-01 10:00:00")

7. Asset Server & Organization Lookup

Browse the PI AF hierarchy.

# List AF Asset Servers
servers = service.get_asset_servers()
print(f"Servers: {servers}")

# List Databases for a specific server
dbs = service.get_organizations("AF_SERVER_DEMO")
print(f"Databases: {dbs}")

8. Connection Utilities

# Test connection status
status = service.test_connection()
print(f"Connection Status: {status}")

# Disconnect cleanly
service.disconnect()

---

License

This project is licensed under the MIT License.