1. PLC が無いまま MC プロトコルのクライアントを書くことになったら
新ラインの立ち上げでは、ソフトの担当だけが先に走り出します。PLC が入るのは来月、それでも上位のツールは先に作れと言われる。手元にあるのはデバイス割付表だけです(本記事では架空の例として、D100 に生産数、D102 に不良数、M20 に異常フラグを割り付けます)。
相手がいなければ、書いたコードは 1 行も確かめられません。かといって、既存ラインの PLC を借りるのも簡単ではありません。
- 稼働中の設備に試験電文を打つのは論外です。打ち間違えれば、読出しのつもりが書込みになります
- 停止中の設備を借りるにも段取りが要ります。計画停止を押さえ、保全と生産の合意を取り、立ち会ってもらう——「ちょっと試したい」の単位では動けません
- 検証機の確保も予算と時間がかかります。PLC 本体・電源・ベース・イーサネットの口、ツールのライセンスまで要ります
そこで要るのが、手元の PC の中だけで完結する「PLC 役」です。ラダーの動作まで模擬する PLC シミュレータではなく、通信だけを模擬する「通信シミュレータ」=クライアントが話しかける相手だけを用意するダミーサーバーになります。
1.1 実機が無いときの選択肢 — PLC シミュレータを含む 4 つの手段
| 手段 | 手に入るもの | 向かないこと・注意点 |
|---|---|---|
| 実機を借りる / 検証機を買う | 唯一の「本物」。応答時間・スキャンタイム・同時接続の実際が分かる | 段取りと費用。試したいタイミングで触れないことが多い |
| エンジニアリングツールのシミュレーション機能 | ラダープログラム自体の動作検証 | 目的が上位通信のテストとは別で、ライセンスが要ります。当サイトでは実機・ツールを所有していないため未検証です |
| 先行 OSS のシミュレータ | 対応コマンドが多く、制御ロジックまで模擬できるものがあります(10 章) | 中身を把握せず入れると、通らなかったときの切り分け先が増えます |
| ダミーサーバーを自作する(本記事) | フレームの中身が全部見える。CI(コードを変えるたびに自動でテストを回す仕組み)で回せる。改造できる | 実機の代わりにはなりません(2 章)。作る手間は 188 行分 |
1.2 Modbus と違って「自作しかない」理由
同じ FA 系プロトコルでも、Modbus なら話が早く済みます。pymodbus はクライアントとサーバーの両方を実装し、シミュレータのコマンドまで同梱しているためです。
SLMP / MC プロトコルは事情が違います。pymcprotocol 0.3.0(MIT ライセンス・2021 年のリリース以降は更新なし)にあるのはクライアント実装だけで、サーバー機能は含まれていません。pymelsec も同じくクライアント側です(選定は親記事)。
つまり SLMP では、実機を用意するか、相手役を自分で書くかの二択です。本記事は後者を標準ライブラリだけで書きます。pip install が通らない閉域網の PC でもサーバー側は動かせるからです(クライアント側の pymcprotocol は別途必要)。
1.3 この記事の到達点
以下はすべて 2026 年 9 月 4 日に自宅の検証機で実測したものです。
- 188 行のサーバーが、
pymcprotocolのワード/ビット各 2 種の一括読出し・一括書込みに応答します - D100 に
[1234, -1, 32767]を書いて読み戻すと同じ値が返り、M0 から 8 点のビットも一致しました - 範囲外アクセス・未対応コマンド・不正な単位指定には終了コードを返し、クライアント側で例外が上がりました
- pytest 8 件が 4.11 秒で PASS。実機なしで CI に載せられます
- フレームを 1 バイトずつ分割して送っても、組み立て直して応答しました(9 章)
1.4 まず動かす(3 ステップ)
フレームの解説より先に手元で動かしたい場合は、ここだけで足ります。
- サーバーを保存する。11 章の完成コードをコピーして
slmp_dummy_server.pyとして保存します。標準ライブラリだけで動くので、サーバー側に追加インストールは要りません(検証は Python 3.12.10。from __future__ import annotationsを入れてあるので 3.8 以降を想定していますが、3.12 以外では未検証です) - クライアント側だけ pip が要ります。
pip install pymcprotocolを実行します。管理者権限が無い PC なら--user、プロキシ経由なら--proxy http://プロキシ:ポートを添えてください。pip が外に出られない閉域網の手順は閉域網・プロキシ・管理者権限なしの Python 環境構築にまとめました(ファイルの持ち込みは社内規程に従ってください) - ターミナルを 2 枚開く。
serve_forever()は止まらずに待ち受け続けるため、1 枚目でサーバー、2 枚目でクライアントを動かします
# 1 枚目: サーバー(listening on ... と出たまま止まりますが、それが正常です)
python slmp_dummy_server.py --host 127.0.0.1 --port 5007
# 2 枚目: クライアント(6.1 のコードを client.py として保存したもの)
python client.py
クライアント側のコードは 6.1 にあります。実行して [1234, -1, 32767] と [1, 0, 1, 1, 0, 0, 0, 1]、つまり書いた値がそのまま表示されれば往復成功です。1 枚目のサーバー側には RECV / SEND の 16 進が並んでいるはずです。
うまくいかないとき
- 同じサーバーを 2 つ起動してもエラーは出ません(実測は 9.5)。別のアプリがそのポートを使っていると
WinError 10013(別記事で実測)が出ることがあります。どちらの場合も--port 5008のように空き番号へ変えます(connect()も同じ番号に) ModuleNotFoundError: pymcprotocolが出たら、ステップ 2 が別の Python に入っています。python -m pip install pymcprotocolで入れ直すのが確実です- 応答を待ったまま返らないときは、1 枚目のサーバーに
RECVが出ているか確認します。出ていなければ、接続先の IP かポートが違うか、別に起動したままのサーバーが受け取っています(9.5)
3〜5 章はフレームの中身と実装の解説なので、動かすだけなら飛ばして 6 章へ進んで構いません。
2. このダミーサーバーでできること・できないこと
先に線引きをします。曖昧なまま「PLC の代わりができた」と社内で言うと、後で必ず食い違います。
| できること | できないこと |
|---|---|
| クライアント側コードの結線確認(デバイス名・点数・引数の型・戻り値の並び) | 応答時間の評価。ループバックの往復は速すぎて、実機の目安になりません(下の実測) |
| 例外処理の分岐を通すこと(終了コードを意図的に返せます) | タイムアウト値・リトライ間隔の妥当性検証。遅延も再送も起きない相手では決められません |
| 実機なしで回帰テストを CI に載せること(8 章) | 同時接続数の上限。実機側の交信コネクション数の制約は再現していません |
| 3E フレームのバイト列を目で見ながら理解すること | スキャンタイムとの同期。値は要求のたびに確定し、演算周期の概念がありません |
| 分割到着(短読み)への耐性を試すこと(9 章) | 実機の終了コードの再現。どの状況でどれを返すかは筆者が決めた実装です |
| 割付表どおりに値を仕込んで、監視画面や帳票を先に作ること(仕込み方は 11.1) | ASCII 交信・4E フレーム・0401 / 1401 以外のコマンド・D と M 以外のデバイス(未実装) |
plctype が Q / L / QnA / iQ-L のクライアントとの結線確認 |
plctype="iQ-R" のクライアント。電文の形が変わり C059h を返します(実測・3.1) |
「速すぎる」の根拠を数字で出します。batchread_wordunits("D0", 10) を 1,000 回繰り返した往復時間の実測です。
localhost round-trip, batchread_wordunits D0 x10, n=1000
min = 0.043 ms
median = 0.050 ms
p95 = 0.065 ms
max = 0.201 ms
中央値 0.050 ミリ秒。同じ PC の中でメモリ上のリストを引いているだけなので、当然こうなります。実機ではここにイーサネットの往復と PLC 側の交信処理が加わります。この環境でタイムアウトを 0.5 秒に決めた、という判断には根拠がありません。タイムアウトやリトライは実機の条件で決めてください(切り分け手順は親記事にあります)。
裏を返せば、「相手が返してくれさえすれば通るコードか」を確かめる用途では十分です。読出し点数が範囲を超えたら何が起きるか、負の値がどう化けるか——実機に触る前に潰せます。
3. SLMP 3E バイナリフレームの最小知識
ここから 5 章まではフレームの中身と実装の解説です。先に動かすなら 1.4 の手順で起動し、必要になってから戻ってきてください。
3.1 SLMP と MC プロトコルの関係
SLMP(Seamless Message Protocol)は、CC-Link IE 対応機器と汎用イーサネット機器がネットワークの階層を意識せずに通信するために、CC-Link 協会が公開している共通プロトコルです。三菱電機の MC プロトコル(MELSEC Communication Protocol)の 3E フレームは、その一部にあたります(用語の整理は親記事の 5-2 節)。
対象は MELSEC iQ-R シリーズなどの SLMP 対応機ですが、本記事のダミーサーバーは特定の機種を模したものではありません。「3E バイナリの要求に、それらしい応答を返す相手役」です。
ひとつ先に断っておきます。同じ 3E バイナリでも、クライアント側の機種指定で電文の形が変わります。Type3E(plctype="iQ-R") で同じ batchread_wordunits("D0", 2) を実行した実測です。
### plctype='Q': batchread_wordunits('D0', 2)
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 00 00 00 a8 02 00
SEND d0 00 00 ff ff 03 00 06 00 00 00 00 00 00 00
### plctype='iQ-R': batchread_wordunits('D0', 2)
RECV 50 00 00 ff ff 03 00 0e 00 04 00 01 04 02 00 00 00 00 00 a8 00 02 00
SEND d0 00 00 ff ff 03 00 02 00 59 c0
iQ-R 指定では要求が 21 バイトから 23 バイトに増えました。サブコマンドが 00 00 から 02 00 に変わり、先頭デバイス番号は 4 バイト、デバイスコードは 2 バイトになります。本サーバーはこの形を受け付けず、終了コード C059h を返しました。相手にできるのは plctype が Q / L / QnA / iQ-L のときの電文だけです。実測したのは "Q" と "iQ-R" の 2 つで、残る 3 つは pymcprotocol 0.3.0 が電文の形を iQ-R かどうかだけで分岐している実装から判断しています。なお plctype に iQ-F の選択肢はありません(指定できるのは Q / L / QnA / iQ-L / iQ-R の 5 種です。機種差は親記事の 5-2 節)。
3.2 出典の扱い — ここは正直に書きます
フレーム構造の一次資料は 2 つあります。CC-Link 協会が公開している SLMP 仕様書と、三菱電機の「MELSEC コミュニケーションプロトコルリファレンスマニュアル」(資料番号 SH-080003)です。前者はダウンロードに会員登録が必要で、後者も入手に利用登録を求められる場合があります。本記事の執筆時点で、筆者はどちらの本文も入手していません。
そのため本章は 2 段構えで書いています。
- ①公開されている技術情報の一致からの帰納(確度は中): 項目の並び・バイト長・コマンド値は、公開された複数の解説が独立に一致する内容をまとめたものです。公式仕様書で確認したわけではありません
- ②今回の実測による裏付け: そのうえで、
pymcprotocol0.3.0 が実際に送ってきたバイト列を 1 バイトずつ突き合わせ、こちらが返した応答をライブラリが正しく解釈したことを確認しました。送られてきた 21 バイトは推測ではありません
言えるのは「①の解釈で組んだサーバーが、②のクライアントと往復した」ところまでで、実機の PLC が同じ解釈で動く保証にはなりません。正確な定義、終了コードの全リスト、機種ごとの制約は公式マニュアル(SH-080003)を参照してください。
3.3 実測した要求フレーム 21 バイト
batchread_wordunits(headdevice="D100", readsize=3) を実行したとき、サーバーが受信したのは次の 21 バイトです。bytes.hex(sep=" ") の出力に、フィールドの区切りだけ空けています。
50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 64 00 00 a8 03 00
フィールドごとに分解すると次のようになります。
| フィールド | バイト数 | 実測値 | 意味 |
|---|---|---|---|
| サブヘッダ | 2 | 50 00 | 3E フレームの要求。ここだけビッグエンディアン送出(5000h がそのまま 50 00) |
| ネットワーク番号 | 1 | 00 | 自局 |
| 局番(PC 番号) | 1 | ff | 自局 |
| 要求先ユニット I/O 番号 | 2 | ff 03 | 03FFh をリトルエンディアン送出したもの(CPU ユニット宛の既定値) |
| 要求先マルチドロップ局番 | 1 | 00 | 自局 |
| 要求データ長 | 2 | 0c 00 | 12。監視タイマ以降のバイト数(監視タイマ 2 + 残り 10 = 12 と実測一致) |
| 監視タイマ | 2 | 04 00 | 4(pymcprotocol 0.3.0 の既定値) |
| コマンド | 2 | 01 04 | 0401h = 一括読出し。ワイヤ上は 01 04 の順 |
| サブコマンド | 2 | 00 00 | 0000h = ワード単位(0001h ならビット単位) |
| 先頭デバイス番号 | 3 | 64 00 00 | 100(3 バイトのリトルエンディアン) |
| デバイスコード | 1 | a8 | D(データレジスタ) |
| デバイス点数 | 2 | 03 00 | 3 点 |
3.4 実装で効くのは 2 点 — バイト配置を図で見る
- 先頭 9 バイトは長さが固定(サブヘッダ 2 + 経路 5 + 要求データ長 2)。ここを読めば、残りが何バイト来るか分かります
- 要求データ長は「監視タイマ以降」のバイト数。フレーム全体でも、コマンド以降でもありません。取り違えると、読み過ぎるか、次のフレームの先頭を食い残します
3.5 応答フレームは「エコーバック+長さ+終了コード」
上の要求に対してサーバーが返した 17 バイトです。
d0 00 00 ff ff 03 00 08 00 00 00 d2 04 ff ff ff 7f
組み立ては単純です。サブヘッダ d0 00 の後ろに要求の経路 5 バイトをそのままエコーバックし、応答データ長(終了コード以降のバイト数)を 2 バイト、終了コード 00 00(正常)、データの順に並べます。この例では終了コード 2 + データ 6 = 8 なので 08 00 です。
位置をずらせない理由があります。pymcprotocol 0.3.0 は応答を固定オフセットで読むためです。ソースでは終了コードがオフセット 9、データ本体がオフセット 11 でした(_get_answerstatus_index() = 9 / _get_answerdata_index() = 11)。経路を 5 バイトきっちり返さないと、クライアントは終了コードとして関係ないバイトを読みます。
3.6 デバイスコードと終了コード(代表例のみ)
使うデバイスコードは 2 つだけです。公開されている解説と一致し、実測でも同じ値が届きました。
| デバイス | コード | 本実装での扱い |
|---|---|---|
| D(データレジスタ) | A8h | ワードデバイス。8,192 点 |
| M(内部リレー) | 90h | ビットデバイス。8,192 点 |
終了コードは、本実装で返す 4 つだけを挙げます。
| 終了コード | 本実装で返す条件 |
|---|---|
0000h | 正常完了 |
C056h | デバイスの範囲を超える読み書き要求 |
C059h | 未対応のコマンド・サブコマンド |
C05Ch | 要求内容の誤り(ワードデバイスへのビット単位アクセスなど) |
これは代表例で、正式な全リストは公式マニュアル(SH-080003)にしかありません。どの状況でどれを返すかも、公開情報から読み取れる意味に沿って筆者が決めた実装です。「実機でもこのコードが来るはず」と読み替えないでください。
4. 実装 — 受信ループとフレームの組み立て
4〜5 章のコードは、クラスの一部を切り出した説明用の抜粋です。インデントが付いたままなので、そのままファイルに貼っても動きません。動かすときは 11 章の完成コードを丸ごと使ってください。
4.1 標準ライブラリの socketserver で書く
サーバー側は socketserver.ThreadingTCPServer です。接続ごとに別スレッドでハンドラが動き、書くのは BaseRequestHandler を継承して handle() を実装する部分だけになります。公式ドキュメントは共有する状態をロックで保護することを求めており、全接続で共有するデバイスメモリがそれに当たります(5.1)。
まず、実測して確定した値を定数に落とします。マジックナンバーを散らすと、後で「この 9 は何の 9 か」が分からなくなります。
SUBHEADER_REQ = 0x5000
SUBHEADER_RES = 0xD000
HEADER_LEN = 9 # subheader 2 + route 5 + request data length 2
CMD_BATCH_READ = 0x0401
CMD_BATCH_WRITE = 0x1401
SUB_WORD = 0x0000
SUB_BIT = 0x0001
END_OK = 0x0000
END_RANGE = 0xC056 # out of device range
END_COMMAND = 0xC059 # unsupported command / subcommand
END_REQUEST = 0xC05C # malformed request (e.g. bit access to a word device)
DEVICE_CODES = {0xA8: "D", 0x90: "M"}
WORD_DEVICES = {"D"}
BIT_DEVICES = {"M"}
DEVICE_POINTS = {"D": 8192, "M": 8192}
HEADER_LEN = 9 が、3.4 と図 1 で確認した「長さが固定の先頭 9 バイト」です。
4.2 受信は 2 段階で行う
TCP はバイトの流れで、メッセージの区切りを持ちません。「送信側が 21 バイトを 1 回で送ったから、受信側も 1 回で受け取れる」保証はどこにもありません。Python 公式のソケットプログラミング HOWTO も、必要な分が揃うまで recv を繰り返し呼ぶのは呼び出し側の責任だと明記しています。受け取り方は 2 段階です。
これをそのままコードにしたのが handle() です。
class Slmp3EHandler(socketserver.BaseRequestHandler):
"""One connection. Reads the 9-byte header, then the declared body."""
def handle(self) -> None:
while True:
header = self._recv_exact(HEADER_LEN)
if header is None:
return
subheader = int.from_bytes(header[0:2], "big")
length = int.from_bytes(header[7:9], "little")
body = self._recv_exact(length)
if body is None:
return
LOG.info("RECV %s", (header + body).hex(" "))
if subheader != SUBHEADER_REQ:
continue # not a 3E request frame: drop it
try:
end_code, data = self._dispatch(body[2:]) # body[0:2] is the monitoring timer
except SlmpError as err:
end_code, data = err.end_code, b""
response = self._build_response(header[2:7], end_code, data)
LOG.info("SEND %s", response.hex(" "))
self.request.sendall(response)
ポイントは 3 つあります。
while Trueで回し、1 本の接続で何度でも要求を受け付けます。クライアントは接続を張ったまま読み書きするためです- サブヘッダが
5000hでなければcontinueで捨てます。3E の要求でないものに応答しないためです body[2:]は先頭 2 バイトの監視タイマを読み飛ばす指定です。ダミーサーバーは待たされないので使いません
短読みを吸収するのが _recv_exact() です。9 行しかありませんが、この記事でいちばん実務的な部分です。
def _recv_exact(self, size: int) -> bytes | None:
"""recv() may return short reads, so loop until size bytes arrive."""
buffer = bytearray()
while len(buffer) < size:
chunk = self.request.recv(size - len(buffer))
if not chunk:
return None
buffer += chunk
return bytes(buffer)
recv() が空のバイト列を返したら、相手が接続を閉じた合図です。ここで「例外が飛ぶはず」と思い込むと、クライアントの正常終了でサーバーが無限ループします。
4.3 応答の組み立ては数行で終わる
応答は、経路のエコーバックと終了コードとデータを、長さを付けて並べるだけです。
@staticmethod
def _build_response(route: bytes, end_code: int, data: bytes) -> bytes:
payload = struct.pack("<H", end_code) + data
return (SUBHEADER_RES.to_bytes(2, "big") + route
+ struct.pack("<H", len(payload)) + payload)
len(payload) が応答データ長です。終了コードもデータも payload に入るので、「終了コード以降のバイト数」という定義を自動的に満たします。長さを別に数えると、エラー応答(データ 0 バイト)で 2 を足し忘れます。
5. デバイスメモリと 0401 / 1401 の処理
5.1 デバイスメモリはリストと辞書で足りる
PLC のデバイスは、上位から見れば「番号で引ける値の並び」です。D は 16 ビットのワード、M は 0 / 1 のビットで、それぞれ 8,192 点のリストで足ります。
class SlmpError(Exception):
"""Carries an SLMP end code back to the response builder."""
def __init__(self, end_code: int) -> None:
super().__init__(f"end code 0x{end_code:04X}")
self.end_code = end_code
class DeviceMemory:
"""In-memory device area. Words are 16-bit unsigned, bits are 0/1."""
def __init__(self) -> None:
self._lock = threading.Lock()
self.words = {name: [0] * DEVICE_POINTS[name] for name in WORD_DEVICES}
self.bits = {name: [0] * DEVICE_POINTS[name] for name in BIT_DEVICES}
def _area(self, name: str, head: int, count: int) -> list[int]:
area = self.words.get(name) or self.bits.get(name)
if area is None or head < 0 or count < 1 or head + count > len(area):
raise SlmpError(END_RANGE)
return area
def read(self, name: str, head: int, count: int) -> list[int]:
with self._lock:
return self._area(name, head, count)[head:head + count]
def write(self, name: str, head: int, values: list[int]) -> None:
with self._lock:
area = self._area(name, head, len(values))
area[head:head + len(values)] = values
設計上のポイントは 2 つです。
- 範囲チェックは
_area()の 1 か所に閉じ込め、違反したらSlmpErrorを送出します。終了コードを運ぶのはこの例外です。読出しと書込みで別々にifを書くと、片方だけ直して食い違います - ロックで保護します。接続ごとにスレッドが立つので、同じ領域への同時アクセスが起こり得ます
点数を 8,192 にしたのは範囲外アクセスのテスト(7 章)を書くためで、特定機種の点数に合わせたものではありません。
5.2 コマンドの振り分け
監視タイマを除いた要求データは、先頭 4 バイトがコマンドとサブコマンドです。続く 3 バイトが先頭デバイス番号、1 バイトがデバイスコード、2 バイトが点数になります。書込みなら後ろにデータが続きます。
def _dispatch(self, request: bytes) -> tuple[int, bytes]:
if len(request) < 4: # too short to even hold a command: answer, do not crash
raise SlmpError(END_REQUEST)
command, subcommand = struct.unpack("<HH", request[0:4])
if command not in (CMD_BATCH_READ, CMD_BATCH_WRITE) or subcommand not in (SUB_WORD, SUB_BIT):
raise SlmpError(END_COMMAND)
if len(request) < 10:
raise SlmpError(END_REQUEST)
head = int.from_bytes(request[4:7], "little")
name = DEVICE_CODES.get(request[7])
count = int.from_bytes(request[8:10], "little")
if name is None or (subcommand == SUB_BIT and name in WORD_DEVICES):
raise SlmpError(END_REQUEST)
memory = self.server.memory
if command == CMD_BATCH_READ:
return END_OK, self._read(memory, name, head, count, subcommand)
self._write(memory, name, head, count, subcommand, request[10:])
return END_OK, b""
冒頭の 2 行はコマンドを取り出す前の長さチェックです。これが無いと、短すぎる要求で struct.error が飛んで接続ごと切れます(実測は 7 章)。
struct.unpack("<H", ...) の < がリトルエンディアン指定です。ここを > にするとコマンド 0401h が 0104h に化け、「未対応コマンド」を返し続けます。いちばん時間を溶かしやすい 1 文字です。
未対応コマンドは C059h、ワードデバイスへのビット単位アクセスは C05Ch を返します。異常系を「無視」や「接続切断」で済ませないことが、テスト相手としての価値を決めます。切断すると、クライアント側は通信エラーとしか受け取れません。
5.3 ビットは 1 バイトに 2 点入る
ビット単位(サブコマンド 0001h)のデータ部は、1 バイトに 2 点が詰め込まれます。実測では M0 から 8 点に [1, 0, 1, 1, 0, 0, 0, 1] を書いたデータ部が 10 11 00 01 の 4 バイトでした。偶数番目のビットが上位ニブル(1 バイトの上位 4 ビット。ここでは bit4)、奇数番目が bit0 に入ります。
def pack_bits(values: list[int]) -> bytes:
"""Two points per byte: even index in the upper nibble, odd in bit 0."""
packed = bytearray((len(values) + 1) // 2)
for i, value in enumerate(values):
packed[i // 2] |= (value & 1) << (4 if i % 2 == 0 else 0)
return bytes(packed)
def unpack_bits(data: bytes, count: int) -> list[int]:
return [(data[i // 2] >> (4 if i % 2 == 0 else 0)) & 1 for i in range(count)]
もうひとつ、ビットデバイスをワード単位で読む経路があります。M0 をワード 1 点で読むと 16 点が 1 ワードに詰められ、実測では 8d 00= 10 進 141(0b10001101)が返りました。いちばん若いデバイス番号が最下位ビットです。
@staticmethod
def _read(memory, name: str, head: int, count: int, subcommand: int) -> bytes:
if subcommand == SUB_BIT:
return pack_bits(memory.read(name, head, count))
if name in BIT_DEVICES: # word-unit access to a bit device: 16 bits per word
bits = memory.read(name, head * 16, count * 16)
words = [sum(bits[w * 16 + b] << b for b in range(16)) for w in range(count)]
else:
words = memory.read(name, head, count)
return b"".join(struct.pack("<H", word) for word in words)
@staticmethod
def _write(memory, name: str, head: int, count: int, subcommand: int, data: bytes) -> None:
needed = (count + 1) // 2 if subcommand == SUB_BIT else count * 2
if len(data) < needed: # the frame promised more points than it carries
raise SlmpError(END_REQUEST)
if subcommand == SUB_BIT:
memory.write(name, head, unpack_bits(data, count))
return
words = [int.from_bytes(data[i * 2:i * 2 + 2], "little") for i in range(count)]
if name in BIT_DEVICES:
bits = [(word >> b) & 1 for word in words for b in range(16)]
memory.write(name, head * 16, bits)
else:
memory.write(name, head, words)
_write() の冒頭 3 行は点数分のデータが本当に入っているかの検算です。「8 点書く」と宣言してデータが空のフレームを送られたとき、検算が無いと unpack_bits() が IndexError で落ち、接続ごと切れます(実測は 7 章)。
ここまでで、0401 と 1401 が D と M の両方に対して動きます。
6. pymcprotocol から往復させる(実測ログ)
6.1 起動してつなぐ
サーバーはファイル 1 つで起動します。ポート 5007 に特別な意味はなく、筆者が空きポートから選んだ番号です。
python slmp_dummy_server.py --host 127.0.0.1 --port 5007
起動したサーバーは待ち受けたまま止まりません。クライアントは別のターミナルで動かします。書き方は、接続先がループバックになる以外は実機に向けるときと同じです。
import pymcprotocol
plc = pymcprotocol.Type3E(plctype="Q")
plc.connect("127.0.0.1", 5007)
plc.batchwrite_wordunits(headdevice="D100", values=[1234, -1, 32767])
print(plc.batchread_wordunits(headdevice="D100", readsize=3))
plc.batchwrite_bitunits(headdevice="M0", values=[1, 0, 1, 1, 0, 0, 0, 1])
print(plc.batchread_bitunits(headdevice="M0", readsize=8))
plc.close()
plctype は、pymcprotocol がデバイス点数の上限などを判定するための指定です。本サーバーはこの値を見ないので "Q" のままで往復します。"iQ-R" のときだけ電文の形が変わり、C059h を返します(3.1)。
以下は実測時の出力を時系列で並べたものです。RECV / SEND の 16 進はサーバー側、RESULT と MATCH はクライアント側の表示になります。MATCH は実測に使ったスクリプトの一致判定で、上のコードには入れていません。
### 1. batchwrite_wordunits D100 <- [1234, -1, 32767]
RECV 50 00 00 ff ff 03 00 12 00 04 00 01 14 00 00 64 00 00 a8 03 00 d2 04 ff ff ff 7f
SEND d0 00 00 ff ff 03 00 02 00 00 00
### 2. batchread_wordunits D100 x3
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 64 00 00 a8 03 00
SEND d0 00 00 ff ff 03 00 08 00 00 00 d2 04 ff ff ff 7f
RESULT read back = [1234, -1, 32767]
MATCH = True
書いた値がそのまま読み戻りました(MATCH = True)。クライアント側のコードが「デバイス名の書式・点数・戻り値の並び」まで正しいと確認できます。
6.2 バイト列で見ると型変換が分かる
書込み要求のデータ部は d2 04 ff ff ff 7f の 6 バイトでした。3 つの値がどう化けているかを並べます。
| Python 側の値 | ワイヤ上のバイト | 読み方 |
|---|---|---|
1234 | d2 04 | 04D2h = 1234。リトルエンディアン |
-1 | ff ff | 16 ビットの 2 の補数。符号なしで読めば 65535 |
32767 | ff 7f | 7FFFh = 32767(符号付き 16 ビットの上限) |
ここが実機に触る前に潰しておきたい典型例です。サーバー側は符号なし 16 ビットで保持し、pymcprotocol は符号付きとして -1 を返します。「PLC 側の 65535 が Python では -1 になる」という食い違いを、実機なしで再現できます。
書込み要求の要求データ長は 12 00= 18 バイト(監視タイマ 2 + コマンド以降 10 + データ 6)、応答は終了コードだけなので 02 00 です。
6.3 ビットの往復と、ワードでまとめて読む経路
M リレーのビット単位も往復しました。
### 3. batchwrite_bitunits M0 <- [1,0,1,1,0,0,0,1]
RECV 50 00 00 ff ff 03 00 10 00 04 00 01 14 01 00 00 00 00 90 08 00 10 11 00 01
SEND d0 00 00 ff ff 03 00 02 00 00 00
### 4. batchread_bitunits M0 x8
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 01 00 00 00 00 90 08 00
SEND d0 00 00 ff ff 03 00 06 00 00 00 10 11 00 01
RESULT read back = [1, 0, 1, 1, 0, 0, 0, 1]
MATCH = True
### 5. batchread_wordunits M0 x1 (word-unit access to a bit device)
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 00 00 00 90 01 00
SEND d0 00 00 ff ff 03 00 04 00 00 00 8d 00
RESULT = [141] expected 0b10001101 = 141
3 番目が 5.3 のワード単位アクセスで、実測どおり 141 が返りました。
7. 異常系 — 終了コードを返す
ダミーサーバーの価値は、正常系よりむしろ異常系にあります。実機では狙って起こしにくい失敗を、こちらから確実に起こせるからです。
| クライアントの操作 | サーバーの判断 | 返した終了コード | クライアント側で発生した例外 |
|---|---|---|---|
batchread_wordunits("D8000", 500) | D は 8,192 点なので範囲外 | C056h | MCProtocolError(メッセージに 0xC056 が入る) |
read_cputype()(コマンド 0101h) | 未対応コマンド | C059h | UnsupportedComandError |
batchread_bitunits("D0", 4) | ワードデバイスへのビット単位アクセス | C05Ch | MCProtocolError |
実測ログは次のとおりです。応答はいずれも「サブヘッダ+経路+応答データ長 02 00+終了コード」の 11 バイトで、データ部はありません。
### 6. out-of-range: batchread_wordunits D8000 x500
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 40 1f 00 a8 f4 01
SEND d0 00 00 ff ff 03 00 02 00 56 c0
RESULT MCProtocolError => mc protocol error: error code 0xC056
### 7. unsupported command: read_cputype (0x0101)
RECV 50 00 00 ff ff 03 00 06 00 04 00 01 01 00 00
SEND d0 00 00 ff ff 03 00 02 00 59 c0
RESULT UnsupportedComandError => This command is not supported by the module you connected.If you connect with CPU module, please use E71 module.
### 8. bit-unit access to a word device: batchread_bitunits D0 x4
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 01 00 00 00 00 a8 04 00
SEND d0 00 00 ff ff 03 00 02 00 5c c0
RESULT MCProtocolError => mc protocol error: error code 0xC05C
### 9. still alive after 3 error responses: batchread_wordunits D100 x3
RECV 50 00 00 ff ff 03 00 0c 00 04 00 01 04 00 00 64 00 00 a8 03 00
SEND d0 00 00 ff ff 03 00 08 00 00 00 d2 04 ff ff ff 7f
RESULT = [1234, -1, 32767]
押さえておきたい点が 3 つあります。
- 終了コードもリトルエンディアンで載ります。C056h はワイヤ上
56 c0です。ログを目で追うときは読み替えが要ります - エラーを返しても接続は切れません。C056 / C059 / C05C を 3 連続で返した直後に、同じ接続で D100 の読出しが通りました(項目 9)。エラー時に
close()する実装だと、クライアント側では通信障害と区別できなくなります - 例外クラス名の綴りに注意。
pymcprotocol0.3.0 の未対応コマンド例外はUnsupportedComandErrorです(Commandではありません)。ライブラリ側の表記がこうなので、except節もこの綴りで書きます
もうひとつ、フレームが宣言した長さと中身が食い違う要求も異常系です。生のソケットから 2 種類を送り、どちらも C05C が返りました。回帰テストは 8 章の test_truncated_request_gets_an_end_code と test_write_with_missing_data_gets_an_end_code です。
- コマンドが入りきらないほど短い要求(要求本体 2 バイト)。初版は長さチェックより先に
struct.unpackを呼び、struct.errorがSlmpErrorに捕まらず接続が切れました。5.2 の冒頭 2 行が塞いだ箇所です - 「8 点書く」と宣言してデータが 0 バイトの書込み要求。
unpack_bits()がIndexErrorで落ち、同じく接続が切れました。5.3 の_write()冒頭 3 行が塞いだ箇所です
どちらも「異常系を接続切断で済ませない」という主張に実装が追いついていなかった箇所です。塞いだあとに往復を測り直し、D / M の往復一致とバイト列が変わらないことを確認しています。
言えるのは「クライアント側の異常系分岐が動くこと」までです。実機の終了コードとの一致は確認していません(13 章)。
8. pytest に載せる(8 テスト)
ここまでの往復を pytest に載せると、実機なしで回る回帰テストになります。テストは 96 行です。
"""pytest: pymcprotocol client <-> self-made SLMP dummy server, over 127.0.0.1."""
import socket
import threading
import time
import pymcprotocol
import pytest
from slmp_dummy_server import SlmpDummyServer
@pytest.fixture()
def server():
srv = SlmpDummyServer("127.0.0.1", 0) # port 0 = let the OS pick a free port
threading.Thread(target=srv.serve_forever, daemon=True).start()
yield srv
srv.shutdown()
srv.server_close()
@pytest.fixture()
def plc(server):
client = pymcprotocol.Type3E(plctype="Q")
client.connect("127.0.0.1", server.server_address[1])
yield client
client.close()
def test_word_roundtrip(plc):
plc.batchwrite_wordunits(headdevice="D100", values=[1234, -1, 32767])
assert plc.batchread_wordunits(headdevice="D100", readsize=3) == [1234, -1, 32767]
def test_bit_roundtrip(plc):
bits = [1, 0, 1, 1, 0, 0, 0, 1]
plc.batchwrite_bitunits(headdevice="M0", values=bits)
assert plc.batchread_bitunits(headdevice="M0", readsize=8) == bits
def test_bit_device_read_as_word(plc):
plc.batchwrite_bitunits(headdevice="M0", values=[1, 0, 1, 1, 0, 0, 0, 1])
assert plc.batchread_wordunits(headdevice="M0", readsize=1) == [0b10001101]
def test_out_of_range_returns_end_code(plc):
with pytest.raises(Exception) as excinfo:
plc.batchread_wordunits(headdevice="D8000", readsize=500)
assert "C056" in str(excinfo.value).upper()
def test_unsupported_command_returns_end_code(plc):
with pytest.raises(Exception):
plc.read_cputype()
# the connection survives the error response
assert plc.batchread_wordunits(headdevice="D0", readsize=1) == [0]
def test_fragmented_frame_is_reassembled(server):
"""Send one request 1 byte at a time: the server must still answer correctly."""
request = bytes.fromhex("5000 00ff ff03 00 0c00 0400 0104 0000 000000 a8 0100".replace(" ", ""))
sock = socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=5)
try:
for i in range(len(request)):
sock.sendall(request[i:i + 1])
time.sleep(0.002)
response = sock.recv(4096)
finally:
sock.close()
assert response[0:2] == b"\xd0\x00"
assert response[9:11] == b"\x00\x00" # end code 0000
def test_truncated_request_gets_an_end_code(server):
"""A body too short to hold a command must be answered, not crash the connection."""
# 9-byte header declaring a 4-byte body = monitoring timer 2 + only 2 bytes of request
request = bytes.fromhex("5000 00 ff ff03 00 0400 0100 0104".replace(" ", ""))
sock = socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=5)
try:
sock.sendall(request)
response = sock.recv(4096)
finally:
sock.close()
assert response[9:11] == b"\x5c\xc0" # end code C05C
def test_write_with_missing_data_gets_an_end_code(server):
"""A write frame that promises 8 bit points but carries no data must not crash."""
# sub route len timer cmd sub dev num code points
request = bytes.fromhex("5000 00 ff ff03 00 0c00 0100 0114 0100 000000 90 0800".replace(" ", ""))
sock = socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=5)
try:
sock.sendall(request)
response = sock.recv(4096)
finally:
sock.close()
assert response[9:11] == b"\x5c\xc0" # end code C05C
実行結果です。
platform win32 -- Python 3.12.10, pytest-9.1.1, pluggy-1.6.0
collected 8 items
test_slmp_dummy_server.py::test_word_roundtrip PASSED [ 12%]
test_slmp_dummy_server.py::test_bit_roundtrip PASSED [ 25%]
test_slmp_dummy_server.py::test_bit_device_read_as_word PASSED [ 37%]
test_slmp_dummy_server.py::test_out_of_range_returns_end_code PASSED [ 50%]
test_slmp_dummy_server.py::test_unsupported_command_returns_end_code PASSED [ 62%]
test_slmp_dummy_server.py::test_fragmented_frame_is_reassembled PASSED [ 75%]
test_slmp_dummy_server.py::test_truncated_request_gets_an_end_code PASSED [ 87%]
test_slmp_dummy_server.py::test_write_with_missing_data_gets_an_end_code PASSED [100%]
============================== 8 passed in 4.11s ==============================
効いた工夫が 3 つあります。
- ポート 0 を指定する。OS が空きポートを割り当て、
server.server_address[1]で番号を取れます。固定ポートだと、前回のテストが残した接続がTIME_WAIT(切断直後にポートが一定時間解放されない OS の状態)でポートを掴み、2 回目が落ちることがあります(ポート 0 で 8 テスト連続 PASS) - fixture(テストの前後で下準備と後始末をする pytest の仕組み)を 2 段にする。サーバーを立てる
serverと接続済みクライアントを渡すplcに分ければ、生のソケットで殴るテスト(分割送信・長さの食い違う要求)だけserverを直接受け取れます - 後始末は
shutdown()→server_close()の順。serve_forever()はブロッキングなのでdaemon=Trueのスレッドに逃がします
CI に載せる構成や pytest 自体の書き方は pytest 入門 — 現場アプリのための単体テスト設計で扱います。
注意がひとつ。このテストが保証するのは「自作サーバーとクライアントコードの辻褄が合っていること」だけです。解釈が実機と違えば、テストは緑のまま実機で落ちます(13 章)。実機に繋ぐ最初の 1 回は、必ず読出しだけ・停止中の設備で行ってください。ただし「停止中」=安全ではありません。実機に向ける前に実機の PLC に向けるときの安全ルールを必ずお読みください。
9. つまずきポイント — 短読み・エンディアン・ビットのパッキング
9.1 recv() の短読みはローカルでは再現しにくい
_recv_exact() を書かず self.request.recv(4096) の 1 回で済ませても、ローカルで試している間はまず動きます。実測でも 21 バイトのフレームは毎回 1 回の recv() で届き、短読みは 1 バイトずつ送る意図的な分割でしか再現しませんでした。素朴な単発 recv() のサーバーに向けた結果です。
naive single-recv server, frame sent 1 byte at a time
naive server: recv() returned 1 bytes (frame is 21 bytes)
naive server: declared data length = 0, actually available = -8
naive server: parsed command = 0x0000
client: send aborted after 2 of 21 bytes -> ConnectionAbortedError: 10053
1 バイトしか受け取れなかったサーバーは、要求データ長を 0、コマンドを 0x0000 と誤読しました。応答を返そうとして接続が壊れています。ConnectionAbortedError: 10053 は Windows のループバックでの観測値で、他の OS で同じ番号になるとは限りません。多点の一括読出しやネットワーク機器を数段挟む条件でも同じ取りこぼしは起こり得ます(未検証)。
対策は 2 つセットです。受信側は必要な分が揃うまで繰り返すループで書き、そのうえで分割して送りつけるテストを 1 本用意します(test_fragmented_frame_is_reassembled)。
9.2 エンディアンは「サブヘッダだけ例外」と覚える
2 バイト以上の値はリトルエンディアンで送出されます。コマンド 0401h はワイヤ上 01 04、要求先ユニット I/O 番号 03FFh は ff 03。例外はサブヘッダで、5000h が 50 00 のまま並びます。実装で int.from_bytes(header[0:2], "big") とビッグエンディアン指定にしているのがその 1 行です。
公開されている解説には、表を「値」で書いたもの(0401h)と「バイト列」で書いたもの(01 04h)が混在します。手元に相手役があれば、実バイト列で決着が付きます。
9.3 ビットのパッキングは「1 バイト 2 点」
ビット単位のデータ部を「1 バイト 8 点」と思い込むと点数が合いません。実測どおり 1 バイトに 2 点で、8 点なら 4 バイト、9 点なら 5 バイトです。pack_bits() の (len(values) + 1) // 2 がこの切り上げです。
9.4 バイト列のログは hex で残す
今回の実測ログはすべて bytes.hex(sep=" ") の出力で、パケットキャプチャのツールは使っていません(sep 引数は Python 3.8 で追加)。送受信の直前直後で 16 進を吐くだけで、切り分けに必要な情報はだいたい揃います。要求と応答をペアで残せば、長さフィールドのずれや終了コードを後から目視で追えます(ログ設計はこちら)。
9.5 二重起動してもエラーになりません
本コードは allow_reuse_address = True なので、同じポートでもう 1 つ起動できてしまいます。実測です。
1st server listening on 65073, allow_reuse_address=True
2nd bind on the SAME port: SUCCEEDED -> no WinError 10048
2nd bind with allow_reuse_address=False: OSError winerror=10048
エラーが出ないので気づけません。クライアントが繋がるのは一方だけなので、「書いたはずの値が読めない」という形で症状が出ます。既定の allow_reuse_address = False なら WinError 10048 で弾かれます(実測)。Windows 11 のループバックでの観測値で、SO_REUSEADDR の意味は OS ごとに違うため一般化しません。
この True を残す以上、同じ PC の別プロセスがポートを奪えます(2 つ目の bind が実例)。127.0.0.1 に閉じ、毎回起動し直す使い捨ての検証用に限った割り切りです。常駐や --host 0.0.0.0 なら外します(外せば二重起動は WinError 10048 で弾かれます)。Windows の常駐なら SO_EXCLUSIVEADDRUSE が要ります(実測 5 パターン)。
10. 先行 OSS(VirtualMelsecR)との違い
「自作しかない」と書きましたが、正確には先行例があります。GitHub の VirtualMelsecR は、MELSEC iQ-R シリーズ向けの SLMP テスト用仮想 PLC を名乗る Python 製の OSS(MIT ライセンス)です。2026 年 9 月 4 日時点でコミット数 8・Star 0 でした。
公開されている説明によれば、3E / 4E のバイナリに対応し、デバイスは M / D / X / Y / B / W ほか多数、通信は asyncio ベースです。起動時やスキャンごとに呼ばれる関数を Python で書けば、PLC の中身の振る舞いまで模擬できる設計になっています。
| 観点 | 本記事のダミーサーバー | VirtualMelsecR |
|---|---|---|
| 狙い | クライアントを試すための最小の相手役 | 制御ロジックまで含めた仮想 PLC |
| 対応コマンド | 0401 / 1401 | 0401 / 1401 / 0403 / 1402 / 0101(公開情報より) |
| コード量 | 188 行(標準ライブラリのみ) | 本記事では計測していません |
| 当サイトでの検証 | 実測あり(本記事) | 動作は未検証 |
どちらが優れているかは述べられません。当サイトは VirtualMelsecR を動かしていないためです。対応コマンドを広げたい・iQ-R の電文を相手にしたい・内部動作まで模擬したいなら先行 OSS を、フレームの中身を自分で説明できる状態にしたいなら自作が向きます。
なお、SLMP のサーバー側(PLC 役)を自作する日本語の解説記事は、筆者が調べた範囲では見つかりませんでした。検索の仕方を変えれば出てくる可能性は残ります。
11. 完成コード(サーバー本体 188 行)
ここまでの抜粋を 1 本にまとめたものです。slmp_dummy_server.py として保存すれば、追加のインストールなしで起動できます。本コードは筆者の独自実装で、社内利用・改変を含めて自由に使っていただけますが、無保証です(利用規約 第 5 条)。実機 PLC での検証は行っていません。
"""SLMP (MC protocol 3E) binary dummy server.
Public specification only. Not affiliated with Mitsubishi Electric / CLPA.
Supported: 0401 batch read / 1401 batch write, word and bit units, D and M.
"""
from __future__ import annotations
import argparse
import logging
import socketserver
import struct
import threading
LOG = logging.getLogger("slmp")
SUBHEADER_REQ = 0x5000
SUBHEADER_RES = 0xD000
HEADER_LEN = 9 # subheader 2 + route 5 + request data length 2
CMD_BATCH_READ = 0x0401
CMD_BATCH_WRITE = 0x1401
SUB_WORD = 0x0000
SUB_BIT = 0x0001
END_OK = 0x0000
END_RANGE = 0xC056 # out of device range
END_COMMAND = 0xC059 # unsupported command / subcommand
END_REQUEST = 0xC05C # malformed request (e.g. bit access to a word device)
DEVICE_CODES = {0xA8: "D", 0x90: "M"}
WORD_DEVICES = {"D"}
BIT_DEVICES = {"M"}
DEVICE_POINTS = {"D": 8192, "M": 8192}
class SlmpError(Exception):
"""Carries an SLMP end code back to the response builder."""
def __init__(self, end_code: int) -> None:
super().__init__(f"end code 0x{end_code:04X}")
self.end_code = end_code
class DeviceMemory:
"""In-memory device area. Words are 16-bit unsigned, bits are 0/1."""
def __init__(self) -> None:
self._lock = threading.Lock()
self.words = {name: [0] * DEVICE_POINTS[name] for name in WORD_DEVICES}
self.bits = {name: [0] * DEVICE_POINTS[name] for name in BIT_DEVICES}
def _area(self, name: str, head: int, count: int) -> list[int]:
area = self.words.get(name) or self.bits.get(name)
if area is None or head < 0 or count < 1 or head + count > len(area):
raise SlmpError(END_RANGE)
return area
def read(self, name: str, head: int, count: int) -> list[int]:
with self._lock:
return self._area(name, head, count)[head:head + count]
def write(self, name: str, head: int, values: list[int]) -> None:
with self._lock:
area = self._area(name, head, len(values))
area[head:head + len(values)] = values
def pack_bits(values: list[int]) -> bytes:
"""Two points per byte: even index in the upper nibble, odd in bit 0."""
packed = bytearray((len(values) + 1) // 2)
for i, value in enumerate(values):
packed[i // 2] |= (value & 1) << (4 if i % 2 == 0 else 0)
return bytes(packed)
def unpack_bits(data: bytes, count: int) -> list[int]:
return [(data[i // 2] >> (4 if i % 2 == 0 else 0)) & 1 for i in range(count)]
class Slmp3EHandler(socketserver.BaseRequestHandler):
"""One connection. Reads the 9-byte header, then the declared body."""
def handle(self) -> None:
while True:
header = self._recv_exact(HEADER_LEN)
if header is None:
return
subheader = int.from_bytes(header[0:2], "big")
length = int.from_bytes(header[7:9], "little")
body = self._recv_exact(length)
if body is None:
return
LOG.info("RECV %s", (header + body).hex(" "))
if subheader != SUBHEADER_REQ:
continue # not a 3E request frame: drop it
try:
end_code, data = self._dispatch(body[2:]) # body[0:2] is the monitoring timer
except SlmpError as err:
end_code, data = err.end_code, b""
response = self._build_response(header[2:7], end_code, data)
LOG.info("SEND %s", response.hex(" "))
self.request.sendall(response)
def _recv_exact(self, size: int) -> bytes | None:
"""recv() may return short reads, so loop until size bytes arrive."""
buffer = bytearray()
while len(buffer) < size:
chunk = self.request.recv(size - len(buffer))
if not chunk:
return None
buffer += chunk
return bytes(buffer)
def _dispatch(self, request: bytes) -> tuple[int, bytes]:
if len(request) < 4: # too short to even hold a command: answer, do not crash
raise SlmpError(END_REQUEST)
command, subcommand = struct.unpack("<HH", request[0:4])
if command not in (CMD_BATCH_READ, CMD_BATCH_WRITE) or subcommand not in (SUB_WORD, SUB_BIT):
raise SlmpError(END_COMMAND)
if len(request) < 10:
raise SlmpError(END_REQUEST)
head = int.from_bytes(request[4:7], "little")
name = DEVICE_CODES.get(request[7])
count = int.from_bytes(request[8:10], "little")
if name is None or (subcommand == SUB_BIT and name in WORD_DEVICES):
raise SlmpError(END_REQUEST)
memory = self.server.memory
if command == CMD_BATCH_READ:
return END_OK, self._read(memory, name, head, count, subcommand)
self._write(memory, name, head, count, subcommand, request[10:])
return END_OK, b""
@staticmethod
def _read(memory, name: str, head: int, count: int, subcommand: int) -> bytes:
if subcommand == SUB_BIT:
return pack_bits(memory.read(name, head, count))
if name in BIT_DEVICES: # word-unit access to a bit device: 16 bits per word
bits = memory.read(name, head * 16, count * 16)
words = [sum(bits[w * 16 + b] << b for b in range(16)) for w in range(count)]
else:
words = memory.read(name, head, count)
return b"".join(struct.pack("<H", word) for word in words)
@staticmethod
def _write(memory, name: str, head: int, count: int, subcommand: int, data: bytes) -> None:
needed = (count + 1) // 2 if subcommand == SUB_BIT else count * 2
if len(data) < needed: # the frame promised more points than it carries
raise SlmpError(END_REQUEST)
if subcommand == SUB_BIT:
memory.write(name, head, unpack_bits(data, count))
return
words = [int.from_bytes(data[i * 2:i * 2 + 2], "little") for i in range(count)]
if name in BIT_DEVICES:
bits = [(word >> b) & 1 for word in words for b in range(16)]
memory.write(name, head * 16, bits)
else:
memory.write(name, head, words)
@staticmethod
def _build_response(route: bytes, end_code: int, data: bytes) -> bytes:
payload = struct.pack("<H", end_code) + data
return (SUBHEADER_RES.to_bytes(2, "big") + route
+ struct.pack("<H", len(payload)) + payload)
class SlmpDummyServer(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = True
def __init__(self, host: str = "127.0.0.1", port: int = 5007) -> None:
super().__init__((host, port), Slmp3EHandler)
self.memory = DeviceMemory()
def main() -> None:
parser = argparse.ArgumentParser(description="SLMP 3E binary dummy server")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=5007)
args = parser.parse_args()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
with SlmpDummyServer(args.host, args.port) as server:
LOG.info("listening on %s:%d", args.host, args.port)
server.serve_forever()
if __name__ == "__main__":
main()
起動はこれだけです。
# 起動(Ctrl+C で停止)
python slmp_dummy_server.py --host 127.0.0.1 --port 5007
127.0.0.1 に閉じておけば外部からは触れません。別 PC から繋げば、上位アプリとサーバーを分けた構成も試せます。ただし先に警告です。⚠️ このサーバーに認証機構はありません。業務ネットワークでそのまま待ち受けさせないでください。接続元を絞る・検証用セグメントに閉じるといった扱いを情報システム部門と決めてから、次のオプションを使ってください。
# 同じ LAN の別 PC から繋ぐ場合(認証なし。検証用セグメントに限ること)
python slmp_dummy_server.py --host 0.0.0.0 --port 5007
11.1 割付表どおりの値を仕込む
起動直後のデバイスは全点 0 です。2 章で挙げた「画面や帳票を先に作る」用途では、割付表の値を入れておくと進みが早くなります。以下は 1 章と同じ架空の割付(D100 に生産数、D102 に不良数、M20 に異常フラグ)を例にした書き方です。方法は 2 つあります。
- クライアント側から書く。6.1 の
batchwrite_wordunits/batchwrite_bitunitsをそのまま使います(実測済みの経路) - サーバー起動時に仕込む。
SlmpDummyServer.__init__の末尾にDeviceMemoryへの書き込みを足します(5.1 のwrite())
self.memory = DeviceMemory()
self.memory.write("D", 100, [1234, 0, 5]) # D100=生産数, D102=不良数
self.memory.write("M", 20, [1]) # M20=異常フラグ ON
生産数をカウントアップさせたいときは、クライアント側から 1 秒おきに batchwrite_wordunits を呼ぶのが手軽です。この追記とカウントアップは実測ログに含まれない応用例です(write() 自体の挙動は 5.1・6.1 で確認しています)。画面の常駐化はWindows サービス化へ。
12. よくある質問(FAQ)
このダミーサーバーで通信タイムアウトの設定値を決められますか?
決められません。ループバックの往復は中央値 0.050 ミリ秒(n=1000)で、実機とは桁が違います。タイムアウトやリトライ間隔は実機の条件で測ってください。接続できないときの切り分けは親記事にあります。
pymcprotocol 以外のクライアントでも動きますか?
3E バイナリで 0401 / 1401 を使うクライアントなら、理屈のうえでは動きます。ただし当サイトで往復を確認したのは pymcprotocol 0.3.0 だけで、pymelsec や他言語のクライアントは試していません。応答の読み取り位置はライブラリごとに違うので、初回はログでバイト列を突き合わせてください。
MELSEC iQ-R を指定(plctype="iQ-R")しても使えますか?
使えません。plctype="iQ-R" では電文の形が変わります。同じ batchread_wordunits("D0", 2) でも要求は 21 バイトから 23 バイトに増え、サブコマンドは 00 00 から 02 00、先頭デバイス番号は 4 バイト、デバイスコードは 2 バイトになります。本サーバーはこの形を受け付けず、終了コード C059h を返します(実測バイト列は 3.1)。対応させるならサブコマンドで分岐し、デバイス指定部の解析を 6 バイト側にも用意することになります。
D と M 以外のデバイスを増やすには?
DEVICE_CODES にコードを足し、ワードかビットかを WORD_DEVICES / BIT_DEVICES に登録し、点数を DEVICE_POINTS に書きます。設計上はそれで動くはずですが、D と M 以外は未実装・未検証です。追加したら往復テストを 1 本書いてください。
ダミーサーバーで動いたコードを実機に向けるとき、何を変えますか?
コード上は接続先の IP アドレスとポート番号、それに Type3E(plctype=...) の機種指定です(本サーバーは見ませんが、実機では合わせる必要があります・3.1)。作業の中心は PLC 側の設定で、手順は親記事の 5-2 節にあります。最初の 1 回は読出しだけ、停止中の設備で行ってください。接続の前に実機に向けるときの安全ルールを必ず確認してください。
終了コードの一覧はどこで確認できますか?
公式マニュアル「MELSEC コミュニケーションプロトコルリファレンスマニュアル」(資料番号 SH-080003)です。本記事の 4 つは代表例で、全リストは載せていません。実機のエラーは必ず公式マニュアルで確認してください。
13. 検証環境と、測れなかったこと
本記事の数値・ログ・バイト列は、次の環境で 2026 年 9 月 4 日に実測したものです。
| 項目 | 内容 |
|---|---|
| 実行環境 | 自宅の検証機(Windows 11 Pro)。会社支給の端末・業務ネットワークは使用していません |
| Python | 3.12.10(64bit) |
| クライアント | pymcprotocol 0.3.0(MIT ライセンス・PyPI) |
| テスト | pytest 9.1.1(8 件 PASS・4.11 秒) |
| 権限 | 管理者権限なし |
| 接続先 | 127.0.0.1 のポート 5007(pytest ではポート 0 = OS 自動割当)。外部ホストへの送信はありません |
| 実機 PLC | 所有しておらず、接続していません |
そのうえで、測れなかったことを並べます。以下は本記事の内容から判断しないでください。
- 実機 PLC での動作。終了コードの割り当ても筆者が決めた実装で、実機と一致する保証はありません
- 実機の応答時間・リトライ挙動・同時接続数の上限
- ASCII 交信(バイナリのみ)
- 4E フレーム(3E のみ)
- 0401 / 1401 以外のコマンド(0403・1402・リモート RUN / STOP など)
- D / M 以外のデバイス(X / Y / W / B / R など。コード表に足せば動く設計ですが未実装)
- iQ-R 形式の電文。実測は
C059hを返すところまでで、対応は実装していません(3.1) - 公式仕様書との突合。CC-Link 協会の SLMP 仕様書も SH-080003 も入手していません
- 先行 OSS(VirtualMelsecR)の動作。インストールも実行もしていません
- エンジニアリングツールのシミュレーション機能との比較。ツールを所有していないため未検証です
実測で言えるのはこの範囲です。188 行の Python サーバーは、pymcprotocol 0.3.0 から見て「読み書きに応答し、異常時に終了コードを返す 3E の相手」として振る舞い、往復は pytest で自動化できます。実機が届くまでの期間を、待つのではなく書いて過ごすには十分です。