DeviceProtocol と PortDevice¶

DeviceProtocolはデバイスの標準インターフェイスを定義し、PortDeviceは実際のシリアルデバイスへの接続を管理します。

概要¶

通常はCommand クラスを使用することをオススメしますが、より低レベルのアクセスが必要な場合は、DeviceProtocolを直接使用してください。

from kazunoko import connect

# デバイスに接続
with connect() as device:
    # 低レベルアクセス
    response = device.query("STATUS")

DeviceProtocol¶

DeviceProtocolは、すべてのデバイス実装が実装する必要があるインターフェイスです。

メソッド一覧¶

`send_command(command: str) -> bool`¶

テキストコマンドをデバイスに送信します。

パラメーター:

command: 送信するテキストコマンド（例: "STATUS", "GET_VERSION"）

戻り値:

True：コマンド送信成功

発生例外:

CommandError：デバイスが接続されていない
CommandTimeout：コマンド送信のタイムアウト

device.send_command("STATUS")

`receive_response(field_type: str) -> Response`¶

デバイスからJSONL形式のレスポンスを受信してパースします。

パラメーター:

field_type: 期待されるレスポンスのタイプ（"response" または "event"）

戻り値:

Response：パース済みレスポンスオブジェクト

発生例外:

ResponseTimeout：タイムアウト内にレスポンスがない
ResponseError：応答が無効な形式

response = device.receive_response("response")
print(response.status)  # "ok"

`query(command: str) -> Response`¶

コマンド送信と応答受信を1つの操作で実行します。

通常、このメソッドを使用することをオススメします。

パラメーター:

command: 送信するテキストコマンド

戻り値:

Response：パース済みレスポンスオブジェクト

発生例外:

CommandError：コマンド送信失敗
ResponseError：応答が無効
ResponseTimeout：応答タイムアウト

response = device.query("STATUS")
print(response.type)     # "response"
print(response.status)   # "ok"

`close() -> None`¶

デバイス接続を閉じます。

device.close()

Context Manager Protocol¶

コンテキストマネージャーをサポートしており、withステートメントで使用できます。

with device:
    response = device.query("STATUS")
# 自動的に close() が呼ばれる

PortDevice¶

PortDeviceは、シリアルポート経由でOSECHI検出器に接続するための実装です。

初期化¶

from kazunoko import PortDevice

# デフォルト設定で接続
device = PortDevice()

# カスタム設定で接続
device = PortDevice(
    port="/dev/ttyUSB0",
    baudrate=115200,
    timeout=0.1
)

パラメーター:

port (str): シリアルポート（デフォルト: "/dev/ttyUSB0"）
baudrate (int): ボーレート（デフォルト: 115200）
timeout (float): 読み込みタイムアウト秒数（デフォルト: 0.1）

発生例外:

DeviceError：接続に失敗した場合

シリアル通信仕様¶

ポート: /dev/ttyUSB0（デフォルト）
ボーレート: 115200 bps
データビット: 8
ストップビット: 1
パリティ: なし
タイムアウト: 0.1秒（デフォルト）

コマンド形式¶

テキストベースのコマンド
改行終端（\n）
UTF-8エンコーディング
例: "STATUS\n", "SET_THRESHOLD 1 300\n"

レスポンス形式¶

単一行JSONL（1行に1つのJSONオブジェクト）
フィールド: type（必須）, status（オプション）, その他のデバイス固有フィールド
例: {"type":"response","status":"ok","version":"1.10.1"}

コネクション管理¶

自動ポート検出¶

connect()関数はシリアルポートを自動検出します。

from kazunoko import connect

# 自動検出
device = connect()

# または明示的に指定
device = connect("auto")

# 特定のポートを指定
device = connect("/dev/ttyUSB0")

ポート検出優先順位¶

/dev/cu.usbserial*（macOSのUSBシリアルアダプター）
/dev/ttyUSB*（LinuxのUSBシリアルアダプター）
/dev/ttyACM*（Arduinoなど）
/dev/ttyS*（組み込みシリアルポート）
利用可能な最初のポート（フォールバック）

from kazunoko import detect_port

# 利用可能なポートを検出
port = detect_port()
print(f"Found port: {port}")

低レベルコマンド送受信の例¶

単純なクエリ¶

from kazunoko import connect

with connect() as device:
    # ステータス確認
    response = device.query("STATUS")
    print(response.model_dump())

複数ステップの操作¶

from kazunoko import connect

with connect() as device:
    # コマンド送信とレスポンス受信を分けて実行
    device.send_command("STATUS")
    response = device.receive_response("response")
    print(response.status)

データストリーム読み込み¶

from kazunoko import connect

with connect() as device:
    # 検出イベントを受け取る
    for i in range(10):
        event = device.receive_response("event")
        print(f"Event {i}: signal1={event.signal1}, adc={event.adc}")

エラーハンドリング¶

from kazunoko import connect
from kazunoko.exceptions import (
    DeviceError,
    CommandError,
    ResponseError,
    ResponseTimeout,
)

try:
    device = connect()
    response = device.query("STATUS")
except DeviceError as e:
    print(f"Device connection failed: {e}")
except CommandError as e:
    print(f"Failed to send command: {e}")
except ResponseTimeout:
    print("No response from device")
except ResponseError as e:
    print(f"Invalid response: {e}")

詳細はError Handlingを参照してください。

MockDevice との互換性¶

PortDeviceと同じインターフェイスを実装したMockDeviceでテストできます。

from kazunoko import MockDevice, Command

# テスト用のモックデバイス
device = MockDevice()

with device:
    cmd = Command(device)
    response = cmd.status()
    print(response.version)  # 実デバイスと同様に動作

詳細はMockDevice Testingを参照してください。

カスタムデバイスの実装¶

DeviceProtocolを実装すれば、カスタムデバイスクラスを作成できます。

from typing import Protocol
from kazunoko.parser import Response

class CustomDevice:
    """Custom device implementation"""

    def send_command(self, command: str) -> bool:
        # カスタム実装
        return True

    def receive_response(self, field_type: str) -> Response:
        # カスタム実装
        pass

    def query(self, command: str) -> Response:
        # カスタム実装
        pass

    def close(self) -> None:
        # クリーンアップ処理
        pass

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


# Command クラスと互換性あり
from kazunoko import Command

device = CustomDevice()
with device:
    cmd = Command(device)
    response = cmd.status()

DeviceProtocol と PortDevice¶

概要¶

DeviceProtocol¶

メソッド一覧¶

send_command(command: str) -> bool¶

receive_response(field_type: str) -> Response¶

query(command: str) -> Response¶

close() -> None¶

Context Manager Protocol¶

PortDevice¶

初期化¶

シリアル通信仕様¶

コマンド形式¶

レスポンス形式¶

コネクション管理¶

自動ポート検出¶

ポート検出優先順位¶

低レベルコマンド送受信の例¶

単純なクエリ¶

複数ステップの操作¶

データストリーム読み込み¶

エラーハンドリング¶

MockDevice との互換性¶

カスタムデバイスの実装¶

参考リンク¶

`send_command(command: str) -> bool`¶

`receive_response(field_type: str) -> Response`¶

`query(command: str) -> Response`¶

`close() -> None`¶