serial_comm.py

"""
IMS-SDデバイス用シリアル通信モジュール
"""

import serial
import serial.tools.list_ports
from typing import List, Optional
import time


class SerialCommunication:
    """シリアル通信管理クラス"""

    def __init__(self):
        self.port: Optional[serial.Serial] = None
        self.port_name: str = ""
        self.baudrate: int = 921600  # IMS-SDのデフォルトボーレート

    def list_available_ports(self) -> List[dict]:
        """
        利用可能なCOMポートをリスト表示

        Returns:
            List[dict]: ポート情報のリスト
                - device: ポート名 (例: COM1)
                - description: ポートの説明
                - hwid: ハードウェアID
        """
        ports = serial.tools.list_ports.comports()
        port_list = []

        for port in ports:
            port_info = {
                'device': port.device,
                'description': port.description,
                'hwid': port.hwid
            }
            port_list.append(port_info)

        return port_list

    def open_port(self, port_name: str, baudrate: int = 921600) -> bool:
        """
        シリアルポートを開く

        Args:
            port_name: COMポート名 (例: "COM3")
            baudrate: ボーレート (デフォルト: 921600)

        Returns:
            bool: 成功した場合True
        """
        try:
            # 既存の接続があれば閉じる
            if self.port and self.port.is_open:
                self.close_port()

            # シリアルポート設定
            self.port = serial.Serial(
                port=port_name,
                baudrate=baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
                timeout=1.0,  # 読み込みタイムアウト
                write_timeout=1.0,  # 書き込みタイムアウト
                xonxoff=False,
                rtscts=True,  # RTS/CTSハンドシェイク有効
                dsrdtr=False
            )

            # 受信バッファサイズを大きく設定（Windows）
            try:
                if hasattr(self.port, 'set_buffer_size'):
                    self.port.set_buffer_size(rx_size=16384, tx_size=4096)
                    print(f"バッファサイズ設定: 受信16KB, 送信4KB")
            except Exception as e:
                print(f"バッファサイズ設定失敗: {e}")

            self.port_name = port_name
            self.baudrate = baudrate

            # ポートオープン後の待機（デバイス安定化）
            time.sleep(0.1)

            # バッファクリア
            self.port.flush()
            self.port.reset_input_buffer()
            self.port.reset_output_buffer()

            return True

        except serial.SerialException as e:
            print(f"ポートオープンエラー: {e}")
            return False
        except Exception as e:
            print(f"予期しないエラー: {e}")
            return False

    def close_port(self) -> bool:
        """
        シリアルポートを閉じる

        Returns:
            bool: 成功した場合True
        """
        try:
            if self.port and self.port.is_open:
                # バッファをフラッシュ
                self.port.flush()
                # ポートを閉じる
                self.port.close()
                self.port = None
                self.port_name = ""
                return True
            return False

        except Exception as e:
            print(f"ポートクローズエラー: {e}")
            return False

    def is_open(self) -> bool:
        """
        ポートが開いているか確認

        Returns:
            bool: 開いている場合True
        """
        return self.port is not None and self.port.is_open

    def write_bytes(self, data: bytes) -> bool:
        """
        バイトデータを送信

        Args:
            data: 送信するバイトデータ

        Returns:
            bool: 成功した場合True
        """
        if not self.is_open():
            return False

        try:
            self.port.write(data)
            self.port.flush()
            return True

        except serial.SerialTimeoutException:
            print("書き込みタイムアウト")
            return False
        except Exception as e:
            print(f"書き込みエラー: {e}")
            return False

    def read_bytes(self, size: int = 1) -> Optional[bytes]:
        """
        指定バイト数読み込み

        Args:
            size: 読み込むバイト数

        Returns:
            bytes: 読み込んだデータ、エラー時はNone
        """
        if not self.is_open():
            return None

        try:
            data = self.port.read(size)
            return data if data else None

        except serial.SerialTimeoutException:
            return None
        except Exception as e:
            print(f"読み込みエラー: {e}")
            return None

    def read_available(self) -> Optional[bytes]:
        """
        受信バッファにあるデータをすべて読み込み

        Returns:
            bytes: 読み込んだデータ、エラー時はNone
        """
        if not self.is_open():
            return None

        try:
            if self.port.in_waiting > 0:
                return self.port.read(self.port.in_waiting)
            return b''

        except Exception as e:
            print(f"読み込みエラー: {e}")
            return None

    def get_bytes_in_buffer(self) -> int:
        """
        受信バッファのバイト数を取得

        Returns:
            int: バッファ内のバイト数
        """
        if not self.is_open():
            return 0

        try:
            return self.port.in_waiting
        except:
            return 0

    def clear_buffers(self):
        """入出力バッファをクリア"""
        if self.is_open():
            try:
                self.port.reset_input_buffer()
                self.port.reset_output_buffer()
            except:
                pass

    def set_timeout(self, timeout: float):
        """
        タイムアウト時間を設定

        Args:
            timeout: タイムアウト時間（秒）
        """
        if self.is_open():
            self.port.timeout = timeout


# テスト用コード
if __name__ == "__main__":
    comm = SerialCommunication()

    # 利用可能なポートを表示
    print("利用可能なCOMポート:")
    ports = comm.list_available_ports()
    for port in ports:
        print(f"  {port['device']}: {port['description']}")

    # ポートを開く例
    # if comm.open_port("COM3"):
    #     print("ポートを開きました")
    #     # 処理...
    #     comm.close_port()
    #     print("ポートを閉じました")