英語の記事で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をインストールします。
具体的には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の方がコードが楽に書けてよいかと思います。
疑問点があればぜひコメントしてください。
