piafsdkconnect 0.1.3

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. Create New Tags (Batch)

Create multiple points on the PI Server efficiently in a single operation.

# Configuration for multiple tags
new_tags = [
    {"tag_name": "DEMO_TAG_1", "point_type": "Float32", "descriptor": "First Batch Tag"},
    {"tag_name": "DEMO_TAG_2", "point_type": "Int32", "engineeringunits": "bar"},
    {"tag_name": "DEMO_TAG_3"} # Uses default Float32
]

# Create tags and get status for each
results = service.create_tags(new_tags)
for tag, success in results.items():
    print(f"Tag {tag} creation: {'Success' if success else 'Failed'}")

# You can still create a single tag
service.create_tag("SINGLE_DEMO_TAG", point_type="String")

8. 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.