RON-V #13 — MeshtasticをUSB接続し、920MHz Mesh通信を車載ノードへ統合する

Last updated at 2026-10-25Posted at 2026-10-18

RON-V 連載リンク
← 前の記事：#12 — GNSS・LTEデータとGeomapの準備
→ 次の記事：#14 — 統合前編：L0／L1／L2と自動配送

L0〜L4の定義追記（2026-10-25）

本文中のコード・試験結果は執筆当時の記録として残しています。画像AI・音声を関連付ける現在の定義は次のとおりです。

レベル	残す内容
L0	Meshtastic向けの23バイト短報。画像AIの候補通知は任意設定
L1	個別の観測・撮影／録音／認識の状態と、詳細への参照
L2	一定時間の観測件数・分類別候補などの集計
L3	画像AIの詳細、録音情報、文字起こし全文
L4	元画像JPEG、元音声WAV、LTEのraw logなど

L0の警報欄は1バイト＝8ビットです。新しいversion=2の定義は、倒木 0x01、冠水 0x02、煙 0x04、道路上の障害物 0x08、路面の穴 0x10。残り3ビットは未割り当てです。0x03は倒木と冠水の組み合わせで、煙の番号ではありません。0x00はL0上では未判定で、安全を示しません。 候補ありでは元イベントの時刻・位置を使い、現在地と混同しません。

発信元はnode_uid、根拠は画像・音声などです。「処理成功」「災害の候補」「人の確認済み」を分けます。初期の確認状態はunreviewed。画像AIの検出0件を「危険なし」とせず、音声の否定も全文へ残します。人が補足・訂正する画面と、音声からの災害自動分類は今後の実装です。

10月25日にL1の関連付け・L2の集計・音声L3/L4の転送処理をコードへ追加しました。新連携の実機確認はこれからです。音声転送とL0候補通知は初期設定では無効で、個別に有効化します。従来のL0 version=1へ、この分類の意味を遡って付けません。

Looker Studioで使用中のSheets summaryはRF測定から直接出力する既存の10列で、L2とは別です。今回の追加で列名・順序・値の意味、既存Grafanaの参照項目は変えません。L1のMesh送信、写真・音声のMesh転送、自動公開は実装していません。

詳細と設定手順：データレベルの定義（リポジトリの閲覧権限が必要です）。

[!NOTE]
2026-09-23 現行GitHub同期メモ
Meshtasticの現行実装は app/mesh/gateway.py、message_store.py、queue_text.py、systemd/ronv-mesh.service が正本です。依存パッケージは個別 pip install ではなく requirements.txt で管理し、現在は meshtastic==2.7.11 / Pypubsub==4.0.7 を固定しています。USBは /dev/serial/by-id/ を使用し、Meshtastic USBを開く常駐プロセスは ronv-mesh.service のみとします。TX/RX/ACKは mesh_messages に保存され、Web v2からのTEXT送信も同じSQLiteキューを使用します。

はじめに

前回の #12 では、

GNSS
  +
LTE
  ↓
SQLite

として、位置情報とLTEセル情報をRON-VのSQLiteへ保存する仕組みを構築しました。

今回は、RON-Vへもう一つの通信経路として、

Meshtastic

を追加します。

構成は、

Meshtastic Node
      ↓
  920MHz Mesh
      ↓
Meshtastic Node
      ↓ USB
Raspberry Pi 5
      ↓
RON-V
      ↓
SQLite

です。

RON-VではBluetoothではなく、

USB Serial

を基本接続とします。

今回の実機では、

RX
TX
ACK
RSSI
SNR
SQLite保存
systemd常駐

まで確認しました。

1. 今回の目的

RON-VへMeshtasticを追加し、

外部Meshtastic Node
        ↓
920MHz Mesh
        ↓
RON-V

の双方向通信を可能にします。

今回実装する範囲は、

USB接続
TEXT_MESSAGE_APP受信
TEXT_MESSAGE_APP送信
ACK確認
SQLite保存
systemd常駐

です。

2. 今回は実装しないもの

既存実装には、

PRIVATE_APP
独自L0 binary packet
Unix Domain Socket
sync_deliveries
sync_objects

など、より上位の配送処理もありました。

しかし今回はそれらを同時に移植しません。

RON-Vではまず、

Meshtastic基本通信

を独立モジュールとして完成させます。

上位の通信制御は後のRON-V統合段階で追加します。

3. RON-V側の構成

今回追加するファイルは、

/home/ronv/ronv/
│
├── app/
│   └── mesh/
│       ├── __init__.py
│       ├── gateway.py
│       ├── message_store.py
│       └── queue_text.py
│
├── config/
│   └── ronv.toml
│
└── systemd/
    └── ronv-mesh.service

です。

SQLiteは既存の、

/var/lib/ronv/ronv.db

を使用します。

4. Meshtastic USB機器を接続する

Meshtastic NodeをRaspberry Pi 5へUSB接続します。

確認：

echo "===== USB ====="

lsusb

echo
echo "===== SERIAL BY-ID ====="

ls -l \
  /dev/serial/by-id/

今回の実機では、

QinHeng Electronics USB Single Serial
VID:PID = 1a86:55d4

として認識されました。

Serial deviceは、

/dev/ttyACM2

でした。

ただしRON-Vでは、

/dev/ttyACM2

を設定に使用しません。

5. `/dev/serial/by-id/` を使用する

Linuxでは、

/dev/ttyACM0
/dev/ttyACM1
/dev/ttyACM2

などの番号は接続順によって変わる可能性があります。

そのためRON-Vでは、

/dev/serial/by-id/

を使用します。

今回の実機では、

/dev/serial/by-id/usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00

として認識されました。

Qiita記事では固有IDを伏せています。

6. 現行のPython依存管理

現在はMeshtastic導入前に pip freeze を個別保存して手作業でpackageを追加するのではなく、GitHubの requirements.txt を正本にします。

現行の関連依存は、

meshtastic==2.7.11
Pypubsub==4.0.7

です。環境構築は install.sh が /home/ronv/.venvs/ronv を作成し、requirements.txt を導入します。

cd /home/ronv/ronv
sudo ./install.sh

既存環境で依存だけ確認する場合は、

/home/ronv/.venvs/ronv/bin/python -m pip show meshtastic Pypubsub

を使用します。

9. PythonからUSB接続を確認する

正式プログラムを作成する前に、Meshtastic Python APIから直接USB機器を開きます。

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python - <<'PY'
from meshtastic.serial_interface import SerialInterface

DEVICE = (
    "/dev/serial/by-id/"
    "usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00"
)

print(
    "===== RON-V MESHTASTIC USB TEST ====="
)

print(
    "device :",
    DEVICE,
)

interface = None

try:
    print(
        "opening..."
    )

    interface = SerialInterface(
        devPath=DEVICE,
        timeout=30,
    )

    node_num = (
        interface
        .localNode
        .nodeNum
    )

    print(
        "result : OK"
    )

    print(
        "node   :",
        f"!{node_num:08x}",
    )

finally:
    if interface is not None:
        interface.close()

        print(
            "closed : OK"
        )
PY

実機では、

===== RON-V MESHTASTIC USB TEST =====

device :
/dev/serial/by-id/usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00

opening...

result : OK

node   : !<LOCAL_NODE_ID>

closed : OK

となりました。

これで、

RON-V Python
   ↓
Meshtastic Python API
   ↓
USB Serial
   ↓
Meshtastic Node

が成立しました。

10. 現行のMesh設定

実環境の config/ronv.toml はGit管理外です。公開テンプレートでは、

[mesh]
device = "/dev/serial/by-id/<MESHTASTIC_SERIAL_DEVICE>"
timeout_sec = 30

とし、実機では <MESHTASTIC_SERIAL_DEVICE> を /dev/serial/by-id/ の実デバイスへ置き換えます。/dev/ttyACM* の番号は固定値として使用しません。

12. Mesh設定を確認する

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python - <<'PY'
from pathlib import Path

from app.common.config import load_config

cfg = load_config()
mesh = cfg["mesh"]

print(
    "===== RON-V MESH CONFIG ====="
)

print(
    "device :",
    mesh["device"],
)

print(
    "exists :",
    Path(
        mesh["device"]
    ).exists(),
)

print(
    "timeout:",
    mesh["timeout_sec"],
)
PY

実機結果：

===== RON-V MESH CONFIG =====

device :
/dev/serial/by-id/usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00

exists : True

timeout: 30

です。

13. `mesh_messages` テーブルを作成する

Meshtastic通信履歴をRON-V SQLiteへ保存します。

保存項目は、

RX / TX
送信元
送信先
packet ID
portnum
text
状態
ACK
RSSI
SNR
error

です。

14. `message_store.py` を作成する

cd /home/ronv/ronv

cat > app/mesh/message_store.py <<'PY'
from __future__ import annotations

from datetime import datetime
from zoneinfo import ZoneInfo

from app.common.config import load_config
from app.common.db import (
    connect_db,
    initialize_database,
)


def initialize_mesh_schema(
    config: dict | None = None,
) -> None:
    cfg = config or load_config()

    initialize_database(cfg)

    con = connect_db(cfg)

    try:
        con.execute(
            """
            CREATE TABLE IF NOT EXISTS mesh_messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,

                timestamp_jst TEXT NOT NULL,
                direction TEXT NOT NULL,

                from_id TEXT,
                to_id TEXT,

                packet_id INTEGER,
                portnum TEXT,

                text TEXT,

                state TEXT NOT NULL,

                ack_type TEXT,
                ack_from TEXT,
                ack_error TEXT,

                rssi REAL,
                snr REAL,

                last_error TEXT,

                created_at_jst TEXT NOT NULL,
                updated_at_jst TEXT NOT NULL
            )
            """
        )

        con.execute(
            """
            CREATE INDEX IF NOT EXISTS
            idx_mesh_messages_state
            ON mesh_messages(
                direction,
                state,
                id
            )
            """
        )

        con.execute(
            """
            CREATE INDEX IF NOT EXISTS
            idx_mesh_messages_timestamp
            ON mesh_messages(
                timestamp_jst
            )
            """
        )

        con.commit()

    finally:
        con.close()


def now_jst(
    config: dict | None = None,
) -> str:
    cfg = config or load_config()

    timezone = ZoneInfo(
        cfg["project"]["timezone"]
    )

    return datetime.now(
        timezone
    ).isoformat(
        timespec="seconds"
    )


def save_received_message(
    *,
    from_id: str | None,
    to_id: str | None,
    packet_id: int | None,
    portnum: str | None,
    text: str | None,
    rssi: float | None,
    snr: float | None,
    config: dict | None = None,
) -> int:
    cfg = config or load_config()

    ts = now_jst(cfg)

    con = connect_db(cfg)

    try:
        cur = con.execute(
            """
            INSERT INTO mesh_messages (
                timestamp_jst,
                direction,
                from_id,
                to_id,
                packet_id,
                portnum,
                text,
                state,
                rssi,
                snr,
                created_at_jst,
                updated_at_jst
            )
            VALUES (
                ?,
                'RX',
                ?, ?, ?, ?, ?,
                'received',
                ?, ?,
                ?, ?
            )
            """,
            (
                ts,
                from_id,
                to_id,
                packet_id,
                portnum,
                text,
                rssi,
                snr,
                ts,
                ts,
            ),
        )

        con.commit()

        return int(
            cur.lastrowid
        )

    finally:
        con.close()


def queue_text_message(
    destination: str,
    text: str,
    config: dict | None = None,
) -> int:
    cfg = config or load_config()

    ts = now_jst(cfg)

    con = connect_db(cfg)

    try:
        cur = con.execute(
            """
            INSERT INTO mesh_messages (
                timestamp_jst,
                direction,
                to_id,
                portnum,
                text,
                state,
                created_at_jst,
                updated_at_jst
            )
            VALUES (
                ?,
                'TX',
                ?,
                'TEXT_MESSAGE_APP',
                ?,
                'queued',
                ?,
                ?
            )
            """,
            (
                ts,
                destination,
                text,
                ts,
                ts,
            ),
        )

        con.commit()

        return int(
            cur.lastrowid
        )

    finally:
        con.close()


def claim_next_tx(
    config: dict | None = None,
):
    cfg = config or load_config()

    con = connect_db(cfg)

    try:
        con.execute(
            "BEGIN IMMEDIATE"
        )

        row = con.execute(
            """
            SELECT *
            FROM mesh_messages
            WHERE direction='TX'
              AND state='queued'
            ORDER BY id
            LIMIT 1
            """
        ).fetchone()

        if row is None:
            con.commit()
            return None

        ts = now_jst(cfg)

        cur = con.execute(
            """
            UPDATE mesh_messages
            SET
                state='sending',
                updated_at_jst=?
            WHERE id=?
              AND state='queued'
            """,
            (
                ts,
                row["id"],
            ),
        )

        if cur.rowcount != 1:
            con.rollback()
            return None

        con.commit()

        return dict(row)

    finally:
        con.close()


def mark_tx_acked(
    *,
    row_id: int,
    packet_id: int | None,
    ack_type: str | None,
    ack_from: str | None,
    ack_error: str | None,
    config: dict | None = None,
) -> None:
    cfg = config or load_config()

    ts = now_jst(cfg)

    con = connect_db(cfg)

    try:
        con.execute(
            """
            UPDATE mesh_messages
            SET
                timestamp_jst=?,
                state='acked',
                packet_id=?,
                ack_type=?,
                ack_from=?,
                ack_error=?,
                last_error=NULL,
                updated_at_jst=?
            WHERE id=?
            """,
            (
                ts,
                packet_id,
                ack_type,
                ack_from,
                ack_error,
                ts,
                row_id,
            ),
        )

        con.commit()

    finally:
        con.close()


def mark_tx_failed(
    *,
    row_id: int,
    error: str,
    config: dict | None = None,
) -> None:
    cfg = config or load_config()

    ts = now_jst(cfg)

    con = connect_db(cfg)

    try:
        con.execute(
            """
            UPDATE mesh_messages
            SET
                state='failed',
                last_error=?,
                updated_at_jst=?
            WHERE id=?
            """,
            (
                error[:500],
                ts,
                row_id,
            ),
        )

        con.commit()

    finally:
        con.close()


def main() -> None:
    cfg = load_config()

    initialize_mesh_schema(
        cfg
    )

    print(
        "Mesh message schema : OK"
    )


if __name__ == "__main__":
    main()
PY

15. 構文確認

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python \
  -m py_compile \
  app/mesh/message_store.py

何も表示されなければ正常です。

16. `mesh_messages` を初期化する

/home/ronv/.venvs/ronv/bin/python \
  -m app.mesh.message_store

正常に実行されればテーブルが作成されます。

17. SQLiteを確認する

以下は Tera Term / bashへそのまま貼り付けるコマンド です。

sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column

SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;

SELECT name
FROM sqlite_master
WHERE type='index'
  AND tbl_name='mesh_messages'
ORDER BY name;

SELECT
    COUNT(*) AS mesh_messages
FROM mesh_messages;

PRAGMA journal_mode;
PRAGMA integrity_check;
SQL

実機では、

lte_measurements
mesh_messages
ronv_meta
sqlite_sequence
vehicle_events

となりました。

INDEX：

idx_mesh_messages_state
idx_mesh_messages_timestamp

初期データ：

mesh_messages
-------------
0

SQLite：

journal_mode
------------
wal

integrity_check
---------------
ok

です。

18. Meshtastic Gatewayを作成する

次に、

app/mesh/gateway.py

を作ります。

Gatewayは、

Meshtastic USB
      ↓
RX / TX
      ↓
mesh_messages

を担当します。

19. `gateway.py` を作成する

cd /home/ronv/ronv

cat > app/mesh/gateway.py <<'PY'
from __future__ import annotations

import signal
import threading
import time

from pubsub import pub
from meshtastic.serial_interface import (
    SerialInterface,
)
from meshtastic.protobuf import (
    portnums_pb2,
)

from app.common.config import load_config
from app.mesh.message_store import (
    claim_next_tx,
    initialize_mesh_schema,
    mark_tx_acked,
    mark_tx_failed,
    save_received_message,
)


TEXT_APP = (
    portnums_pb2
    .PortNum
    .TEXT_MESSAGE_APP
)


class MeshGateway:

    def __init__(
        self,
        config: dict | None = None,
    ):
        self.cfg = (
            config
            or load_config()
        )

        self.interface = None
        self.running = True
        self.tx_lock = threading.Lock()

    def start(self) -> None:
        mesh = self.cfg["mesh"]

        device = str(
            mesh["device"]
        )

        timeout_sec = int(
            mesh["timeout_sec"]
        )

        print(
            "===== RON-V MESHTASTIC ====="
        )

        print(
            "device :",
            device,
        )

        pub.subscribe(
            self.on_receive,
            "meshtastic.receive",
        )

        self.interface = SerialInterface(
            devPath=device,
            timeout=timeout_sec,
        )

        node_num = (
            self.interface
            .localNode
            .nodeNum
        )

        print(
            "local  :",
            f"!{node_num:08x}",
        )

        print(
            "status : running"
        )

    def close(self) -> None:
        if self.interface is not None:
            self.interface.close()
            self.interface = None

        print(
            "status : stopped"
        )

    def on_receive(
        self,
        packet,
        interface,
    ) -> None:
        try:
            decoded = (
                packet.get(
                    "decoded"
                )
                or {}
            )

            portnum = decoded.get(
                "portnum"
            )

            if (
                portnum
                != "TEXT_MESSAGE_APP"
            ):
                return

            text = decoded.get(
                "text"
            )

            if text is None:
                payload = decoded.get(
                    "payload"
                )

                if payload:
                    text = bytes(
                        payload
                    ).decode(
                        "utf-8",
                        errors="replace",
                    )

            from_id = packet.get(
                "fromId"
            )

            to_id = packet.get(
                "toId"
            )

            packet_id = packet.get(
                "id"
            )

            rssi = packet.get(
                "rxRssi"
            )

            snr = packet.get(
                "rxSnr"
            )

            row_id = (
                save_received_message(
                    from_id=from_id,
                    to_id=to_id,
                    packet_id=packet_id,
                    portnum=portnum,
                    text=text,
                    rssi=rssi,
                    snr=snr,
                    config=self.cfg,
                )
            )

            print(
                "MESH RX",
                f"id={row_id}",
                f"from={from_id}",
                f"to={to_id}",
                f"text={text!r}",
                f"rssi={rssi}",
                f"snr={snr}",
            )

        except Exception as exc:
            print(
                "MESH RX WARNING:",
                repr(exc),
            )

    def send_text(
        self,
        *,
        destination: str,
        text: str,
    ) -> dict:
        if self.interface is None:
            raise RuntimeError(
                "Meshtastic interface is not open"
            )

        with self.tx_lock:
            started = time.monotonic()

            ack_info = {
                "type": None,
                "from": None,
                "error": None,
            }

            local_num = (
                self.interface
                .localNode
                .nodeNum
            )

            def on_ack_nak(
                packet,
            ):
                decoded = (
                    packet.get(
                        "decoded"
                    )
                    or {}
                )

                routing = (
                    decoded.get(
                        "routing"
                    )
                    or {}
                )

                error = routing.get(
                    "errorReason",
                    "NONE",
                )

                ack_from = packet.get(
                    "from"
                )

                ack_info["from"] = (
                    ack_from
                )

                ack_info["error"] = (
                    error
                )

                try:
                    from_num = int(
                        ack_from
                    )

                except (
                    TypeError,
                    ValueError,
                ):
                    from_num = None

                if error != "NONE":
                    ack_info["type"] = (
                        "nak"
                    )

                    (
                        self.interface
                        ._acknowledgment
                        .receivedNak
                    ) = True

                    return

                if from_num == local_num:
                    ack_info["type"] = (
                        "implicit"
                    )

                    (
                        self.interface
                        ._acknowledgment
                        .receivedImplAck
                    ) = True

                else:
                    ack_info["type"] = (
                        "real"
                    )

                    (
                        self.interface
                        ._acknowledgment
                        .receivedAck
                    ) = True

            result = (
                self.interface.sendData(
                    text.encode(
                        "utf-8"
                    ),
                    destinationId=destination,
                    portNum=TEXT_APP,
                    wantAck=True,
                    onResponse=on_ack_nak,
                    onResponseAckPermitted=True,
                )
            )

            (
                self.interface
                .waitForAckNak()
            )

            latency_ms = int(
                (
                    time.monotonic()
                    - started
                )
                * 1000
            )

            if (
                ack_info["type"]
                not in (
                    "implicit",
                    "real",
                )
            ):
                raise RuntimeError(
                    "Meshtastic ACK missing"
                )

            if (
                ack_info["error"]
                not in (
                    None,
                    "NONE",
                )
            ):
                raise RuntimeError(
                    "Meshtastic NAK: "
                    + str(
                        ack_info[
                            "error"
                        ]
                    )
                )

            return {
                "packet_id": result.id,
                "ack_type": ack_info["type"],
                "ack_from": ack_info["from"],
                "ack_error": ack_info["error"],
                "latency_ms": latency_ms,
            }

    def process_tx_queue(
        self,
    ) -> bool:
        row = claim_next_tx(
            self.cfg
        )

        if row is None:
            return False

        row_id = int(
            row["id"]
        )

        destination = str(
            row["to_id"]
        )

        text = str(
            row["text"]
            or ""
        )

        try:
            result = self.send_text(
                destination=destination,
                text=text,
            )

            ack_from = result.get(
                "ack_from"
            )

            if ack_from is not None:
                try:
                    ack_from = (
                        f"!{int(ack_from):08x}"
                    )
                except Exception:
                    ack_from = str(
                        ack_from
                    )

            mark_tx_acked(
                row_id=row_id,
                packet_id=result.get(
                    "packet_id"
                ),
                ack_type=result.get(
                    "ack_type"
                ),
                ack_from=ack_from,
                ack_error=result.get(
                    "ack_error"
                ),
                config=self.cfg,
            )

            print(
                "MESH TX",
                f"id={row_id}",
                f"to={destination}",
                f"text={text!r}",
                f"packet="
                f"{result.get('packet_id')}",
                f"ack="
                f"{result.get('ack_type')}",
                f"latency_ms="
                f"{result.get('latency_ms')}",
            )

        except Exception as exc:
            mark_tx_failed(
                row_id=row_id,
                error=str(exc),
                config=self.cfg,
            )

            print(
                "MESH TX ERROR",
                f"id={row_id}",
                repr(exc),
            )

        return True

    def run(self) -> None:
        self.start()

        try:
            while self.running:
                worked = (
                    self
                    .process_tx_queue()
                )

                if not worked:
                    time.sleep(
                        0.5
                    )

        finally:
            self.close()


def main() -> None:
    cfg = load_config()

    initialize_mesh_schema(
        cfg
    )

    gateway = MeshGateway(
        cfg
    )

    def stop_handler(
        signum,
        frame,
    ):
        gateway.running = False

    signal.signal(
        signal.SIGINT,
        stop_handler,
    )

    signal.signal(
        signal.SIGTERM,
        stop_handler,
    )

    gateway.run()


if __name__ == "__main__":
    main()
PY

20. TXキュー登録プログラムを作成する

送信プログラムがUSBを直接開くのではなく、

SQLiteへ送信要求を登録
       ↓
常駐Gatewayが送信

という構成にします。

app/mesh/queue_text.py：

cd /home/ronv/ronv

cat > app/mesh/queue_text.py <<'PY'
from __future__ import annotations

import argparse

from app.common.config import load_config
from app.mesh.message_store import (
    initialize_mesh_schema,
    queue_text_message,
)


def main() -> None:
    parser = argparse.ArgumentParser(
        description=(
            "Queue a Meshtastic text message"
        )
    )

    parser.add_argument(
        "destination",
        help=(
            "Meshtastic node ID"
        ),
    )

    parser.add_argument(
        "text",
        help="Text message",
    )

    args = parser.parse_args()

    cfg = load_config()

    initialize_mesh_schema(
        cfg
    )

    row_id = queue_text_message(
        destination=args.destination,
        text=args.text,
        config=cfg,
    )

    print(
        "===== RON-V MESH TX QUEUE ====="
    )

    print(
        "row_id      :",
        row_id,
    )

    print(
        "destination :",
        args.destination,
    )

    print(
        "text        :",
        args.text,
    )

    print(
        "state       : queued"
    )


if __name__ == "__main__":
    main()
PY

21. Python構文確認

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python \
  -m py_compile \
  app/common/config.py \
  app/common/db.py \
  app/mesh/message_store.py \
  app/mesh/queue_text.py \
  app/mesh/gateway.py

何も表示されなければ正常です。

22. Gatewayを手動起動する

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python \
  -m app.mesh.gateway

実機では、

===== RON-V MESHTASTIC =====

device :
/dev/serial/by-id/usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00

local  : !<LOCAL_NODE_ID>

status : running

となりました。

23. Meshtastic RXを確認する

別のMeshtastic Nodeからテキストを送信しました。

実機では、

MESH RX
id=1
from=!<REMOTE_NODE_ID>
to=^all
text='Yes'
rssi=-55
snr=13.75

を確認しました。

これで、

Remote Meshtastic
      ↓
920MHz
      ↓
RON-V Meshtastic
      ↓ USB
gateway.py

まで成立しました。

24. RXデータをSQLiteへ保存する

確認：

sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column

SELECT
    id,
    timestamp_jst,
    direction,
    from_id,
    to_id,
    packet_id,
    portnum,
    text,
    state,
    rssi,
    snr
FROM mesh_messages
ORDER BY id DESC
LIMIT 10;

PRAGMA integrity_check;
SQL

実機では、

direction
RX

portnum
TEXT_MESSAGE_APP

text
Yes

state
received

RSSI
-55.0

SNR
13.75

としてSQLiteへ保存されました。

25. Meshtastic TXを確認する

次にRON-Vから別ノードへ送信します。

Gatewayを動かした状態で、別ターミナルから実行します。

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python \
  -m app.mesh.queue_text \
  '!<REMOTE_NODE_ID>' \
  'RONV TX TEST 01'

Gateway側では、

MESH TX
id=2
to=!<REMOTE_NODE_ID>
text='RONV TX TEST 01'
packet=<PACKET_ID>
ack=real
latency_ms=2401

となりました。

26. `real ACK` を確認する

SQLiteを確認します。

sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column

SELECT
    id,
    direction,
    to_id,
    packet_id,
    text,
    state,
    ack_type,
    ack_from,
    ack_error
FROM mesh_messages
ORDER BY id DESC
LIMIT 10;
SQL

実機では、

direction
TX

text
RONV TX TEST 01

state
acked

ack_type
real

ack_from
!<REMOTE_NODE_ID>

ack_error
NONE

となりました。

つまり単に送信要求を発行しただけではなく、

RON-V
  ↓
Meshtastic送信
  ↓
相手Node受信
  ↓
ACK
  ↓
RON-V

まで成立しています。

27. systemd用serviceを作成する

手動動作が確認できたため、Gatewayをsystemd化します。

手動Gatewayを、

Ctrl + C

で終了します。

次に、

systemd/ronv-mesh.service

を作成します。

cd /home/ronv/ronv

cat > systemd/ronv-mesh.service <<'UNIT'
[Unit]
Description=RON-V Meshtastic USB Gateway
After=local-fs.target
Wants=local-fs.target

[Service]
Type=simple

User=ronv
WorkingDirectory=/home/ronv/ronv

Environment=PYTHONUNBUFFERED=1
Environment=RONV_CONFIG=/home/ronv/ronv/config/ronv.toml

ExecStart=/home/ronv/.venvs/ronv/bin/python -m app.mesh.gateway

Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
UNIT

28. systemd unitを検証する

systemd-analyze verify \
  /home/ronv/ronv/systemd/ronv-mesh.service

エラーが表示されなければ正常です。

29. systemdへ登録する

既存serviceがある場合はバックアップします。

if [ -e /etc/systemd/system/ronv-mesh.service ]; then

    STAMP=$(date +%Y%m%d_%H%M%S)

    sudo cp -a \
      /etc/systemd/system/ronv-mesh.service \
      "/etc/systemd/system/ronv-mesh.service.bak_${STAMP}"

else

    echo "new service"

fi

正式配置：

sudo install \
  -m 0644 \
  /home/ronv/ronv/systemd/ronv-mesh.service \
  /etc/systemd/system/ronv-mesh.service

再読込み：

sudo systemctl daemon-reload

30. systemdからMesh Gatewayを起動する

sudo systemctl start \
  ronv-mesh.service

確認：

systemctl status \
  ronv-mesh.service \
  --no-pager

実機では、

Loaded:
loaded

Active:
active (running)

となりました。

31. systemdログを確認する

journalctl \
  -u ronv-mesh.service \
  -n 50 \
  --no-pager

実機では、

===== RON-V MESHTASTIC =====

device :
/dev/serial/by-id/usb-1a86_USB_Single_Serial_<DEVICE_ID>-if00

local :
!<LOCAL_NODE_ID>

status :
running

となりました。

これで、

systemd
   ↓
RON-V Python venv
   ↓
app.mesh.gateway
   ↓
Meshtastic USB

が成立しました。

32. systemd稼働中にRXを確認する

systemd Gatewayを動かした状態で、別Meshtastic Nodeからメッセージを送ります。

実機では、

MESH RX
id=3
from=!<REMOTE_NODE_ID>
to=!<LOCAL_NODE_ID>
text='Hi'
rssi=-40
snr=10.0

となりました。

つまりsystemd常駐状態でも受信できます。

33. systemd稼働中にTXする

別ターミナルから、

cd /home/ronv/ronv

/home/ronv/.venvs/ronv/bin/python \
  -m app.mesh.queue_text \
  '!<REMOTE_NODE_ID>' \
  'RONV SYSTEMD TX'

を実行しました。

Gatewayのjournalでは、

MESH TX
id=4
to=!<REMOTE_NODE_ID>
text='RONV SYSTEMD TX'
packet=<PACKET_ID>
ack=real
latency_ms=2501

となりました。

34. systemd TX/RXをSQLiteで確認する

以下は Tera Term / bashで実行するSQL確認コマンド です。

sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column

SELECT
    id,
    timestamp_jst,
    direction,
    from_id,
    to_id,
    packet_id,
    portnum,
    text,
    state,
    ack_type,
    ack_from,
    ack_error,
    rssi,
    snr,
    last_error
FROM mesh_messages
ORDER BY id DESC
LIMIT 10;

PRAGMA integrity_check;
SQL

実機では、

RX

text      = Hi
state     = received
RSSI      = -40.0
SNR       = 10.0

および、

TX

text      = RONV SYSTEMD TX
state     = acked
ack_type  = real
ack_from  = !<REMOTE_NODE_ID>
ack_error = NONE

を確認しました。

SQLiteも、

integrity_check
---------------
ok

でした。

35. 自動起動を有効化する

systemdでTX/RXが成功したため、自動起動を有効にします。

sudo systemctl enable \
  ronv-mesh.service

確認：

systemctl is-enabled \
  ronv-mesh.service

systemctl is-active \
  ronv-mesh.service

実機結果：

enabled
active

です。

36. RON-Vサービス全体を確認する

echo "===== RON-V SERVICES ====="

for UNIT in \
  ronv-vehicle.service \
  ronv-web.service \
  ronv-mesh.service \
  grafana-server.service \
  gpsd.socket

do
    printf "%-28s " "$UNIT"

    systemctl is-active "$UNIT" \
      2>/dev/null \
      || true
done

実機結果：

ronv-vehicle.service         active
ronv-web.service             active
ronv-mesh.service            active
grafana-server.service       active
gpsd.socket                  active

となりました。

37. 完成したMeshtasticデータ経路

受信：

Remote Meshtastic Node
         ↓
      920MHz
         ↓
RON-V Meshtastic Node
         ↓
        USB
         ↓
 app.mesh.gateway
         ↓
TEXT_MESSAGE_APP
         ↓
 app.mesh.message_store
         ↓
   mesh_messages
         ↓
 /var/lib/ronv/ronv.db

送信：

queue_text.py
      ↓
mesh_messages
state=queued
      ↓
gateway.py
      ↓
Meshtastic USB
      ↓
920MHz Mesh
      ↓
Remote Node
      ↓
ACK
      ↓
gateway.py
      ↓
mesh_messages
state=acked

38. なぜ送信要求をSQLiteキューにするのか

RON-VではMeshtastic USBを、

ronv-mesh.service

だけが開きます。

他のプログラムがUSBを直接開く構成にはしません。

送信したいプログラムは、

mesh_messages

へ、

state = queued

として登録します。

常駐Gatewayがそれを取得し、

queued
   ↓
sending
   ↓
acked

と状態を更新します。

これによりUSB Serialを複数プロセスが奪い合うことを防げます。

39. `received` と `acked`

RX時：

direction = RX
state     = received

TX成功時：

direction = TX
state     = acked

となります。

TXではさらに、

packet_id
ack_type
ack_from
ack_error

を保存します。

これにより、

送信要求
実送信
ACK
失敗

の状態をSQLite上で追跡できます。

40. RSSIとSNR

受信時には、

RSSI
SNR

も保存します。

実機では例えば、

RSSI = -55
SNR  = 13.75

や、

RSSI = -40
SNR  = 10.0

を取得できました。

これらを後からGrafanaで表示すれば、

Mesh通信状態

の可視化にも利用できます。

41. 今回旧実装から整理した部分

既存実装には、

Meshtastic USB
TEXT RX/TX
ACK
SQLite
PRIVATE_APP
L0 packet
Unix socket
同期配送

などが一つの構成に含まれていました。

RON-Vでは機能を分離し、今回は、

USB
TEXT RX
TEXT TX
ACK
SQLite
systemd

だけをMeshモジュールとして独立させました。

これにより、

Meshtastic通信機能

と、

RON-V上位配送機能

を分離できます。

42. 今回確認できたこと

[OK] Meshtastic USB認識

[OK] QinHeng USB Serial

[OK] /dev/serial/by-id

[OK] meshtastic 2.7.11

[OK] protobuf 7.36.0

[OK] Pypubsub 4.0.7

[OK] pyserial 3.5

[OK] SerialInterface

[OK] Local Node ID取得

[OK] ronv.toml [mesh]

[OK] mesh_messages

[OK] SQLite WAL

[OK] TEXT_MESSAGE_APP RX

[OK] from / to

[OK] packet_id

[OK] RSSI

[OK] SNR

[OK] SQLite RX保存

[OK] TX queue

[OK] TEXT_MESSAGE_APP TX

[OK] real ACK

[OK] ACK送信元確認

[OK] systemd Gateway

[OK] systemd RX

[OK] systemd TX

[OK] ronv-mesh.service active

[OK] ronv-mesh.service enabled

[OK] SQLite integrity_check

43. 現在のRON-V構成

ここまでで、

Raspberry Pi 5
│
├─ GNSS
│    └─ gpsd
│
├─ Vehicle
│    └─ AtomS3R
│         ↓
│       SQLite
│
├─ Camera
│    └─ Logitech C270
│
├─ LTE
│    └─ RTL-SDR Blog V4
│         ↓
│       SoapySDR
│         ↓
│       srsRAN
│
├─ Meshtastic
│    └─ USB
│         ↓
│       RX / TX / ACK
│         ↓
│       SQLite
│
├─ SQLite
│    ├─ vehicle_events
│    ├─ lte_measurements
│    └─ mesh_messages
│
├─ Web
│    └─ Flask + Gunicorn
│
├─ Grafana
│
└─ systemd
     ├─ ronv-vehicle.service
     ├─ ronv-web.service
     └─ ronv-mesh.service

となりました。

44. 再起動試験について

ronv-mesh.service は、

enabled

まで確認済みです。

ただし、現在はRON-V全体を構築している途中のため、

sudo reboot

による最終自動起動試験はまだ行いません。

Web UIや統合機能まで構築した後に、RON-V全体を一度まとめて再起動して確認します。

45. 次回

次回は、現在の簡易Flask APIから進めて、

RON-V Web UI

を構築します。

現在は、

と、

/health

だけのWeb基盤ですが、次回は、

Vehicle
GNSS
LTE
Camera
Meshtastic
systemd status

などを車載端末から確認できるダッシュボードへ発展させます。

まとめ

今回はRON-VへMeshtasticをUSB接続し、

RX
TX
ACK
RSSI
SNR
SQLite
systemd

まで実装しました。

特に送信では、

state = queued

としてSQLiteへ登録し、

ronv-mesh.service

が送信処理を担当します。

ACK受信後は、

state = acked
ack_type = real

としてSQLiteへ保存します。

これにより、

Meshtastic
     ↓
USB
     ↓
RON-V
     ↓
SQLite

という通信経路が正式にRON-Vへ追加されました。

次回は、

RON-V #14 — Web UIで車載観測ノードの状態を一画面に表示する

へ進みます。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

RON-V #13 — MeshtasticをUSB接続し、920MHz Mesh通信を車載ノードへ統合する

L0〜L4の定義追記（2026-10-25）

はじめに

1. 今回の目的

2. 今回は実装しないもの

3. RON-V側の構成

4. Meshtastic USB機器を接続する

5. /dev/serial/by-id/ を使用する

6. 現行のPython依存管理

9. PythonからUSB接続を確認する

10. 現行のMesh設定

12. Mesh設定を確認する

13. mesh_messages テーブルを作成する

14. message_store.py を作成する

15. 構文確認

16. mesh_messages を初期化する

17. SQLiteを確認する

18. Meshtastic Gatewayを作成する

19. gateway.py を作成する

20. TXキュー登録プログラムを作成する

21. Python構文確認

22. Gatewayを手動起動する

23. Meshtastic RXを確認する

24. RXデータをSQLiteへ保存する

25. Meshtastic TXを確認する

26. real ACK を確認する

27. systemd用serviceを作成する

28. systemd unitを検証する

29. systemdへ登録する

30. systemdからMesh Gatewayを起動する

31. systemdログを確認する

32. systemd稼働中にRXを確認する

33. systemd稼働中にTXする

34. systemd TX/RXをSQLiteで確認する

35. 自動起動を有効化する

36. RON-Vサービス全体を確認する

37. 完成したMeshtasticデータ経路

38. なぜ送信要求をSQLiteキューにするのか

39. received と acked

40. RSSIとSNR

41. 今回旧実装から整理した部分

42. 今回確認できたこと

43. 現在のRON-V構成

44. 再起動試験について

45. 次回

まとめ

5. `/dev/serial/by-id/` を使用する

13. `mesh_messages` テーブルを作成する

14. `message_store.py` を作成する

16. `mesh_messages` を初期化する

19. `gateway.py` を作成する

26. `real ACK` を確認する

39. `received` と `acked`