Types

This module contains type definitions, enums, and data classes used throughout PIPolars.

Type Aliases

pipolars.core.types.PITimestamp

Type alias for timestamps: Union[datetime, str, AFTime]

Accepts: - Python datetime objects - PI time expression strings (e.g., "*-1d", "t") - AFTime objects

pipolars.core.types.TagName

Type alias for tag names: str

pipolars.core.types.TagPath

Type alias for AF attribute paths: str

AFTime Class

class pipolars.AFTime[source]

Bases: object

Represents a PI AF Time specification.

Supports both absolute timestamps and relative time expressions like “*” (now), “*-1d” (1 day ago), “t” (today), etc.

Examples

>>> AFTime("*")  # Now
>>> AFTime("*-1h")  # 1 hour ago
>>> AFTime("2024-01-01")  # Absolute date
>>> AFTime("t")  # Today at midnight
>>> AFTime("y")  # Yesterday at midnight

Class Methods

classmethod now()[source]

Create an AFTime representing the current time.

classmethod today()[source]

Create an AFTime representing today at midnight.

classmethod yesterday()[source]

Create an AFTime representing yesterday at midnight.

classmethod ago(**kwargs)[source]

Create an AFTime relative to now.

Parameters:

**kwargs (int) – Time units (days, hours, minutes, seconds)

Returns:

AFTime representing the relative time.

Return type:

AFTime

Example

>>> AFTime.ago(days=1, hours=2)  # 1 day and 2 hours ago
classmethod from_datetime(dt)[source]

Create an AFTime from a Python datetime object.

expression: str

The time expression string.

classmethod now()[source]

Create an AFTime representing the current time.

classmethod today()[source]

Create an AFTime representing today at midnight.

classmethod yesterday()[source]

Create an AFTime representing yesterday at midnight.

classmethod ago(**kwargs)[source]

Create an AFTime relative to now.

Parameters:

**kwargs (int) – Time units (days, hours, minutes, seconds)

Returns:

AFTime representing the relative time.

Return type:

AFTime

Example

>>> AFTime.ago(days=1, hours=2)  # 1 day and 2 hours ago
classmethod from_datetime(dt)[source]

Create an AFTime from a Python datetime object.

__init__(expression)

Usage:

from pipolars import AFTime

# Create from expression
now = AFTime("*")
yesterday = AFTime("y")

# Use class methods
now = AFTime.now()
today = AFTime.today()
one_hour_ago = AFTime.ago(hours=1)
complex_ago = AFTime.ago(days=1, hours=6, minutes=30)

# From datetime
from datetime import datetime
dt = datetime(2024, 1, 15, 12, 0)
af_time = AFTime.from_datetime(dt)

PIValue Class

class pipolars.PIValue[source]

Bases: object

Represents a single PI value with timestamp and quality.

This is the fundamental data unit returned from PI queries.

timestamp: datetime

The timestamp of the value.

value: Any

The actual value (can be numeric, string, or digital state).

quality: DataQuality

The quality flag for this value.

is_good: bool

Convenience property indicating if the value is good quality.

to_dict()[source]

Convert to a dictionary for DataFrame construction.

__init__(timestamp, value, quality=DataQuality.GOOD)

TimeRange Class

class pipolars.core.types.TimeRange[source]

Bases: object

Represents a time range for PI queries.

start

Start time of the range

Type:

datetime.datetime | str | pipolars.core.types.AFTime

end

End time of the range

Type:

datetime.datetime | str | pipolars.core.types.AFTime

Class Methods

classmethod last(**kwargs)[source]

Create a time range from now to the past.

Example

>>> TimeRange.last(days=7)  # Last 7 days
>>> TimeRange.last(hours=24)  # Last 24 hours
classmethod today()[source]

Create a time range for today.

start: datetime | str | AFTime
end: datetime | str | AFTime
classmethod last(**kwargs)[source]

Create a time range from now to the past.

Example

>>> TimeRange.last(days=7)  # Last 7 days
>>> TimeRange.last(hours=24)  # Last 24 hours
classmethod today()[source]

Create a time range for today.

__init__(start, end)

Usage:

from pipolars.core.types import TimeRange, AFTime

# Create time range
time_range = TimeRange(
    start=AFTime("*-1d"),
    end=AFTime("*")
)

# Use convenience methods
last_week = TimeRange.last(days=7)
today = TimeRange.today()

PointConfig Class

class pipolars.core.types.PointConfig[source]

Bases: object

Configuration for a PI Point (tag).

Contains metadata about a PI Point retrieved from the server.

name: str

The PI Point name (tag name).

__init__(name, point_id, point_type, description='', engineering_units='', zero=0.0, span=100.0, display_digits=-5, typical_value=None)
point_id: int

The unique point ID in the PI Data Archive.

point_type: PointType

The data type of the point.

description: str

Description of the point.

engineering_units: str

Engineering units for the point.

zero: float

Zero value for scaling.

span: float

Span value for scaling.

display_digits: int

Number of display digits.

typical_value: float | None

Typical value for this point.

SummaryResult Class

class pipolars.core.types.SummaryResult[source]

Bases: object

Result of a summary calculation.

Contains the calculated summary values for a time range.

__init__(tag, start, end, average=None, minimum=None, maximum=None, total=None, count=None, std_dev=None, range=None, percent_good=None)
tag: str

The PI Point name.

start: datetime

Start time of the summary period.

end: datetime

End time of the summary period.

average: float | None
minimum: float | None
maximum: float | None
total: float | None
count: int | None
std_dev: float | None
range: float | None
percent_good: float | None

Enumerations

RetrievalMode

class pipolars.RetrievalMode[source]

Bases: str, Enum

Data retrieval modes for PI Point queries.

These modes determine how data is retrieved from the PI Data Archive.

Data retrieval modes for PI queries.

RECORDED = "recorded"

Return actual recorded values as stored in the archive.

INTERPOLATED = "interpolated"

Return interpolated values at regular intervals.

PLOT = "plot"

Return values optimized for plotting (reduced data density).

SUMMARY = "summary"

Return summary statistics (min, max, avg, etc.).

COMPRESSED = "compressed"

Return compressed data using exception/compression settings.

RECORDED = 'recorded'

Return actual recorded values as stored in the archive.

INTERPOLATED = 'interpolated'

Return interpolated values at regular intervals.

PLOT = 'plot'

Return values optimized for plotting (reduced data density).

SUMMARY = 'summary'

Return summary statistics (min, max, avg, etc.).

COMPRESSED = 'compressed'

Return compressed data using exception/compression settings.

__new__(value)

SummaryType

class pipolars.SummaryType[source]

Bases: IntEnum

Summary calculation types for PI data.

These correspond to OSIsoft AF SDK AFSummaryTypes enumeration.

Summary calculation types for PI data.

TOTAL = 1

Sum of values.

AVERAGE = 2

Time-weighted average.

MINIMUM = 4

Minimum value.

MAXIMUM = 8

Maximum value.

RANGE = 16

Range (max - min).

STD_DEV = 32

Standard deviation.

POP_STD_DEV = 64

Population standard deviation.

COUNT = 128

Number of values.

PERCENT_GOOD = 8192

Percentage of good values.

NONE = 0
TOTAL = 1
AVERAGE = 2
MINIMUM = 4
MAXIMUM = 8
RANGE = 16
STD_DEV = 32
POP_STD_DEV = 64
COUNT = 128
PERCENT_GOOD = 8192
TOTAL_WITH_UOM = 16384
ALL = 24831
ALL_FOR_NON_NUMERIC = 8320
__new__(value)

Usage:

from pipolars import SummaryType

# Single summary type
df = client.summary("TAG", "*-1d", "*", summary_types=SummaryType.AVERAGE)

# Multiple summary types
df = client.summary(
    "TAG", "*-1d", "*",
    summary_types=[
        SummaryType.AVERAGE,
        SummaryType.MINIMUM,
        SummaryType.MAXIMUM,
        SummaryType.STD_DEV
    ]
)

TimestampMode

class pipolars.TimestampMode[source]

Bases: str, Enum

Timestamp handling modes for summary calculations.

Timestamp handling modes for summary calculations.

AUTO = "auto"

Automatically determine timestamp placement.

START = "start"

Use interval start time.

END = "end"

Use interval end time.

MIDDLE = "middle"

Use interval midpoint.

AUTO = 'auto'

Automatically determine timestamp placement.

START = 'start'

Use interval start time.

END = 'end'

Use interval end time.

MIDDLE = 'middle'

Use interval midpoint.

__new__(value)

DataQuality

class pipolars.DataQuality[source]

Bases: IntEnum

PI data quality flags.

These flags indicate the quality and reliability of PI values.

PI data quality flags.

GOOD = 0

Value is good and reliable.

SUBSTITUTED = 1

Value was manually substituted.

QUESTIONABLE = 2

Value quality is questionable.

BAD = 3

Value is bad or unreliable.

NO_DATA = 4

No data available for the requested time.

CALC_FAILED = 5

Calculation failed to produce a value.

GOOD = 0

Value is good and reliable.

SUBSTITUTED = 1

Value was manually substituted.

QUESTIONABLE = 2

Value quality is questionable.

BAD = 3

Value is bad or unreliable.

NO_DATA = 4

No data available for the requested time.

CALC_FAILED = 5

Calculation failed to produce a value.

__new__(value)

DigitalState

class pipolars.core.types.DigitalState[source]

Bases: str, Enum

Common PI digital states.

NO_DATA = 'No Data'
BAD_INPUT = 'Bad Input'
CALC_OFF = 'Calc Off'
COMM_FAIL = 'Comm Fail'
CONFIGURE = 'Configure'
I_O_TIMEOUT = 'I/O Timeout'
NO_SAMPLE = 'No Sample'
SHUTDOWN = 'Shutdown'
SCAN_OFF = 'Scan Off'
OVER_RANGE = 'Over Range'
UNDER_RANGE = 'Under Range'
__new__(value)

PointType

class pipolars.core.types.PointType[source]

Bases: str, Enum

PI Point data types.

FLOAT16 = 'float16'
FLOAT32 = 'float32'
FLOAT64 = 'float64'
INT16 = 'int16'
INT32 = 'int32'
DIGITAL = 'digital'
TIMESTAMP = 'timestamp'
STRING = 'string'
BLOB = 'blob'
__new__(value)

BoundaryType

class pipolars.core.types.BoundaryType[source]

Bases: str, Enum

Boundary handling for time range queries.

INSIDE = 'inside'

Only include values strictly inside the time range.

OUTSIDE = 'outside'

Include boundary values outside the range.

INTERPOLATED = 'interpolated'

Interpolate values at boundaries.

__new__(value)

Schema Definitions

PIPolars defines Polars schemas for consistent DataFrame structures:

pipolars.core.types.PI_VALUE_SCHEMA

Schema for single-tag value DataFrames:

{
    "timestamp": pl.Datetime("us", "UTC"),
    "value": pl.Float64(),
    "quality": pl.Int8(),
}
pipolars.core.types.PI_VALUE_WITH_TAG_SCHEMA

Schema for multi-tag value DataFrames:

{
    "tag": pl.Utf8(),
    "timestamp": pl.Datetime("us", "UTC"),
    "value": pl.Float64(),
    "quality": pl.Int8(),
}
pipolars.core.types.SUMMARY_SCHEMA

Schema for summary DataFrames:

{
    "tag": pl.Utf8(),
    "start": pl.Datetime("us", "UTC"),
    "end": pl.Datetime("us", "UTC"),
    "average": pl.Float64(),
    "minimum": pl.Float64(),
    "maximum": pl.Float64(),
    "total": pl.Float64(),
    "count": pl.Int64(),
    "std_dev": pl.Float64(),
    "percent_good": pl.Float64(),
}

See Also