エラーハンドリング¶

構造化された例外階層を使用して、さまざまエラー状況を明確に区別できます。

例外の階層¶

kazunokoのすべての例外（exception）はKazunokoErrorを継承しています。

KazunokoError (base class)
├── DeviceError
├── CommandError
│   └── CommandTimeout
├── ResponseError
│   └── ResponseTimeout
└── ProtocolError

例外の種類¶

`KazunokoError`¶

kazunokoのすべての例外の基底クラスです。 kazunoko固有のすべてのエラーをキャッチするために使用します。

from kazunoko import connect
from kazunoko.exceptions import KazunokoError

try:
    device = connect()
    response = device.query("STATUS")
except KazunokoError as e:
    print(f"kazunoko error: {e}")

`DeviceError`¶

デバイスの初期化または接続時のエラーを表します。

発生状況:

シリアルポートが見つからない
ポートを開くことができない
ボーレート設定に失敗した
ポート自動検出に失敗した

from kazunoko import connect
from kazunoko.exceptions import DeviceError

try:
    device = connect("/dev/ttyUSB0")
except DeviceError as e:
    print(f"Failed to connect to device: {e}")

対応方法:

ポートの接続を確認
デバイスが接続されているか確認
別のポートを試す
detect_port()で利用可能なポートを確認

from kazunoko import detect_port, connect
from kazunoko.exceptions import DeviceError

try:
    port = detect_port()
    device = connect(port)
except DeviceError as e:
    print(f"No device found: {e}")

`CommandError`¶

コマンド送信エラーを表します。

発生状況:

デバイスが接続されていない
コマンド送信に失敗した
シリアルポートの書き込みに失敗した

from kazunoko import CommandError

try:
    response = device.query("STATUS")
except CommandError as e:
    print(f"Failed to send command: {e}")

対応方法:

デバイスの接続を再確認
デバイスを再起動してみる
ポートの接続状態を確認

`CommandTimeout`¶

コマンド送信のタイムアウトを表します。 CommandErrorのサブクラスです。

発生状況:

コマンド送信後、応答がない

from kazunoko import CommandTimeout

try:
    response = device.query("STATUS")
except CommandTimeout as e:
    print(f"Command timed out: {e}")

対応方法:

タイムアウト時間を増やす
デバイスが応答しているか確認
デバイスを再起動

from kazunoko import PortDevice

# タイムアウト時間を長くしてデバイスを接続
device = PortDevice(timeout=2.0)  # デフォルト: 0.1秒

try:
    response = device.query("STATUS")
except CommandTimeout:
    print("Device did not respond")

`ResponseError`¶

レスポンスが無効またはフォーマットが正しくないエラーを表します。

発生状況:

レスポンスがJSON形式ではない
必須フィールド（type）がない
予期したレスポンスタイプが返されなかった
UTF-8デコードエラー

from kazunoko import ResponseError

try:
    response = device.receive_response("response")
except ResponseError as e:
    print(f"Invalid response: {e}")

対応方法:

デバイスのファームウェアを確認
シリアル通信ケーブルの接続を確認
ボーレート設定を確認

`ResponseTimeout`¶

レスポンス受信のタイムアウトを表します。ResponseErrorのサブクラスです。

発生状況:

コマンド送信後、タイムアウト時間内にレスポンスがない

from kazunoko import ResponseTimeout

try:
    response = device.query("STATUS")
except ResponseTimeout as e:
    print(f"No response from device: {e}")

対応方法:

デバイスが応答しているか確認
タイムアウト時間を増やす
デバイスを再起動

from kazunoko import PortDevice, ResponseTimeout

device = PortDevice(timeout=2.0)  # タイムアウトを2秒に設定

try:
    response = device.query("STATUS")
except ResponseTimeout:
    print("Device did not respond within timeout")
    device.close()

`ProtocolError`¶

プロトコルの解析または検証エラーを表します。

発生状況:

JSONのパース失敗
フィールドの型が不正
予期しないプロトコルエラー

from kazunoko import ProtocolError

try:
    response = device.query("STATUS")
except ProtocolError as e:
    print(f"Protocol error: {e}")

エラーハンドリングのパターン¶

基本的なエラーハンドリング¶

from kazunoko import connect
from kazunoko.exceptions import KazunokoError

try:
    device = connect()
    response = device.query("STATUS")
    print(response.status)
except KazunokoError as e:
    print(f"Error: {e}")

詳細なエラーハンドリング¶

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

try:
    device = connect()
    response = device.query("STATUS")
    print(response.status)

except DeviceError as e:
    print(f"Device connection failed: {e}")
    print("Make sure the device is connected")

except CommandError as e:
    print(f"Failed to send command: {e}")
    print("Try reconnecting the device")

except ResponseTimeout:
    print("Device did not respond in time")
    print("The device may be busy or unresponsive")

except ResponseError as e:
    print(f"Received invalid response: {e}")
    print("Check the device firmware version")

リトライロジック¶

from kazunoko import connect
from kazunoko.exceptions import ResponseTimeout

MAX_RETRIES = 3

def query_with_retry(device, command: str, max_retries: int = MAX_RETRIES):
    """Execute query with automatic retry on timeout"""
    for attempt in range(max_retries):
        try:
            return device.query(command)
        except ResponseTimeout:
            if attempt < max_retries - 1:
                print(f"Timeout on attempt {attempt + 1}, retrying...")
                continue
            else:
                print(f"Failed after {max_retries} attempts")
                raise

with connect() as device:
    try:
        response = query_with_retry(device, "STATUS")
        print(response.status)
    except ResponseTimeout:
        print("Device unresponsive")

コンテキストマネージャーによる安全な接続管理¶

from kazunoko import connect
from kazunoko.exceptions import DeviceError, KazunokoError

try:
    with connect() as device:
        # コマンド実行
        response = device.query("STATUS")
        print(response.status)
        # 自動的に接続を閉じる
except DeviceError as e:
    print(f"Connection failed: {e}")
except KazunokoError as e:
    print(f"Error: {e}")

タイムアウト設定とエラーハンドリング¶

from kazunoko import PortDevice
from kazunoko.exceptions import ResponseTimeout

# 長時間の操作用にタイムアウトを長くする
device = PortDevice(timeout=3.0)

try:
    with device:
        response = device.query("STATUS")
        print(response.status)
except ResponseTimeout:
    print("Operation timed out")

デバイス接続のトラブルシューティング¶

ポートが見つからない¶

from kazunoko import detect_port
from kazunoko.exceptions import DeviceError

try:
    port = detect_port()
    print(f"Found device on port: {port}")
except DeviceError as e:
    print(f"No device found: {e}")
    print("Available options:")
    print("1. Check if device is connected")
    print("2. Check USB cable connection")
    print("3. Install USB driver if needed")

デバイスが応答しない¶

from kazunoko import connect
from kazunoko.exceptions import ResponseTimeout

try:
    device = connect()
    response = device.query("STATUS")
except ResponseTimeout as e:
    print(f"Device not responding: {e}")
    print("Try:")
    print("1. Power cycle the device")
    print("2. Check the baud rate setting")
    print("3. Try a different USB port")
    print("4. Increase timeout: PortDevice(timeout=2.0)")

無効なレスポンス¶

from kazunoko import connect
from kazunoko.exceptions import ResponseError

try:
    device = connect()
    response = device.query("STATUS")
except ResponseError as e:
    print(f"Invalid response: {e}")
    print("This may indicate:")
    print("1. Incompatible firmware version")
    print("2. Corrupted serial communication")
    print("3. Electromagnetic interference")
    print("Try reconnecting and restarting the device")

ロギングとデバッギング¶

詳細なエラー情報を記録¶

import logging
from kazunoko import connect
from kazunoko.exceptions import KazunokoError

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

try:
    device = connect()
    response = device.query("STATUS")
except KazunokoError as e:
    logger.error(f"kazunoko error: {e}", exc_info=True)

チェーンされた例外（Exception Chaining）¶

デバッグ時には詳細なエラー情報を取得することが重要です。 e.__cause__を使って、大元のエラー情報にアクセスできます。

from kazunoko import connect
from kazunoko.exceptions import DeviceError

try:
    device = connect("/dev/ttyUSB0")
except DeviceError as e:
    # 元の例外情報を保持
    print(f"Original error: {e.__cause__}")
    raise

エラーハンドリング¶

例外の階層¶

例外の種類¶

KazunokoError¶

DeviceError¶

CommandError¶

CommandTimeout¶

ResponseError¶

ResponseTimeout¶

ProtocolError¶

エラーハンドリングのパターン¶

基本的なエラーハンドリング¶

詳細なエラーハンドリング¶

リトライロジック¶

コンテキストマネージャーによる安全な接続管理¶

タイムアウト設定とエラーハンドリング¶

デバイス接続のトラブルシューティング¶

ポートが見つからない¶

デバイスが応答しない¶

無効なレスポンス¶

ロギングとデバッギング¶

詳細なエラー情報を記録¶

チェーンされた例外（Exception Chaining）¶

参考リンク¶

`KazunokoError`¶

`DeviceError`¶

`CommandError`¶

`CommandTimeout`¶

`ResponseError`¶

`ResponseTimeout`¶

`ProtocolError`¶