Metadata-Version: 2.4
Name: piwebapiconnect
Version: 0.1.8
Summary: A professional library for PI Web API integration
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.26.0
Requires-Dist: pydantic-settings>=2.1.0
Requires-Dist: python-dotenv>=1.0.0

# 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

```bash
pip install piwebapiconnect
```

## Setup & Connecting

### 1. Using a `.env` file (Recommended)

Create a `.env` file in your root folder:

```env
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
```

```python
from piwebapiconnect import get_service
service = get_service()
```

### 2. Passing credentials directly

```python
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

```python
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.

```python
# 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.

```python
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)

```python
# 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.

```python
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.

```python
# 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)

```python
# 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

```python
# 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.

```python
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.
