PythonでOPC UAで通信してみた

はじめに

こんにちは、omkです。
検証作業にOPC UAサーバが必要になったのでPythonで簡単に作っていきます。

やってみた

今回はAmazon EC2の仮想マシンを2台用意してそれぞれクライアントとサーバとして、
クライアントマシンのCPU使用率の情報をサーバに集約するシチュエーションを想定します（？）。

環境準備

クライアント・サーバそれぞれでPythonが動作するようにセットアップして、
python-opcuaをインストールします。

pip install opcua

これで環境の準備は完了です。
ではプログラムコードを作成していきます。

サーバ

Gitからサンプルを確認しつつサーバのコードを作成します。
出来上がったものがこちらです。

import sys
import time
from opcua import ua, Server

namespace_uri = "urn:omk:client"
client_ip = "0.0.0.0"
port = "4840"

if __name__ == "__main__":

    # サーバエンドポイントを設定
    server = Server()
    server.set_endpoint(f"opc.tcp://{client_ip}:{port}/server/")

    # ネームスペースを登録
    namespace_index = server.register_namespace(namespace_uri)

    # ネームスペースの確認
    if namespace_index is None:
        print("Error: Namespace index is None")
    else:
        # オブジェクトにClient_Aオブジェクトを追加
        objects = server.get_objects_node()
        client_a = objects.add_object(namespace_index, "Client_A")

        # CPU_USAGE変数を追加して、クライアントから書き込み可能に設定
        cpu_usage = client_a.add_variable(namespace_index, "CPU_USAGE", 0.0)
        cpu_usage.set_writable()

        # 起動
        server.start()

    try:
        while True:
            time.sleep(1)
    finally:
        server.stop()

クライアント全体用の名前空間を作成して、Objectsノード配下にクライアントごとのオブジェクトを作成（Client_A）、クライアントマシンのCPU使用率（CPU_USAGE）を書き込めるようにしています。

クライアント

同じようにクライアント用のコードも作成します。
以下のプログラムを作成しました。

import sys
import psutil
from opcua import Client, ua

namespace = "urn:omk:client"
server_ip = "18.182.113.257" #念の為嘘のIPを入れています
port = "4840"

if __name__ == "__main__":

    client = Client(f"opc.tcp://{server_ip}:{port}/server/")

    try:
        # 接続
        client.connect()

        # 階層のルートを取得
        root = client.get_root_node()

        # 対象のネームスペースインデックスを取得
        namespace_array = client.get_namespace_array()
        ns_index = namespace_array.index(namespace)

        # CPU使用率を取得
        cpu_percent = psutil.cpu_percent(percpu=False)
        print("CPU USAGE:", cpu_percent)

        # ノードを取得
        node = root.get_child([
            f"0:Objects",
            f"{ns_index}:Client_A",
            f"{ns_index}:CPU_USAGE"
        ])

        # 値を書き込み
        value_to_write = ua.Variant(cpu_percent, ua.VariantType.Double)
        node.set_value(value_to_write)

        # サーバから値を取得
        value = node.get_value()
        print("VALUE:", value)

    finally:
        client.disconnect()

サーバに接続したら対象の名前空間と階層からノードを取得、
次に取得したノードにCPU使用率を書き込み、
最後に書き込まれた値を確認、
という内容になっています。

実行

ではサーバ・クライアントでそれぞれ作成したコードを実行して通信出来ることを確認します。

サーバ

なんか色々出ましたがサーバとしてListenしてくれていることがわかります。

cryptography is not installed, use of crypto disabled
cryptography is not installed, use of crypto disabled
Endpoints other than open requested but private key and certificate are not set.
Listening on 0.0.0.0:4840

クライアント

こちらも読み書き出来ました。

cryptography is not installed, use of crypto disabled
cryptography is not installed, use of crypto disabled
CPU USAGE: 51.4
VALUE: 51.4

以下で書き込み時の値を出力しています。

CPU USAGE: 51.4

以下で書き込み後に実際にサーバから値を取得して出力しています。

VALUE: 51.4

ということで同じ値が取れたので正常にサーバに書き込まれたことがわかります。
（もちろん読み取りだけのクライアントコードを実行しても正常に値が取れるのでローカル上の値を利用しているわけではありません）

おわりに

簡易的にOPC UAの通信が出来る環境が作成出来ました。
サンプルを見るかぎりもっと色々なことが出来そうなので今後も試していきたいです。

以上、最後までお付き合いありがとうございました。