1. はじめに — なぜ TCP/IP の自前実装が必要なのか
産業機器との通信は、Modbus TCP や OPC UA のような標準プロトコルが使えれば話は早く、専用ライブラリで済ませられます。しかしながら、現場では「メーカー独自のバイナリプロトコル」「機器ベンダーの仕様書（PDF）通りに socket でバイト列を組み立てる必要がある」という案件が、いまだに数多く存在します。
たとえば、画像検査装置・特殊計測器・受託開発の制御装置・メーカー独自プロトコルの専用機など、汎用プロトコルを実装していない機器は珍しくありません。こうした機器との通信は、Python の標準 socket モジュールで自前実装するのが最も汎用性が高い解になります。
本記事は、最小サンプルから入って、現場で 24 時間運用するレベルまで段階的に積み上げる構成です。すべて自宅環境（Python の socket ベースの簡易シミュレータ）で再現確認できます。
2. TCP/IP 通信の基本構造
TCP/IP 通信は、サーバー側（受信待ち）とクライアント側（接続要求）の役割分担で動きます。産業機器との通信では、機器がサーバーになるパターンと、機器がクライアントになるパターンの両方があります。
2.1 サーバーとクライアントの役割
- サーバー側:
bindでアドレスを確保 →listenで待機 →acceptで接続受付 →recv / sendでやり取り - クライアント側:
connectで接続 →send / recvでやり取り →close
機器が「常時待ち受けるサーバー」である場合は Python 側がクライアントとして接続しに行きます。一方、機器側がイベント発生時に通知を「送信してくる」場合は Python 側がサーバーとして待ち受ける必要があります。仕様書を読む際は、まずどちら側が bind するのかを確認してください。なお、Python 側がサーバーとして待ち受ける構成を会社 PC でやる場合、Windows ファイアウォールの受信許可が必要です。管理者権限のない PC では自分で許可を追加できないため、情シスへの依頼を見込んでおいてください。
2.2 最小実装（サーバーとクライアント）
まずは動かすことを優先した最小実装です。シミュレータとして、サーバー側スクリプトを 1 つ用意しておけば、クライアントの実装テストは PC 単体で完結します。
# server_simulator.py — 自宅 PC で動かす機器シミュレータ
import socket
HOST, PORT = "127.0.0.1", 9000
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((HOST, PORT))
s.listen()
print(f"listening on {HOST}:{PORT}")
conn, addr = s.accept()
with conn:
print(f"connected: {addr}")
while True:
data = conn.recv(1024)
if not data:
break
print(f"recv: {data!r}")
conn.sendall(b"ACK\n")# client.py — Python から機器に接続する側
import socket
HOST, PORT = "127.0.0.1", 9000
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((HOST, PORT))
s.sendall(b"PING\n")
response = s.recv(1024)
print(f"response: {response!r}")動かし方: PowerShell（またはコマンドプロンプト）を 2 つ開き、1 つ目で python server_simulator.py を実行（listening on ... と表示されて待機状態になります）、2 つ目で python client.py を実行します。クライアント側に response: b'ACK\n' と表示されれば成功です。なお、この最小サーバーは 1 回の接続で終了します。もう一度試すときは、サーバー側も再実行してください。
この時点で、Python での TCP/IP 通信は「動く」状態になります。問題はこの先です。現場ではこのコードのままでは、ほぼ確実に運用に耐えません。
3. 現場でハマる 5 つの落とし穴
本セクションでは、業務で実際に遭遇する代表的な失敗パターンを 5 つに整理します。いずれも「最小サンプルでは見えないが、長時間運用で必ず顕在化する」種類の問題です。
落とし穴 1: パケット境界が保証されない
TCP はストリームプロトコルです。送信側で sendall(b"FRAME1") / sendall(b"FRAME2") と 2 回送っても、受信側に b"FRAME1FRAME2" として 1 回でまとめて届く（あるいは逆に、1 つのフレームが分割して届く）可能性があります。「1 回の send = 1 回の recv」が成立すると思って実装すると、必ず破綻します。
対策: フレーム境界をプロトコル側で明示する。代表的な方式は次の 2 つです。
- 区切り文字方式: 改行
\nや独自バイトでフレーム終端を表す（テキスト系プロトコルで多用） - 長さプレフィックス方式: 先頭に「以降のデータ長」を入れる（バイナリ系プロトコルの定石）。本記事では 4 バイトの big-endian（上位バイトから先に並べるバイト順。PLC の Modbus レジスタでおなじみの「バイト順問題」と同じ話）で長さを表します
落とし穴 2: recv は要求バイト数を返してくれない
recv(1024) は「最大 1024 バイト読む」のであって「ちょうど 1024 バイト返す」わけではありません。実際には、1 バイトしか返ってこない場合もあります。
対策: 必要なバイト数が揃うまでループで読む受信ヘルパーを用意します。
def recv_exact(sock: socket.socket, n: int) -> bytes:
"""正確に n バイト受信するまでブロックする。途中切断は例外で通知。"""
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise ConnectionError("connection closed during recv")
buf.extend(chunk)
return bytes(buf)
def recv_frame(sock: socket.socket) -> bytes:
"""長さプレフィックス（4 バイト big-endian）方式でフレームを受信する。"""
header = recv_exact(sock, 4)
length = int.from_bytes(header, "big")
return recv_exact(sock, length)この recv_exact / recv_frame をすべての受信箇所で使うことで、パケット境界とバイト数の問題が同時に解決します。
落とし穴 3: タイムアウトを設定しないと永遠にブロックする
デフォルトの socket はブロッキングモードです。機器が応答しないまま無音になった場合、recv が永遠にブロックし、監視アプリが「ハングしているのか動いているのか」分からなくなります。
対策: 接続直後に settimeout を必ず指定します。タイムアウト値は機器仕様の応答時間 + マージン（一般に 2〜5 倍）が目安です。
# ※断片コードです（sock・logger は定義済みの前提。完全版は 4 章の DeviceClient に統合）
sock.settimeout(5.0) # 5 秒で socket.timeout 例外
try:
data = recv_frame(sock)
except socket.timeout:
# Python 3.10+ では socket.timeout は TimeoutError の別名（どちらでも捕捉できる）
# タイムアウトを業務ログに記録し、必要なら再送・再接続へ
logger.warning("recv timeout, will reconnect")
raise落とし穴 4: 切断検知が遅い・できない
ケーブルが抜けた、ハブの電源が落ちた、相手機器が再起動した——こうした物理切断は、TCP の仕様上「すぐには検知されない」ことがあります。TCP Keep-Alive はデフォルトで無効なので、お互いに何も送らなければ切断は検知されないままになり得ます。Keep-Alive を有効化しても、既定ではアイドル 2 時間後にようやく検知が始まります。
対策: 「アプリ層での Keep-Alive（定期 PING）」または「TCP Keep-Alive のチューニング」のいずれか、可能なら両方を入れます。
def enable_tcp_keepalive(sock: socket.socket,
idle: int = 30,
interval: int = 10,
count: int = 3) -> None:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
# TCP_KEEPIDLE / KEEPINTVL / KEEPCNT は Windows でも使える
# （Python 3.7+ / Windows 10 1709 以降。非対応環境では定数自体が
# 定義されないため、hasattr 判定でクロスプラットフォームに書ける）
if hasattr(socket, "TCP_KEEPIDLE"):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, idle)
if hasattr(socket, "TCP_KEEPINTVL"):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, interval)
if hasattr(socket, "TCP_KEEPCNT"):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, count)かつてこれらのオプションは Linux 系 OS 専用で、Windows では SIO_KEEPALIVE_VALS という別 API が必要でしたが、現在は上記のとおり Windows でも setsockopt() で設定できます。それでも古い Windows（10 の 1709 より前、Server 2019 より前）が残る現場や OS 依存性を気にする場合は、アプリ層で 30 秒おきに PING を送り、応答が来なければ再接続する設計のほうが移植性が高くなります。
落とし穴 5: スレッド設計を間違えると全体が止まる
受信ループと UI（tkinter など）を同じスレッドで回すと、受信が詰まった瞬間に画面が固まります。逆に、安易にスレッドを増やすと、socket への並行アクセスでデータが混ざります。
対策: 「socket 専用スレッド 1 本 + queue.Queue で UI 側に渡す」が基本形です。socket には 1 スレッドしかアクセスしない設計にします。ここでのスレッドとは、アプリの中で並行して動くもう 1 本の処理の流れのこと。queue（キュー）は、スレッド間でデータを安全に受け渡すベルトコンベアのような仕組みです。
4. 24 時間運用に耐える完全実装
落とし穴対策をまとめた、再接続・タイムアウト・フレーム境界・スレッド分離をすべて含む実装パターンを示します。これは 24 時間運用に耐える監視アプリの典型的な設計を、自宅環境向けに整理したものです。
4.1 接続マネージャ
import logging
import queue
import socket
import threading
import time
from typing import Optional
logger = logging.getLogger(__name__)
class DeviceClient:
"""産業機器との TCP 接続を管理するクライアント。
設計方針:
- 切断は例外ではなく状態として扱う（自動再接続）
- 受信は専用スレッド、データは queue 経由で外に渡す
- 外部からは start / stop / send だけ呼べばよい
"""
def __init__(self, host: str, port: int,
reconnect_interval: float = 3.0,
recv_timeout: float = 5.0):
self.host = host
self.port = port
self.reconnect_interval = reconnect_interval
self.recv_timeout = recv_timeout
self.recv_queue: queue.Queue[bytes] = queue.Queue(maxsize=1000)
self._sock: Optional[socket.socket] = None
self._stop_event = threading.Event()
self._thread: Optional[threading.Thread] = None
self._send_lock = threading.Lock()
def start(self) -> None:
self._stop_event.clear()
self._thread = threading.Thread(target=self._run, daemon=True)
self._thread.start()
def stop(self) -> None:
self._stop_event.set()
self._close_socket()
if self._thread:
self._thread.join(timeout=5)
def send(self, payload: bytes) -> None:
with self._send_lock:
sock = self._sock # 再接続スレッドとの競合に備えてローカル変数に取る
if sock is None:
raise ConnectionError("not connected")
length = len(payload).to_bytes(4, "big")
sock.sendall(length + payload)
def _run(self) -> None:
while not self._stop_event.is_set():
try:
self._connect()
self._recv_loop()
except (ConnectionError, OSError, socket.timeout) as e:
logger.warning("connection error: %s", e)
finally:
self._close_socket()
if not self._stop_event.is_set():
time.sleep(self.reconnect_interval)
def _connect(self) -> None:
sock = socket.create_connection((self.host, self.port), timeout=5.0)
sock.settimeout(self.recv_timeout)
self._sock = sock
logger.info("connected to %s:%s", self.host, self.port)
def _recv_loop(self) -> None:
while not self._stop_event.is_set():
frame = self._recv_frame()
try:
self.recv_queue.put_nowait(frame)
except queue.Full:
logger.warning("recv_queue is full, dropping frame")
def _recv_frame(self) -> bytes:
header = self._recv_exact(4)
length = int.from_bytes(header, "big")
if length > 1024 * 1024:
raise ConnectionError(f"frame too large: {length}")
return self._recv_exact(length)
def _recv_exact(self, n: int) -> bytes:
sock = self._sock # stop() との競合に備えてローカル変数に取る
if sock is None:
raise ConnectionError("not connected")
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise ConnectionError("peer closed")
buf.extend(chunk)
return bytes(buf)
def _close_socket(self) -> None:
if self._sock:
try:
self._sock.close()
except OSError:
pass
self._sock = Noneこのクラスは、外から見ると start / stop / send / recv_queue の 4 つだけで使えます。再接続・タイムアウト・フレーム境界処理・例外処理は、内部で完結しています。send() は切断中に ConnectionError / OSError を投げるため、呼び出し側で捕捉してください。
1 つ重要な設計前提があります。この実装は recv_timeout（既定 5 秒）の間に何も受信しないと切断とみなして再接続します。つまり定周期でデータを送ってくる機器向けの設計です。「異常時だけ通知してくる」ようなイベント駆動型の機器に対しては、健全な接続を 5 秒ごとに切断・再接続し続けてしまいます（同時接続数に制限のある機器では、この再接続連打が原因で接続拒否に陥る機種もあります）。イベント駆動型の機器では、recv_timeout を十分長くするか、落とし穴 4 で述べたアプリ層 PING を併用して「無通信＝異常」と判定できるようにしてください。
4.2 利用側のコード
以下は 4.1 の DeviceClient と同じファイル（例: main.py）の末尾に続けて書きます。
logging.basicConfig(level=logging.INFO)
client = DeviceClient("127.0.0.1", 9000)
client.start()
try:
while True:
try:
frame = client.recv_queue.get(timeout=1.0)
except queue.Empty:
continue
# 業務処理（パース、ログ記録、UI 更新の通知）
print(f"frame: {frame!r}")
except KeyboardInterrupt:
pass
finally:
client.stop()UI（tkinter / PyQt / Streamlit）から使う場合は、UI 側の定期コールバックで recv_queue.get_nowait() を呼ぶ設計にすると、画面側はノンブロッキングで動き続けます。
5. シミュレータでの動作確認
本記事のクライアント実装は、対応するサーバーシミュレータと組み合わせると、PC 単体で完全に再現確認できます。シミュレータは長さプレフィックス方式に合わせた送受信をするだけのシンプルな構造です。
# server_with_length_prefix.py
import socket
import struct
import time
HOST, PORT = "127.0.0.1", 9000
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((HOST, PORT))
s.listen()
while True:
conn, addr = s.accept()
with conn:
print(f"connected: {addr}")
counter = 0
try:
while True:
payload = f"DATA-{counter}".encode("utf-8")
conn.sendall(struct.pack(">I", len(payload)) + payload)
counter += 1
time.sleep(0.5)
except ConnectionError:
# BrokenPipe / ConnectionReset / ConnectionAborted をまとめて捕捉
# （Windows では ConnectionAbortedError も起こり得る）
print("disconnected")このシミュレータを起動した上で前述の DeviceClient を動かすと、0.5 秒ごとに DATA-0, DATA-1, ... が recv_queue に積まれていきます。試しにシミュレータを Ctrl+C で落とし、再起動してみてください。クライアント側がログに「connection error → reconnect」を出しながら、自動復旧することを確認できます。
実機につなぐときは
シミュレータで動きを確認できたら、実機への切り替えは HOST を機器の IP アドレスに変えるだけです（IP とポートは機器の仕様書または設定画面で確認し、まず ping で疎通を確かめてから接続します）。ただしその前に、必ず守ってほしいことが 3 つあります。
- 制御ネットワークに PC を勝手につながない: IP アドレスの重複は設備停止に直結します。IP の払い出しと接続の可否は、設備管理責任者・情シスに必ず確認を
- 最初の接続テストは計画停止・保全日に: 稼働中のラインで初接続を試さない
- 閉域ネットワーク内でのみ使う: 本記事の通信は平文・認証なしの前提です。事務 LAN やインターネットに露出する経路では使わないでください
6. 24 時間運用設計のチェックリスト
長期運用を見据えるとき、上記の実装に加えて確認すべきポイントを整理します。
- ログ:
loggingでファイル出力、ローテーション（RotatingFileHandlerもしくはTimedRotatingFileHandler）、文字コード明示（Windows ならencoding="utf-8"を必ず指定） - メモリ:
queue.Queue(maxsize=...)で上限を必ず設定。無制限にすると、消費側が止まったときにメモリリーク - 例外: 受信ループで「予想外の例外でスレッドが死ぬ」と無音故障になる。最外で
try/except Exceptionを入れて再接続フローへ流す - 監視: 「最後に受信した時刻」を変数に記録しておき、別スレッド（または UI の定期処理）で
time.time() - last_recv_time > 60のような判定をしてアラートを出す。UI と組み合わせた実装例は tkinter 監視アプリの記事 を参照 - 停止:
KeyboardInterruptやサービス停止シグナルで、open している socket を確実に閉じる（finally句必須） - 時刻: 受信ログには「受信時刻」と「フレームの送信時刻（ペイロードに含まれているなら）」の両方を残す。遅延解析の必須情報
7. パフォーマンスの考え方
「Python の socket は遅いのでは」という質問をよく受けますが、産業機器との通信（毎秒数件〜数百フレーム程度）で標準 socket がボトルネックになることは、まずありません。ループバック（同じ PC 内での折り返し通信。127.0.0.1 のこと）・数十バイト程度の小さなフレームという条件なら、毎秒数千フレーム規模の送受信も標準 socket で扱える水準です（実際の上限はフレームサイズ・NIC・CPU に依存します）。
むしろパフォーマンス問題の多くは次の 3 点に起因します。
- UI スレッドで重い処理（描画、ファイル IO、pandas 集計）を同期的に実行している
- queue を経由せずに socket から直接 UI を更新している
- 不要な
printやlogger.debugが高頻度で動き、コンソールがボトルネックになっている
1 秒あたり 1 万フレーム以上扱う必要があるケースでは、asyncio ベースに切り替える、またはバッファリングしてバッチ処理するなどの設計検討に入ります。それ以下のレートでは、本記事の同期スレッド設計で十分実用になります。
8. おわりに
TCP/IP 通信は「動かす」のと「24 時間止まらず動かし続ける」の間に、見えにくいが大きなギャップがあります。本記事の DeviceClient パターン（接続マネージャ + 受信スレッド + queue + 自動再接続）は、現場で使えるサイズの最小骨格です。実プロジェクトでは、ここに「フレーム種別ごとのパーサ」「メトリクス収集」「監視アラート連携」を足していくことになります。
本サイトの監視アプリ記事と組み合わせると、socket 通信から UI 表示まで一気通貫で構築できます。