1. はじめに — なぜ最初に学ぶべきは Modbus なのか

「PLC や産業機器を Python から扱いたい」と考えたとき、最初に学ぶプロトコルとしてもっとも費用対効果が高いのが Modbus です。理由はシンプルで、多くのメーカーがメーカー固有プロトコルとは別に Modbus 対応をオプションで持っているため、メーカー横断で同じ知識が使い回せるからです。

三菱、キーエンス、オムロン、シーメンス、シュナイダー、富士電機、横河——主要 PLC メーカーは何らかの形で Modbus に対応しています。インバータ、温調器、電力計、流量計のような周辺機器でも Modbus 対応は標準的です。

本記事は、Modbus が初めての方でも自宅 PC だけで実装と動作確認まで完了できるように構成しています。pymodbus のサーバー機能で簡易シミュレータを自作し、サーバーとクライアントを同一 PC で動かす形で TCP 編の全コードを再現できます（なお pymodbus には Web UI 付きの公式シミュレータ pymodbus.simulator コマンドもあります。pip install "pymodbus[simulator]" で入り、TCP ポート 5020 / Web UI 8081 で起動します。本記事では挙動を理解しやすいコード自作方式で進めます）。

2. Modbus プロトコルの基礎

Modbus はリクエスト・レスポンス型のシンプルなプロトコルです。「どのアドレスのデータを、どれだけ、どの操作で」を 1 リクエストで送り、サーバー側が結果を返します。

2.1 TCP / RTU / ASCII の違い

  • Modbus TCP: Ethernet 上で動く。ホスト名・ポート（既定 502）でアクセス。初学者はこれから始めるのが一番楽
  • Modbus RTU: シリアル（RS-232C / RS-485）で動くバイナリ形式。CRC 付き。物理層がシリアルなので、USB-シリアル変換ケーブルなどが必要
  • Modbus ASCII: シリアルでテキスト形式。実用では RTU のほうが一般的

2.2 4 種類のデータ領域

Modbus には扱うデータの種類が 4 つあり、それぞれ別のファンクションコードでアクセスします。

  • コイル（Coil）: 1 ビット、読み書き可能。主にデジタル出力
  • ディスクリート入力（Discrete Input）: 1 ビット、読み取り専用。主にデジタル入力
  • 保持レジスタ（Holding Register）: 16 ビット、読み書き可能。主にアナログ出力・設定値
  • 入力レジスタ（Input Register）: 16 ビット、読み取り専用。主にアナログ入力

業務でもっとも頻繁に使うのは保持レジスタです。「設定値の書き込み」「測定値の読み取り」「制御指令の書き込み」がすべて保持レジスタで完結することが多いはずです。

2.3 主要なファンクションコード

  • FC01: コイル読み取り
  • FC02: ディスクリート入力読み取り
  • FC03: 保持レジスタ読み取り
  • FC04: 入力レジスタ読み取り
  • FC05: 単一コイル書き込み
  • FC06: 単一保持レジスタ書き込み
  • FC15: 複数コイル書き込み
  • FC16: 複数保持レジスタ書き込み

pymodbus ではこれらが read_holding_registers、write_register のようにメソッド名で表現されており、ファンクションコードを覚えていなくても使えます。

3. 環境構築

3.1 pymodbus のインストール

python -m pip install "pymodbus>=3.15"

pymodbus は 3.x 系の中でも API が激しく変わり続けています。特に「相手局を指定するキーワード」は 2.x で unit= → 3.3.0 以降 slave= → 3.10.0 以降 device_id= と 2 回改名されており、後方互換エイリアスはありません（3.15 で slave= を渡すと TypeError になります）。Web 上の古い記事のコードが動かないときは、まず pip show pymodbus でバージョンを確認してください。本記事のコードは 2026 年 8 月時点の最新版である 3.15.0（2026-08-13 リリース）で動作確認しています。なお、社内プロキシ・閉域網で pip が使えない環境への持ち込み方は 学習ロードマップ STEP 2 を参照してください。

3.2 動作確認用シミュレータ

pymodbus には Modbus サーバー機能が同梱されており、サーバーとクライアントを同一 PC で起動できます。実機がなくても、TCP 編のコードはすべて自宅 PC で動作確認できます。

4. シミュレータサーバーの起動

まずは「クライアントから接続して読み書きできる」サーバーを 1 つ立ち上げます。テスト用に保持レジスタへダミーデータを入れておきます。なおサーバー側は非同期処理（asyncio）で書くのが現行 3.x 系の作法です。async def / await は深入りせず、このままコピペで動かして OK です。

# server.py — Modbus TCP サーバー（シミュレータ、pymodbus 3.15）
import asyncio
import logging

from pymodbus.simulator import SimData, SimDevice, DataType
from pymodbus.server import StartAsyncTcpServer

logging.basicConfig(level=logging.INFO)


async def main() -> None:
    # SimData の address はクライアントのリクエストアドレスと 1:1 対応（オフセットなし）
    block = [
        SimData(0, values=[25, 30, 50], datatype=DataType.REGISTERS),  # 0〜2 番地: 温度っぽい初期値
        SimData(3, count=97, values=0, datatype=DataType.REGISTERS),   # 3〜99 番地: 0 埋め
    ]
    device = SimDevice(id=1, simdata=block)
    await StartAsyncTcpServer(context=device, address=("127.0.0.1", 5020))


if __name__ == "__main__":
    asyncio.run(main())

ここで使っている SimData / SimDevice は pymodbus 3.13 以降の現行サーバー API です（3.15.0 で動作確認済み）。Web の古い記事にある ModbusSequentialDataBlock + ModbusSlaveContext 方式は 3.10 で改名・3.13 で非推奨になっており、3.15 では ModbusSlaveContext の import 自体が ImportError になるうえ、開始アドレス 0 が例外になる制約も残っているため、そのままでは動きません（なお 3.13 系には holding register と input register が入れ替わる不具合もありましたが、こちらは 3.14.0 で修正済みです。いずれも 3.15.0 で実測確認）。現行 3.x 系でサーバーを書くなら SimData / SimDevice 一択です。

ポート番号について: Linux では 1024 未満のポート（Modbus 標準の 502 を含む）の bind に root 権限が必要ですが、Windows にはこの制限はありません。それでも、実機 PLC やほかのソフトとの衝突を避ける意味で、学習用途ではどの OS でも安全な 5020 を使うのがおすすめです。このスクリプトを実行すると、ターミナルで Modbus TCP サーバーが起動します（停止は Ctrl+C）。

5. クライアント側の実装パターン

5.1 接続と切断の基本形

# client_basic.py
from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient("127.0.0.1", port=5020)
if not client.connect():
    raise ConnectionError("failed to connect")

try:
    # 0 番地から 3 個読む
    result = client.read_holding_registers(address=0, count=3, device_id=1)
    if result.isError():
        print(f"modbus error: {result}")
    else:
        print(f"values: {result.registers}")
finally:
    client.close()

server.py を起動したターミナルとは別のターミナルで実行してください。values: [25, 30, 50] と表示されれば成功です。

ポイント:

  • client.connect() の戻り値を必ず確認。False なら通信前にエラーで止める
  • レスポンスに isError() が用意されている。例外ではなくこちらで判定する
  • finally で close()。例外時にも socket を閉じる

5.2 単一レジスタの書き込み

※ 5.2〜5.4 のコードは断片です。5.1 の client.connect() 成功後〜 client.close() の間に挿入して実行してください。

client.write_register(address=10, value=100, device_id=1)

たとえば「アドレス 10 に温度しきい値 100 を書き込む」のような単純なケース。書き込み後は読み戻して値が反映されたことを確認すると安全です。

5.3 複数レジスタの一括書き込み

# 32bit 値（例: 設備の稼働時間カウンタ）を 2 レジスタに分けて書く
seconds = 123456
high = (seconds >> 16) & 0xFFFF
low = seconds & 0xFFFF
client.write_registers(address=20, values=[high, low], device_id=1)

16 ビットでは表現しきれない値（32 ビット整数、浮動小数）を扱う際は 2 レジスタに分けます。エンディアンとワード順は機器仕様により異なるため、必ず仕様書を確認してください。

5.4 32 ビット浮動小数の読み書き

製造機器のアナログ値は IEEE 754 単精度浮動小数（4 バイト = 2 レジスタ）で扱われることがよくあります。かつては pymodbus.payload（BinaryPayloadBuilder / BinaryPayloadDecoder）が定番でしたが、3.10 で削除済みで、3.15 でも import した時点で ModuleNotFoundError になります。現行はクライアント自身が持つ変換メソッド convert_to_registers() / convert_from_registers() を使います。

# 書き込み: float を 2 レジスタに変換して書く
regs = client.convert_to_registers(
    123.45, data_type=client.DATATYPE.FLOAT32, word_order="big"
)
client.write_registers(address=30, values=regs, device_id=1)

# 読み取り: 2 レジスタを読んで float に戻す
result = client.read_holding_registers(address=30, count=2, device_id=1)
value = client.convert_from_registers(
    result.registers, data_type=client.DATATYPE.FLOAT32, word_order="big"
)
print(f"float: {value}")  # 123.44999694824219（単精度の量子化誤差込み）

data_type には client.DATATYPE の列挙（INT16 / UINT16 / INT32 / UINT32 / FLOAT32 / FLOAT64 / STRING など）を指定します。ワード順は word_order="big" / "little" の 2 通りで、レジスタ内のバイト順は Modbus 標準のビッグエンディアン固定です。機器によってワード順が異なるため、値が異常に大きい・小さい・NaN になる場合はまず word_order を疑ってください。もう 1 つ大事な注意として、float32 は単精度の量子化誤差を含むため（123.45 → 123.44999…）、読み戻した値を == で比較しないでください（math.isclose() や許容差での比較にします）。

6. 現場で使える実装パターン

パターン A: 定周期ポーリング

1 秒に 1 回温度・圧力・回転数を読みに行き、CSV に書き出す典型的なロガーです。

# logger.py
import csv
import datetime as dt
import time
from pathlib import Path

from pymodbus.client import ModbusTcpClient

INTERVAL = 1.0
LOG_PATH = Path("data.csv")


def main() -> None:
    client = ModbusTcpClient("127.0.0.1", port=5020)
    if not client.connect():
        raise ConnectionError("failed to connect")

    write_header = not LOG_PATH.exists()
    with LOG_PATH.open("a", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        if write_header:
            writer.writerow(["timestamp", "temperature", "pressure", "rpm"])

        try:
            while True:
                t0 = time.monotonic()
                res = client.read_holding_registers(address=0, count=3, device_id=1)
                if res.isError():
                    print(f"error: {res}")
                else:
                    writer.writerow([
                        dt.datetime.now().isoformat(timespec="seconds"),
                        *res.registers,
                    ])
                    f.flush()
                # 周期維持（処理時間を差し引いて sleep）
                elapsed = time.monotonic() - t0
                time.sleep(max(0.0, INTERVAL - elapsed))
        finally:
            client.close()


if __name__ == "__main__":
    main()

ポイント:

  • time.monotonic() 基準で「処理時間を差し引いた sleep」にすることで周期がずれにくい
  • encoding="utf-8" 明示。CP932 環境でも UTF-8 で統一すれば後段の解析が楽になる
  • f.flush() で書き込みを即時反映。アプリが落ちてもデータが残る

パターン B: しきい値監視 → アラート

異常値を検知してアラートを出す監視。チャタリング対策と通知重複防止が現場の鍵です。

import logging
import time

from pymodbus.client import ModbusTcpClient

THRESHOLD = 80
COOLDOWN = 60  # アラート間の最小間隔（秒）
HYSTERESIS = 5  # 復帰しきい値（80 で発報、75 で復帰）

logger = logging.getLogger(__name__)


def monitor_loop() -> None:
    client = ModbusTcpClient("127.0.0.1", port=5020)
    if not client.connect():
        raise ConnectionError("failed to connect")
    in_alert = False
    last_alert = 0.0

    try:
        while True:
            res = client.read_holding_registers(address=0, count=1, device_id=1)
            if res.isError():
                logger.warning("read error: %s", res)
                time.sleep(1.0)
                continue

            value = res.registers[0]
            now = time.time()

            if not in_alert and value > THRESHOLD and (now - last_alert) > COOLDOWN:
                logger.warning("ALERT: value=%s exceeds threshold=%s", value, THRESHOLD)
                in_alert = True
                last_alert = now
            elif in_alert and value < (THRESHOLD - HYSTERESIS):
                logger.info("RECOVERED: value=%s back to normal", value)
                in_alert = False

            time.sleep(1.0)
    finally:
        client.close()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)  # これが無いとアラートが画面に出ない
    monitor_loop()

ヒステリシス（発報しきい値と復帰しきい値を分ける）と、クールダウン（連続発報の抑止）の組み合わせは、現場運用で必須の設計です。これがないと、しきい値付近で値が振動するたびにアラートが大量発報し、現場の信頼を失います。

パターン C: レシピ書き込み（一括設定）

生産品種切替時に、しきい値や設定値を一気に書き込みたいケース。「人間が間違えやすい順番で書き込まないこと」がポイントです。

# ※ 断片コードです。5.1 のように接続済みの client を渡して使います
RECIPE_A = {
    10: 100,   # 温度しきい値
    11: 50,    # 圧力しきい値
    12: 1500,  # 設備回転数
}


def apply_recipe(client, recipe: dict[int, int]) -> None:
    for addr, value in recipe.items():
        rsp = client.write_register(address=addr, value=value, device_id=1)
        if rsp.isError():
            raise RuntimeError(f"failed to write addr={addr}: {rsp}")
    # 書き込み後、1 点ずつ読み戻して検算
    # （アドレスが飛び飛びのレシピでも正しく検算できるよう、一括読みにしない）
    for addr, expected in recipe.items():
        rsp = client.read_holding_registers(address=addr, count=1, device_id=1)
        if rsp.isError():
            raise RuntimeError(f"verify read failed at addr={addr}: {rsp}")
        actual = rsp.registers[0]
        if actual != expected:
            raise RuntimeError(f"verify failed at addr={addr}: expected={expected} actual={actual}")

レシピ書き込みは現場の安全に直結するため、書き込み後の読み戻し検算は必ず実装してください。書き込み成功でも、機器側で値域外の値が拒否されるケースがあります。

7. RTU（シリアル）の場合

pymodbus は RTU もほぼ同じインターフェースで使えます。違うのは接続先の指定です。

from pymodbus.client import ModbusSerialClient

client = ModbusSerialClient(
    port="COM3",          # Linux なら "/dev/ttyUSB0"
    baudrate=9600,
    bytesize=8,
    parity="N",
    stopbits=1,
    timeout=1,
)
if not client.connect():
    raise ConnectionError("failed to open serial port")
res = client.read_holding_registers(address=0, count=3, device_id=1)
if res.isError():
    print(f"modbus error: {res}")
else:
    print(f"values: {res.registers}")
client.close()

シリアル通信特有の注意点:

  • 機器側の通信設定（ボーレート・パリティ・ストップビット・スレーブアドレス）と完全一致させること
  • RS-485 の場合は終端抵抗の有無、半二重の方向制御を確認。マルチドロップ接続では同一スレーブアドレスの機器が 2 台いないかも確認（応答が化ける定番原因）
  • USB-シリアル変換ケーブルのドライバ（FTDI 系・Prolific 系など）を現場 PC にインストール
  • 自分の PC の COM ポート番号は、デバイスマネージャーの「ポート (COM と LPT)」で確認できる。番号は再起動・差し替えで変わることがあり、固定もデバイスマネージャーから設定
  • シリアルポートの無い PC で RTU の練習をしたい場合は、仮想 COM ポートペアを作るツール（com0com など）でクライアント・サーバーを同居させる方法がある

8. 24 時間運用設計のチェックリスト

本記事のサンプルは「動く」段階のコードです。長期運用にもっていくには、以下を加えていきます。

  • 例外時の自動再接続: connect 失敗・タイムアウト・通信エラー時にバックオフ付きで再接続
  • ログのファイル出力 + ローテーション: logging.handlers.TimedRotatingFileHandler
  • 同時接続制限: 多くの PLC は同時接続数に制限あり。ポーリングプロセスは 1 つに集約する
  • ポーリング周期の調整: 機器の応答性能を超える周期を設定すると、レスポンスが詰まり全体が遅くなる
  • 書き込みの権限分離: 「読み取り専用クライアント」と「書き込み権限ありクライアント」を別プロセスにし、誤書き込みリスクを下げる
  • サマータイム / 時刻同期: 海外現場では時刻ずれが集計に影響。NTP 同期を必ず入れる

9. おわりに

Modbus は仕様がシンプルで、Python との相性も極めて良いプロトコルです。一度 pymodbus でこのパターンを身に付けておけば、PLC・インバータ・電力計・温調器・流量計など、現場のほぼ全ての機器が同じインターフェースで扱えるようになります。

本記事のサンプルは、TCP 編を自宅環境のシミュレータで再現できる構成にしています。「実機が触れない」という制約は、もはや学習を止める理由にはなりません。

関連記事