以前の記事では、Modbus TCPを使ってPLCの通信を模倣する基礎的な部分を実装してきました。
ただ、SCADAを本気で開発しようと思うと、「PLCが1台だけ元気に動いている」という理想的な状況だけでは不十分なわけです。
実際の現場では、複数のPLCが動き、センサーが連動し、時にはネットワークが遅延したり機器が突然死したりするようです。。。
こうした「現場の空気感」をシミュレートするため、システムをバラバラのプロセスに分割し、それらを束ねる司令塔、Orchestratorを実装することにしました。
今回は、このOrchestratorがどうやって複数のプロセスを操り、さらには「障害」というスパイスを注入しているのか、そのあたりを整理・アウトプットしておこうと思います。
コード類はこちらに公開しています。
遊んでみたい方がいらっしゃれば使ってみてください。
ポイント1：独立した「命」を吹き込むプロセス起動
まず、各シミュレータ（PLCやデバイス）を同一プログラムの関数として動かすのではなく、完全に独立したプロセスとして起動させています。
# orchestrator.py のエッセンス
for svc in start_order:
name = svc["name"]
# 各シミュレータを別プロセスとして立ち上げる
cmd = [sys.executable] + svc["command"] + svc.get("args", [])
processes[name] = subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
logger.log(f"Launching {name}...")
subprocess.Popen を使うことで、Orchestratorが終了しても（あるいはOrchestrator自体がバグっても）、シミュレータたちは自分のサイクルを回し続けます。
この「独立性」が、後のカオス注入で生きている感じです。
ポイント2：現場を操る「対話型CLI」の実装
Orchestratorをただのランチャーで終わらせないために、input() ループを使ったCLIを搭載しました。
これがなかなか便利で、動いているシステムに対してリアルタイムに命令を飛ばせるようになっています。
# メインループでのCLI受付
while running:
cmd_input = input("orchestrator > ").strip().lower()
if not cmd_input: continue
parts = cmd_input.split()
cmd = parts[0]
if cmd in ["status", "ps"]:
show_status(start_order) # 全プロセスのPIDやREADY状態を表示
elif cmd == "chaos":
# 悪意ある「障害」を意図的に引き起こすコマンド
execute_chaos(parts[1], parts[2], logger, args=parts[3:])
Pythonの標準機能だけで、簡易的なシェルを作っている感覚ですね。
input() でコマンドを受け取り、それに応じて背後のプロセスを kill したり、Modbus経由で設定を書き換えたりしています。
ポイント3：意図的な「嫌がらせ」を実現するカオス機能
SCADA側の耐久テストをするために実装したのが chaos delay です。
これは単にプロセスを止めるのではなく、 「生きてるけど返事がめちゃくちゃ遅い」 という一番厄介な状況を作り出します。
elif subcmd == "delay":
sec = int(args[0])
# 対象PLCのシステム領域(10005)に遅延秒数を書き込む
client = ModbusTcpClient(rc["host"], port=rc["port"])
if client.connect():
client.write_register(10005, sec)
client.close()
Orchestrator自身がModbusクライアントとなって、対象PLCの「内部レジスタ」を直接書き換える。
自作シミュレータならではの、やりたい放題なデバッグ手法というわけです。
結び
Orchestratorという「司令塔」を作ったことで、単なるプログラムの集まりが、一つの「生産ライン」として命を吹き込まれたような気がします。
SCADAを開発するための土台はこれで完璧に整いました。
次は、この司令塔が操る「心臓部」、Larkパーサーを使ったラダーロジックの実装について整理・アウトプットしておこうと思います。