OSI PI AF SDK query

The PI AF SDK seems to be a OSIsoft.AFSDK.dll library implemented in .Net

You can use it through C# directly, or use it through a Python wrapper pythonnet.

The way it queries  asset framework is through navigating from top element to the target element and then get the attribute

It seems to need the connection to asset framework set up in advance, e.g. having the PI system explorer installed and have the AF servers setup? The dll library file needs to be available in the Program Files folder.

It's not like RTQP that can just connect to the URL through ODBC.

Prerequisites: (1) install pythonnet (2) install PI System Explorer 

#conda install pythonnet, the pip install may not work with clr

import sys

import clr

sys.path.append(r'C:\Program Files (x86)\PIPC\AF\PublicAssemblies\4.0')  

clr.AddReference('OSIsoft.AFSDK')

from OSIsoft.AF import PISystems

from OSIsoft.AF.Time import AFTimeRange

import pandas as pd

#Connect to AF

af_servers = PISystems()

pi_system = af_servers['Your AF Server']

print(pi_system.ConnectionInfo.Host)

#navigate to the attribute

pi_db = pi_system.Databases.get_Item('You AF Database')

element1= pi_db.Elements.get_Item('the top element in the AF')

element2= element1.Elements.get_Item('the second element in the AF')

element3= element2.Elements.get_Item('the third element in the AF')

attr= element3.Attributes.get_Item('target attribute')

print(attr.GetValue())

#query a time range

query_range = AFTimeRange('*-24h', '*')

values = attr.GetValues(query_range, 0, None)

for v in values:

    print("Annotated = {0}".format(v.Annotated))

    print("IsGood = {0}".format(v.IsGood))

    print("Questionable = {0}".format(v.Questionable))

    print("Substituted = {0}".format(v.Substituted))

    print("TimeStamp = {0}".format(v.Timestamp.LocalTime))

    print("UOM = {0}".format(v.UOM))

    print("Value = {0}".format(v.Value))

#convert to pandas dataframe

values_list = [[v.Timestamp.LocalTime, v.Value] for v in values if (v.IsGood==1 and v.Questionable==0)]

df = pd.DataFrame(values_list, columns = ['TimeStamp', 'Value'])

Query the current value for a specific timestamp 

Note there may be no recorded value at the time. If after the interpolated value, use the GetValue() method below, otherwise the RecordedValue() method can return the immediately prior value.

The PI AF SDK references:

https://docs.osisoft.com/bundle/af-sdk/page/html/M_OSIsoft_AF_Asset_AFAttribute_GetValue_3.htm

https://docs.osisoft.com/bundle/af-sdk/page/html/M_OSIsoft_AF_Data_AFData_RecordedValue.htm

#conda install pythonnet

import sys

import clr

import pandas as pd

sys.path.append(r'C:\Program Files (x86)\PIPC\AF\PublicAssemblies\4.0')  

clr.AddReference('OSIsoft.AFSDK')

from OSIsoft.AF import PISystems

from OSIsoft.AF.Time import AFTime, AFTimeRange 

from OSIsoft.AF.Data import AFRetrievalMode, AFBoundaryType

#Connect to AF

af_servers = PISystems()

pi_system = af_servers['AF SERVER NAME']

print(pi_system.ConnectionInfo.Host)

#navigate to the attribute

pi_db = pi_system.Databases.get_Item('AF DATABASE NAME')

element1 = pi_db.Elements.get_Item('1st element name')

element2 = element1.Elements.get_Item('2nd element name')

attribute = element2.Attributes.get_Item('ATTRIBUTE NAME')

query_time = AFTime('2022-09-27 21:42:06')

value1 = attribute.GetValue(query_time)

value2 = attribute.Data.RecordedValue(query_time, AFRetrievalMode.AtOrBefore, attribute.DefaultUOM)

print(value1)

print(value2)

Query the values for given a time range

By setting the AFBoundaryType to 'outside' it also returns the immediate values before and after the range. Therefore even there is no recorded value within the range, it still returns the current value for the range.

The PI AF SDK reference:

https://docs.osisoft.com/bundle/af-sdk/page/html/M_OSIsoft_AF_Data_AFData_RecordedValues.htm

#conda install pythonnet

import sys

import clr

import pandas as pd

sys.path.append(r'C:\Program Files (x86)\PIPC\AF\PublicAssemblies\4.0')  

clr.AddReference('OSIsoft.AFSDK')

from OSIsoft.AF import PISystems

from OSIsoft.AF.Time import AFTimeRange 

from OSIsoft.AF.Data import AFBoundaryType

#Connect to AF

af_servers = PISystems()

pi_system = af_servers['AF SERVER NAME']

#navigate to the attribute

pi_db = pi_system.Databases.get_Item('AF DATABASE NAME')

element1 = pi_db.Elements.get_Item('1st element name')

element2 = element1.Elements.get_Item('2nd element name')

attribute = element2.Attributes.get_Item('ATTRIBUTE NAME')

#query data by time range, also return the immediate values before and after the range

time_range = AFTimeRange('2022-09-27 21:42:06', '2022-09-27 21:42:07')

values = attribute.Data.RecordedValues(time_range, AFBoundaryType.Outside, attribute.DefaultUOM, None, True, 0)

values_list = [[v.Timestamp.LocalTime, v.Value] for v in values if (v.IsGood==1 and v.Questionable==0)]

df = pd.DataFrame(values_list, columns = ['TimeStamp', 'Value'])