Raspberry Pi と Arduino の信頼できる Python シリアル通信

信頼できるシリアル接続に必要なのは、ボーレートを揃えることだけではありません。アプリケーションコマンドを作る前に、物理接続、フレーム境界、エンコーディング、最大メッセージ長、タイムアウト、確認応答、リセット動作を定義します。

この2026年メンテナンス層では、元の5行のメモを、Arduino の公式 Serial API と pySerial を使う完全な USB シリアル手順へ拡張しました。2019年のエクスポート全文は出典アーカイブとして末尾に残しています。

1. USB シリアルか GPIO UART を選ぶ

最初の実装では、Arduino の通常の USB データポートを Raspberry Pi の USB ポートへ接続するのが簡単です。Arduino は一般に /dev/ttyACM0 または /dev/ttyUSB0 として現れます。この方法では Raspberry Pi の GPIO UART を有効にしたり、TX/RX ピンを配線したりする必要はありません。ポートを開くと DTR によってリセットされる Arduino ボードもあるため、ソフトウェア側で短い再起動を許容します。

TTL UART の直結は別の電気設計です。TX と RX を交差させ、GND を共有します。Raspberry Pi の UART ピンは 3.3 V ロジックであり、公式文書は 5 V 信号を接続すると破損すると警告しています。相手機器が 3.3 V 対応でなければ、適切なレベルシフターまたは USB-3.3 V シリアルアダプターを使います。UART を直結する場合は raspi-config で UART ハードウェアを有効にし、シリアルのログインコンソールを無効にして、モデル固有の /dev/serial* 対応を確認します。

回路全体とボード動作をレビューしていない限り、USB シリアルと TX/RX の直結を同時に使わないでください。

2. 先に小さなプロトコルを定義する

この例は意図的に狭い契約を使います。

項目契約
転送USB シリアル、115200 baud、8 data bits、no parity、1 stop bit
フレーム境界1行に1つの ASCII コマンド。LF(n)で終端し、CR は無視
最大コマンド終端文字の前に63バイトまで
リクエストPING <10進リクエストID>
成功OK <同じリクエストID>
失敗ERR bad_commandERR invalid_byteERR line_too_long
起動通知READY 1。ホストがポートを開く前に送られる場合があるため任意

リクエスト ID があれば、ホストは応答を対応付けられます。PING は副作用がないので安全に再試行できます。ハードウェアの移動、課金、状態書き込みを行うコマンドには明示的な冪等性設計が必要です。無条件に再試行すると動作が重複します。

プロトコルのキーワードと数値 ID には ASCII で十分です。後で人間向けテキストを運ぶなら UTF-8 と明記し、バイト長を制限して、不正入力の拒否方法を決めます。両端で偶然有効になっているエンコーディングに依存してはいけません。

3. 上限付き Arduino パーサーをアップロードする

次のスケッチは、上限のない String を使わず、loop() 内でも待機しません。1行をバッファリングし、ASCII 以外の制御/データバイトを拒否し、長すぎるフレームは改行まで破棄します。

#include <Arduino.h>
#include <string.h>

constexpr unsigned long BAUD_RATE = 115200;
constexpr size_t MAX_LINE = 64;

enum class DropReason {
  none,
  invalid_byte,
  line_too_long
};

char line_buffer[MAX_LINE];
size_t line_length = 0;
DropReason drop_reason = DropReason::none;

bool is_request_id(const char *text) {
  if (*text == '') {
    return false;
  }

  while (*text != '') {
    if (*text < '0' || *text > '9') {
      return false;
    }
    ++text;
  }
  return true;
}

void handle_line(const char *line) {
  if (strncmp(line, "PING ", 5) == 0 && is_request_id(line + 5)) {
    Serial.print(F("OK "));
    Serial.println(line + 5);
  } else {
    Serial.println(F("ERR bad_command"));
  }
}

void finish_line() {
  if (drop_reason == DropReason::invalid_byte) {
    Serial.println(F("ERR invalid_byte"));
  } else if (drop_reason == DropReason::line_too_long) {
    Serial.println(F("ERR line_too_long"));
  } else {
    line_buffer[line_length] = '';
    handle_line(line_buffer);
  }

  line_length = 0;
  drop_reason = DropReason::none;
}

void setup() {
  Serial.begin(BAUD_RATE);
  Serial.println(F("READY 1"));
}

void loop() {
  while (Serial.available() > 0) {
    const int raw = Serial.read();
    if (raw < 0) {
      break;
    }

    const char value = static_cast<char>(raw);
    if (value == 'r') {
      continue;
    }
    if (value == 'n') {
      finish_line();
      continue;
    }
    if (drop_reason != DropReason::none) {
      continue;
    }
    if (raw < 0x20 || raw > 0x7e) {
      drop_reason = DropReason::invalid_byte;
      continue;
    }
    if (line_length + 1 >= MAX_LINE) {
      drop_reason = DropReason::line_too_long;
      continue;
    }

    line_buffer[line_length++] = value;
  }
}

Serial.available() は受信済みのバイト数を報告し、Serial.read() は次のバイトを返します。データがなければ -1 です。パーサーは戻り値も確認します。ボーレートはホストと一致させますが、部分読み取りが部分コマンドになるのを防ぐのはフレーム契約です。

4. pySerial をインストールしてデバイスを特定する

Raspberry Pi OS では、システム Python からディストリビューションパッケージを使えます。

sudo apt update
sudo apt install --yes python3-serial
python3 -c 'import serial; print(serial.VERSION)'

または、プロジェクト仮想環境へ依存関係を分離します。

python3 -m venv .venv
. .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install pyserial
python -c 'import serial; print(serial.VERSION)'

Arduino の接続前後でポートを一覧します。

python -m serial.tools.list_ports --verbose
ls -l /dev/serial/by-id/ 2>/dev/null || true

利用できる場合、/dev/serial/by-id/... のリンクは、ボードが常に /dev/ttyACM0 だと仮定するより安定しています。最初のポートを無条件に選ばず、ベンダー、製品、シリアル番号を記録します。

5. シリアルポートの権限を確認する

選んだノードと、実際に開くプロセスの有効グループを確認します。

serial_port="/dev/ttyACM0"
ls -l "$serial_port"
id
id -nG

Raspberry Pi OS の USB シリアルデバイスは、一般に dialout グループから利用できます。対象アカウントにそのグループがなければ、そのアカウントだけを追加し、実際に新しいログインセッションを開始します。

serial_user="$(id -un)"
sudo usermod -a -G dialout "$serial_user"

完全にログアウトして再ログインし、id -nG をもう一度実行します。アプリを root で動かしたり、デバイスを全員書き込み可能にしたりしないでください。systemd サービスやコンテナは対話端末とは別のアカウントとデバイスポリシーを使う場合があるので、そのコンテキストを直接診断します。

6. リセットを許容する Python クライアントを実行する

次を serial_ping.py として保存します。一致する応答を受けるまで同じ冪等リクエストを送るため、USB シリアルを開いたときにリセットされるボードも、推測した固定待ち時間に頼らず扱えます。

import argparse
import time

import serial


BAUD_RATE = 115200
MAX_RESPONSE = 80


class ProtocolError(RuntimeError):
    pass


def read_ascii_line(port):
    raw = port.read_until(b"n", size=MAX_RESPONSE)
    if not raw:
        return None
    if not raw.endswith(b"n"):
        raise ProtocolError("response exceeded the frame limit")

    try:
        return raw.rstrip(b"rn").decode("ascii")
    except UnicodeDecodeError as error:
        raise ProtocolError("response was not ASCII") from error


def ping(port_name, request_id):
    if not request_id.isdecimal() or len(request_id) > 32:
        raise ValueError("request ID must contain 1 to 32 decimal digits")

    request = f"PING {request_id}n".encode("ascii")
    expected = f"OK {request_id}"

    with serial.Serial(
        port=port_name,
        baudrate=BAUD_RATE,
        timeout=0.4,
        write_timeout=1.0,
        exclusive=True,
    ) as port:
        port.reset_input_buffer()

        for _ in range(12):
            port.write(request)
            port.flush()

            deadline = time.monotonic() + 0.6
            while time.monotonic() < deadline:
                response = read_ascii_line(port)
                if response is None:
                    break
                if response == "READY 1":
                    continue
                if response == expected:
                    return response
                if response.startswith("ERR "):
                    raise ProtocolError(response)

        raise TimeoutError(f"no matching response from {port_name}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("port")
    parser.add_argument("--request-id", default="1")
    arguments = parser.parse_args()
    print(ping(arguments.port, arguments.request_id))


if __name__ == "__main__":
    main()

安定したデバイスリンクまたは確認済みノードで実行します。

python -m py_compile serial_ping.py
python serial_ping.py /dev/ttyACM0 --request-id 42

期待する出力:

OK 42

exclusive=True は POSIX 上で、このクライアントが所有している間、別のオープンを拒否するよう要求します。同一プロセス内のスレッド間調整を置き換えるものではありません。対象プラットフォームやドライバーが排他モードに対応しない場合は、アプリが使う所有権の仕組みを文書化してテストします。

7. ハードウェアなしでバイトフレームをテストする

pySerial には loop:// URL ハンドラーがあります。バイトをエコーするため、ホスト側のフレーム境界をローカルで確認できます。

import serial


frame = b"PING 42n"
with serial.serial_for_url("loop://", timeout=1) as port:
    written = port.write(frame)
    reply = port.read_until(b"n", size=64)

assert written == len(frame)
assert reply == frame
print("loop framing test passed")

このテストで分かるのは、Python 環境が pySerial の loop ハンドラーを開き、同じバイト列を往復できることだけです。USB ケーブル、Arduino ファームウェア、リセット動作、UART の電気特性、実デバイスの権限はテストしません。

8. バイト、テキスト、整数を区別する

pySerial の write() は bytes-like データを受け取り、読み取りは bytes を返します。エンコードとデコードはプロトコル境界だけで行います。

text = "hello"
frame = (text + "n").encode("utf-8")
decoded = frame.rstrip(b"n").decode("utf-8")

one_byte = bytes([65])
integer_value = one_byte[0]

chr(65) はテキスト文字 "A" を返し、シリアルのバイトフレームは作りません。bytes([value])0 <= value <= 255 を要求し、意図が明確です。複数バイト整数では、両端で int.to_bytes()int.from_bytes() または struct を使ってバイト順と幅を定義します。

変数名に strbytesserialtime を使わないでください。組み込み型やインポートしたモジュールを隠します。制御プロトコルでは errors="strict" でデコードし、破損を置換文字で隠さず証拠として扱います。

9. リセット、タイムアウト、バッファを理解する

  • 多くの Arduino USB ボードは、ホストがシリアルポートを開くとリセットされます。上限付き再試行のハンドシェイクは、固定の sleep(2) より堅牢です。
  • pySerial の正の timeout は読み取り時間を制限します。タイムアウトがないと、改行が届かない場合に永久に待つことがあります。
  • 書き込みタイムアウトを設定しない限り、write() はブロックします。flush() は待機中の出力が書かれるまで待ちますが、入出力バッファを消去しません。
  • reset_input_buffer() は受信済みバイトを意図的に捨てます。意味のある応答が到着する可能性がある途中ではなく、既知のセッション境界だけで使います。
  • シリアル配送はバイトストリームです。1回の write() がデバイス側の1回の read に対応する保証はなく、メッセージを定めるのはフレームプロトコルです。
  • Arduino の受信バッファは有限です。loop() で長くブロックせず、フレームを制限し、通信量を増やす前にフロー制御またはアプリ層のバックプレッシャーを設計します。

10. レイヤー別にトラブルシュートする

症状可能性の高いレイヤー集める証拠
ポートが現れないUSB ケーブルが給電専用、デバイス未給電、カーネルドライバー未接続、USB 電力不足python -m serial.tools.list_ports --verbosedmesg、既知のデータケーブル/別ポート
Permission denied有効アカウントにデバイスグループがない、またはサービス/コンテナポリシーが拒否ノードの ls -lid -nG、サービス/コンテナ設定
ポートを開いた後に切断ボードのリセット、不安定な USB 電源/ケーブル、ファームウェア再起動カーネルログ、ボード LED/リセット動作、ハンドシェイク時刻
ランダムな文字ボーレート/設定の不一致、電圧レベル/ノイズ問題両端設定、物理配線、直結 UART ならロジックアナライザー
応答なしでタイムアウトポート違い、改行欠落、ファームウェア未稼働、リセット待ち、リクエスト拒否送信した生フレーム、Arduino スケッチ/版、上限付きデバッグログ
応答が1リクエストずれる古い入力、リクエスト ID 欠落、複数クライアントポート所有者、プロトコルトレース、応答 ID
Device or resource busy別プロセスがポートを所有許可があれば lsoffuser、IDE Serial Monitor、サービス一覧

Python を実行する前に Arduino IDE の Serial Monitor を閉じてください。この単純なプロトコル端点は1プロセスだけが所有します。

11. プロトコルを意図的に拡張する

コマンドが数個を超えるならプロトコルに版を付け、各リクエストについて応答、単位、範囲、副作用、タイムアウト、再試行規則を表にします。並行処理の前にリクエスト ID を追加します。状態変更には重複排除などの冪等性機構が必要です。

バイナリデータでは、版、メッセージ種別、明示的な長さ、ペイロード、checksum/CRC を持つフレームを使います。確保や書き込みの前に長さを検証します。チェックサムは破損を検出しますが、コマンドを認証しません。外部公開または安全上重要な接続には別のセキュリティ設計が必要です。

ボーレートを上げる前に、スループットと最悪時の遅延を測定します。USB シリアル、Arduino のメインループ、センサー処理、Raspberry Pi のスケジューリングがすべて影響します。締切超過で機器が破損するなら、安全制御ループを適切なコントローラーへ移し、デスクトップ型のシリアルプロセスには任せないでください。

12. 完全な2019年エクスポートのアーカイブ

以下は、メタデータと短い本文を含む2019年の WordPress エクスポート全文です。行末空白だけを正規化し、出典として保存しています。元コードの bytesstr は Python の組み込み型を隠し、chr(int) はバイトフレームではなくテキストを作ります。現行作業には上のメンテナンス版を使ってください。

---
id: 1933
title: 'RaspberryPi and Arduino Talk with Serial Port Using Python'
slug: 'raspberrypi-and-arduino-talk-with-serial-port'
date: '2019-05-09T13:37:55'
modified: '2019-05-09T13:57:59'
status: 'publish'
link: 'https://blog.lazying.art/en/html/computer_internet/hardware_system/raspberry-pi/1933/raspberrypi-and-arduino-talk-with-serial-port.html'
author: 'Lachlan Chen'
categories:
  - 'Raspberry Pi'
---

char, str; bytes, unicode, string

bytes = b"string"
str = chr(int)


Arduino

Raspberry Pi

一次資料

Leave a Reply