0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@NKKTechGlobal(NKKTech Global)

工場PLCのModbus/TCP通信をMQTT経由でクラウドに安全転送するエッジゲートウェイの設計とPython実装

0
Posted at

工場PLCのModbus/TCP通信をMQTT経由でクラウドに安全転送するエッジゲートウェイの設計とPython実装

Feature(167).png

はじめに

こんにちは、NKKTech Global 技術チームです。

工場の稼働監視や予兆保全（PdM）を実現するためには、PLC内のセンサー値やカウンタ値をクラウド上の時系列DBへ収集する必要があります。しかし、産業用ネットワークで長年使われている Modbus/TCP (Port 502) には暗号化も認証も存在しません。

PLCが接続されているOTネットワークをクラウドへ直結させるのではなく、中間に**「エッジゲートウェイ（産業用PC / Raspberry Pi等）」**を配置し、データを安全なプロトコルである MQTT over TLS (Port 8883) に変換して片方向通信で送信する設計が業界標準です。

本記事では、ネットワークの切断（瞬断）にも耐えうるエッジゲートウェイの設計パターンとPythonコードを完全公開します。

1. エッジゲートウェイのアーキテクチャ設計

OT（現場制御）とIT（クラウド）を分離し、セキュアな中継を行うための構成図です。

設計上の3大原則

  1. デュアルNIC構成による物理/論理分離: エッジゲートウェイにNICを2ポート持たせ、OT網とインターネット網の間でパケットルーティングを一切許可しない（フォワーディング無効化）。
  2. 片方向（アウトバウンドのみ）接続: クラウド側から工場内へのポート開放（インバウンド）は完全遮断し、エッジ側からクラウドへ確立したMQTTコネクション経由でメッセージを発行（Publish）する。
  3. Store & Forward（オフライン耐性）: 工場のネットワークが切断されてもデータを欠損させないよう、一時的にローカルストレージへバッファリングする。

2. Pythonによるゲートウェイ実装

pymodbus でPLCの保持レジスタ（Holding Registers）を読み出し、paho-mqtt でTLS暗号化してクラウドへ送信する実用的なスクリプトです。

必要なライブラリのインストール

import time
import json
import ssl
import logging
from pymodbus.client import ModbusTcpClient
import paho.mqtt.client as mqtt

# ログ設定
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("EdgeGateway")

# --- 設定情報 ---
PLC_IP = "192.168.1.50"         # PLCのローカルIP (OT側)
PLC_PORT = 502
PLC_UNIT_ID = 1

MQTT_BROKER = "iot.nkktech.com" # クラウドMQTTブローカーのエンドポイント
MQTT_PORT = 8883
MQTT_TOPIC = "factory/line_1/plc_data"
CLIENT_ID = "edge-gw-osaka-01"

# TLS証明書のパス (mTLS用)
CA_CERT = "./certs/rootCA.pem"
CLIENT_CERT = "./certs/client.crt"
CLIENT_KEY = "./certs/client.key"

# --- MQTTクライアント初期化 ---
def create_mqtt_client() -> mqtt.Client:
    client = mqtt.Client(client_id=CLIENT_ID, protocol=mqtt.MQTTv311)
    
    # TLS設定 (相互認証 mTLS)
    client.tls_set(
        ca_certs=CA_CERT,
        certfile=CLIENT_CERT,
        keyfile=CLIENT_KEY,
        tls_version=ssl.PROTOCOL_TLSv1_2
    )
    
    def on_connect(c, userdata, flags, rc):
        if rc == 0:
            logger.info("Connected to Cloud MQTT Broker successfully.")
        else:
            logger.error(f"Failed to connect to MQTT broker, return code: {rc}")

    client.on_connect = on_connect
    return client

# --- メインループ ---
def main():
    mqtt_client = create_mqtt_client()
    try:
        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
        mqtt_client.loop_start()
    except Exception as e:
        logger.error(f"Initial MQTT connection failed: {e}")

    modbus_client = ModbusTcpClient(PLC_IP, port=PLC_PORT)

    while True:
        try:
            # 1. Modbus/TCP接続確認
            if not modbus_client.connected:
                if not modbus_client.connect():
                    logger.warning("PLC connection failed. Retrying in 5 seconds...")
                    time.sleep(5)
                    continue

            # 2. 保持レジスタの読み出し (例: アドレス0から4ワード読み出し)
            # アドレス0: 温度 (x10), アドレス1: 圧力, アドレス2-3: 製造累計数 (32bit)
            result = modbus_client.read_holding_registers(address=0, count=4, slave=PLC_UNIT_ID)

            if result.isError():
                logger.error(f"Modbus read error: {result}")
            else:
                regs = result.registers
                
                # 生データのスケーリング変換
                payload = {
                    "gateway_id": CLIENT_ID,
                    "timestamp": int(time.time()),
                    "metrics": {
                        "temperature": regs[0] / 10.0,
                        "pressure_kpa": regs[1],
                        "total_production_count": (regs[2] << 16) | regs[3]
                    }
                }

                json_payload = json.dumps(payload)

                # 3. MQTTでクラウドへ送信 (QoS 1: 確実に1回以上届ける)
                msg_info = mqtt_client.publish(MQTT_TOPIC, json_payload, qos=1)
                msg_info.wait_for_publish(timeout=2.0)
                
                logger.info(f"Published: {json_payload}")

        except Exception as ex:
            logger.error(f"Unexpected error in gateway loop: {ex}")

        # 収集周期 (例: 1秒インターバル)
        time.sleep(1.0)

if __name__ == "__main__":
    main()

3. 実運用で必須となる3つの耐障害性対策

プロトタイプから本番環境へ移行する際、以下の耐久テストを必ずクリアする必要があります。

① ネットワーク瞬断時のローカル永続化（Store and Forward）

インターネット接続が数時間遮断された場合、MQTTのPublishは失敗します。

  • msg_info.is_published() を監視し、送信失敗時はメモリ上ではなく ローカルの軽量DB（SQLiteやDuckDB） にJSONをINSERTします。
  • 回線復旧を検知したバックグラウンドスレッドが、未送信データをタイムスタンプ順にクラウドへ再送するロジックを組み込みます。

② PLCのEndian（バイト順）差異への対応

PLCメーカーや型番によって、32bit整数や浮動小数点（Float）のバイト順（Big Endian / Little Endian / Mid-Little Endian）が異なります。

  • pymodbus.payload.BinaryPayloadDecoder を利用し、機種ごとのエンディアン差を吸収するマッピング定義層を設けます。

③ プロセスの死活監視（systemd / Docker Restart Policy）

エッジデバイスの電源断やPLCの一時停止に備え、OSレベルで自動再起動を設定します。

# /etc/systemd/system/edge-gateway.service
[Unit]
Description=PLC to Cloud MQTT Edge Gateway
After=network.target

[Service]
ExecStart=/usr/bin/python3 /opt/gateway/edge_gateway.py
Restart=always
RestartSec=10
User=edgeuser

[Install]
WantedBy=multi-user.target

まとめ：OTセキュリティを守り抜くクラウド連携

工場のスマート化において、「PLCをネットワークに繋ぐ」こと自体は難しくありません。しかし、**「現場の稼働と安全を脅かさずにデータを外へ持ち出す」**ことには、OTネットワークのプロトコル特性とITのセキュリティ技術双方の深い理解が求められます。

エッジゲートウェイによるプロトコル変換とmTLS暗号化は、製造現場の安心を担保しながらクラウドの恩恵を最大化するための唯一無二の防壁です。

お問い合わせ先

工場IoTの導入、PLC/SCADAからの安全なデータ収集パイプラインの構築、エッジデバイスの量産運用設計でお困りの際は、下記よりお気軽にご相談ください。

著者：NKKTech Global 技術チーム
私たちは、OTとITを高いエンジニアリング力で繋ぎ、製造業の次世代DXをグローバルに支援します。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?