[TEC] Pythonによるシリアル通信サンプルプログラム（hdlSerial）

# HuMANDATA LTD. hdlSerial_v21.py
import sys
import threading
import serial
from serial import SerialException


class hdlSerialComm:
    DELIMITERS = {"CR": b"\r", "LF": b"\n", "CRLF": b"\r\n"}

    @staticmethod
    def _notify(callback, message):
        if callback:
            callback(message.replace("\r\n", "\n").replace("\r", "\n").rstrip("\n"))
        else:
            print(message)

    def __init__(self, port=None, baudrate=9600, eol_tx="CR", eol_rx="CR",
                 encoding="utf-8", timeout=1, callback=None, log_callback=None):

        self.port_num = port
        self.baudrate = int(baudrate)
        self.eol_tx = self.DELIMITERS.get(eol_tx, b"\r")
        self.eol_rx = self.DELIMITERS.get(eol_rx, b"\r")
        self.encoding = encoding
        self.timeout = timeout
        self.serial = None
        self.running = False
        self.thread = None
        self.callback = callback
        self.log_callback = log_callback

    def start(self):
        self.running = True
        self.thread = threading.Thread(target=self.recvLoop)  # 受信用スレッド作成、関数指定
        self.thread.start()

    def open(self):
        if self.port_num is None:
            hdlSerialComm._notify(self.log_callback, "Open Error: Port not specified")
            return False
        try:
            self.serial = serial.Serial(
                port=self.port_num,
                baudrate=self.baudrate,
                timeout=self.timeout
            )
            hdlSerialComm._notify(self.log_callback, "Success: Open COM, "+self.port_num)
            return True
        except SerialException as e:
            hdlSerialComm._notify(self.log_callback, "Open Error: "+str(e))
            return False

    def recvLoop(self):
        buf = b""
        while self.running and self.serial.is_open:
            try:
                if self.serial.in_waiting > 0:
                    chunk = self.serial.read(self.serial.in_waiting)  # まとめて読む
                    if not chunk:
                        continue
                    buf += chunk
                while self.eol_rx in buf:
                    line, buf = buf.split(self.eol_rx, 1)  # バッファ内のデータを1行ずつ取り出す
                    data = line.decode(self.encoding, errors="replace")
                    hdlSerialComm._notify(self.callback, data)
            except SerialException as e:
                hdlSerialComm._notify(self.log_callback, "Receive Error: "+str(e))
                break

    def sendData(self, data: str):
        if self.running and self.serial.is_open:
            try:
                self.serial.write(data.encode(self.encoding) + self.eol_tx)
            except SerialException as e:
                hdlSerialComm._notify(self.log_callback, "Send Error: "+str(e))

    def close(self):
        self.running = False
        if self.serial and self.serial.is_open:
            self.serial.close()
        if self.thread:
            self.thread.join(timeout=2)
        hdlSerialComm._notify(self.log_callback, "Port closed")


def main():
    hdlCom = hdlSerialComm(port="COM10", baudrate=9600, )
    if not hdlCom.open():
        sys.exit(1)  # 接続エラーで終了
    hdlCom.start()
    try:
        while True:
            hdlCom.sendData(input())
    except KeyboardInterrupt:
        print("\nEnd: KeyboardInterrupt")
    finally:
        hdlCom.close()


if __name__ == "__main__":
    main()

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

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

接続先は例としてUSB-045A（2CH 独立絶縁型 USB 4-20mA 電流モニタ）を使用しております。

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

1. ターミナルでhdlSerial_v21.pyを実行すると、プログラム内で指定したCOMポートへの接続を試みます。
   ここで接続が成功すれば、「Success: Open COM」が返ってきます。
2. 上記の実行例では次に「CST,1」を入力し、送信しています。（「CST」はUSB-045Aの接続確認コマンド）
3. 次の「OK,CST,1」は接続先（USB-045A）から返ってきた受信データです。

この他、一例として、USB-003、USB-010、USB-045VなどのCOMポート使用してシリアル通信を行う製品とも同様に接続、通信の確認が可能です。接続・通信時にはそれぞれの製品、環境に応じてCOM番号やシリアル通信の設定の変更が必要になる場合があります。
詳しくは各製品ページやマニュアルをご参照ください。