[TEC] Pythonによるソケット通信サンプルプログラム（hdlSocket）

# HuMANDATA LTD. hdlSocket_v11.py
import socket
import threading
import sys


class TcpClient:
    def __init__(self, host, port, bufsize, timeout):
        self.host = host
        self.port = port
        self.bufsize = bufsize
        self.timeout = timeout
        self.hdlSocket = None
        self.running = False
        self.lock = threading.Lock()

    def connect(self):
        try:
            self.hdlSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.hdlSocket.settimeout(self.timeout)
            self.hdlSocket.connect((self.host, self.port))
            self.running = True
            print("OK: Connection")
            return True
        except socket.timeout:
            print("Error: Connection timeout")
        except socket.error as e:
            print(f"Error: Connection failed ({e})")
        return False

    def recvLoop(self):
        buf = b""
        while self.running:
            try:
                data = self.hdlSocket.recv(self.bufsize)
                if not data:
                    print("Error: Connection closed")  # 0バイトデータ受信=相手側のclose
                    break
                buf += data
                lines = buf.split(b"\r")
                for line in lines[:-1]:
                    print(line.decode("utf-8", errors="replace"))  # 例外文字は？に置換
                buf = lines[-1]
            except socket.timeout:
                continue  # タイムアウトは再トライ
            except socket.error as e:
                print(f"Error: Receiving ({e})")
                break
        self.stop()

    def sendLoop(self):
        try:
            while self.running:
                data = input().encode("utf-8") + b"\r"
                with self.lock:
                    self.hdlSocket.sendall(data)
        except socket.error as e:
            print(f"Error: Sending ({e})")
        except KeyboardInterrupt:
            print("\nKeyboardInterrupt")
        finally:
            self.stop()

    def stop(self):
        if self.running:
            self.running = False
            with self.lock:
                try:
                    self.hdlSocket.shutdown(socket.SHUT_RDWR)  # 送受信を終了
                except Exception:
                    pass
                self.hdlSocket.close()
            print("Connection closed")


def main():
    hdlClient = TcpClient("192.168.100.1", 10001, 128, 2.0)  # host,port,bufsize,timeout

    if not hdlClient.connect():
        sys.exit(1)  # 接続エラーで終了

    recvThread = threading.Thread(target=hdlClient.recvLoop)  # 受信用スレッド作成、関数指定
    recvThread.start()

    hdlClient.sendLoop()
    recvThread.join()


if __name__ == "__main__":
    main()

• Pythonプログラムの実行には、あらかじめPython実行環境のインストールが必要です。
• 本プログラムは、Pythonバージョン3.11.9で作成・確認しております。バージョンが異なる場合の動作については未確認です。

以下は、windowsターミナルを用いたプログラムの実行例です。

接続先は例としてLNX-210A-W24（4CH 独立絶縁型 Wi-Fi 4-20mA 電流モニタ）を使用しております。
実行前に、あらかじめ操作するPCなどとサーバ側（LNX-210A-W24）を接続しておく必要があります。
（LNX-210A-W24の場合はWi-Fiで接続）

• windowsターミナルはMicrosoftの無料ターミナルアプリです。

1. ターミナルでhdlSocket_v11.pyを実行すると、プログラム内で指定したホスト・ポートに接続を試みます。
   ここでTCP接続が成功すれば、「OK: Connection」が返ってきます。
2. 上記の実行例では次に「CST,1」を入力し、送信しています。（「CST」はLNX-210A-W24の接続確認コマンド）
3. 次の「OK,CST,1」はサーバ側（LNX-210A-W24）から返ってきた受信データです。

この他、一例としてLNX-006A、LNX-210A-L24、LNX-211V-L24、LNX-211V-W24などのLAN経由で通信を行う製品とも同様に接続、通信の確認が可能です。接続・通信時にはそれぞれの製品、環境に応じてIPアドレスやポート、送信コマンドのなどの変更が必要になる場合があります。
詳しくは各製品ページやマニュアルをご参照ください。