MCPLCWatcher

MCPLCWatcher は、MC プロトコル対応機器のデバイス値を読み取り、値の変化を監視する Python ライブラリです。 三菱電機 PLC に加え、MC プロトコルに対応した他社機器にも利用できます。 安全のため、デバイスへの書き込み機能は意図的に提供していません。

特徴

• MC プロトコル 3E / 4E フレームに対応
• D / M / X / Y デバイスの読み取りに対応
• 個別デバイス監視と連続デバイスの一括監視に対応
• デバイス値や通信状態が変化したときに動作するコールバック関数を設定可能
• PLC 停止やネットワーク断を通信状態コールバックで通知
• 通信断後はバックグラウンドで自動再接続
• 外部依存なし、Python 標準ライブラリのみで動作

対応範囲

インストール

pip install mcplcwatcher

基本的な使い方

import time

from mcplcwatcher import PLCWatcher


def on_value_changed(device, old_value, new_value, values):
    # device: 変化したデバイス名（例: "D2000"）
    # old_value: 前回値。初回通知時は None
    # new_value: 今回読み取った新しい値
    # values: 同回読み取りのスナップショット（デバイス名→値の Mapping）
    #   - watch_device では当該デバイスのみ
    #   - watch_device_group ではグループ内の全点が含まれる
    print(f"{device}: {old_value} -> {new_value}")
    print(f"snapshot={values}")


watcher = PLCWatcher(
    ip_address="192.168.10.130",
    port=5000,
    plc_type="Q",
    protocol_type="3E",
    interval=0.3,
)

watcher.watch_device("D2000", callback=on_value_changed)
watcher.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    watcher.stop()

よく使う機能

監視間隔

生成時は interval で指定します（既定は 0.3 秒）。監視の開始後も interval プロパティで変更できます。

watcher = PLCWatcher("192.168.10.130", interval=0.3)
watcher.start()
watcher.interval = 1.0  # 次の待機から 1 秒間隔になる

一括監視

連続するデバイスは watch_device_group() でまとめて監視できます。 通信回数を減らしたい場合に使用します。

from mcplcwatcher import PLCWatcher


def on_change(device, old_value, new_value, values):
    print(f"{device}: {old_value} -> {new_value}, snapshot={values}")


watcher = PLCWatcher("192.168.10.130")
watcher.watch_device_group("D2000", 10, on_change)
watcher.watch_device_group("M100", 32, on_change)
watcher.start()

通信状態の監視

PLC との接続状態が変化したときは、通信状態コールバックが呼び出されます。 start() 時点で PLC に接続できない場合も監視スレッドは起動し、再接続を継続します。

from mcplcwatcher import PLCWatcher


def on_change(device, old_value, new_value, values):
    print(f"{device}: {old_value} -> {new_value}")


def on_communication_status(connected, error):
    if connected:
        print("PLC通信が復旧しました")
    else:
        print(f"PLC通信に失敗しました: {error}")


watcher = PLCWatcher("192.168.10.130")
watcher.set_communication_status_callback(on_communication_status)
watcher.watch_device("D2000", callback=on_change)
watcher.start()

直接読み取り

監視ではなく現在値だけを読み取る場合は MCClient を使用します。

from mcplcwatcher import MCClient


with MCClient("192.168.10.130") as client:
    d_value = client.read_device("D2000")
    x_values = client.read_device("X0", 16)
    print(d_value, x_values)

詳細ドキュメント

• 使用方法ガイド: 監視、直接読み取り、通信状態、ログ、トラブルシュート
• API リファレンス: 公開クラス、メソッド、例外
• システム設計: 現行アーキテクチャと内部構成

サンプル

examples/ に用途別のサンプルがあります。

制限事項

• このライブラリはPLCに対して読み取りのみを行います。書き込み機能は提供していません。
• 通信方式は IPv4 の TCP/IP のみです。IPv6、UDP、シリアル通信には対応していません。
• 監視方式はポーリングです。PLC 側イベント通知ではありません。
• 1 つの PLCWatcher インスタンスは 1 つの MCClient 接続を使用します。
• MC プロトコル自体に認証や暗号化はないため、信頼できるネットワーク内で使用してください。