PLCからゲートウェイでデータを取得し、データベースにJSONで保存します。複数回に分けて、サンプルを用いて解説します。
前回は、データを保存するデータベースとテーブルを準備しました。

PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (4)

今回は、MQTTブローカーからデータを受信するサブスクライバーを実装します。少々駆け足になりますが、サブスクライバーには、データを受信しつつデータベースに書き込むまでを実装します。

プログラミング

Pythonを使用して実装します。Pythonでなければならない合理的な理由はありませんが、私の環境では、Pythonではだめだという理由もありません。

PythonでMQTTをハンドリングには、paho-mqttというライブラリを使うと、新規コーディング量を減らすことができます。paho-mqttの導入については、このサイトをおすすめします。
Python で MQTT (Paho)

同様に、PythonからMySQLをアクセスするには、mysql-connector-python-rfというライブラリを使うと、新規コーディング量を減らすことができます。mysql-connector-python-rfの導入については、このサイトをおすすめします。
Python 3 から MySQL を触る

最低限の機能を実装してみました。

import paho.mqtt.client as mqtt
import json
import mysql.connector

#-----------------------------
#    mqtt
host = '127.0.0.1'
port = 1883
topic = 'sample'

def on_connect(client, userdata, flags, respons_code):
    print('status {0}'.format(respons_code))
    client.subscribe(topic)

def on_message(client, userdata, msg):
    json_text = json.dumps(json.loads(msg.payload))
    print(json_text)
    try:
        #-----------------------------
        #    database
        conn = mysql.connector.connect(
            host='127.0.0.1',
            port=3306,
            user='hoge',
            password='hoge001',
            database='hoge'
        )
        cur = conn.cursor()
        sql = "insert into from_plc (body) values (%s);"
        cur.execute(sql, (json_text,))
        conn.commit()
        cur.close()
        conn.close()
    except Exception as e:
        print(e)
        print("Debug: Error at Database access.")
        return

if __name__ == '__main__':
    client = mqtt.Client(protocol=mqtt.MQTTv311)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(host, port=port, keepalive=60)
    client.loop_forever()

サンプル・プログラムは、MQTTブローカー（=mosquitto）に届く、Topic = sample のメッセージを待ち受けます。メッセージが届くと、メッセージ中のデータをJSON文字列に変換し、これをMySQLのテーブル"from_plc"に追加します。この動作をひたすら繰り返します。

テスト

Pythonスクリプトを起動しておきます。

c:\Users\hoge>py sample.py
status 0

これまでと同様に、PLCには、本物の代わりにModbus/TCPシミュレーターを使用します。
Modbus/TCPシミュレーター"mod_RSsim.exe"を起動します。

ゲートウェイのシミュレーターを起動するために、EasyBuilderのメニュー[プロジェクト]-[オンラインシミュレーション]をクリックします。

シミュレーターの起動を確認します。

Modbus/TCPシミュレータの、アドレス40001（注意: +0です）をクリックし、Valueに任意の数値を入力し値を変更します。[OK]ボタンで閉じます。

Modbus/TCPシミュレータの、アドレス40001の値を変更する都度、Pythonスクリプトがこの情報をJSONで受け取り、コンソールに表示します。
Pythonスクリプトを止めるには、キーボードからCtrl+Cを押してください。

c:\Users\hoge>py sample.py
status 0
{"value": 0, "ts": "2022-07-21T15:27:43.676079"}
{"value": 1, "ts": "2022-07-21T15:27:49.704662"}
{"value": 2, "ts": "2022-07-21T15:27:57.641088"}
{"value": 3, "ts": "2022-07-21T15:28:07.644020"}
{"value": 4, "ts": "2022-07-21T15:28:11.627588"}
{"value": 5, "ts": "2022-07-21T15:28:14.659965"}
Traceback (most recent call last):
  File "c:\Users\hoge\sample.py", line 44, in <module>
    client.loop_forever()
  File "C:\Users\hoge\AppData\Local\Programs\Python\Python310\lib\site-packages\paho\mqtt\client.py", line 1756, in loop_forever
    rc = self._loop(timeout)
  File "C:\Users\hoge\AppData\Local\Programs\Python\Python310\lib\site-packages\paho\mqtt\client.py", line 1150, in _loop
    socklist = select.select(rlist, wlist, [], timeout)
KeyboardInterrupt
^C
c:\Users\hoge>py sample.py

MySQLのテーブル"from_plc"を確認すると、ゲートウェイからのJSONデータが保存されているのがわかります。

c:\Users\hoge>mysql -u hoge -p
Enter password: *******
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 24
Server version: 8.0.29 MySQL Community Server - GPL

Copyright (c) 2000, 2022, Oracle and/or its affiliates.

Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql> use hoge
Database changed
mysql> select * from from_plc;
+--------------------------------------------------+---------------------+
| body                                             | time_insert         |
+--------------------------------------------------+---------------------+
| {"ts": "2022-07-21T15:27:43.676079", "value": 0} | 2022-07-21 23:27:43 |
| {"ts": "2022-07-21T15:27:49.704662", "value": 1} | 2022-07-21 23:27:49 |
| {"ts": "2022-07-21T15:27:57.641088", "value": 2} | 2022-07-21 23:27:57 |
| {"ts": "2022-07-21T15:28:07.644020", "value": 3} | 2022-07-21 23:28:07 |
| {"ts": "2022-07-21T15:28:11.627588", "value": 4} | 2022-07-21 23:28:11 |
| {"ts": "2022-07-21T15:28:14.659965", "value": 5} | 2022-07-21 23:28:14 |
+--------------------------------------------------+---------------------+
6 rows in set (0.00 sec)

mysql>

次回

PLCのデータをデータベースにJSONで書きました。もう次回は不要かもしれませんが、せっかくデータベースに書いたので、次回はこれをブラウザに表示させます。くどいなあ。

PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (1)
PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (2)
PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (3)
PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (4)
PLCからゲートウェイでデータを取得しデータベースにJSONで保存 (6)