pi-web-sdk

Name	pi-web-sdk JSON
Version	0.1.17 JSON
	download
home_page	None
Summary	Python client SDK for the AVEVA PI Web API with typed controllers and request helpers
upload_time	2025-10-08 18:53:15
maintainer	None
docs_url	None
author	PI Web API SDK Maintainers
requires_python	>=3.9
license	Proprietary
keywords	osi pi web api sdk
VCS
bugtrack_url
requirements	No requirements were recorded.
Travis-CI	No Travis.
coveralls test coverage	No coveralls.

            # PI Web API Python SDK

A modular Python SDK for interacting with the OSIsoft PI Web API. The codebase has been reorganised from a single monolithic module into a structured package that groups related controllers, configuration primitives, and the HTTP client.

## Project Description
pi_web_sdk delivers a consistently structured Python interface for AVEVA PI Web API deployments. It wraps the REST endpoints with typed controllers, rich client helpers, and practical defaults so you can query PI data, manage assets, and orchestrate analytics without hand-crafting HTTP calls. The package is organised for extensibility: add new controllers or override behaviours while keeping a cohesive developer experience.

## Features
- **Typed configuration** via `PIWebAPIConfig` and enums for authentication and WebID formats
- **Reusable HTTP client** `PIWebAPIClient` wrapper around `requests.Session` with centralised error handling
- **Domain-organized controllers** split by functionality (system, assets, data, streams, OMF, etc.) for easier navigation
- **Stream Updates** for incremental data retrieval without websockets (marker-based polling)
- **OMF support** with ORM-style API for creating types, containers, assets, and hierarchies
- **Comprehensive CRUD operations** for all major PI Web API endpoints
- **Event Frame helpers** - High-level convenience methods for complex event frame operations
- **Parsed responses** - Type-safe response wrappers with generic data classes
- **Advanced search** - AFSearch syntax support for querying elements and attributes
- **Real-time streaming** - WebSocket/SSE channel support for live data updates
- **Bulk operations** - Get/update multiple resources in single API calls
- **108 tests** - Comprehensive test coverage for all controllers
- **Backwards-compatible** `aveva_web_api.py` re-export for existing imports

## Installation
This project depends on `requests`. Install it with:

```bash
pip install requests
```

## Quick Start

### Basic Usage
```python
from pi_web_sdk import AuthMethod, PIWebAPIClient, PIWebAPIConfig

config = PIWebAPIConfig(
    base_url="https://your-pi-server/piwebapi",
    auth_method=AuthMethod.ANONYMOUS,
    verify_ssl=False,  # enable in production
)

client = PIWebAPIClient(config)
print(client.home.get())
```

### Working with Assets
```python
from pi_web_sdk.models.responses import ItemsResponse
from pi_web_sdk.models.asset import Element, Attribute

# List asset servers (parsed response)
servers: ItemsResponse = client.asset_server.list()
server_web_id = servers.items[0].web_id

# Get databases
databases = client.asset_server.get_databases(server_web_id)
db_web_id = databases.items[0].web_id

# Get all elements from a database
elements: ItemsResponse[Element] = client.asset_database.get_elements(
    db_web_id,
    search_full_hierarchy=True,  # Include nested elements
    max_count=1000
)
for elem in elements.items:
    print(f"Element: {elem.name} at {elem.path}")

# Get elements with filters
pumps: ItemsResponse[Element] = client.asset_database.get_elements(
    db_web_id,
    name_filter="Pump*",
    template_name="Equipment"
)

# Get all attributes from an element
attributes: ItemsResponse[Attribute] = client.element.get_attributes(
    element_web_id,
    search_full_hierarchy=False,  # Only direct attributes
    max_count=100
)
for attr in attributes.items:
    print(f"Attribute: {attr.name} = {attr.value}")

# Get attributes with filters
temp_attributes = client.element.get_attributes(
    element_web_id,
    name_filter="Temp*",
    value_type="Float64"
)

# Create an element
element = Element(
    name="MyElement",
    description="Test element",
    template_name="MyTemplate"
)
new_element = client.asset_database.create_element(db_web_id, element)
element_web_id = new_element.web_id

# Create static attributes (values stored in AF)
from pi_web_sdk.models.attribute import Attribute, AttributeType

# String attribute
string_attr = Attribute(
    name="Description",
    type=AttributeType.STRING,
    value="Production Line A"
)
client.element.create_attribute(element_web_id, string_attr)

# Integer attribute
int_attr = Attribute(
    name="MaxCapacity",
    type=AttributeType.INT32,
    value=1000
)
client.element.create_attribute(element_web_id, int_attr)

# Float attribute
float_attr = Attribute(
    name="Efficiency",
    type=AttributeType.DOUBLE,
    value=95.5
)
client.element.create_attribute(element_web_id, float_attr)

# Create dynamic attributes (PI Point references)
# These store data in PI Data Archive

# Get PI Point web IDs first
point_web_id_temp = client.point.get_by_path(r"\\PI_SERVER\Temperature_Tag").web_id
point_web_id_pressure = client.point.get_by_path(r"\\PI_SERVER\Pressure_Tag").web_id

# Temperature attribute (Float, PI Point reference)
temp_attr = Attribute(
    name="Temperature",
    type=AttributeType.DOUBLE,
    data_reference_plug_in="PI Point",
    config_string=point_web_id_temp  # Reference to PI Point
)
client.element.create_attribute(element_web_id, temp_attr)

# Pressure attribute (Integer, PI Point reference)
pressure_attr = Attribute(
    name="Pressure",
    type=AttributeType.INT32,
    data_reference_plug_in="PI Point",
    config_string=point_web_id_pressure
)
client.element.create_attribute(element_web_id, pressure_attr)

# Status attribute (String, PI Point reference)
status_point_web_id = client.point.get_by_path(r"\\PI_SERVER\Status_Tag").web_id
status_attr = Attribute(
    name="Status",
    type=AttributeType.STRING,
    data_reference_plug_in="PI Point",
    config_string=status_point_web_id
)
client.element.create_attribute(element_web_id, status_attr)
```

### Attribute Helper Functions
```python
from typing import Union, Optional
from pi_web_sdk.models.attribute import Attribute, AttributeType

def add_static_attribute(
    client,
    element_web_id: str,
    name: str,
    value: Union[str, int, float],
    description: Optional[str] = None
) -> Attribute:
    """Add a static attribute to an element (value stored in AF)."""

    # Determine type from value
    if isinstance(value, str):
        attr_type = AttributeType.STRING
    elif isinstance(value, int):
        attr_type = AttributeType.INT32
    elif isinstance(value, float):
        attr_type = AttributeType.DOUBLE
    else:
        raise ValueError(f"Unsupported value type: {type(value)}")

    attribute = Attribute(
        name=name,
        type=attr_type,
        value=value,
        description=description
    )

    return client.element.create_attribute(element_web_id, attribute)

def add_dynamic_attribute(
    client,
    element_web_id: str,
    name: str,
    pi_point_path: str,
    value_type: Union[str, int, float] = float,
    description: Optional[str] = None
) -> Attribute:
    """Add a dynamic attribute to an element (PI Point reference)."""

    # Get PI Point web ID
    point = client.point.get_by_path(pi_point_path)
    point_web_id = point.web_id

    # Determine type from value_type parameter
    if value_type == str or value_type == "string":
        attr_type = AttributeType.STRING
    elif value_type == int or value_type == "int":
        attr_type = AttributeType.INT32
    elif value_type == float or value_type == "float":
        attr_type = AttributeType.DOUBLE
    else:
        raise ValueError(f"Unsupported value type: {value_type}")

    attribute = Attribute(
        name=name,
        type=attr_type,
        data_reference_plug_in="PI Point",
        config_string=point_web_id,
        description=description
    )

    return client.element.create_attribute(element_web_id, attribute)

# Usage examples
# Add static attributes
add_static_attribute(client, element_web_id, "Location", "Building 1")
add_static_attribute(client, element_web_id, "Capacity", 5000)
add_static_attribute(client, element_web_id, "Efficiency", 98.7)

# Add dynamic attributes with PI Point references
add_dynamic_attribute(
    client,
    element_web_id,
    "Temperature",
    r"\\PI_SERVER\Temperature_Tag",
    value_type=float,
    description="Process temperature in Celsius"
)

add_dynamic_attribute(
    client,
    element_web_id,
    "Pressure",
    r"\\PI_SERVER\Pressure_Tag",
    value_type=int,
    description="Process pressure in PSI"
)

add_dynamic_attribute(
    client,
    element_web_id,
    "Status",
    r"\\PI_SERVER\Status_Tag",
    value_type=str,
    description="Equipment status"
)
```

### Working with Streams
```python
from pi_web_sdk.models.stream import StreamValue, TimedValue
from datetime import datetime

# Get stream value
value: StreamValue = client.stream.get_value(stream_web_id)
print(f"Current value: {value.value} at {value.timestamp}")

# Get recorded data
recorded = client.stream.get_recorded(
    web_id=stream_web_id,
    start_time="*-7d",
    end_time="*",
    max_count=1000
)

# Get latest value
latest: StreamValue = client.stream.get_end(stream_web_id)

# Get value at specific time
value_at_time = client.stream.get_recorded_at_time(
    stream_web_id,
    "2024-01-01T12:00:00Z",
    retrieval_mode="AtOrBefore"
)

# Get values at multiple times
values = client.stream.get_recorded_at_times(
    stream_web_id,
    ["2024-01-01T00:00:00Z", "2024-01-01T12:00:00Z", "2024-01-02T00:00:00Z"]
)

# Get interpolated values
interpolated = client.stream.get_interpolated_at_times(
    stream_web_id,
    ["2024-01-01T06:00:00Z", "2024-01-01T18:00:00Z"]
)

# Open real-time streaming channel
channel = client.stream.get_channel(
    stream_web_id,
    include_initial_values=True,
    heartbeat_rate=30
)

# Update stream value
new_value = TimedValue(
    timestamp=datetime(2024, 1, 1, 0, 0, 0),
    value=42.5
)
client.stream.update_value(stream_web_id, new_value)

# Bulk operations for multiple streams
latest_values = client.streamset.get_end([stream_id1, stream_id2, stream_id3])
```

### Stream Updates (Incremental Data Retrieval)
```python
import time
from pi_web_sdk.models.stream import StreamUpdateRegistration, StreamUpdates

# Register for stream updates
registration: StreamUpdateRegistration = client.stream.register_update(stream_web_id)
marker = registration.latest_marker

# Poll for incremental updates
while True:
    time.sleep(5)  # Wait between polls

    # Retrieve only new data since last marker
    updates: StreamUpdates = client.stream.retrieve_update(marker)

    for item in updates.items:
        print(f"{item.timestamp}: {item.value}")

    # Update marker for next poll
    marker = updates.latest_marker

# For multiple streams, use streamset
registration = client.streamset.register_updates([stream_id1, stream_id2, stream_id3])
marker = registration.latest_marker

updates = client.streamset.retrieve_updates(marker)
for stream_update in updates.items:
    stream_id = stream_update.web_id
    for item in stream_update.items:
        print(f"Stream {stream_id}: {item.timestamp} = {item.value}")
```

See [examples/README_STREAM_UPDATES.md](examples/README_STREAM_UPDATES.md) for comprehensive Stream Updates documentation.

### OMF (OSIsoft Message Format) Support
```python
from pi_web_sdk.omf import OMFManager
from pi_web_sdk.models.omf import OMFType, OMFProperty, OMFContainer, OMFData
from datetime import datetime

# Initialize OMF manager
omf_manager = OMFManager(client, data_server_web_id)

# Create a type definition
sensor_type = OMFType(
    id="TempSensorType",
    classification="dynamic",
    type="object",
    properties=[
        OMFProperty("timestamp", "string", is_index=True, format="date-time"),
        OMFProperty("temperature", "number", name="Temperature")
    ]
)
omf_manager.create_type(sensor_type)

# Create a container
container = OMFContainer(
    id="sensor1",
    type_id="TempSensorType"
)
omf_manager.create_container(container)

# Send data
data_point = OMFData(
    container_id="sensor1",
    values=[{
        "timestamp": datetime(2024, 1, 1, 0, 0, 0).isoformat() + "Z",
        "temperature": 25.5
    }]
)
omf_manager.send_data(data_point)
```

### OMF Hierarchies
```python
from pi_web_sdk.models.omf import OMFHierarchy, OMFHierarchyNode

# Create hierarchy from paths
hierarchy = OMFHierarchy()
hierarchy.add_path("Plant/Area1/Line1")
hierarchy.add_path("Plant/Area1/Line2")
hierarchy.add_path("Plant/Area2/Line3")

# Or create nodes explicitly
root = OMFHierarchyNode("Plant", "Root")
area1 = OMFHierarchyNode("Area1", "Area", parent=root)
line1 = OMFHierarchyNode("Line1", "Line", parent=area1)

# Deploy hierarchy
omf_manager.create_hierarchy(hierarchy)
```

### Event Frame Helpers
```python
from pi_web_sdk.models.event import EventFrame, EventFrameAttribute
from datetime import datetime

# Create event frame with attributes in one operation
event = EventFrame(
    name="Batch Run 001",
    description="Production batch",
    start_time=datetime(2024, 1, 1, 8, 0, 0),
    end_time=datetime(2024, 1, 1, 16, 0, 0),
    attributes=[
        EventFrameAttribute(name="Temperature", value=95.5),
        EventFrameAttribute(name="Pressure", value=1013.25),
        EventFrameAttribute(name="Status", value="Complete")
    ]
)
created_event = client.event_frame_helpers.create_event_frame_with_attributes(
    db_web_id, event
)

# Create child event frame
child = EventFrame(
    name="Quality Check",
    description="QC inspection",
    start_time=datetime(2024, 1, 1, 15, 30, 0),
    end_time=datetime(2024, 1, 1, 15, 45, 0),
    attributes=[
        EventFrameAttribute(name="Result", value="Pass"),
        EventFrameAttribute(name="Inspector", value="John Doe")
    ]
)
client.event_frame_helpers.create_child_event_frame_with_attributes(
    created_event.web_id, child
)

# Create complete hierarchy
from pi_web_sdk.models.event import EventFrameHierarchy, ChildEventFrame

hierarchy = EventFrameHierarchy(
    root=EventFrame(
        name="Production Run",
        description="Full production cycle",
        start_time=datetime(2024, 1, 1, 8, 0, 0),
        end_time=datetime(2024, 1, 1, 18, 0, 0),
        attributes=[EventFrameAttribute(name="Batch", value="B-2024-001")]
    ),
    children=[
        ChildEventFrame(
            name="Mixing",
            start_time=datetime(2024, 1, 1, 8, 0, 0),
            end_time=datetime(2024, 1, 1, 10, 0, 0),
            attributes=[EventFrameAttribute(name="Speed", value=1200)]
        ),
        ChildEventFrame(
            name="Heating",
            start_time=datetime(2024, 1, 1, 10, 0, 0),
            end_time=datetime(2024, 1, 1, 14, 0, 0),
            attributes=[EventFrameAttribute(name="Target", value=95.0)]
        )
    ]
)
client.event_frame_helpers.create_event_frame_hierarchy(db_web_id, hierarchy)

# Get event frame with all attribute values
event_data = client.event_frame_helpers.get_event_frame_with_attributes(
    event_web_id,
    include_values=True
)
```

See [docs/event_frame_helpers.md](docs/event_frame_helpers.md) for complete documentation.

### Parsed Responses (Type-Safe)
```python
from pi_web_sdk.models.data import DataServer, Point
from pi_web_sdk.models.responses import ItemsResponse

# Get parsed response with type safety
server: DataServer = client.data_server.get_parsed(web_id)
print(f"Server: {server.name}")
print(f"Version: {server.server_version}")
print(f"Connected: {server.is_connected}")

# List with type safety and iteration
servers: ItemsResponse[DataServer] = client.data_server.list_parsed()
for server in servers:
    print(f"{server.name}: {server.path}")

# Get points with type safety
points: ItemsResponse[Point] = client.data_server.get_points_parsed(server_web_id)
for point in points:
    print(f"{point.name} ({point.point_type}): {point.engineering_units}")
```

### Advanced Search (AFSearch Syntax)
```python
from pi_web_sdk.models.asset import Element, AttributeSearch
from pi_web_sdk.models.responses import ItemsResponse

# Query elements by attributes
elements: ItemsResponse[Element] = client.element.get_elements_query(
    database_web_id,
    query="Name:='Pump*' Type:='Equipment'"
)

# Create persistent attribute search
search = AttributeSearch(
    database_web_id=database_web_id,
    query="Name:='Temperature' Type:='Float64'"
)
search_result = client.element.create_search_by_attribute(search)
search_id = search_result.web_id

# Execute search later
results: ItemsResponse[Element] = client.element.execute_search_by_attribute(search_id)

# Bulk get multiple elements
elements: ItemsResponse[Element] = client.element.get_multiple(
    [web_id1, web_id2, web_id3],
    selected_fields="Name;Path;Description"
)
```

### Analysis Operations
```python
from pi_web_sdk.models.analysis import Analysis, AnalysisTemplate, SecurityEntry

# Get analysis with security
analysis: Analysis = client.analysis.get(analysis_web_id)

# Get security entries
entries: ItemsResponse[SecurityEntry] = client.analysis.get_security_entries(analysis_web_id)

# Create security entry
entry = SecurityEntry(
    name="Operators",
    security_identity_web_id=identity_web_id,
    allow_rights=["Read", "Execute"]
)
client.analysis.create_security_entry(analysis_web_id, entry)

# Work with analysis templates
template: AnalysisTemplate = client.analysis_template.get_by_path(
    "\\\\AnalysisTemplate\\MyTemplate"
)
template.description = "Updated"
client.analysis_template.update(template.web_id, template)

# Get analysis categories
categories = client.analysis.get_categories(analysis_web_id)
```

## Available Controllers
All controller instances are available as attributes on `PIWebAPIClient`:

### System & Configuration
- `client.home` - Home endpoint
- `client.system` - System information and status
- `client.configuration` - System configuration

### Asset Model
- `client.asset_server` - Asset servers
- `client.asset_database` - Asset databases
- `client.element` - Elements
- `client.element_category` - Element categories
- `client.element_template` - Element templates
- `client.attribute` - Attributes
- `client.attribute_category` - Attribute categories
- `client.attribute_template` - Attribute templates

### Data & Streams
- `client.data_server` - Data servers
- `client.point` - PI Points
- `client.stream` - Stream data operations (including Stream Updates)
- `client.streamset` - Batch stream operations (including Stream Set Updates)

### Analysis & Events
- `client.analysis` - PI Analyses
- `client.analysis_category` - Analysis categories
- `client.analysis_rule` - Analysis rules
- `client.analysis_rule_plugin` - Analysis rule plugins
- `client.analysis_template` - Analysis templates
- `client.event_frame` - Event frames
- `client.event_frame_helpers` - High-level event frame operations
- `client.table` - PI Tables
- `client.table_category` - Table categories

### OMF
- `client.omf` - OSIsoft Message Format endpoint

### Batch & Advanced
- `client.batch` - Batch operations
- `client.calculation` - Calculations
- `client.channel` - Channels

### Supporting Resources
- `client.enumeration_set` - Enumeration sets
- `client.enumeration_value` - Enumeration values
- `client.unit` - Units of measure
- `client.time_rule` - Time rules
- `client.security` - Security operations
- `client.notification` - Notification rules
- `client.metrics` - System metrics

## Package Layout
- `pi_web_sdk/config.py` - Enums and configuration dataclass
- `pi_web_sdk/exceptions.py` - Custom exception types
- `pi_web_sdk/client.py` - Session management and HTTP helpers
- `pi_web_sdk/controllers/` - Individual controller modules grouped by domain
  - `controllers/base.py` - Base controller with shared utilities
  - `controllers/system.py` - System and configuration controllers
  - `controllers/asset.py` - Asset servers, databases, elements, templates
  - `controllers/attribute.py` - Attributes, categories, templates
  - `controllers/data.py` - Data servers and points (with parsed methods)
  - `controllers/stream.py` - Stream and streamset operations (enhanced)
  - `controllers/analysis.py` - Analysis controllers (fully enhanced)
  - `controllers/event.py` - Event frame controller and high-level helpers
  - `controllers/omf.py` - OMF controller and manager
  - Additional controllers for tables, enumerations, units, security, notifications, etc.
- `pi_web_sdk/models/` - Data models and response classes
  - `models/responses.py` - Generic ItemsResponse[T] and PIResponse[T]
  - `models/data.py` - DataServer and Point models
  - `models/stream.py` - Stream enums (BufferOption, UpdateOption)
  - `models/omf.py` - OMF data models
  - `models/attribute.py` - Attribute models
- `pi_web_sdk/omf/` - OMF support with ORM-style API
  - `omf/orm.py` - Core OMF classes (Type, Container, Asset, Data)
  - `omf/hierarchy.py` - Hierarchy builder utilities
- `docs/` - Comprehensive documentation
- `examples/` - Working code examples
- `tests/` - 108 tests covering all controllers
- `aveva_web_api.py` - Compatibility shim for existing imports

## Extending the SDK
Each controller inherits from `BaseController`, which exposes helper methods and the configured client session. Add new endpoint support by:

1. Create a new controller module under `pi_web_sdk/controllers/`
2. Define data models in `pi_web_sdk/models/`
3. Register controller in `pi_web_sdk/controllers/__init__.py`
4. Add it to `pi_web_sdk/client.py` in the `PIWebAPIClient.__init__` method

Example:
```python
from __future__ import annotations
from typing import Optional
from dataclasses import dataclass
from .base import BaseController
from ..models.responses import PIResponse

@dataclass
class MyResource:
    web_id: str
    name: str
    description: Optional[str] = None

class MyController(BaseController):
    def get(self, web_id: str) -> PIResponse[MyResource]:
        response = self.client.get(f"myresource/{web_id}")
        return PIResponse(MyResource(**response))
```

## Testing
Run the test suite:

```bash
# Run all tests
pytest tests/

# Run specific test files
pytest tests/test_omf_endpoint.py -v

# Run with integration marker
pytest -m integration
```

## Deployment

### Quick Deployment

```bash
# Validate package
python deploy.py --check

# Deploy to TestPyPI (recommended first)
python deploy.py --test

# Deploy to PyPI
python deploy.py --prod
```

### Prerequisites
- PyPI and TestPyPI accounts
- API tokens configured in `~/.pypirc`
- Build tools: `pip install build twine`

See [DEPLOYMENT_QUICKSTART.md](DEPLOYMENT_QUICKSTART.md) for quick start guide or [DEPLOYMENT.md](DEPLOYMENT.md) for comprehensive instructions.

## Documentation & Examples
- [PI Web API Reference](https://docs.aveva.com/bundle/pi-web-api-reference/page/help/getting-started.html)
- [OMF Documentation](https://docs.aveva.com/)
- [Stream Updates Guide](examples/README_STREAM_UPDATES.md) - Comprehensive guide for incremental data retrieval
- [Stream Updates Examples](examples/stream_updates_example.py) - Working code examples
- [Event Frame Helpers Documentation](docs/event_frame_helpers.md) - High-level event frame operations
- [Parsed Responses Documentation](docs/parsed_responses.md) - Type-safe response wrappers
- [Controller Additions Summary](docs/CONTROLLER_ADDITIONS_SUMMARY.md) - Complete list of all enhancements
- See `examples/` directory for more usage examples

## Recent Additions

### Comprehensive Controller Enhancements (v2025.10)
Major enhancement to the SDK with 78 new methods and 108 tests covering the full PI Web API surface:

**Analysis Controllers** (46 methods)
- Full CRUD operations for analyses, templates, categories, rules, and plugins
- Security management (get, create, update, delete security entries)
- Category associations
- Analysis rule management

**Element Enhancements** (7 methods)
- `get_multiple()` - Bulk retrieval of elements
- `get_elements_query()` - AFSearch syntax support for powerful queries
- `create_search_by_attribute()` / `execute_search_by_attribute()` - Persistent searches
- `add_referenced_element()` / `remove_referenced_element()` - Reference management
- `get_notification_rules()` / `create_notification_rule()` - Notification support

**Stream Enhancements** (11 methods)
- `get_end()` - Latest recorded value
- `get_recorded_at_time()` / `get_recorded_at_times()` - Point-in-time retrieval
- `get_interpolated_at_times()` - Interpolated values at specific times
- `get_channel()` - Real-time WebSocket/SSE streaming channels
- All methods available for single streams (`StreamController`) and multiple streams (`StreamSetController`)

**Event Frame Helpers** (6 methods)
- `create_event_frame_with_attributes()` - Create event frame and attributes in one call
- `create_child_event_frame_with_attributes()` - Create child with attributes
- `create_event_frame_hierarchy()` - Build complete hierarchies
- `get_event_frame_with_attributes()` - Retrieve with all attribute values
- `update_event_frame_attributes()` - Bulk attribute updates
- `close_event_frame()` - Close with optional value capture

**Parsed Responses** (Type-safe data classes)
- Generic `ItemsResponse[T]` and `PIResponse[T]` wrappers
- Support for iteration, indexing, and len()
- Added `*_parsed()` methods to DataServer and Point controllers
- Automatic deserialization to typed objects

See [docs/CONTROLLER_ADDITIONS_SUMMARY.md](docs/CONTROLLER_ADDITIONS_SUMMARY.md) for complete details.

### Stream Updates (v2025.01)
Stream Updates provides an efficient way to retrieve incremental data updates without websockets. Key features:
- **Marker-based tracking** - Maintains position in data stream
- **Single or multiple streams** - Support for individual streams and stream sets
- **Metadata change detection** - Notifies when data is invalidated
- **Unit conversion** - Convert values during retrieval
- **Selected fields** - Filter response data
- **Type-safe data classes** - Strongly typed response models

```python
from pi_web_sdk.models.stream import StreamUpdateRegistration, StreamUpdates

# Register once
registration: StreamUpdateRegistration = client.stream.register_update(stream_web_id)
marker = registration.latest_marker

# Poll repeatedly for new data only
while True:
    time.sleep(5)
    updates: StreamUpdates = client.stream.retrieve_update(marker)
    # Process updates.items with type safety
    for item in updates.items:
        print(f"{item.timestamp}: {item.value}")
    marker = updates.latest_marker
```

**Requirements**: PI Web API 2019+ with Stream Updates feature enabled

See [examples/README_STREAM_UPDATES.md](examples/README_STREAM_UPDATES.md) for complete documentation.

## License
See LICENSE file for details.


            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "pi-web-sdk",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.9",
    "maintainer_email": null,
    "keywords": "osi, pi, web, api, sdk",
    "author": "PI Web API SDK Maintainers",
    "author_email": null,
    "download_url": "https://files.pythonhosted.org/packages/94/1a/7dbdfd2a8a55f4f90ceb1599c8ba1cc89ebf89da6f0e95d9976f394e1f5e/pi_web_sdk-0.1.17.tar.gz",
    "platform": null,
    "description": "# PI Web API Python SDK\r\n\r\nA modular Python SDK for interacting with the OSIsoft PI Web API. The codebase has been reorganised from a single monolithic module into a structured package that groups related controllers, configuration primitives, and the HTTP client.\r\n\r\n## Project Description\r\npi_web_sdk delivers a consistently structured Python interface for AVEVA PI Web API deployments. It wraps the REST endpoints with typed controllers, rich client helpers, and practical defaults so you can query PI data, manage assets, and orchestrate analytics without hand-crafting HTTP calls. The package is organised for extensibility: add new controllers or override behaviours while keeping a cohesive developer experience.\r\n\r\n## Features\r\n- **Typed configuration** via `PIWebAPIConfig` and enums for authentication and WebID formats\r\n- **Reusable HTTP client** `PIWebAPIClient` wrapper around `requests.Session` with centralised error handling\r\n- **Domain-organized controllers** split by functionality (system, assets, data, streams, OMF, etc.) for easier navigation\r\n- **Stream Updates** for incremental data retrieval without websockets (marker-based polling)\r\n- **OMF support** with ORM-style API for creating types, containers, assets, and hierarchies\r\n- **Comprehensive CRUD operations** for all major PI Web API endpoints\r\n- **Event Frame helpers** - High-level convenience methods for complex event frame operations\r\n- **Parsed responses** - Type-safe response wrappers with generic data classes\r\n- **Advanced search** - AFSearch syntax support for querying elements and attributes\r\n- **Real-time streaming** - WebSocket/SSE channel support for live data updates\r\n- **Bulk operations** - Get/update multiple resources in single API calls\r\n- **108 tests** - Comprehensive test coverage for all controllers\r\n- **Backwards-compatible** `aveva_web_api.py` re-export for existing imports\r\n\r\n## Installation\r\nThis project depends on `requests`. Install it with:\r\n\r\n```bash\r\npip install requests\r\n```\r\n\r\n## Quick Start\r\n\r\n### Basic Usage\r\n```python\r\nfrom pi_web_sdk import AuthMethod, PIWebAPIClient, PIWebAPIConfig\r\n\r\nconfig = PIWebAPIConfig(\r\n    base_url=\"https://your-pi-server/piwebapi\",\r\n    auth_method=AuthMethod.ANONYMOUS,\r\n    verify_ssl=False,  # enable in production\r\n)\r\n\r\nclient = PIWebAPIClient(config)\r\nprint(client.home.get())\r\n```\r\n\r\n### Working with Assets\r\n```python\r\nfrom pi_web_sdk.models.responses import ItemsResponse\r\nfrom pi_web_sdk.models.asset import Element, Attribute\r\n\r\n# List asset servers (parsed response)\r\nservers: ItemsResponse = client.asset_server.list()\r\nserver_web_id = servers.items[0].web_id\r\n\r\n# Get databases\r\ndatabases = client.asset_server.get_databases(server_web_id)\r\ndb_web_id = databases.items[0].web_id\r\n\r\n# Get all elements from a database\r\nelements: ItemsResponse[Element] = client.asset_database.get_elements(\r\n    db_web_id,\r\n    search_full_hierarchy=True,  # Include nested elements\r\n    max_count=1000\r\n)\r\nfor elem in elements.items:\r\n    print(f\"Element: {elem.name} at {elem.path}\")\r\n\r\n# Get elements with filters\r\npumps: ItemsResponse[Element] = client.asset_database.get_elements(\r\n    db_web_id,\r\n    name_filter=\"Pump*\",\r\n    template_name=\"Equipment\"\r\n)\r\n\r\n# Get all attributes from an element\r\nattributes: ItemsResponse[Attribute] = client.element.get_attributes(\r\n    element_web_id,\r\n    search_full_hierarchy=False,  # Only direct attributes\r\n    max_count=100\r\n)\r\nfor attr in attributes.items:\r\n    print(f\"Attribute: {attr.name} = {attr.value}\")\r\n\r\n# Get attributes with filters\r\ntemp_attributes = client.element.get_attributes(\r\n    element_web_id,\r\n    name_filter=\"Temp*\",\r\n    value_type=\"Float64\"\r\n)\r\n\r\n# Create an element\r\nelement = Element(\r\n    name=\"MyElement\",\r\n    description=\"Test element\",\r\n    template_name=\"MyTemplate\"\r\n)\r\nnew_element = client.asset_database.create_element(db_web_id, element)\r\nelement_web_id = new_element.web_id\r\n\r\n# Create static attributes (values stored in AF)\r\nfrom pi_web_sdk.models.attribute import Attribute, AttributeType\r\n\r\n# String attribute\r\nstring_attr = Attribute(\r\n    name=\"Description\",\r\n    type=AttributeType.STRING,\r\n    value=\"Production Line A\"\r\n)\r\nclient.element.create_attribute(element_web_id, string_attr)\r\n\r\n# Integer attribute\r\nint_attr = Attribute(\r\n    name=\"MaxCapacity\",\r\n    type=AttributeType.INT32,\r\n    value=1000\r\n)\r\nclient.element.create_attribute(element_web_id, int_attr)\r\n\r\n# Float attribute\r\nfloat_attr = Attribute(\r\n    name=\"Efficiency\",\r\n    type=AttributeType.DOUBLE,\r\n    value=95.5\r\n)\r\nclient.element.create_attribute(element_web_id, float_attr)\r\n\r\n# Create dynamic attributes (PI Point references)\r\n# These store data in PI Data Archive\r\n\r\n# Get PI Point web IDs first\r\npoint_web_id_temp = client.point.get_by_path(r\"\\\\PI_SERVER\\Temperature_Tag\").web_id\r\npoint_web_id_pressure = client.point.get_by_path(r\"\\\\PI_SERVER\\Pressure_Tag\").web_id\r\n\r\n# Temperature attribute (Float, PI Point reference)\r\ntemp_attr = Attribute(\r\n    name=\"Temperature\",\r\n    type=AttributeType.DOUBLE,\r\n    data_reference_plug_in=\"PI Point\",\r\n    config_string=point_web_id_temp  # Reference to PI Point\r\n)\r\nclient.element.create_attribute(element_web_id, temp_attr)\r\n\r\n# Pressure attribute (Integer, PI Point reference)\r\npressure_attr = Attribute(\r\n    name=\"Pressure\",\r\n    type=AttributeType.INT32,\r\n    data_reference_plug_in=\"PI Point\",\r\n    config_string=point_web_id_pressure\r\n)\r\nclient.element.create_attribute(element_web_id, pressure_attr)\r\n\r\n# Status attribute (String, PI Point reference)\r\nstatus_point_web_id = client.point.get_by_path(r\"\\\\PI_SERVER\\Status_Tag\").web_id\r\nstatus_attr = Attribute(\r\n    name=\"Status\",\r\n    type=AttributeType.STRING,\r\n    data_reference_plug_in=\"PI Point\",\r\n    config_string=status_point_web_id\r\n)\r\nclient.element.create_attribute(element_web_id, status_attr)\r\n```\r\n\r\n### Attribute Helper Functions\r\n```python\r\nfrom typing import Union, Optional\r\nfrom pi_web_sdk.models.attribute import Attribute, AttributeType\r\n\r\ndef add_static_attribute(\r\n    client,\r\n    element_web_id: str,\r\n    name: str,\r\n    value: Union[str, int, float],\r\n    description: Optional[str] = None\r\n) -> Attribute:\r\n    \"\"\"Add a static attribute to an element (value stored in AF).\"\"\"\r\n\r\n    # Determine type from value\r\n    if isinstance(value, str):\r\n        attr_type = AttributeType.STRING\r\n    elif isinstance(value, int):\r\n        attr_type = AttributeType.INT32\r\n    elif isinstance(value, float):\r\n        attr_type = AttributeType.DOUBLE\r\n    else:\r\n        raise ValueError(f\"Unsupported value type: {type(value)}\")\r\n\r\n    attribute = Attribute(\r\n        name=name,\r\n        type=attr_type,\r\n        value=value,\r\n        description=description\r\n    )\r\n\r\n    return client.element.create_attribute(element_web_id, attribute)\r\n\r\ndef add_dynamic_attribute(\r\n    client,\r\n    element_web_id: str,\r\n    name: str,\r\n    pi_point_path: str,\r\n    value_type: Union[str, int, float] = float,\r\n    description: Optional[str] = None\r\n) -> Attribute:\r\n    \"\"\"Add a dynamic attribute to an element (PI Point reference).\"\"\"\r\n\r\n    # Get PI Point web ID\r\n    point = client.point.get_by_path(pi_point_path)\r\n    point_web_id = point.web_id\r\n\r\n    # Determine type from value_type parameter\r\n    if value_type == str or value_type == \"string\":\r\n        attr_type = AttributeType.STRING\r\n    elif value_type == int or value_type == \"int\":\r\n        attr_type = AttributeType.INT32\r\n    elif value_type == float or value_type == \"float\":\r\n        attr_type = AttributeType.DOUBLE\r\n    else:\r\n        raise ValueError(f\"Unsupported value type: {value_type}\")\r\n\r\n    attribute = Attribute(\r\n        name=name,\r\n        type=attr_type,\r\n        data_reference_plug_in=\"PI Point\",\r\n        config_string=point_web_id,\r\n        description=description\r\n    )\r\n\r\n    return client.element.create_attribute(element_web_id, attribute)\r\n\r\n# Usage examples\r\n# Add static attributes\r\nadd_static_attribute(client, element_web_id, \"Location\", \"Building 1\")\r\nadd_static_attribute(client, element_web_id, \"Capacity\", 5000)\r\nadd_static_attribute(client, element_web_id, \"Efficiency\", 98.7)\r\n\r\n# Add dynamic attributes with PI Point references\r\nadd_dynamic_attribute(\r\n    client,\r\n    element_web_id,\r\n    \"Temperature\",\r\n    r\"\\\\PI_SERVER\\Temperature_Tag\",\r\n    value_type=float,\r\n    description=\"Process temperature in Celsius\"\r\n)\r\n\r\nadd_dynamic_attribute(\r\n    client,\r\n    element_web_id,\r\n    \"Pressure\",\r\n    r\"\\\\PI_SERVER\\Pressure_Tag\",\r\n    value_type=int,\r\n    description=\"Process pressure in PSI\"\r\n)\r\n\r\nadd_dynamic_attribute(\r\n    client,\r\n    element_web_id,\r\n    \"Status\",\r\n    r\"\\\\PI_SERVER\\Status_Tag\",\r\n    value_type=str,\r\n    description=\"Equipment status\"\r\n)\r\n```\r\n\r\n### Working with Streams\r\n```python\r\nfrom pi_web_sdk.models.stream import StreamValue, TimedValue\r\nfrom datetime import datetime\r\n\r\n# Get stream value\r\nvalue: StreamValue = client.stream.get_value(stream_web_id)\r\nprint(f\"Current value: {value.value} at {value.timestamp}\")\r\n\r\n# Get recorded data\r\nrecorded = client.stream.get_recorded(\r\n    web_id=stream_web_id,\r\n    start_time=\"*-7d\",\r\n    end_time=\"*\",\r\n    max_count=1000\r\n)\r\n\r\n# Get latest value\r\nlatest: StreamValue = client.stream.get_end(stream_web_id)\r\n\r\n# Get value at specific time\r\nvalue_at_time = client.stream.get_recorded_at_time(\r\n    stream_web_id,\r\n    \"2024-01-01T12:00:00Z\",\r\n    retrieval_mode=\"AtOrBefore\"\r\n)\r\n\r\n# Get values at multiple times\r\nvalues = client.stream.get_recorded_at_times(\r\n    stream_web_id,\r\n    [\"2024-01-01T00:00:00Z\", \"2024-01-01T12:00:00Z\", \"2024-01-02T00:00:00Z\"]\r\n)\r\n\r\n# Get interpolated values\r\ninterpolated = client.stream.get_interpolated_at_times(\r\n    stream_web_id,\r\n    [\"2024-01-01T06:00:00Z\", \"2024-01-01T18:00:00Z\"]\r\n)\r\n\r\n# Open real-time streaming channel\r\nchannel = client.stream.get_channel(\r\n    stream_web_id,\r\n    include_initial_values=True,\r\n    heartbeat_rate=30\r\n)\r\n\r\n# Update stream value\r\nnew_value = TimedValue(\r\n    timestamp=datetime(2024, 1, 1, 0, 0, 0),\r\n    value=42.5\r\n)\r\nclient.stream.update_value(stream_web_id, new_value)\r\n\r\n# Bulk operations for multiple streams\r\nlatest_values = client.streamset.get_end([stream_id1, stream_id2, stream_id3])\r\n```\r\n\r\n### Stream Updates (Incremental Data Retrieval)\r\n```python\r\nimport time\r\nfrom pi_web_sdk.models.stream import StreamUpdateRegistration, StreamUpdates\r\n\r\n# Register for stream updates\r\nregistration: StreamUpdateRegistration = client.stream.register_update(stream_web_id)\r\nmarker = registration.latest_marker\r\n\r\n# Poll for incremental updates\r\nwhile True:\r\n    time.sleep(5)  # Wait between polls\r\n\r\n    # Retrieve only new data since last marker\r\n    updates: StreamUpdates = client.stream.retrieve_update(marker)\r\n\r\n    for item in updates.items:\r\n        print(f\"{item.timestamp}: {item.value}\")\r\n\r\n    # Update marker for next poll\r\n    marker = updates.latest_marker\r\n\r\n# For multiple streams, use streamset\r\nregistration = client.streamset.register_updates([stream_id1, stream_id2, stream_id3])\r\nmarker = registration.latest_marker\r\n\r\nupdates = client.streamset.retrieve_updates(marker)\r\nfor stream_update in updates.items:\r\n    stream_id = stream_update.web_id\r\n    for item in stream_update.items:\r\n        print(f\"Stream {stream_id}: {item.timestamp} = {item.value}\")\r\n```\r\n\r\nSee [examples/README_STREAM_UPDATES.md](examples/README_STREAM_UPDATES.md) for comprehensive Stream Updates documentation.\r\n\r\n### OMF (OSIsoft Message Format) Support\r\n```python\r\nfrom pi_web_sdk.omf import OMFManager\r\nfrom pi_web_sdk.models.omf import OMFType, OMFProperty, OMFContainer, OMFData\r\nfrom datetime import datetime\r\n\r\n# Initialize OMF manager\r\nomf_manager = OMFManager(client, data_server_web_id)\r\n\r\n# Create a type definition\r\nsensor_type = OMFType(\r\n    id=\"TempSensorType\",\r\n    classification=\"dynamic\",\r\n    type=\"object\",\r\n    properties=[\r\n        OMFProperty(\"timestamp\", \"string\", is_index=True, format=\"date-time\"),\r\n        OMFProperty(\"temperature\", \"number\", name=\"Temperature\")\r\n    ]\r\n)\r\nomf_manager.create_type(sensor_type)\r\n\r\n# Create a container\r\ncontainer = OMFContainer(\r\n    id=\"sensor1\",\r\n    type_id=\"TempSensorType\"\r\n)\r\nomf_manager.create_container(container)\r\n\r\n# Send data\r\ndata_point = OMFData(\r\n    container_id=\"sensor1\",\r\n    values=[{\r\n        \"timestamp\": datetime(2024, 1, 1, 0, 0, 0).isoformat() + \"Z\",\r\n        \"temperature\": 25.5\r\n    }]\r\n)\r\nomf_manager.send_data(data_point)\r\n```\r\n\r\n### OMF Hierarchies\r\n```python\r\nfrom pi_web_sdk.models.omf import OMFHierarchy, OMFHierarchyNode\r\n\r\n# Create hierarchy from paths\r\nhierarchy = OMFHierarchy()\r\nhierarchy.add_path(\"Plant/Area1/Line1\")\r\nhierarchy.add_path(\"Plant/Area1/Line2\")\r\nhierarchy.add_path(\"Plant/Area2/Line3\")\r\n\r\n# Or create nodes explicitly\r\nroot = OMFHierarchyNode(\"Plant\", \"Root\")\r\narea1 = OMFHierarchyNode(\"Area1\", \"Area\", parent=root)\r\nline1 = OMFHierarchyNode(\"Line1\", \"Line\", parent=area1)\r\n\r\n# Deploy hierarchy\r\nomf_manager.create_hierarchy(hierarchy)\r\n```\r\n\r\n### Event Frame Helpers\r\n```python\r\nfrom pi_web_sdk.models.event import EventFrame, EventFrameAttribute\r\nfrom datetime import datetime\r\n\r\n# Create event frame with attributes in one operation\r\nevent = EventFrame(\r\n    name=\"Batch Run 001\",\r\n    description=\"Production batch\",\r\n    start_time=datetime(2024, 1, 1, 8, 0, 0),\r\n    end_time=datetime(2024, 1, 1, 16, 0, 0),\r\n    attributes=[\r\n        EventFrameAttribute(name=\"Temperature\", value=95.5),\r\n        EventFrameAttribute(name=\"Pressure\", value=1013.25),\r\n        EventFrameAttribute(name=\"Status\", value=\"Complete\")\r\n    ]\r\n)\r\ncreated_event = client.event_frame_helpers.create_event_frame_with_attributes(\r\n    db_web_id, event\r\n)\r\n\r\n# Create child event frame\r\nchild = EventFrame(\r\n    name=\"Quality Check\",\r\n    description=\"QC inspection\",\r\n    start_time=datetime(2024, 1, 1, 15, 30, 0),\r\n    end_time=datetime(2024, 1, 1, 15, 45, 0),\r\n    attributes=[\r\n        EventFrameAttribute(name=\"Result\", value=\"Pass\"),\r\n        EventFrameAttribute(name=\"Inspector\", value=\"John Doe\")\r\n    ]\r\n)\r\nclient.event_frame_helpers.create_child_event_frame_with_attributes(\r\n    created_event.web_id, child\r\n)\r\n\r\n# Create complete hierarchy\r\nfrom pi_web_sdk.models.event import EventFrameHierarchy, ChildEventFrame\r\n\r\nhierarchy = EventFrameHierarchy(\r\n    root=EventFrame(\r\n        name=\"Production Run\",\r\n        description=\"Full production cycle\",\r\n        start_time=datetime(2024, 1, 1, 8, 0, 0),\r\n        end_time=datetime(2024, 1, 1, 18, 0, 0),\r\n        attributes=[EventFrameAttribute(name=\"Batch\", value=\"B-2024-001\")]\r\n    ),\r\n    children=[\r\n        ChildEventFrame(\r\n            name=\"Mixing\",\r\n            start_time=datetime(2024, 1, 1, 8, 0, 0),\r\n            end_time=datetime(2024, 1, 1, 10, 0, 0),\r\n            attributes=[EventFrameAttribute(name=\"Speed\", value=1200)]\r\n        ),\r\n        ChildEventFrame(\r\n            name=\"Heating\",\r\n            start_time=datetime(2024, 1, 1, 10, 0, 0),\r\n            end_time=datetime(2024, 1, 1, 14, 0, 0),\r\n            attributes=[EventFrameAttribute(name=\"Target\", value=95.0)]\r\n        )\r\n    ]\r\n)\r\nclient.event_frame_helpers.create_event_frame_hierarchy(db_web_id, hierarchy)\r\n\r\n# Get event frame with all attribute values\r\nevent_data = client.event_frame_helpers.get_event_frame_with_attributes(\r\n    event_web_id,\r\n    include_values=True\r\n)\r\n```\r\n\r\nSee [docs/event_frame_helpers.md](docs/event_frame_helpers.md) for complete documentation.\r\n\r\n### Parsed Responses (Type-Safe)\r\n```python\r\nfrom pi_web_sdk.models.data import DataServer, Point\r\nfrom pi_web_sdk.models.responses import ItemsResponse\r\n\r\n# Get parsed response with type safety\r\nserver: DataServer = client.data_server.get_parsed(web_id)\r\nprint(f\"Server: {server.name}\")\r\nprint(f\"Version: {server.server_version}\")\r\nprint(f\"Connected: {server.is_connected}\")\r\n\r\n# List with type safety and iteration\r\nservers: ItemsResponse[DataServer] = client.data_server.list_parsed()\r\nfor server in servers:\r\n    print(f\"{server.name}: {server.path}\")\r\n\r\n# Get points with type safety\r\npoints: ItemsResponse[Point] = client.data_server.get_points_parsed(server_web_id)\r\nfor point in points:\r\n    print(f\"{point.name} ({point.point_type}): {point.engineering_units}\")\r\n```\r\n\r\n### Advanced Search (AFSearch Syntax)\r\n```python\r\nfrom pi_web_sdk.models.asset import Element, AttributeSearch\r\nfrom pi_web_sdk.models.responses import ItemsResponse\r\n\r\n# Query elements by attributes\r\nelements: ItemsResponse[Element] = client.element.get_elements_query(\r\n    database_web_id,\r\n    query=\"Name:='Pump*' Type:='Equipment'\"\r\n)\r\n\r\n# Create persistent attribute search\r\nsearch = AttributeSearch(\r\n    database_web_id=database_web_id,\r\n    query=\"Name:='Temperature' Type:='Float64'\"\r\n)\r\nsearch_result = client.element.create_search_by_attribute(search)\r\nsearch_id = search_result.web_id\r\n\r\n# Execute search later\r\nresults: ItemsResponse[Element] = client.element.execute_search_by_attribute(search_id)\r\n\r\n# Bulk get multiple elements\r\nelements: ItemsResponse[Element] = client.element.get_multiple(\r\n    [web_id1, web_id2, web_id3],\r\n    selected_fields=\"Name;Path;Description\"\r\n)\r\n```\r\n\r\n### Analysis Operations\r\n```python\r\nfrom pi_web_sdk.models.analysis import Analysis, AnalysisTemplate, SecurityEntry\r\n\r\n# Get analysis with security\r\nanalysis: Analysis = client.analysis.get(analysis_web_id)\r\n\r\n# Get security entries\r\nentries: ItemsResponse[SecurityEntry] = client.analysis.get_security_entries(analysis_web_id)\r\n\r\n# Create security entry\r\nentry = SecurityEntry(\r\n    name=\"Operators\",\r\n    security_identity_web_id=identity_web_id,\r\n    allow_rights=[\"Read\", \"Execute\"]\r\n)\r\nclient.analysis.create_security_entry(analysis_web_id, entry)\r\n\r\n# Work with analysis templates\r\ntemplate: AnalysisTemplate = client.analysis_template.get_by_path(\r\n    \"\\\\\\\\AnalysisTemplate\\\\MyTemplate\"\r\n)\r\ntemplate.description = \"Updated\"\r\nclient.analysis_template.update(template.web_id, template)\r\n\r\n# Get analysis categories\r\ncategories = client.analysis.get_categories(analysis_web_id)\r\n```\r\n\r\n## Available Controllers\r\nAll controller instances are available as attributes on `PIWebAPIClient`:\r\n\r\n### System & Configuration\r\n- `client.home` - Home endpoint\r\n- `client.system` - System information and status\r\n- `client.configuration` - System configuration\r\n\r\n### Asset Model\r\n- `client.asset_server` - Asset servers\r\n- `client.asset_database` - Asset databases\r\n- `client.element` - Elements\r\n- `client.element_category` - Element categories\r\n- `client.element_template` - Element templates\r\n- `client.attribute` - Attributes\r\n- `client.attribute_category` - Attribute categories\r\n- `client.attribute_template` - Attribute templates\r\n\r\n### Data & Streams\r\n- `client.data_server` - Data servers\r\n- `client.point` - PI Points\r\n- `client.stream` - Stream data operations (including Stream Updates)\r\n- `client.streamset` - Batch stream operations (including Stream Set Updates)\r\n\r\n### Analysis & Events\r\n- `client.analysis` - PI Analyses\r\n- `client.analysis_category` - Analysis categories\r\n- `client.analysis_rule` - Analysis rules\r\n- `client.analysis_rule_plugin` - Analysis rule plugins\r\n- `client.analysis_template` - Analysis templates\r\n- `client.event_frame` - Event frames\r\n- `client.event_frame_helpers` - High-level event frame operations\r\n- `client.table` - PI Tables\r\n- `client.table_category` - Table categories\r\n\r\n### OMF\r\n- `client.omf` - OSIsoft Message Format endpoint\r\n\r\n### Batch & Advanced\r\n- `client.batch` - Batch operations\r\n- `client.calculation` - Calculations\r\n- `client.channel` - Channels\r\n\r\n### Supporting Resources\r\n- `client.enumeration_set` - Enumeration sets\r\n- `client.enumeration_value` - Enumeration values\r\n- `client.unit` - Units of measure\r\n- `client.time_rule` - Time rules\r\n- `client.security` - Security operations\r\n- `client.notification` - Notification rules\r\n- `client.metrics` - System metrics\r\n\r\n## Package Layout\r\n- `pi_web_sdk/config.py` - Enums and configuration dataclass\r\n- `pi_web_sdk/exceptions.py` - Custom exception types\r\n- `pi_web_sdk/client.py` - Session management and HTTP helpers\r\n- `pi_web_sdk/controllers/` - Individual controller modules grouped by domain\r\n  - `controllers/base.py` - Base controller with shared utilities\r\n  - `controllers/system.py` - System and configuration controllers\r\n  - `controllers/asset.py` - Asset servers, databases, elements, templates\r\n  - `controllers/attribute.py` - Attributes, categories, templates\r\n  - `controllers/data.py` - Data servers and points (with parsed methods)\r\n  - `controllers/stream.py` - Stream and streamset operations (enhanced)\r\n  - `controllers/analysis.py` - Analysis controllers (fully enhanced)\r\n  - `controllers/event.py` - Event frame controller and high-level helpers\r\n  - `controllers/omf.py` - OMF controller and manager\r\n  - Additional controllers for tables, enumerations, units, security, notifications, etc.\r\n- `pi_web_sdk/models/` - Data models and response classes\r\n  - `models/responses.py` - Generic ItemsResponse[T] and PIResponse[T]\r\n  - `models/data.py` - DataServer and Point models\r\n  - `models/stream.py` - Stream enums (BufferOption, UpdateOption)\r\n  - `models/omf.py` - OMF data models\r\n  - `models/attribute.py` - Attribute models\r\n- `pi_web_sdk/omf/` - OMF support with ORM-style API\r\n  - `omf/orm.py` - Core OMF classes (Type, Container, Asset, Data)\r\n  - `omf/hierarchy.py` - Hierarchy builder utilities\r\n- `docs/` - Comprehensive documentation\r\n- `examples/` - Working code examples\r\n- `tests/` - 108 tests covering all controllers\r\n- `aveva_web_api.py` - Compatibility shim for existing imports\r\n\r\n## Extending the SDK\r\nEach controller inherits from `BaseController`, which exposes helper methods and the configured client session. Add new endpoint support by:\r\n\r\n1. Create a new controller module under `pi_web_sdk/controllers/`\r\n2. Define data models in `pi_web_sdk/models/`\r\n3. Register controller in `pi_web_sdk/controllers/__init__.py`\r\n4. Add it to `pi_web_sdk/client.py` in the `PIWebAPIClient.__init__` method\r\n\r\nExample:\r\n```python\r\nfrom __future__ import annotations\r\nfrom typing import Optional\r\nfrom dataclasses import dataclass\r\nfrom .base import BaseController\r\nfrom ..models.responses import PIResponse\r\n\r\n@dataclass\r\nclass MyResource:\r\n    web_id: str\r\n    name: str\r\n    description: Optional[str] = None\r\n\r\nclass MyController(BaseController):\r\n    def get(self, web_id: str) -> PIResponse[MyResource]:\r\n        response = self.client.get(f\"myresource/{web_id}\")\r\n        return PIResponse(MyResource(**response))\r\n```\r\n\r\n## Testing\r\nRun the test suite:\r\n\r\n```bash\r\n# Run all tests\r\npytest tests/\r\n\r\n# Run specific test files\r\npytest tests/test_omf_endpoint.py -v\r\n\r\n# Run with integration marker\r\npytest -m integration\r\n```\r\n\r\n## Deployment\r\n\r\n### Quick Deployment\r\n\r\n```bash\r\n# Validate package\r\npython deploy.py --check\r\n\r\n# Deploy to TestPyPI (recommended first)\r\npython deploy.py --test\r\n\r\n# Deploy to PyPI\r\npython deploy.py --prod\r\n```\r\n\r\n### Prerequisites\r\n- PyPI and TestPyPI accounts\r\n- API tokens configured in `~/.pypirc`\r\n- Build tools: `pip install build twine`\r\n\r\nSee [DEPLOYMENT_QUICKSTART.md](DEPLOYMENT_QUICKSTART.md) for quick start guide or [DEPLOYMENT.md](DEPLOYMENT.md) for comprehensive instructions.\r\n\r\n## Documentation & Examples\r\n- [PI Web API Reference](https://docs.aveva.com/bundle/pi-web-api-reference/page/help/getting-started.html)\r\n- [OMF Documentation](https://docs.aveva.com/)\r\n- [Stream Updates Guide](examples/README_STREAM_UPDATES.md) - Comprehensive guide for incremental data retrieval\r\n- [Stream Updates Examples](examples/stream_updates_example.py) - Working code examples\r\n- [Event Frame Helpers Documentation](docs/event_frame_helpers.md) - High-level event frame operations\r\n- [Parsed Responses Documentation](docs/parsed_responses.md) - Type-safe response wrappers\r\n- [Controller Additions Summary](docs/CONTROLLER_ADDITIONS_SUMMARY.md) - Complete list of all enhancements\r\n- See `examples/` directory for more usage examples\r\n\r\n## Recent Additions\r\n\r\n### Comprehensive Controller Enhancements (v2025.10)\r\nMajor enhancement to the SDK with 78 new methods and 108 tests covering the full PI Web API surface:\r\n\r\n**Analysis Controllers** (46 methods)\r\n- Full CRUD operations for analyses, templates, categories, rules, and plugins\r\n- Security management (get, create, update, delete security entries)\r\n- Category associations\r\n- Analysis rule management\r\n\r\n**Element Enhancements** (7 methods)\r\n- `get_multiple()` - Bulk retrieval of elements\r\n- `get_elements_query()` - AFSearch syntax support for powerful queries\r\n- `create_search_by_attribute()` / `execute_search_by_attribute()` - Persistent searches\r\n- `add_referenced_element()` / `remove_referenced_element()` - Reference management\r\n- `get_notification_rules()` / `create_notification_rule()` - Notification support\r\n\r\n**Stream Enhancements** (11 methods)\r\n- `get_end()` - Latest recorded value\r\n- `get_recorded_at_time()` / `get_recorded_at_times()` - Point-in-time retrieval\r\n- `get_interpolated_at_times()` - Interpolated values at specific times\r\n- `get_channel()` - Real-time WebSocket/SSE streaming channels\r\n- All methods available for single streams (`StreamController`) and multiple streams (`StreamSetController`)\r\n\r\n**Event Frame Helpers** (6 methods)\r\n- `create_event_frame_with_attributes()` - Create event frame and attributes in one call\r\n- `create_child_event_frame_with_attributes()` - Create child with attributes\r\n- `create_event_frame_hierarchy()` - Build complete hierarchies\r\n- `get_event_frame_with_attributes()` - Retrieve with all attribute values\r\n- `update_event_frame_attributes()` - Bulk attribute updates\r\n- `close_event_frame()` - Close with optional value capture\r\n\r\n**Parsed Responses** (Type-safe data classes)\r\n- Generic `ItemsResponse[T]` and `PIResponse[T]` wrappers\r\n- Support for iteration, indexing, and len()\r\n- Added `*_parsed()` methods to DataServer and Point controllers\r\n- Automatic deserialization to typed objects\r\n\r\nSee [docs/CONTROLLER_ADDITIONS_SUMMARY.md](docs/CONTROLLER_ADDITIONS_SUMMARY.md) for complete details.\r\n\r\n### Stream Updates (v2025.01)\r\nStream Updates provides an efficient way to retrieve incremental data updates without websockets. Key features:\r\n- **Marker-based tracking** - Maintains position in data stream\r\n- **Single or multiple streams** - Support for individual streams and stream sets\r\n- **Metadata change detection** - Notifies when data is invalidated\r\n- **Unit conversion** - Convert values during retrieval\r\n- **Selected fields** - Filter response data\r\n- **Type-safe data classes** - Strongly typed response models\r\n\r\n```python\r\nfrom pi_web_sdk.models.stream import StreamUpdateRegistration, StreamUpdates\r\n\r\n# Register once\r\nregistration: StreamUpdateRegistration = client.stream.register_update(stream_web_id)\r\nmarker = registration.latest_marker\r\n\r\n# Poll repeatedly for new data only\r\nwhile True:\r\n    time.sleep(5)\r\n    updates: StreamUpdates = client.stream.retrieve_update(marker)\r\n    # Process updates.items with type safety\r\n    for item in updates.items:\r\n        print(f\"{item.timestamp}: {item.value}\")\r\n    marker = updates.latest_marker\r\n```\r\n\r\n**Requirements**: PI Web API 2019+ with Stream Updates feature enabled\r\n\r\nSee [examples/README_STREAM_UPDATES.md](examples/README_STREAM_UPDATES.md) for complete documentation.\r\n\r\n## License\r\nSee LICENSE file for details.\r\n\r\n",
    "bugtrack_url": null,
    "license": "Proprietary",
    "summary": "Python client SDK for the AVEVA PI Web API with typed controllers and request helpers",
    "version": "0.1.17",
    "project_urls": {
        "Homepage": "https://example.com/pi-web-sdk",
        "Repository": "https://example.com/pi-web-sdk/source"
    },
    "split_keywords": [
        "osi",
        " pi",
        " web",
        " api",
        " sdk"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "a51b103ed95877ab3e9ee4521cb01bd8c0bc1322a5f231514dad669f1985bdda",
                "md5": "f11161d1e73662adec66124ecb10b679",
                "sha256": "d32af94e58ec2f6907596993ffcaf34da79062521d3773ebb3c21931a63bf8d4"
            },
            "downloads": -1,
            "filename": "pi_web_sdk-0.1.17-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "f11161d1e73662adec66124ecb10b679",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.9",
            "size": 72491,
            "upload_time": "2025-10-08T18:53:13",
            "upload_time_iso_8601": "2025-10-08T18:53:13.994086Z",
            "url": "https://files.pythonhosted.org/packages/a5/1b/103ed95877ab3e9ee4521cb01bd8c0bc1322a5f231514dad669f1985bdda/pi_web_sdk-0.1.17-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "941a7dbdfd2a8a55f4f90ceb1599c8ba1cc89ebf89da6f0e95d9976f394e1f5e",
                "md5": "2ace1352dc8a7a23f1cda557bc112cd9",
                "sha256": "2026a6ecc161cd374d4ba4d88c2c4f8fbb6102752005d6d9eb057465474d80d6"
            },
            "downloads": -1,
            "filename": "pi_web_sdk-0.1.17.tar.gz",
            "has_sig": false,
            "md5_digest": "2ace1352dc8a7a23f1cda557bc112cd9",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.9",
            "size": 107134,
            "upload_time": "2025-10-08T18:53:15",
            "upload_time_iso_8601": "2025-10-08T18:53:15.250651Z",
            "url": "https://files.pythonhosted.org/packages/94/1a/7dbdfd2a8a55f4f90ceb1599c8ba1cc89ebf89da6f0e95d9976f394e1f5e/pi_web_sdk-0.1.17.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-10-08 18:53:15",
    "github": false,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "lcname": "pi-web-sdk"
}