英語の記事でpythonでAFSDKを使用する投稿があります。(Rafael Borges が作成した投稿です。)

PI and Python? PIthon!

PI Systemのトレーニングを実施しても、お客様の中でPythonからPIのデータを取得したいという要望はあります。

通常はおそらくPI Web APIを使用すると思うのですが、AFSDKも上記ポストの方法で呼び出せるのでその紹介です。

なお、PythonにてPI Web APIからデータ取得する例は以下のポストをご参照ください。

PythonにてPI Web APIからデータを収集の例 - GET

 

まず、Pythonをダウンロードします。

https://www.python.org/downloads/

2.7と3.6がありますが、今回は3.6で行いました。

pythonnetをインストールします。

http://pythonnet.github.io/

具体的にはpipを使用してインストールすると、楽です。

コマンドプロンプトでpython 3.6のフォルダまで移動し、

cd scripts

pip install pythonnetを実行します。(以下スクリーンショットはPython3.6をC:\Python36にインストールした場合です)

あとは以下コードをPIthon.pyとして保存し、Python実行フォルダに入れて、

python.exe PIthon.pyをコマンドプロンプトで実行します。

 

GitHubにコードも上がっています。

GitHub - kenji0711/PIthon: Python and AFSDK project

 

*注意 PI Tagへの書き込みの機能が入っています。

writept = PIPoint.FindPIPoint(piServer,"PleaseEnterWriteTagName")

の"PleaseEnterWriteTagName"を書き込みを実施するタグ名に変更してください。

また、AFはNuGreenの階層を元に作成しています。ご自身のAFのエレメント、属性に変更してください。

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 *
from OSIsoft.AF.PI import *
from OSIsoft.AF.Asset import *
from OSIsoft.AF.Data import *
from OSIsoft.AF.Time import *
from OSIsoft.AF.UnitsOfMeasure import *


print("Welcome to PIthon!!")
# PI Data Archive
piServers = PIServers()  
piServer = piServers.DefaultPIServer;


pt = PIPoint.FindPIPoint(piServer, "sinusoid")
name = pt.Name.lower()
# CurrentValue
print('\nShowing PI Tag CurrentValue from {0}'.format(name))
current_value = pt.CurrentValue()
print( '{0}\'s Current Value: {1}'.format(name, current_value.Value))


#recordedvalues
timerange = AFTimeRange("*-3h", "*")
recorded = pt.RecordedValues(timerange, AFBoundaryType.Inside, "", False)
print('\nShowing PI Tag RecordedValues from {0}'.format(name))
for event in recorded:
    print('{0} value: {1}'.format(event.Timestamp.LocalTime, event.Value))


#plotValues
plotvalues = pt.PlotValues(timerange, 100)
print('\nShowing PI Tag PlotValues from {0}'.format(name))
for event in plotvalues:
    print('{0} value: {1}'.format(event.Timestamp.LocalTime, event.Value))


#interpolatedvalues
span = AFTimeSpan.Parse("1h")
interpolated = pt.InterpolatedValues(timerange, span, "", False)
print('\nShowing PI Tag InterpolatedValues from {0}'.format(name))
for event in interpolated:
    print('{0} value: {1}'.format(event.Timestamp.LocalTime, event.Value))


#summariesvalues
summaries = pt.Summaries(timerange, span, AFSummaryTypes.Average, AFCalculationBasis.TimeWeighted, AFTimestampCalculation.Auto)
print('\nShowing PI Tag SummariesValues(Average) from {0}'.format(name))
for summary in summaries:
    for event in summary.Value:
        print('{0} value: {1}'.format(event.Timestamp.LocalTime, event.Value))


#writeValue
writept = PIPoint.FindPIPoint(piServer,"PleaseEnterWriteTagName")
writeptname = writept.Name.lower()
val = AFValue()
val.Value = 20
#val.Timestamp = AFTime("t+9h")


print('\nWrite value to {0} value: {1}'.format(writeptname, val.Value))
writept.UpdateValue(val, AFUpdateOption.Replace, AFBufferOption.BufferIfPossible)


#Connect to AF
print ('\nConnect to AF')
afServers = PISystems()
afServer = afServers.DefaultPISystem
DB = afServer.Databases.DefaultDatabase
#DB = afServer.Databases.get_Item("NuGreen")
element = DB.Elements.get_Item("NuGreen").Elements.get_Item("Little Rock").Elements.get_Item("Extruding Process").Elements.get_Item("Equipment").Elements.get_Item("K-435")


attribute = element.Attributes.get_Item("Steam Flow")
attval = attribute.GetValue()


print ('Element Name: {0}'.format(element.Name))
print ('Attribute Name: {0} | Value : {1} {2}'.format(attribute.Name, attval.Value, attribute.DefaultUOM))


#create element with attribute
print('\nCreate Element with Attribute')
if DB.Elements.get_Item("Test New Element") is not None:
    print("Already Existing Element: Test New Element")
else:
    newelement = DB.Elements.Add("Test New Element")
    newelement.Description = "Created element from PIthon"
    newattribute = newelement.Attributes.Add("Test Attribute")
    newattribute.DataReferencePlugIn = afServer.DataReferencePlugIns.get_Item("PI Point")
    newattribute.DataReference.ConfigString = "cdt158"
    DB.CheckIn()
    print("Created new Element : Test New Element")











 

以下のように実行できます。

ちなみにPython 3.6, 2.7どちらも実行できることを確認しています。

Python + AFSDKはHelpファイルがなく、Python 2.7だとAFValueを生成する際など、

val = AFValue(30) が動作しないなど癖がありますが、手さぐりで上記コードを作成してみました。

ちなみにPython 3.6ではval = AFValue(30)でも動作します。

Python3.6の方がコードが楽に書けてよいかと思います。

 

疑問点があればぜひコメントしてください。