RON-V 連載リンク
← 前の記事：#13 — Meshtastic 920MHz Mesh通信
→ 次の記事：#15 — 統合後編：自動配送と再起動後の復帰
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同期メモ
現在のIntegration Layerはapp/integration/が正本です。ControllerはL0を生成し、Vehicle / LTE / RF の新規データをL1へ変換し、L2 summaryではVehicle・LTE・RF・Meshを集約します。L0のMesh配送はsync_deliveries→mesh_messages→ronv-mesh.serviceの順で行い、USBの単一所有を維持します。ronv-controller.timerは10分周期、ronv-delivery.timerは1分周期です。
はじめに
RON-V（Radio Observation Node – Vehicle）では、ここまでRaspberry Pi 5を中心に、車載観測ノードを構成する各機能を個別に構築してきました。
これまでに実装した主な機能は次のとおりです。
- Raspberry Pi 5 基盤
- SQLiteデータベース
- GNSS
- AtomS3Rによる車両イベント取得
- USBカメラ
- RTL-SDR + LTE Band 28測定
- Flask Web UI
- Grafana
- Meshtastic 920MHz
- systemdによる常駐化
ここまでの段階では、それぞれの機能は単独で動作しています。
しかし、RON-Vを「車載観測ノード」として完成させるためには、
観測
↓
データ保存
↓
情報の整理
↓
配送管理
↓
無線送信
↓
ACK確認
までを一つの仕組みとして動かす必要があります。
そこで #14 では、既存モジュールを作り直すのではなく、その上に RON-V Integration Layer（統合層） を追加します。
今回の前編では、最終的に次の経路を自動化します。
各観測モジュール
│
├─ AtomS3R
│ ↓
│ vehicle_events
│
├─ RTL-SDR / LTE
│ ↓
│ lte_measurements
│
├─ RTL-SDR / RF
│ ↓
│ rf_measurements
│
├─ gpsd
│
└─ Meshtastic Gateway
│
▼
RON-V Integration
│
├─ L0 : Node Beacon
├─ L1 : Sensor Fact
└─ L2 : Cycle Summary
│
▼
sync_objects
│
▼
sync_deliveries
│
▼
ronv-delivery.timer
│
▼
mesh_messages
│
▼
ronv-mesh.service
│
▼
Meshtastic 920MHz
│
▼
real ACK
最終的には、人がコマンドを実行しなくても、
ronv-controller.timer
↓
L0生成
↓
L1取込
↓
L2生成
↓
Delivery登録
↓
ronv-delivery.timer
↓
920MHz送信
↓
real ACK
まで一周することを目標にします。
公開記事では、Meshtastic Node ID、USB固有ID、SSID、プライベートIPアドレスなどの実機固有情報はマスクしています。
現行Integration Cycleの要点
現在の app/integration/cycle.py は1回のcycleで、
L0生成
↓
L0 Mesh delivery登録
↓
Vehicle / LTE / RF の新規データをL1化
↓
Vehicle / LTE / RF / MeshをL2 summary化
を行います。L0は R0:<packet_hex> のTEXT_MESSAGE_APP表現で sync_deliveries に登録され、Delivery Workerが mesh_messages へqueueし、ronv-mesh.service が実送信とACK取得を担当します。
1. RON-V統合の設計方針
今回の統合で最も重要なのは、既に動いている機能を巨大なControllerへまとめ直さないことです。
RON-Vでは、各機能の責務を分離します。
ronv-vehicle.service
└─ AtomS3Rからイベント取得
↓
vehicle_eventsへ保存
LTE measure_once
└─ RTL-SDR + srsRANでLTE測定
↓
lte_measurementsへ保存
gpsd.socket
└─ GNSS位置情報を供給
ronv-mesh.service
└─ Meshtastic USBを唯一所有
↓
mesh_messagesを送受信
統合Controllerは、これらのデバイスを直接操作しません。
既にSQLiteへ保存されたデータやgpsdの状態を読み、
L0
L1
L2
という共通形式へ整理します。
さらに、データ生成と無線配送も分離します。
ronv-controller.timer
│
└─ 統合処理
├─ L0生成
├─ 新規L1生成
├─ L2生成
└─ L0 delivery登録
ronv-delivery.timer
│
└─ pending / retryを処理
↓
mesh_messages
ronv-mesh.service
│
└─ 実際の920MHz送信
↓
ACK取得
この構造にする理由の一つが、Meshtastic USBの単一所有です。
USBシリアルデバイスを、
Controller
Delivery Worker
Mesh Gateway
の複数プロセスが同時に開く構成にはしません。
実際にUSBを所有するのは、
ronv-mesh.service
だけです。
ControllerやDelivery WorkerはSQLite経由で連携します。
この構造により、
観測処理
統合処理
配送処理
無線処理
を独立して確認・再起動・改修できるようになります。
2. 前提環境
今回使用しているRON-V実機は、Raspberry Pi 5です。
主な環境は次のとおりです。
Raspberry Pi 5
Debian GNU/Linux 13 (trixie)
arm64
Python 3.13系
SQLite 3.46系
Meshtastic Python 2.7.11
RON-Vのディレクトリ構成は次のようにしています。
Project
/home/ronv/ronv
Python venv
/home/ronv/.venvs/ronv
Config
/home/ronv/ronv/config/ronv.toml
SQLite DB
/var/lib/ronv/ronv.db
既に動作している主なサービスは次のとおりです。
ronv-vehicle.service
ronv-web.service
ronv-mesh.service
gpsd.socket
grafana-server.service
この時点では、今回追加する
ronv-controller.service
ronv-controller.timer
ronv-delivery.service
ronv-delivery.timer
はまだ存在していません。
3. 統合前のバックアップ
統合処理では既存SQLiteへ新しいテーブルを追加します。
そのため、最初にデータベースをバックアップします。
RON-VのSQLiteは各サービスが使用中なので、単純な
cp ronv.db ronv.db.bak
ではなく、SQLiteの .backup を使用します。
まずプロジェクトディレクトリへ移動します。
cd /home/ronv/ronv
バックアップ用ディレクトリを作成します。
mkdir -p backups
日時付きファイル名を作ります。
STAMP=$(date +%Y%m%d_%H%M%S)
DB_BAK="backups/ronv_pre_sync_${STAMP}.db"
SQLiteのオンラインバックアップを実行します。
sqlite3 /var/lib/ronv/ronv.db \
".backup '$DB_BAK'"
バックアップ後に整合性を確認します。
sqlite3 "$DB_BAK" \
'PRAGMA integrity_check;'
結果：
ok
これで、統合前のSQLiteを安全に保存できました。
この時点で既に存在している主なデータテーブルは、
vehicle_events
lte_measurements
mesh_messages
ronv_meta
です。
今回の統合では、これらの既存テーブルを置き換えません。
新たに、
sync_objects
sync_deliveries
sync_cursors
l0_beacons
などの統合用テーブルを追加していきます。
4. RON-VのL0 / L1 / L2とは
実装へ進む前に、RON-Vで使用する情報レベルを整理します。
今回の設計では、観測情報を次のように分類します。
L0 : Node Beacon
L1 : Sensor Fact
L2 : Summary / Change
L3 : Evidence
L4 : Raw file / Raw log
L0：Node Beacon
RON-Vノード自身の基本状態です。
例えば、
Node ID
Sequence
時刻
緯度
経度
Battery
Network状態
Alarm
Need
などを、非常に小さい固定長packetとして表現します。
RON-Vでは23-byte固定packetにします。
L1：Sensor Fact
センサーから得られた「事実」です。
例えば、
AtomS3R
button
shock
shutdown_knock
LTE
Band
EARFCN
PCI
PSS受信レベル
などです。
L2：Summary
一定時間の観測結果をまとめた情報です。
例えば、
直近600秒の
Vehicle event数
Shock数
LTE測定数
確認PCI数
最強PSS
Mesh送信数
Service状態
などをまとめます。
L3 / L4
今回はまだ本格実装しません。
将来的には、
L3
イベント証拠
LTE解析結果
重要イベント要約
L4
JPEG
LTE raw_log
生ログ
などを扱う予定です。
ここで重要なのは、
L0
↓
L1
↓
L2
↓
L3
↓
L4
という順番で処理するパイプラインではない、という点です。
L0～L4は 情報量と用途による分類 として扱います。
今回の前編では、
L0
L1
L2
までを実装します。
5. 統合用ディレクトリを作る
ここから実際にRON-Vの統合層を追加していきます。
まず、統合処理専用のPythonパッケージを作ります。
cd /home/ronv/ronv
mkdir -p app/integration
touch app/integration/__init__.py
確認します。
find app/integration \
-maxdepth 1 \
-type f \
-o \
-type d
最初は、
app/integration
app/integration/__init__.py
だけです。
ここへ、
sync_store.py
l0_packet.py
l0_store.py
l1_store.py
l2_summary.py
cycle.py
delivery_worker.py
を順番に追加していきます。
6. sync_objects / sync_deliveries / sync_cursors を作る
RON-Vの各観測データを統合するため、SQLiteへ3つの共通テーブルを追加します。
sync_objects
sync_deliveries
sync_cursors
それぞれの役割は次のとおりです。
sync_objects
観測情報そのものを管理
sync_deliveries
その情報をどこへ送るか、
どこまで配送できたかを管理
sync_cursors
既存テーブルのどこまで
L1へ取り込んだかを管理
特に重要なのは、
観測データ
と、
配送状態
を別テーブルにしたことです。
例えばL0が正常に生成されたあと、Mesh送信に失敗した場合でも、
sync_objects
ready
sync_deliveries
retry
のように分離して管理できます。
7. sync_store.pyを作る
app/integration/sync_store.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/sync_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
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 initialize_sync_schema(
config: dict | None = None,
) -> None:
cfg = config or load_config()
con = connect_db(cfg)
try:
con.executescript(
"""
CREATE TABLE IF NOT EXISTS
sync_objects (
object_uid TEXT PRIMARY KEY,
source_table TEXT,
source_id INTEGER,
data_type TEXT NOT NULL,
level INTEGER NOT NULL
CHECK (
level BETWEEN 0 AND 4
),
priority INTEGER NOT NULL
DEFAULT 50
CHECK (
priority BETWEEN 0 AND 100
),
event_time_jst TEXT,
created_at_jst TEXT NOT NULL,
source_node TEXT NOT NULL
DEFAULT 'ronv',
payload_json TEXT,
file_path TEXT,
mime_type TEXT,
bytes INTEGER
CHECK (
bytes IS NULL
OR bytes >= 0
),
sha256 TEXT,
ttl_sec INTEGER
CHECK (
ttl_sec IS NULL
OR ttl_sec >= 0
),
expires_at_jst TEXT,
supersedes_uid TEXT,
state TEXT NOT NULL
DEFAULT 'ready'
CHECK (
state IN (
'ready',
'held',
'superseded',
'cancelled'
)
),
metadata_json TEXT,
FOREIGN KEY (
supersedes_uid
)
REFERENCES sync_objects (
object_uid
)
);
CREATE INDEX IF NOT EXISTS
idx_sync_objects_source
ON sync_objects (
source_table,
source_id
);
CREATE INDEX IF NOT EXISTS
idx_sync_objects_level_state
ON sync_objects (
level,
state,
priority
);
CREATE TABLE IF NOT EXISTS
sync_deliveries (
id INTEGER
PRIMARY KEY AUTOINCREMENT,
object_uid TEXT NOT NULL,
representation TEXT NOT NULL
DEFAULT 'native',
target TEXT NOT NULL,
transport TEXT NOT NULL,
priority INTEGER NOT NULL
DEFAULT 50
CHECK (
priority BETWEEN 0 AND 100
),
state TEXT NOT NULL
DEFAULT 'pending'
CHECK (
state IN (
'pending',
'retry',
'inflight',
'sent',
'acked',
'held',
'failed',
'cancelled'
)
),
attempt_count INTEGER
NOT NULL
DEFAULT 0
CHECK (
attempt_count >= 0
),
created_at_jst TEXT NOT NULL,
next_retry_at_jst TEXT,
last_attempt_at_jst TEXT,
sent_at_jst TEXT,
acked_at_jst TEXT,
remote_id TEXT,
ack_token TEXT,
bytes INTEGER
CHECK (
bytes IS NULL
OR bytes >= 0
),
sha256 TEXT,
last_status_code INTEGER,
last_latency_ms REAL,
last_error TEXT,
metadata_json TEXT,
FOREIGN KEY (
object_uid
)
REFERENCES sync_objects (
object_uid
)
ON DELETE CASCADE,
UNIQUE (
object_uid,
representation,
target,
transport
)
);
CREATE INDEX IF NOT EXISTS
idx_sync_deliveries_schedule
ON sync_deliveries (
state,
next_retry_at_jst,
priority,
id
);
CREATE INDEX IF NOT EXISTS
idx_sync_deliveries_transport
ON sync_deliveries (
transport,
state,
priority
);
CREATE INDEX IF NOT EXISTS
idx_sync_deliveries_object
ON sync_deliveries (
object_uid
);
CREATE TABLE IF NOT EXISTS
sync_cursors (
source_name TEXT
PRIMARY KEY,
last_source_id INTEGER
NOT NULL
DEFAULT 0,
updated_at_jst TEXT
NOT NULL
);
"""
)
con.commit()
finally:
con.close()
def print_status(
config: dict | None = None,
) -> None:
cfg = config or load_config()
con = connect_db(cfg)
try:
print(
"===== RON-V SYNC ====="
)
for table in (
"sync_objects",
"sync_deliveries",
"sync_cursors",
):
count = con.execute(
f"""
SELECT COUNT(*)
FROM {table}
"""
).fetchone()[0]
print(
f"{table:20s}",
count,
)
finally:
con.close()
def main() -> None:
cfg = load_config()
initialize_sync_schema(
cfg
)
print_status(
cfg
)
if __name__ == "__main__":
main()
PY
8. sync_objectsの役割
sync_objectsは、RON-Vが扱う情報そのものを管理します。
主なカラムは次のとおりです。
object_uid
source_table
source_id
data_type
level
priority
event_time_jst
created_at_jst
source_node
payload_json
file_path
mime_type
bytes
sha256
state
metadata_json
object_uid
RON-V内部で各データを一意に識別するIDです。
例えば、
ronv-l0-0001-xxxxxxxxxx-00001
ronv-l1-vehicle-00000058
ronv-l1-lte-00000003
ronv-l2-xxxxxxxxxxxxx
のような形式を使います。
SQLiteの数値IDだけに依存せず、RON-V内で意味のあるIDを付けます。
source_table / source_id
どの元データから生成されたかを記録します。
例えばVehicleイベントなら、
source_table = vehicle_events
source_id = 58
LTE観測なら、
source_table = lte_measurements
source_id = 3
となります。
これにより、
sync_objects
↓
元のセンサーデータ
を追跡できます。
data_type
情報の種類です。
今回使用する主な値は、
node_beacon
vehicle_event
lte_cell_observation
cycle_summary
です。
level
先ほど定義した、
0 ～ 4
の情報レベルです。
SQLite側でも、
CHECK (
level BETWEEN 0 AND 4
)
を付けています。
priority
配送や処理の優先度です。
0 高優先
100 低優先
という方向で使用します。
今回の例では、
Vehicle button / shock 10
その他Vehicle 20
LTE 30
L2 Summary 40
L0 Beacon 50
のようにしています。
payload_json
L1やL2などの軽量データはJSONとして保存します。
一方、将来のL4では、
JPEG
raw_log
大きなログ
などを直接JSONへ詰め込まず、
file_path
mime_type
bytes
sha256
を使用することを想定しています。
state
sync_objectsのstateは、
ready
held
superseded
cancelled
です。
これは配送状態ではありません。
例えば、
ready
は、
この観測オブジェクト自体は有効
という意味です。
送信済みかどうかはsync_deliveries側で管理します。
9. sync_deliveriesの役割
sync_deliveriesは、各sync_objectの配送状態を管理します。
例えば、
L0 Beacon
↓
Meshで特定Nodeへ送る
という配送1件が、sync_deliveriesの1行になります。
主な状態は、
pending
retry
inflight
sent
acked
held
failed
cancelled
です。
今回のMesh配送では主に、
pending
↓
inflight
↓
acked
を使います。
送信に失敗した場合は、
inflight
↓
retry
へ戻します。
UNIQUE制約
重要なのが次の制約です。
UNIQUE (
object_uid,
representation,
target,
transport
)
つまり、
同じobject
同じ表現形式
同じ宛先
同じtransport
の配送は1件しか登録できません。
例えば、
object_uid ronv-l0-...
representation l0-text
target !<REMOTE_NODE_ID>
transport mesh
という組み合わせは一意です。
Controllerが何らかの理由で同じdelivery登録処理を再度呼んでも、同一配送が無制限に増えることを防ぎます。
10. sync_cursorsの役割
VehicleやLTEでは、既にSQLiteへ過去データが保存されています。
今回の統合開始時に、
過去の数十件・数百件を
全部L1へ変換する
ことはしません。
そこで、
どこから先を新規データとして扱うか
をsync_cursorsへ保存します。
構造は非常に単純です。
source_name
last_source_id
updated_at_jst
例えば、
vehicle_events 57
lte_measurements 2
で統合を開始した場合、
id > 57
id > 2
のデータだけをL1化します。
この仕組みは後ほどl1_store.pyで使用します。
11. Python構文を確認する
SQLiteへ反映する前に、まずPythonの構文を確認します。
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/sync_store.py
何も表示されなければ構文エラーはありません。
12. sync schemaをSQLiteへ作成する
実行します。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.sync_store
初回実行時の結果：
===== RON-V SYNC =====
sync_objects 0
sync_deliveries 0
sync_cursors 0
まだデータは入っていないため、すべて0件です。
13. SQLiteでテーブルを確認する
SQLite側でも確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
name
FROM sqlite_master
WHERE type='table'
AND name LIKE 'sync_%'
ORDER BY name;
PRAGMA integrity_check;
SQL
結果：
name
---------------
sync_cursors
sync_deliveries
sync_objects
integrity_check
---------------
ok
既存DBを壊さず、
sync_objects
sync_deliveries
sync_cursors
を追加できました。
14. 既存テーブルが残っていることも確認する
統合テーブルだけでなく、既存の観測テーブルがそのまま残っていることも確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
PRAGMA integrity_check;
SQL
この時点では少なくとも、
vehicle_events
lte_measurements
mesh_messages
ronv_meta
sync_objects
sync_deliveries
sync_cursors
が共存します。
RON-Vの統合では、
既存テーブルを統合テーブルへ置き換える
のではなく、
既存センサーデータ
↓
統合層から参照
↓
sync_objectsへ整理
という構成にしています。
15. DBトランザクションの考え方
RON-Vでは、1つの論理処理に複数のSQLite更新が必要な場合、
BEGIN IMMEDIATE
を使用して同一transaction内で処理します。
例えば後ほどL1を生成するときは、
Vehicle rowを読む
↓
sync_objectsへINSERT
↓
sync_cursorsをUPDATE
を1つのtransactionで処理します。
もし、
sync_objectsへのINSERT成功
↓
プロセス停止
↓
cursor更新失敗
となると、次回同じ元データを再度処理する可能性があります。
そこで、
BEGIN IMMEDIATE
↓
sync_objects INSERT
↓
cursor UPDATE
↓
COMMIT
とします。
配送でも同じ考え方を使います。
mesh_messages INSERT
↓
sync_deliveries UPDATE
も同一transactionで処理します。
RON-Vでは、SQLiteを単なるログ保存先ではなく、
各サービス間の状態共有
にも利用しています。
16. 共通sync基盤が完成
ここまででRON-Vの統合層の土台ができました。
既存観測データ
│
▼
sync_objects
│
├─ level
├─ priority
├─ data_type
├─ payload
└─ source
│
▼
sync_deliveries
│
├─ target
├─ transport
├─ pending
├─ inflight
├─ retry
└─ acked
さらに、
sync_cursors
でVehicle/LTEの新規データ位置を管理できます。
次は、この共通基盤へ最初の実データとして、
L0 Node Beacon
を載せます。
RON-VではL0を、
23-byte固定長
CRC16付き
の小型packetとして実装します。
17. L0 Node Beaconを23-byte固定長にする
RON-Vで最も小さい情報単位として、L0 Node Beaconを作ります。
L0は、
現在時刻
Node ID
Sequence
GNSS位置
Battery
Network状態
Alarm
Need
など、ノード自身の基本状態だけを持つ小型packetです。
RON-VではL0を 23-byte固定長 とします。
packet構造は次のとおりです。
version u8
node_id u16
sequence u16
unix_time u32
latitude i32 × 1e7
longitude i32 × 1e7
battery u8
network_flags u8
alarm_flags u8
need_flags u8
CRC16 u16
合計：
1
+ 2
+ 2
+ 4
+ 4
+ 4
+ 1
+ 1
+ 1
+ 1
+ 2
= 23 bytes
バイトオーダーはBig Endianとします。
Pythonのstructでは、
">BHHIiiBBBB"
を使います。
CRCを付ける前のbodyは21 bytesです。
最後にCRC16を2 bytes付け、
21 + 2 = 23 bytes
になります。
18. GNSS不明値を定義する
車両が屋内にいる場合や、起動直後などはGNSS Fixできないことがあります。
しかしL0は固定長なので、
latitude = None
longitude = None
をそのまま送ることはできません。
そこでGNSS不明値として、
GPS_UNKNOWN = -2147483648
を使用します。
これはsigned 32-bit整数の最小値です。
緯度・経度が取得できた場合は、
degree × 10,000,000
で整数化します。
例えば、
35.0000000°
なら、
350000000
です。
これによりfloatを直接送らず、4 bytesのsigned integerとして扱えます。
Batteryについても、まだ車両電源状態をL0へ接続していないため、
BATTERY_UNKNOWN = 255
を未取得値として使用します。
19. l0_packet.pyを作る
app/integration/l0_packet.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/l0_packet.py <<'PY'
from __future__ import annotations
import binascii
import struct
VERSION = 1
GPS_UNKNOWN = -2147483648
BATTERY_UNKNOWN = 255
PACKET_SIZE = 23
def crc16(
data: bytes,
) -> int:
return binascii.crc_hqx(
data,
0xFFFF,
)
def coord_encode(
value: float | None,
) -> int:
if value is None:
return GPS_UNKNOWN
return int(
round(
float(value)
* 10_000_000
)
)
def coord_decode(
value: int,
) -> float | None:
if value == GPS_UNKNOWN:
return None
return (
value
/ 10_000_000.0
)
def pack_l0(
*,
node_id: int,
sequence: int,
unix_time: int,
latitude: float | None = None,
longitude: float | None = None,
battery: int = BATTERY_UNKNOWN,
network_flags: int = 0,
alarm_flags: int = 0,
need_flags: int = 0,
) -> bytes:
body = struct.pack(
">BHHIiiBBBB",
VERSION,
node_id,
sequence,
unix_time,
coord_encode(latitude),
coord_encode(longitude),
battery,
network_flags,
alarm_flags,
need_flags,
)
crc = crc16(
body
)
packet = (
body
+ struct.pack(
">H",
crc,
)
)
if len(packet) != PACKET_SIZE:
raise RuntimeError(
"unexpected L0 packet size"
)
return packet
def unpack_l0(
packet: bytes,
) -> dict:
if len(packet) != PACKET_SIZE:
raise ValueError(
f"invalid packet size: "
f"{len(packet)}"
)
body = packet[:-2]
received_crc = struct.unpack(
">H",
packet[-2:],
)[0]
calculated_crc = crc16(
body
)
values = struct.unpack(
">BHHIiiBBBB",
body,
)
return {
"version":
values[0],
"node_id":
values[1],
"sequence":
values[2],
"unix_time":
values[3],
"latitude":
coord_decode(
values[4]
),
"longitude":
coord_decode(
values[5]
),
"battery":
values[6],
"network_flags":
values[7],
"alarm_flags":
values[8],
"need_flags":
values[9],
"crc":
received_crc,
"crc_ok":
(
received_crc
== calculated_crc
),
}
PY
20. l0_packet.pyの構文確認
実行前に構文確認します。
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/l0_packet.py
何も表示されなければOKです。
21. L0 packetを単体試験する
まず実GNSSを使わず、既知の値でpacket生成とdecodeを確認します。
/home/ronv/.venvs/ronv/bin/python - <<'PY'
from app.integration.l0_packet import (
PACKET_SIZE,
pack_l0,
unpack_l0,
)
packet = pack_l0(
node_id=1,
sequence=123,
unix_time=1700000000,
latitude=35.0,
longitude=137.0,
battery=255,
network_flags=0x04,
)
data = unpack_l0(
packet
)
print(
"packet_bytes:",
len(packet),
)
print(
"packet_hex:",
packet.hex(),
)
print(
"crc_ok:",
data["crc_ok"],
)
print(
"node_id:",
data["node_id"],
)
print(
"sequence:",
data["sequence"],
)
print(
"latitude:",
data["latitude"],
)
print(
"longitude:",
data["longitude"],
)
assert len(packet) == PACKET_SIZE
assert data["crc_ok"] is True
assert data["node_id"] == 1
assert data["sequence"] == 123
print(
"RON-V L0 PACKET TEST OK"
)
PY
実機では、
packet_bytes: 23
crc_ok: True
node_id: 1
sequence: 123
latitude: 35.0
longitude: 137.0
RON-V L0 PACKET TEST OK
となりました。
これで、
23-byte固定長
Big Endian
緯度・経度整数化
CRC16
encode
decode
まで確認できました。
22. CRC16-CCITT
今回使用しているCRCは、
binascii.crc_hqx(
data,
0xFFFF,
)
です。
CRC計算対象は、
version
node_id
sequence
unix_time
latitude
longitude
battery
network_flags
alarm_flags
need_flags
までの21 bytesです。
最後のCRC16自体はCRC計算には含めません。
受信側では、
受信packet
↓
先頭21 bytesからCRC再計算
↓
末尾2 bytesと比較
します。
unpack_l0()では、その結果を、
"crc_ok": True
または、
"crc_ok": False
で返します。
920MHzなどの無線経路へ載せる前に、
L0 packet自身の整合性
を確認できるようにしました。
23. L0をSQLiteへ保存する
次に、このpacketを実際のRON-V状態から作ります。
L0専用テーブルとして、
l0_beacons
を追加します。
保存する内容は、
object_uid
created_at_jst
version
node_id
sequence
unix_time
latitude
longitude
battery
network_flags
alarm_flags
need_flags
CRC
packet_bytes
packet_hex
です。
さらに同じL0を、
sync_objects
にも登録します。
構造としては、
l0_beacons
│
└─ L0のbinary packet情報
sync_objects
│
└─ RON-V統合層から見た
Node Beacon object
となります。
24. l0_store.pyを作る
app/integration/l0_store.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/l0_store.py <<'PY'
from __future__ import annotations
import hashlib
import inspect
import json
import subprocess
import time
from datetime import datetime
from zoneinfo import ZoneInfo
from app.common.config import load_config
from app.common.db import connect_db
from app.gnss.gpsd_reader import (
get_gnss_position,
)
from app.integration.l0_packet import (
pack_l0,
unpack_l0,
)
def now_jst(
config: dict,
) -> str:
timezone = ZoneInfo(
config["project"]["timezone"]
)
return datetime.now(
timezone
).isoformat(
timespec="seconds"
)
def initialize_l0_schema(
config: dict | None = None,
) -> None:
cfg = config or load_config()
con = connect_db(
cfg
)
try:
con.executescript(
"""
CREATE TABLE IF NOT EXISTS
l0_beacons (
id INTEGER
PRIMARY KEY AUTOINCREMENT,
object_uid TEXT
NOT NULL UNIQUE,
created_at_jst TEXT
NOT NULL,
version_u8 INTEGER
NOT NULL,
node_id_u16 INTEGER
NOT NULL,
sequence_u16 INTEGER
NOT NULL,
unix_time_u32 INTEGER
NOT NULL,
latitude_deg REAL,
longitude_deg REAL,
battery_pct_u8 INTEGER
NOT NULL,
network_flags_u8 INTEGER
NOT NULL,
alarm_flags_u8 INTEGER
NOT NULL,
need_flags_u8 INTEGER
NOT NULL,
crc_u16 INTEGER
NOT NULL,
packet_bytes INTEGER
NOT NULL,
packet_hex TEXT
NOT NULL
);
CREATE INDEX IF NOT EXISTS
idx_l0_beacons_node_sequence
ON l0_beacons (
node_id_u16,
sequence_u16
);
CREATE INDEX IF NOT EXISTS
idx_l0_beacons_created
ON l0_beacons (
created_at_jst
);
"""
)
con.commit()
finally:
con.close()
def get_position(
config: dict,
) -> tuple[
str,
float | None,
float | None,
]:
try:
parameters = inspect.signature(
get_gnss_position
).parameters
if len(parameters) == 0:
result = (
get_gnss_position()
)
else:
result = (
get_gnss_position(
config
)
)
except Exception as exc:
return (
f"error:{type(exc).__name__}",
None,
None,
)
if not isinstance(
result,
dict,
):
return (
"invalid",
None,
None,
)
status = str(
result.get(
"gps_status",
"unknown",
)
)
latitude = result.get(
"latitude",
result.get(
"lat"
),
)
longitude = result.get(
"longitude",
result.get(
"lon"
),
)
try:
latitude = (
float(latitude)
if latitude is not None
else None
)
except Exception:
latitude = None
try:
longitude = (
float(longitude)
if longitude is not None
else None
)
except Exception:
longitude = None
return (
status,
latitude,
longitude,
)
def network_flags() -> int:
flags = 0
try:
result = subprocess.run(
[
"nmcli",
"-t",
"-f",
"TYPE,STATE",
"device",
"status",
],
capture_output=True,
text=True,
timeout=3,
check=False,
)
if (
result.returncode == 0
and "wifi:connected"
in result.stdout
):
flags |= 0x04
except Exception:
pass
return flags
def next_sequence(
con,
node_id: int,
) -> int:
row = con.execute(
"""
SELECT
sequence_u16
FROM l0_beacons
WHERE node_id_u16=?
ORDER BY id DESC
LIMIT 1
""",
(
node_id,
),
).fetchone()
if row is None:
return 1
return (
int(
row["sequence_u16"]
)
+ 1
) & 0xFFFF
def produce_l0(
config: dict | None = None,
) -> dict:
cfg = config or load_config()
initialize_l0_schema(
cfg
)
node_id = int(
cfg.get(
"integration",
{},
).get(
"node_id",
1,
)
)
(
gps_status,
latitude,
longitude,
) = get_position(
cfg
)
flags = network_flags()
unix_time = int(
time.time()
)
created = now_jst(
cfg
)
con = connect_db(
cfg
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
sequence = next_sequence(
con,
node_id,
)
packet = pack_l0(
node_id=node_id,
sequence=sequence,
unix_time=unix_time,
latitude=latitude,
longitude=longitude,
battery=255,
network_flags=flags,
alarm_flags=0,
need_flags=0,
)
decoded = unpack_l0(
packet
)
object_uid = (
f"ronv-l0-"
f"{node_id:04x}-"
f"{unix_time}-"
f"{sequence:05d}"
)
packet_hex = (
packet.hex()
)
cur = con.execute(
"""
INSERT INTO l0_beacons (
object_uid,
created_at_jst,
version_u8,
node_id_u16,
sequence_u16,
unix_time_u32,
latitude_deg,
longitude_deg,
battery_pct_u8,
network_flags_u8,
alarm_flags_u8,
need_flags_u8,
crc_u16,
packet_bytes,
packet_hex
)
VALUES (
?,?,?,?,?,?,?,?,?,?,?,?,?,?,?
)
""",
(
object_uid,
created,
decoded["version"],
decoded["node_id"],
decoded["sequence"],
decoded["unix_time"],
decoded["latitude"],
decoded["longitude"],
decoded["battery"],
decoded["network_flags"],
decoded["alarm_flags"],
decoded["need_flags"],
decoded["crc"],
len(packet),
packet_hex,
),
)
source_id = int(
cur.lastrowid
)
payload = json.dumps(
{
"schema":
"ronv.l0.v1",
"packet_hex":
packet_hex,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
raw = payload.encode(
"utf-8"
)
sha256 = hashlib.sha256(
raw
).hexdigest()
metadata = json.dumps(
{
"generator":
"l0_store",
"schema_version":
1,
"gps_status":
gps_status,
"network_flags":
flags,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
con.execute(
"""
INSERT INTO sync_objects (
object_uid,
source_table,
source_id,
data_type,
level,
priority,
event_time_jst,
created_at_jst,
source_node,
payload_json,
bytes,
sha256,
state,
metadata_json
)
VALUES (
?,
'l0_beacons',
?,
'node_beacon',
0,
50,
?,
?,
'ronv',
?,
?,
?,
'ready',
?
)
""",
(
object_uid,
source_id,
created,
created,
payload,
len(raw),
sha256,
metadata,
),
)
con.commit()
except Exception:
con.rollback()
raise
finally:
con.close()
return {
"object_uid":
object_uid,
"sequence":
sequence,
"gps_status":
gps_status,
"latitude":
latitude,
"longitude":
longitude,
"network_flags":
flags,
"packet_bytes":
len(packet),
"packet_hex":
packet_hex,
"mesh_text":
"R0:" + packet_hex,
"crc_ok":
decoded["crc_ok"],
}
def main() -> None:
result = produce_l0()
print(
"===== RON-V L0 ====="
)
for key, value in result.items():
print(
f"{key:15s}:",
value,
)
if __name__ == "__main__":
main()
PY
25. gps_statusの扱い
ここで、実装中に1点修正が必要になりました。
GNSS readerの実際の返却値を確認すると、
{
"gps_status": "no_fix",
"latitude": None,
"longitude": None,
}
という形式でした。
そのためL0側でも、
result.get(
"gps_status",
"unknown",
)
を見る必要があります。
最初に、
result["status"]
のように参照してしまうと、実際にはGNSS readerが返しているgps_statusを取得できません。
最終版では、
status = str(
result.get(
"gps_status",
"unknown",
)
)
としました。
26. Wi-Fi接続状態をnetwork_flagsへ入れる
L0ではNetwork状態も1 byte持たせています。
今回はまずWi-Fi接続だけを実装しました。
NetworkManagerの状態を、
nmcli \
-t \
-f TYPE,STATE \
device status
で取得します。
Wi-Fiが、
wifi:connected
なら、
flags |= 0x04
とします。
つまり現時点では、
network_flags bit 2
をWi-Fi接続として使用しています。
将来的には、
LTE
Wi-Fi
Mesh
Backhaul
など複数の通信状態をbit flagとしてまとめることもできます。
27. sequenceを自動採番する
L0には、
sequence_u16
を持たせています。
最新L0をSQLiteから取得し、
return (
int(
row["sequence_u16"]
)
+ 1
) & 0xFFFF
で次のsequenceを決めています。
最初のpacketなら、
1
です。
その後、
1
2
3
4
...
と増加します。
u16なので最大値は、
65535
です。
その次は、
0
へwrapします。
28. l0_store.pyの構文確認
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/l0_store.py
何も表示されなければOKです。
29. 実GNSS状態でL0を生成する
実行します。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l0_store
実機では屋内だったためGNSSはFixしていませんでした。
結果は、
gps_status : no_fix
latitude : None
longitude : None
network_flags : 4
packet_bytes : 23
crc_ok : True
となりました。
これはエラーではありません。
GNSSがFixしていない場合でも、
latitude
longitude
にはGPS_UNKNOWNがbinary packet上で格納されます。
30. GNSS no_fix時のpacket
実機で生成されたL0の例です。
sequence : 2
gps_status : no_fix
latitude : None
longitude : None
network_flags : 4
packet_bytes : 23
crc_ok : True
packet例：
01000100026a9ab9018000000080000000ff040000ef4c
これを分解すると、
01
version
0001
node_id
0002
sequence
6a9ab901
unix_time
80000000
latitude = GPS_UNKNOWN
80000000
longitude = GPS_UNKNOWN
ff
battery = unknown
04
network_flags
00
alarm_flags
00
need_flags
ef4c
CRC16
となります。
GNSS未取得であっても、
packet formatそのものは正常
です。
31. l0_beaconsを確認する
SQLiteへ保存されたL0を確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
id,
object_uid,
sequence_u16,
latitude_deg,
longitude_deg,
network_flags_u8,
packet_bytes,
packet_hex
FROM l0_beacons
ORDER BY id DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
確認するポイントは、
sequence_u16
packet_bytes
packet_hex
です。
packet_bytesは必ず、
23
になります。
また、
integrity_check
---------------
ok
であることも確認します。
32. sync_objectsにも登録されていることを確認する
L0生成時には、
l0_beacons
だけでなく、
sync_objects
にも同一object_uidで登録しています。
確認：
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
object_uid,
source_table,
source_id,
data_type,
level,
priority,
bytes,
state,
metadata_json
FROM sync_objects
WHERE level=0
ORDER BY created_at_jst DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
L0の場合、
source_table l0_beacons
data_type node_beacon
level 0
priority 50
state ready
となります。
metadataには、
{
"generator": "l0_store",
"gps_status": "no_fix",
"network_flags": 4,
"schema_version": 1
}
のような情報を保存しています。
33. L0生成とsync_objects登録を同一transactionにする
L0生成では、
l0_beacons INSERT
↓
sync_objects INSERT
を別々の処理にはしていません。
BEGIN IMMEDIATE
後に、
l0_beacons
sync_objects
の両方をINSERTし、
COMMIT
します。
構造は、
BEGIN IMMEDIATE
↓
sequence決定
↓
23-byte L0生成
↓
l0_beacons INSERT
↓
sync_objects INSERT
↓
COMMIT
です。
例えば途中で、
l0_beaconsだけ作成
sync_objects登録前に停止
という状態になることを防ぎます。
RON-Vでは、こうした
1つの論理処理
をできるだけ1transactionへまとめています。
34. L0の現在の構成
ここまででL0は次の状態になりました。
gpsd
│
├─ gps_status
├─ latitude
└─ longitude
│
▼
l0_store.py
│
├─ node_id
├─ sequence
├─ unix_time
├─ battery
├─ network_flags
├─ alarm_flags
└─ need_flags
│
▼
l0_packet.py
│
├─ 23 bytes
└─ CRC16
│
├─────────────┐
│ │
▼ ▼
l0_beacons sync_objects
GNSSが取れない場合も、
GPS_UNKNOWN
を使用することで固定長packetを維持できます。
35. Meshへ送る形式を決める
L0 packetそのものは23 bytesです。
しかし既存のRON-V Meshtastic Gatewayは、
TEXT_MESSAGE_APP
での送受信とreal ACKが既に確認できています。
この段階で別の独自binary portを追加すると、
既に安定しているMesh Gateway
へ変更を加えることになります。
そこで今回の統合では、まず既存Gatewayをそのまま使います。
23-byte packetをhex文字列へ変換します。
23 bytes
↓
46文字hex
さらに識別用prefix、
R0:
を付けます。
結果、
R0:<46文字hex>
となります。
例えば、
R0:01000100026a9ab9018000000080000000ff040000ef4c
です。
サイズは、
3文字 + 46文字
= 49 bytes
程度です。
この形式なら既存の、
TEXT_MESSAGE_APP
をそのまま利用できます。
次は、このL0を実際にMeshtasticへ送り、
queued
↓
sending
↓
acked
まで確認します。
36. L0をMeshtasticへ実送信する
L0 packetを、
R0:<46文字hex>
というTEXT形式へ変換できたので、実際にMeshtasticへ送信します。
この段階で重要なのは、RON-VからMeshtastic USBを直接開かないことです。
既に、
ronv-mesh.service
がMeshtasticデバイスを所有しています。
そのためL0側は、
L0
↓
mesh_messages
↓
ronv-mesh.service
↓
Meshtastic
という経路を使います。
実際の送信先Node IDは公開記事では、
!<REMOTE_NODE_ID>
としてマスクします。
37. 最初のL0配送を登録する
最初の実送信では、L0 sequence 2を使用しました。
生成されたpacket：
01000100026a9ab9018000000080000000ff040000ef4c
Meshtasticへ送る文字列：
R0:01000100026a9ab9018000000080000000ff040000ef4c
文字数は49 bytes程度です。
このL0を、
sync_deliveries
へ登録します。
配送情報の考え方は次のとおりです。
object_uid
L0のobject_uid
representation
l0-text
target
!<REMOTE_NODE_ID>
transport
mesh
state
pending
まだ送信していないので、
state = pending
attempt_count = 0
から開始します。
38. 既存Mesh Gatewayへqueueする
配送対象を既存の、
mesh_messages
へ登録します。
概念的には、
sync_deliveries
pending
↓
mesh_messages
queued
↓
ronv-mesh.service
です。
ronv-mesh.serviceは既に#13で、
TEXT_MESSAGE_APP
real ACK
まで動作確認済みです。
そのため今回の統合では、
sendData()
SerialInterface()
を新しいプロセスから直接呼びません。
39. 実際にreal ACKまで確認する
L0 sequence 2を送信した結果、
mesh_messages id = 8
として登録されました。
状態は、
queued
↓
sending
↓
acked
と遷移しました。
Meshtastic側のpacket ID：
1614524260
ACK：
ack_type real
ack_error NONE
ronv-mesh.serviceのjournalでは、
MESH TX id=8 ... ack=real latency_ms=2604
を確認できました。
つまり、
L0
↓
TEXT_MESSAGE_APP
↓
Meshtastic 920MHz
↓
相手Node
↓
real ACK
まで実際の電波経路を通過しています。
40. SQLiteで配送結果を確認する
配送状態を確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
id,
object_uid,
state,
attempt_count,
remote_id,
last_error
FROM sync_deliveries
ORDER BY id DESC
LIMIT 5;
SELECT
id,
direction,
to_id,
text,
state,
packet_id,
ack_type,
ack_error
FROM mesh_messages
WHERE direction='TX'
ORDER BY id DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
実機では、
sync_deliveries
state acked
attempt_count 1
remote_id meshtastic:1614524260
last_error NULL
となりました。
mesh_messages：
state acked
packet_id 1614524260
ack_type real
ack_error NONE
SQLite：
integrity_check
---------------
ok
これでL0を実際にMeshへ載せられることを確認できました。
41. 次はL1 Sensor Factを作る
L0は、
RON-Vノード自身の状態
でした。
次に作るL1は、
センサーから得られた事実
です。
今回L1へ変換する元データは2種類です。
vehicle_events
lte_measurements
構造：
AtomS3R
↓
vehicle_events
↓
L1 Vehicle Fact
RTL-SDR
↓
lte_measurements
↓
L1 LTE Fact
既存テーブルを消したり変更したりせず、
元データ
↓
sync_objects
へL1として追加します。
42. 過去データを全部L1化しない
統合開始時点で、SQLiteには既にデータがありました。
実機では、
vehicle_events 57
lte_measurements 2
でした。
ここで何も考えず、
SELECT *
FROM vehicle_events;
して全件L1化すると、過去の試験データまで大量に同期対象になります。
今回は、
統合開始後に発生した新しいデータ
だけをL1化したいので、
sync_cursors
を使用します。
初期cursor：
vehicle_events 57
lte_measurements 2
つまり、
vehicle_events
id > 57
lte_measurements
id > 2
だけを新規データとして扱います。
43. l1_store.pyを作る
app/integration/l1_store.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/l1_store.py <<'PY'
from __future__ import annotations
import argparse
import hashlib
import json
from app.common.config import load_config
from app.common.db import connect_db
from app.integration.sync_store import (
initialize_sync_schema,
now_jst,
)
VEHICLE_CURSOR = "vehicle_events"
LTE_CURSOR = "lte_measurements"
def initialize_cursor(
con,
*,
source_name: str,
source_table: str,
config: dict,
) -> int:
row = con.execute(
"""
SELECT
last_source_id
FROM sync_cursors
WHERE source_name=?
""",
(
source_name,
),
).fetchone()
if row is not None:
return int(
row["last_source_id"]
)
max_id = con.execute(
f"""
SELECT
COALESCE(
MAX(id),
0
)
FROM {source_table}
"""
).fetchone()[0]
max_id = int(
max_id
)
con.execute(
"""
INSERT INTO sync_cursors (
source_name,
last_source_id,
updated_at_jst
)
VALUES (
?, ?, ?
)
""",
(
source_name,
max_id,
now_jst(config),
),
)
print(
"CURSOR INITIALIZED:",
source_name,
max_id,
)
return max_id
def bootstrap(
config: dict | None = None,
) -> None:
cfg = config or load_config()
initialize_sync_schema(
cfg
)
con = connect_db(
cfg
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
initialize_cursor(
con,
source_name=VEHICLE_CURSOR,
source_table="vehicle_events",
config=cfg,
)
initialize_cursor(
con,
source_name=LTE_CURSOR,
source_table="lte_measurements",
config=cfg,
)
con.commit()
except Exception:
con.rollback()
raise
finally:
con.close()
def vehicle_priority(
row,
) -> int:
event_type = str(
row["event_type"]
or ""
).lower()
event_name = str(
row["event_name"]
or ""
).lower()
if (
event_name in (
"shock",
"marker",
)
or
event_type in (
"button",
"marker",
)
):
return 10
return 20
def vehicle_payload(
row,
object_uid: str,
) -> str:
payload = {
"schema":
"ronv.fact.vehicle.v1",
"level":
1,
"object_uid":
object_uid,
"event_time_jst":
row["received_at_jst"],
"source":
row["source"],
"event_type":
row["event_type"],
"event_name":
row["event_name"],
"acc_g":
row["acc"],
"knock_count":
row["knock_count"],
"knock_target":
row["knock_target"],
}
return json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
def lte_payload(
row,
object_uid: str,
) -> str:
payload = {
"schema":
"ronv.fact.lte.v1",
"level":
1,
"object_uid":
object_uid,
"event_time_jst":
row["timestamp_jst"],
"event_type":
"lte_cell",
"band":
row["band"],
"frequency_mhz":
row["frequency_mhz"],
"earfcn":
row["earfcn"],
"pci":
row["pci"],
"prb":
row["prb"],
"ports":
row["ports"],
"pss_power_dbm":
row["pss_power_dbm"],
"pci_count":
row["pci_count"],
"latitude":
row["latitude"],
"longitude":
row["longitude"],
"gps_status":
row["gps_status"],
"status":
row["status"],
}
return json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
def insert_object(
con,
*,
object_uid: str,
source_table: str,
source_id: int,
data_type: str,
priority: int,
event_time_jst: str,
payload_json: str,
generator: str,
config: dict,
) -> None:
raw = payload_json.encode(
"utf-8"
)
sha256 = hashlib.sha256(
raw
).hexdigest()
metadata = json.dumps(
{
"generator":
generator,
"schema_version":
1,
},
separators=(",", ":"),
sort_keys=True,
)
con.execute(
"""
INSERT OR IGNORE INTO
sync_objects (
object_uid,
source_table,
source_id,
data_type,
level,
priority,
event_time_jst,
created_at_jst,
source_node,
payload_json,
bytes,
sha256,
state,
metadata_json
)
VALUES (
?, ?, ?, ?, 1, ?, ?, ?,
'ronv', ?, ?, ?, 'ready', ?
)
""",
(
object_uid,
source_table,
source_id,
data_type,
priority,
event_time_jst,
now_jst(config),
payload_json,
len(raw),
sha256,
metadata,
),
)
def process_vehicle(
con,
*,
config: dict,
limit: int = 100,
) -> int:
cursor = con.execute(
"""
SELECT
last_source_id
FROM sync_cursors
WHERE source_name=?
""",
(
VEHICLE_CURSOR,
),
).fetchone()
if cursor is None:
raise RuntimeError(
"vehicle cursor not initialized"
)
last_id = int(
cursor["last_source_id"]
)
rows = con.execute(
"""
SELECT
id,
received_at_jst,
source,
event_type,
event_name,
acc,
knock_count,
knock_target
FROM vehicle_events
WHERE id > ?
ORDER BY id ASC
LIMIT ?
""",
(
last_id,
limit,
),
).fetchall()
count = 0
for row in rows:
source_id = int(
row["id"]
)
object_uid = (
f"ronv-l1-vehicle-"
f"{source_id:08d}"
)
payload = vehicle_payload(
row,
object_uid,
)
insert_object(
con,
object_uid=object_uid,
source_table="vehicle_events",
source_id=source_id,
data_type="vehicle_event",
priority=vehicle_priority(row),
event_time_jst=
row["received_at_jst"],
payload_json=payload,
generator="l1_store.vehicle",
config=config,
)
con.execute(
"""
UPDATE sync_cursors
SET
last_source_id=?,
updated_at_jst=?
WHERE source_name=?
""",
(
source_id,
now_jst(config),
VEHICLE_CURSOR,
),
)
count += 1
print(
"L1 VEHICLE:",
source_id,
row["event_type"],
row["event_name"],
)
return count
def process_lte(
con,
*,
config: dict,
limit: int = 100,
) -> int:
cursor = con.execute(
"""
SELECT
last_source_id
FROM sync_cursors
WHERE source_name=?
""",
(
LTE_CURSOR,
),
).fetchone()
if cursor is None:
raise RuntimeError(
"LTE cursor not initialized"
)
last_id = int(
cursor["last_source_id"]
)
rows = con.execute(
"""
SELECT
id,
timestamp_jst,
band,
frequency_mhz,
earfcn,
pci,
prb,
ports,
pss_power_dbm,
pci_count,
latitude,
longitude,
gps_status,
status,
raw_log
FROM lte_measurements
WHERE id > ?
ORDER BY id ASC
LIMIT ?
""",
(
last_id,
limit,
),
).fetchall()
count = 0
for row in rows:
source_id = int(
row["id"]
)
object_uid = (
f"ronv-l1-lte-"
f"{source_id:08d}"
)
payload = lte_payload(
row,
object_uid,
)
insert_object(
con,
object_uid=object_uid,
source_table=
"lte_measurements",
source_id=source_id,
data_type=
"lte_cell_observation",
priority=30,
event_time_jst=
row["timestamp_jst"],
payload_json=payload,
generator="l1_store.lte",
config=config,
)
con.execute(
"""
UPDATE sync_cursors
SET
last_source_id=?,
updated_at_jst=?
WHERE source_name=?
""",
(
source_id,
now_jst(config),
LTE_CURSOR,
),
)
count += 1
print(
"L1 LTE:",
source_id,
"PCI=",
row["pci"],
"status=",
row["status"],
)
return count
def process_new(
config: dict | None = None,
) -> tuple[int, int]:
cfg = config or load_config()
con = connect_db(
cfg
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
vehicle_count = (
process_vehicle(
con,
config=cfg,
)
)
lte_count = (
process_lte(
con,
config=cfg,
)
)
con.commit()
except Exception:
con.rollback()
raise
finally:
con.close()
return (
vehicle_count,
lte_count,
)
def main() -> None:
parser = argparse.ArgumentParser()
mode = (
parser.add_mutually_exclusive_group(
required=True
)
)
mode.add_argument(
"--bootstrap",
action="store_true",
)
mode.add_argument(
"--process",
action="store_true",
)
args = parser.parse_args()
cfg = load_config()
if args.bootstrap:
bootstrap(
cfg
)
print(
"RON-V L1 CURSOR BOOTSTRAP OK"
)
return
(
vehicle_count,
lte_count,
) = process_new(
cfg
)
print()
print(
"===== RON-V L1 RESULT ====="
)
print(
"vehicle :",
vehicle_count,
)
print(
"lte :",
lte_count,
)
if __name__ == "__main__":
main()
PY
44. l1_store.pyの構文確認
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/l1_store.py
何も表示されなければOKです。
45. cursorをbootstrapする
統合開始時点では、
vehicle_events 57
lte_measurements 2
まで既存データがありました。
これを初期cursorにします。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l1_store \
--bootstrap
実機結果：
CURSOR INITIALIZED: vehicle_events 57
CURSOR INITIALIZED: lte_measurements 2
RON-V L1 CURSOR BOOTSTRAP OK
SQLiteで確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
source_name,
last_source_id,
updated_at_jst
FROM sync_cursors
ORDER BY source_name;
PRAGMA integrity_check;
SQL
結果：
source_name last_source_id
---------------- --------------
lte_measurements 2
vehicle_events 57
46. bootstrap直後はL1が0件になる
cursorを初期化した直後にprocessします。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l1_store \
--process
結果：
===== RON-V L1 RESULT =====
vehicle : 0
lte : 0
これは正常です。
理由は、
vehicle_events
id > 57
lte_measurements
id > 2
の新規データがまだ存在しないからです。
この確認により、
既存の過去データを勝手にL1化しない
ことが確認できました。
47. Vehicleイベントを実際にL1化する
AtomS3Rのボタンを操作し、新しいVehicleイベントを発生させます。
SQLiteへ、
id 58
source atoms3r
event_type button
event_name button
が追加されました。
L1 processを実行します。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l1_store \
--process
結果：
L1 VEHICLE: 58 button button
===== RON-V L1 RESULT =====
vehicle : 1
lte : 0
48. Vehicle L1を確認する
SQLiteで確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
object_uid,
source_table,
source_id,
data_type,
level,
priority,
state,
payload_json
FROM sync_objects
WHERE source_table='vehicle_events'
ORDER BY source_id DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
新しいL1：
object_uid
ronv-l1-vehicle-00000058
source_table
vehicle_events
source_id
58
data_type
vehicle_event
level
1
priority
10
state
ready
Vehicleイベントを、
元データ
vehicle_events
から、
統合データ
sync_objects L1
へ変換できました。
49. Vehicleイベントのpriority
Vehicleイベントはすべて同じ重要度ではありません。
今回の初期実装では、
button
shock
marker
などを優先度10としました。
その他は、
priority = 20
です。
実機試験中には、
58 button
59 shock
60 shutdown_knock
61 shutdown_knock
62 shutdown_knock
が追加されました。
この場合、
button 10
shock 10
shutdown_knock 20
となります。
priorityの数字は、
小さいほど高優先
として扱っています。
将来的に、
SHOCK
災害イベント
通信要求
通常状態
などで配送順を変えるための基礎になります。
50. cursorも同じtransactionで更新する
Vehicle L1を作る処理では、
sync_objects INSERT
だけではなく、
sync_cursors UPDATE
も同じtransactionで処理します。
BEGIN IMMEDIATE
↓
vehicle_events取得
↓
sync_objects INSERT
↓
sync_cursors UPDATE
↓
COMMIT
例えば、
L1生成成功
cursor更新前にプロセス停止
すると、次回同じデータを再処理する可能性があります。
そのため、
L1生成
cursor更新
を1つの論理処理として扱います。
51. LTEを新しく1回測定する
次にLTEのL1化を確認します。
LTE測定を1回手動実行します。
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m app.lte.measure_once
実機ではBand 28を取得できました。
band 28
frequency_mhz 788.0
earfcn 9510
pci 123
prb 50
ports 2
pss_power_dbm -23.3
pci_count 1
gps_status no_fix
status cell_found
SQLite：
lte_measurements id = 3
となりました。
52. LTEをL1化する
L1 processを実行します。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l1_store \
--process
結果：
L1 LTE: 3 PCI= 123 status= cell_found
このデータは、
object_uid
ronv-l1-lte-00000003
として登録されます。
53. LTE L1を確認する
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
object_uid,
source_table,
source_id,
data_type,
level,
priority,
bytes,
state,
payload_json
FROM sync_objects
WHERE source_table='lte_measurements'
ORDER BY source_id DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
結果：
object_uid
ronv-l1-lte-00000003
source_table
lte_measurements
source_id
3
data_type
lte_cell_observation
level
1
priority
30
state
ready
これで、
RTL-SDR
↓
srsRAN cell_search
↓
lte_measurements
↓
L1 LTE Fact
までつながりました。
54. LTE raw_logをL1から外す
最初の試作では、
lte_measurements.raw_log
もL1 payloadへ入れていました。
しかし実際のサイズを見ると、
LTE raw_log
1121 chars
L1 payload
1504 bytes
となりました。
L1の目的は、
軽量な観測Fact
です。
そこへcell_searchの生ログをすべて入れる必要はありません。
そこでL1 payloadから、
raw_log
を外しました。
55. raw_logを削除したわけではない
ここは重要です。
L1からraw_logを外しても、
lte_measurements.raw_log
自体は削除していません。
構造は、
lte_measurements
│
├─ LTE Fact用カラム
│
└─ raw_log
↑
保持
sync_objects L1
│
└─ 軽量Factだけ
です。
修正後のL1 payloadは、
336 bytes
程度まで小さくなりました。
つまり、
1504 bytes
↓
336 bytes
まで軽量化できました。
raw_logは将来的に、
L4 Raw Log
として扱う予定です。
56. LTE L1に残す情報
L1へ残した主なLTE情報は、
timestamp
band
frequency_mhz
earfcn
pci
prb
ports
pss_power_dbm
pci_count
latitude
longitude
gps_status
status
です。
例えば、
{
"band": 28,
"earfcn": 9510,
"pci": 123,
"frequency_mhz": 788.0,
"pss_power_dbm": -23.3,
"pci_count": 1,
"gps_status": "no_fix",
"status": "cell_found"
}
のようなFactとして扱います。
これなら、
Mesh
Web API
Grafana
別ノード同期
などで扱いやすくなります。
57. L1 cursorを確認する
VehicleとLTEを処理したあとのcursorを確認します。
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
source_name,
last_source_id,
updated_at_jst
FROM sync_cursors
ORDER BY source_name;
SELECT
level,
data_type,
COUNT(*) AS rows
FROM sync_objects
WHERE level=1
GROUP BY
level,
data_type
ORDER BY data_type;
PRAGMA integrity_check;
SQL
実機では、
vehicle_events
62
lte_measurements
3
までcursorが進みました。
L1としては、
Vehicle L1
5件
LTE L1
1件
が生成されました。
58. L1処理の現在構成
ここまでで、
AtomS3R
↓
vehicle_events
↓
cursor
↓
Vehicle L1
↓
sync_objects
RTL-SDR
↓
lte_measurements
↓
cursor
↓
LTE L1
↓
sync_objects
が完成しました。
重要なのは、L1 processがセンサー測定そのものを行わないことです。
例えば、
L1 processを実行
↓
LTE cell_searchを開始
とはしません。
L1は、
既にSQLiteへ保存された
新しいデータだけを読む
という役割です。
59. L0とL1が揃った
ここまでで統合層には、
L0
Node Beacon
L1
Vehicle Fact
LTE Fact
が揃いました。
現在の構造：
gpsd
↓
L0 Beacon
↓
sync_objects
AtomS3R
↓
vehicle_events
↓
L1 Vehicle Fact
↓
sync_objects
RTL-SDR
↓
lte_measurements
↓
L1 LTE Fact
↓
sync_objects
次は、
直近600秒の状態
を1つにまとめる、
L2 Cycle Summary
を作ります。
L2では、
Vehicle event数
Shock数
Button数
LTE測定数
PCI数
最強PSS
最新L0
Mesh状態
systemdサービス状態
などをまとめます。
60. L2 — 600秒のCycle Summaryを作る
L0はRON-V自身の状態、L1は個々のセンサーから得られたFactでした。
次に、それらを一定時間ごとにまとめる、
L2 Cycle Summary
を作ります。
RON-Vの設定には、
[controller]
cycle_sec = 600
があります。
ここで重要なのは、この600が、
LTEを600秒ごとに測定する
という意味ではないことです。
同様に、
Vehicleを600秒ごとに読み取る
という意味でもありません。
今回の設計では、
既にSQLiteへ保存されている観測結果を
直近600秒の窓で集約する
という意味で使います。
61. L2で集約する情報
L2では、直近600秒の状態を次のようにまとめます。
Vehicle
event数
shock数
button数
shutdown_knock数
first_id
last_id
LTE
測定数
unique PCI数
strongest PSS
first_id
last_id
latest LTE
L0
latest beacon
Mesh
TX数
RX数
ACK済みTX数
latest TX
System
ronv-vehicle.service
ronv-web.service
ronv-mesh.service
gpsd.socket
grafana-server.service
つまりL2を見ることで、
このRON-Vノードが
直近10分で何を観測し、
通信状態がどうだったか
を1つのJSONで確認できます。
62. l2_summary.pyを作る
app/integration/l2_summary.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/l2_summary.py <<'PY'
from __future__ import annotations
import hashlib
import json
import subprocess
import time
from datetime import (
datetime,
timedelta,
)
from zoneinfo import ZoneInfo
from app.common.config import (
load_config,
)
from app.common.db import (
connect_db,
)
from app.integration.sync_store import (
initialize_sync_schema,
)
def service_state(
unit: str,
) -> str:
try:
result = subprocess.run(
[
"systemctl",
"is-active",
unit,
],
capture_output=True,
text=True,
timeout=3,
check=False,
)
state = (
result.stdout.strip()
or result.stderr.strip()
or "unknown"
)
return state
except Exception as exc:
return (
"error:"
+ type(exc).__name__
)
def make_summary(
config: dict | None = None,
) -> dict:
cfg = (
config
or load_config()
)
initialize_sync_schema(
cfg
)
cycle_sec = int(
cfg["controller"][
"cycle_sec"
]
)
timezone = ZoneInfo(
cfg["project"][
"timezone"
]
)
end_dt = datetime.now(
timezone
)
start_dt = (
end_dt
- timedelta(
seconds=cycle_sec
)
)
start_jst = (
start_dt.isoformat(
timespec="seconds"
)
)
end_jst = (
end_dt.isoformat(
timespec="seconds"
)
)
con = connect_db(
cfg
)
try:
vehicle = con.execute(
"""
SELECT
COUNT(*) AS total,
SUM(
CASE
WHEN event_name='shock'
THEN 1
ELSE 0
END
) AS shock_count,
SUM(
CASE
WHEN event_type='button'
THEN 1
ELSE 0
END
) AS button_count,
SUM(
CASE
WHEN event_type=
'shutdown_knock'
THEN 1
ELSE 0
END
) AS shutdown_knock_count,
MIN(id) AS first_id,
MAX(id) AS last_id
FROM vehicle_events
WHERE
julianday(
received_at_jst
)
>= julianday(?)
AND
julianday(
received_at_jst
)
< julianday(?)
""",
(
start_jst,
end_jst,
),
).fetchone()
lte = con.execute(
"""
SELECT
COUNT(*) AS total,
COUNT(
DISTINCT pci
) AS pci_count,
MAX(
pss_power_dbm
) AS strongest_pss_dbm,
MIN(id) AS first_id,
MAX(id) AS last_id
FROM lte_measurements
WHERE
julianday(
timestamp_jst
)
>= julianday(?)
AND
julianday(
timestamp_jst
)
< julianday(?)
""",
(
start_jst,
end_jst,
),
).fetchone()
latest_lte = con.execute(
"""
SELECT
id,
timestamp_jst,
band,
frequency_mhz,
earfcn,
pci,
pss_power_dbm,
pci_count,
gps_status,
status
FROM lte_measurements
ORDER BY id DESC
LIMIT 1
"""
).fetchone()
latest_l0 = con.execute(
"""
SELECT
l.id,
l.object_uid,
l.sequence_u16,
l.latitude_deg,
l.longitude_deg,
l.network_flags_u8,
l.packet_bytes,
json_extract(
o.metadata_json,
'$.gps_status'
) AS gps_status
FROM l0_beacons AS l
LEFT JOIN
sync_objects AS o
ON
o.object_uid =
l.object_uid
ORDER BY
l.id DESC
LIMIT 1
"""
).fetchone()
mesh = con.execute(
"""
SELECT
SUM(
CASE
WHEN direction='TX'
THEN 1
ELSE 0
END
) AS tx_count,
SUM(
CASE
WHEN direction='RX'
THEN 1
ELSE 0
END
) AS rx_count,
SUM(
CASE
WHEN direction='TX'
AND state='acked'
THEN 1
ELSE 0
END
) AS tx_acked
FROM mesh_messages
WHERE
julianday(
timestamp_jst
)
>= julianday(?)
AND
julianday(
timestamp_jst
)
< julianday(?)
""",
(
start_jst,
end_jst,
),
).fetchone()
latest_mesh_tx = (
con.execute(
"""
SELECT
id,
timestamp_jst,
to_id,
state,
packet_id,
ack_type,
ack_error
FROM mesh_messages
WHERE
direction='TX'
ORDER BY
id DESC
LIMIT 1
"""
).fetchone()
)
finally:
con.close()
def row_dict(
row,
):
if row is None:
return None
return dict(
row
)
payload = {
"schema":
"ronv.summary.v1",
"level":
2,
"window": {
"start_jst":
start_jst,
"end_jst":
end_jst,
"cycle_sec":
cycle_sec,
},
"vehicle": {
"count":
int(
vehicle["total"]
or 0
),
"shock_count":
int(
vehicle[
"shock_count"
]
or 0
),
"button_count":
int(
vehicle[
"button_count"
]
or 0
),
"shutdown_knock_count":
int(
vehicle[
"shutdown_knock_count"
]
or 0
),
"first_id":
vehicle[
"first_id"
],
"last_id":
vehicle[
"last_id"
],
},
"lte": {
"count":
int(
lte["total"]
or 0
),
"unique_pci_count":
int(
lte[
"pci_count"
]
or 0
),
"strongest_pss_dbm":
lte[
"strongest_pss_dbm"
],
"first_id":
lte[
"first_id"
],
"last_id":
lte[
"last_id"
],
"latest":
row_dict(
latest_lte
),
},
"l0": {
"latest":
row_dict(
latest_l0
),
},
"mesh": {
"tx_count":
int(
mesh[
"tx_count"
]
or 0
),
"rx_count":
int(
mesh[
"rx_count"
]
or 0
),
"tx_acked":
int(
mesh[
"tx_acked"
]
or 0
),
"latest_tx":
row_dict(
latest_mesh_tx
),
},
"services": {
"ronv_vehicle":
service_state(
"ronv-vehicle.service"
),
"ronv_web":
service_state(
"ronv-web.service"
),
"ronv_mesh":
service_state(
"ronv-mesh.service"
),
"gpsd_socket":
service_state(
"gpsd.socket"
),
"grafana":
service_state(
"grafana-server.service"
),
},
}
return payload
def save_summary(
config: dict | None = None,
) -> str:
cfg = (
config
or load_config()
)
payload = make_summary(
cfg
)
payload_json = json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
raw = payload_json.encode(
"utf-8"
)
sha256 = hashlib.sha256(
raw
).hexdigest()
object_uid = (
"ronv-l2-"
+ str(
time.time_ns()
// 1_000_000
)
)
created_at = (
payload["window"][
"end_jst"
]
)
metadata_json = json.dumps(
{
"generator":
"l2_summary",
"schema_version":
1,
"cycle_sec":
int(
cfg[
"controller"
][
"cycle_sec"
]
),
},
sort_keys=True,
separators=(",", ":"),
)
con = connect_db(
cfg
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
con.execute(
"""
INSERT INTO sync_objects (
object_uid,
source_table,
source_id,
data_type,
level,
priority,
event_time_jst,
created_at_jst,
source_node,
payload_json,
bytes,
sha256,
state,
metadata_json
)
VALUES (
?,
'integration',
NULL,
'cycle_summary',
2,
40,
?,
?,
'ronv',
?,
?,
?,
'ready',
?
)
""",
(
object_uid,
created_at,
created_at,
payload_json,
len(raw),
sha256,
metadata_json,
),
)
con.commit()
except Exception:
con.rollback()
raise
finally:
con.close()
return object_uid
def main() -> None:
cfg = load_config()
object_uid = save_summary(
cfg
)
con = connect_db(
cfg
)
try:
row = con.execute(
"""
SELECT
object_uid,
bytes,
payload_json
FROM sync_objects
WHERE
object_uid=?
""",
(
object_uid,
),
).fetchone()
finally:
con.close()
payload = json.loads(
row["payload_json"]
)
print(
"===== RON-V L2 SUMMARY ====="
)
print(
"object_uid :",
object_uid,
)
print(
"bytes :",
row["bytes"],
)
print(
"window_sec :",
payload[
"window"
][
"cycle_sec"
],
)
print(
"vehicle :",
payload[
"vehicle"
],
)
print(
"lte :",
payload[
"lte"
],
)
print(
"mesh :",
payload[
"mesh"
],
)
print(
"services :",
payload[
"services"
],
)
if __name__ == "__main__":
main()
PY
63. L2の構文確認
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/l2_summary.py
何も表示されなければOKです。
64. L2を手動実行する
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.l2_summary
実機では、
===== RON-V L2 SUMMARY =====
window_sec : 600
として生成されました。
Vehicleについては、
vehicle : {
'button_count': 0,
'count': 0,
'first_id': None,
'last_id': None,
'shock_count': 0,
'shutdown_knock_count': 0
}
LTEについては、
lte : {
'count': 1,
'first_id': 3,
'last_id': 3,
'strongest_pss_dbm': -23.3,
'unique_pci_count': 1
}
Service状態：
services : {
'gpsd_socket': 'active',
'grafana': 'active',
'ronv_mesh': 'active',
'ronv_vehicle': 'active',
'ronv_web': 'active'
}
となりました。
65. Vehicleが0件でも異常ではない
今回のL2手動試験では、
Vehicle count = 0
でした。
しかしこれは異常ではありません。
Vehicleイベント自体は、
button
shock
shutdown_knock
として既にSQLiteへ保存されています。
ただしL2は、
実行時刻から過去600秒
だけを集計します。
今回のVehicleイベントは、その600秒窓から少し外れていたため、
count = 0
となりました。
つまりL2は、
SQLite全履歴の統計
ではなく、
直近600秒の状態
です。
66. latestとwindow集計は分ける
L2では、
直近600秒内の件数
だけでなく、
最新LTE
最新L0
最新Mesh TX
も別に保持しています。
例えばLTE測定が直近600秒に0件でも、
latest LTE
を見ることで、
最後に観測できたCell
PCI
EARFCN
PSS
を確認できます。
つまり、
window情報
と、
latest情報
を両方持つ設計にしています。
67. L2をsync_objectsへ保存する
生成されたL2は、
sync_objects
へ、
level = 2
data_type = cycle_summary
priority = 40
state = ready
として保存します。
確認：
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
object_uid,
data_type,
level,
priority,
bytes,
state,
created_at_jst
FROM sync_objects
WHERE level=2
ORDER BY created_at_jst DESC
LIMIT 5;
PRAGMA integrity_check;
SQL
L2はL0と違い、
固定長binary packet
ではありません。
RON-V内部ではJSON Summaryとして保持します。
68. L0 / L1 / L2が揃った
ここまでで、
L0
Node Beacon
L1
Vehicle Fact
LTE Fact
L2
Cycle Summary
が揃いました。
構造は、
gpsd
↓
L0
↓
sync_objects
vehicle_events
↓
L1 Vehicle
↓
sync_objects
lte_measurements
↓
L1 LTE
↓
sync_objects
直近600秒
↓
L2 Summary
↓
sync_objects
です。
次は、この3つを1回の統合処理としてまとめます。
69. Integration Cycleを作る
RON-Vでは、
L0生成
L1取込
L2生成
を1回のIntegration Cycleとして処理します。
ただし、このCycleから、
LTE測定を開始する
AtomS3Rを読み始める
わけではありません。
各センサーは独立しています。
Controllerは、
既に存在する観測情報
を統合するだけです。
70. Integration設定をconfigへ追加する
config/ronv.tomlへIntegration設定を追加します。
公開記事ではMeshtasticの実Node IDをマスクします。
[integration]
node_id = 1
mesh_destination = "!<REMOTE_NODE_ID>"
確認：
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python - <<'PY'
from app.common.config import load_config
cfg = load_config()
print(
"node_id :",
cfg["integration"][
"node_id"
]
)
print(
"mesh_destination :",
cfg["integration"][
"mesh_destination"
]
)
print(
"cycle_sec :",
cfg["controller"][
"cycle_sec"
]
)
PY
71. cycle.pyを作る
app/integration/cycle.pyを作成します。
最終的なCycleでは、
L0生成
↓
L0配送登録
↓
L1新規取込
↓
L2生成
を行います。
cd /home/ronv/ronv
cat > app/integration/cycle.py <<'PY'
from __future__ import annotations
import time
from app.common.config import (
load_config,
)
from app.integration.l0_store import (
produce_l0,
)
from app.integration.l1_store import (
process_new,
)
from app.integration.l2_summary import (
save_summary,
)
from app.integration.delivery_worker import (
register_l0_delivery,
)
def run_cycle(
config: dict | None = None,
) -> dict:
cfg = (
config
or load_config()
)
started = (
time.monotonic()
)
# ---------------------------------
# L0
# Current node position/status beacon
# ---------------------------------
l0_result = produce_l0(
cfg
)
l0_delivery_id = (
register_l0_delivery(
object_uid=
l0_result[
"object_uid"
],
config=cfg,
)
)
# ---------------------------------
# L1
# Convert only newly stored
# sensor facts
# ---------------------------------
(
vehicle_count,
lte_count,
) = process_new(
cfg
)
# ---------------------------------
# L2
# Summarize observations over
# controller.cycle_sec
# ---------------------------------
l2_object_uid = (
save_summary(
cfg
)
)
elapsed_ms = int(
(
time.monotonic()
- started
)
* 1000
)
return {
"l0_object_uid":
l0_result[
"object_uid"
],
"l0_sequence":
l0_result[
"sequence"
],
"l0_gps_status":
l0_result[
"gps_status"
],
"l0_crc_ok":
l0_result[
"crc_ok"
],
"l0_delivery_id":
l0_delivery_id,
"vehicle_l1_count":
vehicle_count,
"lte_l1_count":
lte_count,
"l2_object_uid":
l2_object_uid,
"cycle_sec":
int(
cfg[
"controller"
][
"cycle_sec"
]
),
"elapsed_ms":
elapsed_ms,
}
def main() -> None:
result = run_cycle()
print(
"===== RON-V INTEGRATION CYCLE ====="
)
for key, value in (
result.items()
):
print(
f"{key:20s}:",
value,
)
if __name__ == "__main__":
main()
PY
72. cycle.pyの構文確認
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/cycle.py \
app/integration/l0_store.py \
app/integration/l1_store.py \
app/integration/l2_summary.py \
app/integration/delivery_worker.py
何も表示されなければOKです。
73. Integration Cycleを手動実行する
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.cycle
実機では、
===== RON-V INTEGRATION CYCLE =====
l0_sequence : 15
l0_gps_status : no_fix
l0_crc_ok : True
l0_delivery_id : 3
vehicle_l1_count : 0
lte_l1_count : 0
l2_object_uid : ronv-l2-...
cycle_sec : 600
elapsed_ms : 5228
となりました。
74. Cycle処理に約5秒かかった理由
実行時間は、
elapsed_ms : 5228
つまり約5.2秒でした。
主な理由はGNSSです。
屋内でGNSS Fixしていないため、
gpsd_reader
がtimeoutまで待ちます。
現在のGNSS timeoutは約5秒なので、
Cycle全体 ≒ 5秒
となっています。
これはCPU処理に5秒必要という意味ではありません。
75. Controllerは無線送信しない
今回のCycleでは、
l0_delivery_id : 3
が生成されました。
しかし、この時点では配送状態は、
state pending
attempt_count 0
です。
つまり、
Controller
↓
L0生成
↓
Delivery登録
までは行いますが、
Meshtasticへ実送信
はしません。
実際の無線配送は、
Delivery Worker
が担当します。
この役割分離が今回の統合の重要な部分です。
76. Controllerをsystemd oneshotにする
手動で動作したIntegration Cycleをsystemd化します。
systemd/ronv-controller.serviceを作ります。
cd /home/ronv/ronv
cat > systemd/ronv-controller.service <<'EOF'
[Unit]
Description=RON-V Integration Controller
After=network.target gpsd.socket ronv-vehicle.service ronv-mesh.service
Wants=gpsd.socket
[Service]
Type=oneshot
User=ronv
Group=ronv
WorkingDirectory=/home/ronv/ronv
Environment=RONV_CONFIG=/home/ronv/ronv/config/ronv.toml
ExecStart=/home/ronv/.venvs/ronv/bin/python -m app.integration.cycle
TimeoutStartSec=60
EOF
Controllerは常駐プロセスにしません。
Type=oneshot
です。
つまり、
timer発火
↓
Controller起動
↓
1 Cycle処理
↓
終了
という動作です。
77. Controller timerを作る
最終版のronv-controller.timerは次の内容です。
cat > systemd/ronv-controller.timer <<'EOF'
[Unit]
Description=RON-V Integration Controller Timer
[Timer]
OnActiveSec=10min
OnUnitActiveSec=10min
AccuracySec=1s
Unit=ronv-controller.service
[Install]
WantedBy=timers.target
EOF
78. OnUnitActiveSecを使う理由
最初の試作では、
OnUnitInactiveSec=10min
を使用しました。
この場合、
Controller開始
↓
約5～6秒処理
↓
Controller終了
↓
そこから10分待つ
↓
次回Controller
となります。
実際に一晩動かしてみると、
約610秒間隔
になりました。
一方、L2 Summaryの集約窓は、
600秒
です。
このままだと理論上、
Controller周期 約610秒
L2 window 600秒
となり、Summary窓の間に小さな隙間ができます。
79. start-to-startで600秒へ合わせる
そこで最終版では、
OnUnitActiveSec=10min
へ変更しました。
これは、
前回Controllerが開始した時刻
↓
600秒
↓
次回Controller開始
という基準になります。
さらに、
AccuracySec=1s
としました。
実機では、
09:08:16
Controller start
09:18:17
Controller start
となりました。
差は約601秒です。
L2の、
cycle_sec = 600
とほぼ一致しました。
80. systemdファイルを構文確認する
cd /home/ronv/ronv
systemd-analyze verify \
/home/ronv/ronv/systemd/ronv-controller.service \
/home/ronv/ronv/systemd/ronv-controller.timer
何も表示されなければOKです。
81. Controllerをsystemdへ登録する
sudo install -m 0644 \
systemd/ronv-controller.service \
/etc/systemd/system/ronv-controller.service
sudo install -m 0644 \
systemd/ronv-controller.timer \
/etc/systemd/system/ronv-controller.timer
sudo systemctl daemon-reload
82. Controller serviceを手動試験する
timerを有効にする前に、service単体を確認します。
sudo systemctl start \
ronv-controller.service
状態確認：
systemctl status \
ronv-controller.service \
--no-pager -l
Type=oneshotなので処理終了後、
Active: inactive (dead)
になります。
これは異常ではありません。
83. Controllerのjournalを確認する
journalctl \
-u ronv-controller.service \
-n 40 \
--no-pager
実機では、
===== RON-V INTEGRATION CYCLE =====
l0_sequence : 4
l0_gps_status : no_fix
l0_crc_ok : True
vehicle_l1_count : 0
lte_l1_count : 0
cycle_sec : 600
elapsed_ms : 5571
となりました。
systemd経由でもIntegration Cycleが正常に実行できています。
84. Controller timerを有効化する
sudo systemctl enable --now \
ronv-controller.timer
確認：
systemctl is-enabled \
ronv-controller.timer
systemctl is-active \
ronv-controller.timer
期待値：
enabled
active
timer一覧：
systemctl list-timers \
ronv-controller.timer \
--no-pager
これでRON-Vは、
約10分ごと
↓
Integration Cycle
を自動実行するようになりました。
85. ここまでの自動化範囲
この段階で、
ronv-controller.timer
↓
ronv-controller.service
↓
L0生成
↓
L0 delivery登録
↓
L1新規データ取込
↓
L2 Summary生成
まで自動化できました。
ただし、
delivery=pending
を実際に、
Meshへ送信
する部分はまだ自動化していません。
次は、
sync_deliveries
↓
Delivery Worker
↓
mesh_messages
↓
ronv-mesh.service
↓
920MHz
↓
real ACK
を作ります。
ここでRON-Vの、
観測
統合
配送
無線
がすべて分離されます。
86. Delivery Workerを作る
ここまででControllerは、
L0生成
↓
Delivery登録
まで行うようになりました。
しかしController自身はMeshtasticへ送信しません。
次に、
sync_deliveries
へ登録された配送要求を処理する、
Delivery Worker
を作ります。
全体の役割は次のとおりです。
sync_deliveries
│
│ pending / retry
▼
Delivery Worker
│
▼
mesh_messages
│
│ queued
▼
ronv-mesh.service
│
▼
Meshtastic 920MHz
│
▼
real ACK
│
▼
sync_deliveries
│
└─ acked
Delivery WorkerもMeshtastic USBを直接開きません。
USBを所有するのは、これまでどおり、
ronv-mesh.service
だけです。
87. Deliveryの状態遷移
RON-VのMesh配送では、主に次の状態を使います。
pending
↓
inflight
↓
acked
失敗した場合は、
inflight
↓
retry
へ移します。
一定時間後に再度、
retry
↓
inflight
へ進みます。
ただし、
queued
sending
の状態を単純にretryすることはしません。
理由は後ほど説明します。
88. L0の配送表現
L0本体は23-byte binaryです。
しかし今回のMeshtastic配送では、
TEXT_MESSAGE_APP
を利用します。
そのため、
23-byte L0
↓
hex
↓
R0:<packet_hex>
という表現を使います。
Delivery Worker内部では、
REPRESENTATION = "l0-text"
TRANSPORT = "mesh"
とします。
retryまでの初期間隔は、
RETRY_SEC = 60
です。
89. delivery_worker.pyを作る
app/integration/delivery_worker.pyを作成します。
cd /home/ronv/ronv
cat > app/integration/delivery_worker.py <<'PY'
from __future__ import annotations
import argparse
import hashlib
import json
import time
from datetime import (
datetime,
timedelta,
)
from zoneinfo import ZoneInfo
from app.common.config import (
load_config,
)
from app.common.db import (
connect_db,
)
from app.integration.l0_packet import (
unpack_l0,
)
from app.integration.sync_store import (
now_jst,
)
REPRESENTATION = "l0-text"
TRANSPORT = "mesh"
RETRY_SEC = 60
def retry_time(
config: dict,
) -> str:
timezone = ZoneInfo(
config["project"][
"timezone"
]
)
dt = (
datetime.now(
timezone
)
+ timedelta(
seconds=RETRY_SEC
)
)
return dt.isoformat(
timespec="seconds"
)
def make_mesh_text(
packet_hex: str,
) -> str:
packet = bytes.fromhex(
packet_hex
)
decoded = unpack_l0(
packet
)
if not decoded[
"crc_ok"
]:
raise RuntimeError(
"L0 CRC invalid"
)
text = (
"R0:"
+ packet_hex
)
if len(
text.encode(
"utf-8"
)
) > 180:
raise RuntimeError(
"L0 mesh payload too large"
)
return text
def register_l0_delivery(
*,
object_uid: str,
config: dict,
) -> int:
destination = str(
config[
"integration"
][
"mesh_destination"
]
)
con = connect_db(
config
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
row = con.execute(
"""
SELECT
o.object_uid,
o.priority,
l.packet_hex
FROM sync_objects AS o
JOIN l0_beacons AS l
ON l.object_uid =
o.object_uid
WHERE
o.object_uid=?
AND
o.level=0
AND
o.data_type=
'node_beacon'
AND
o.state='ready'
""",
(
object_uid,
),
).fetchone()
if row is None:
raise RuntimeError(
"L0 object not found"
)
mesh_text = (
make_mesh_text(
str(
row[
"packet_hex"
]
)
)
)
raw = mesh_text.encode(
"utf-8"
)
sha256 = hashlib.sha256(
raw
).hexdigest()
ts = now_jst(
config
)
metadata = json.dumps(
{
"kind":
"l0",
"mesh_message_id":
None,
},
sort_keys=True,
separators=(",", ":"),
)
con.execute(
"""
INSERT OR IGNORE INTO
sync_deliveries (
object_uid,
representation,
target,
transport,
priority,
state,
attempt_count,
created_at_jst,
bytes,
sha256,
metadata_json
)
VALUES (
?,
?,
?,
?,
?,
'pending',
0,
?,
?,
?,
?
)
""",
(
object_uid,
REPRESENTATION,
destination,
TRANSPORT,
int(
row[
"priority"
]
),
ts,
len(raw),
sha256,
metadata,
),
)
delivery = con.execute(
"""
SELECT
id,
state
FROM sync_deliveries
WHERE
object_uid=?
AND
representation=?
AND
target=?
AND
transport=?
""",
(
object_uid,
REPRESENTATION,
destination,
TRANSPORT,
),
).fetchone()
if delivery is None:
raise RuntimeError(
"delivery registration failed"
)
con.commit()
return int(
delivery[
"id"
]
)
except Exception:
con.rollback()
raise
finally:
con.close()
def queue_delivery(
*,
delivery_id: int,
config: dict,
) -> int | None:
con = connect_db(
config
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
row = con.execute(
"""
SELECT
d.*,
l.packet_hex
FROM sync_deliveries AS d
JOIN l0_beacons AS l
ON l.object_uid =
d.object_uid
JOIN sync_objects AS o
ON o.object_uid =
d.object_uid
WHERE
d.id=?
AND
d.transport='mesh'
AND
d.representation=
'l0-text'
AND
o.state='ready'
""",
(
delivery_id,
),
).fetchone()
if row is None:
con.commit()
return None
state = str(
row[
"state"
]
)
if state not in (
"pending",
"retry",
):
con.commit()
return None
if (
state == "retry"
and
row[
"next_retry_at_jst"
]
):
now = datetime.now(
ZoneInfo(
config[
"project"
][
"timezone"
]
)
)
retry_at = (
datetime.fromisoformat(
str(
row[
"next_retry_at_jst"
]
)
)
)
if now < retry_at:
con.commit()
return None
mesh_text = (
make_mesh_text(
str(
row[
"packet_hex"
]
)
)
)
ts = now_jst(
config
)
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,
row[
"target"
],
mesh_text,
ts,
ts,
),
)
mesh_id = int(
cur.lastrowid
)
try:
metadata = json.loads(
row[
"metadata_json"
]
or "{}"
)
except Exception:
metadata = {}
metadata[
"mesh_message_id"
] = mesh_id
metadata[
"mesh_state"
] = "queued"
metadata_json = json.dumps(
metadata,
sort_keys=True,
separators=(",", ":"),
)
cur = con.execute(
"""
UPDATE sync_deliveries
SET
state='inflight',
attempt_count=
attempt_count + 1,
last_attempt_at_jst=?,
next_retry_at_jst=NULL,
remote_id=?,
metadata_json=?,
last_error=NULL
WHERE
id=?
AND
state IN (
'pending',
'retry'
)
""",
(
ts,
f"mesh-message:{mesh_id}",
metadata_json,
delivery_id,
),
)
if cur.rowcount != 1:
raise RuntimeError(
"delivery claim failed"
)
con.commit()
return mesh_id
except Exception:
con.rollback()
raise
finally:
con.close()
def reconcile_delivery(
*,
delivery_id: int,
config: dict,
) -> str:
con = connect_db(
config
)
try:
con.execute(
"BEGIN IMMEDIATE"
)
delivery = con.execute(
"""
SELECT *
FROM sync_deliveries
WHERE id=?
""",
(
delivery_id,
),
).fetchone()
if delivery is None:
raise RuntimeError(
"delivery not found"
)
state = str(
delivery[
"state"
]
)
if state != "inflight":
con.commit()
return state
try:
metadata = json.loads(
delivery[
"metadata_json"
]
or "{}"
)
except Exception:
metadata = {}
mesh_id = metadata.get(
"mesh_message_id"
)
if mesh_id is None:
con.commit()
return "inflight"
mesh = con.execute(
"""
SELECT *
FROM mesh_messages
WHERE id=?
""",
(
int(
mesh_id
),
),
).fetchone()
if mesh is None:
con.commit()
return "inflight"
mesh_state = str(
mesh[
"state"
]
)
metadata.update(
{
"mesh_message_id":
int(
mesh_id
),
"mesh_state":
mesh_state,
"packet_id":
mesh[
"packet_id"
],
"ack_type":
mesh[
"ack_type"
],
"ack_from":
mesh[
"ack_from"
],
"ack_error":
mesh[
"ack_error"
],
}
)
metadata_json = (
json.dumps(
metadata,
sort_keys=True,
separators=(",", ":"),
)
)
if mesh_state == "acked":
packet_id = (
mesh[
"packet_id"
]
)
acked_at = str(
mesh[
"updated_at_jst"
]
)
remote_id = (
"meshtastic:"
+ str(
packet_id
)
)
con.execute(
"""
UPDATE sync_deliveries
SET
state='acked',
sent_at_jst=
COALESCE(
sent_at_jst,
?
),
acked_at_jst=?,
remote_id=?,
ack_token=?,
last_error=NULL,
metadata_json=?
WHERE
id=?
AND
state='inflight'
""",
(
acked_at,
acked_at,
remote_id,
str(
packet_id
),
metadata_json,
delivery_id,
),
)
con.commit()
return "acked"
if mesh_state == "failed":
error = str(
mesh[
"last_error"
]
or "mesh_failed"
)
con.execute(
"""
UPDATE sync_deliveries
SET
state='retry',
next_retry_at_jst=?,
last_error=?,
metadata_json=?
WHERE
id=?
AND
state='inflight'
""",
(
retry_time(
config
),
error[:1000],
metadata_json,
delivery_id,
),
)
con.commit()
return "retry"
# queued / sending は
# 自動再送しない。
#
# 実送信済みかどうかが
# 未確定だから。
con.commit()
return "inflight"
except Exception:
con.rollback()
raise
finally:
con.close()
def reconcile_all(
config: dict,
) -> None:
con = connect_db(
config
)
try:
rows = con.execute(
"""
SELECT
id
FROM sync_deliveries
WHERE
transport='mesh'
AND
state='inflight'
ORDER BY
id
"""
).fetchall()
finally:
con.close()
for row in rows:
delivery_id = int(
row[
"id"
]
)
state = (
reconcile_delivery(
delivery_id=
delivery_id,
config=config,
)
)
print(
"RECONCILE",
delivery_id,
state,
)
def next_due_delivery(
config: dict,
) -> int | None:
con = connect_db(
config
)
try:
rows = con.execute(
"""
SELECT
id,
state,
next_retry_at_jst
FROM sync_deliveries
WHERE
transport='mesh'
AND
representation=
'l0-text'
AND
state IN (
'pending',
'retry'
)
ORDER BY
priority ASC,
id ASC
"""
).fetchall()
finally:
con.close()
now = datetime.now(
ZoneInfo(
config[
"project"
][
"timezone"
]
)
)
for row in rows:
if (
row[
"state"
]
== "retry"
and
row[
"next_retry_at_jst"
]
):
retry_at = (
datetime.fromisoformat(
str(
row[
"next_retry_at_jst"
]
)
)
)
if now < retry_at:
continue
return int(
row[
"id"
]
)
return None
def run_once(
*,
config: dict,
wait_sec: float,
) -> None:
reconcile_all(
config
)
delivery_id = (
next_due_delivery(
config
)
)
if delivery_id is None:
print(
"DELIVERY: nothing due"
)
return
mesh_id = queue_delivery(
delivery_id=
delivery_id,
config=config,
)
print(
"DELIVERY:",
delivery_id,
"mesh_id:",
mesh_id,
)
if mesh_id is None:
return
deadline = (
time.monotonic()
+ wait_sec
)
last_state = None
while (
time.monotonic()
< deadline
):
state = (
reconcile_delivery(
delivery_id=
delivery_id,
config=config,
)
)
if state != last_state:
print(
"STATE:",
state,
)
last_state = state
if state in (
"acked",
"retry",
"failed",
"cancelled",
):
return
time.sleep(
0.5
)
def main() -> None:
parser = (
argparse.ArgumentParser()
)
parser.add_argument(
"--register-l0",
)
parser.add_argument(
"--run-once",
action="store_true",
)
parser.add_argument(
"--wait-sec",
type=float,
default=10.0,
)
args = parser.parse_args()
cfg = load_config()
if args.register_l0:
delivery_id = (
register_l0_delivery(
object_uid=
args.register_l0,
config=cfg,
)
)
print(
"REGISTERED DELIVERY:",
delivery_id,
)
if args.run_once:
run_once(
config=cfg,
wait_sec=
args.wait_sec,
)
if __name__ == "__main__":
main()
PY
90. Delivery Workerの構文確認
cd /home/ronv/ronv
/home/ronv/.venvs/ronv/bin/python \
-m py_compile \
app/integration/delivery_worker.py
何も表示されなければOKです。
91. register_l0_deliveryの役割
register_l0_delivery()は、生成済みL0を、
sync_deliveries
へ登録します。
まず、
sync_objects
l0_beacons
をJOINして、
level = 0
data_type = node_beacon
state = ready
であることを確認します。
その後、
R0:<packet_hex>
を生成します。
この段階でも、
CRC16
を再確認しています。
つまり、
L0生成時
↓
CRC確認
配送登録時
↓
CRC再確認
という形です。
92. 二重Delivery登録を防ぐ
Delivery登録では、
INSERT OR IGNORE
を使用します。
さらにsync_deliveries側には、
UNIQUE (
object_uid,
representation,
target,
transport
)
があります。
そのため同じ、
object_uid
representation
target
transport
を再登録しても、同一配送が大量に増えることはありません。
93. queue_deliveryの役割
queue_delivery()では、
pending
retry
の配送を、
mesh_messages
へ登録します。
例えば、
delivery_id = 2
を処理すると、
mesh_messages
direction = TX
portnum = TEXT_MESSAGE_APP
state = queued
という新しい行が作られます。
その直後、
sync_deliveries
側を、
state = inflight
へ変更します。
94. 二重送信防止で最も重要なtransaction
今回のDelivery Workerで特に重要なのが、
mesh_messages INSERT
と、
sync_deliveries UPDATE
を同じSQLite transactionで処理することです。
BEGIN IMMEDIATE
↓
mesh_messages
INSERT
state=queued
↓
sync_deliveries
UPDATE
state=inflight
↓
COMMIT
とします。
もしこれを別transactionにすると、
mesh_messagesにはqueuedが入った
↓
プロセス停止
↓
sync_deliveriesはpendingのまま
という状態が発生する可能性があります。
次回Workerが、
pendingだから未送信
と判断すると、同じL0を再びqueueしてしまいます。
そこで、
queue登録
Delivery状態変更
を同時に確定させます。
95. queued / sendingを自動再送しない
無線配送ではもう1つ重要な問題があります。
例えば、
mesh_messages = sending
の瞬間にWorkerが停止した場合です。
このとき、
実際には送信済み
かもしれません。
一方で、
まだ送信前
かもしれません。
この状態を機械的に、
retry
へすると、同じpacketを二重送信する可能性があります。
そこでRON-Vでは、
queued
sending
の場合は、
inflight
のまま保持します。
つまり、
送ったかもしれない状態
を自動的に再送しません。
96. reconcile_deliveryの役割
Meshtastic Gatewayは、
mesh_messages
の状態を更新します。
例えば成功すれば、
state acked
ack_type real
packet_id ...
となります。
Delivery Workerは、
reconcile_delivery()
でこの状態を読み取り、
sync_deliveries
へ反映します。
97. real ACKをsync_deliveriesへ反映する
Mesh側が、
state = acked
になった場合、
sync_deliveries
も、
state = acked
へ変更します。
さらに、
acked_at_jst
remote_id
ack_token
metadata_json
を更新します。
remote_idは、
meshtastic:<packet_id>
という形です。
例えば、
meshtastic:2758949733
のようになります。
98. Mesh失敗時はretryへ移す
Mesh側が、
state = failed
になった場合は、
sync_deliveries
を、
retry
へ変更します。
次回時刻は、
現在時刻 + 60秒
です。
RETRY_SEC = 60
としているため、
next_retry_at_jst
へ60秒後の時刻が入ります。
99. run_onceの動作
Delivery Workerは常駐ループではありません。
1回起動すると、
inflightをreconcile
↓
pending/retryを1件選択
↓
mesh_messagesへqueue
↓
最大10秒ACK待ち
↓
終了
します。
つまりControllerと同じく、
oneshot
として使える構造です。
100. Delivery Workerを実送信試験する
最新L0を取得します。
ここで注意点があります。
Bashには、
UID
というreadonly変数があります。
そのため、
UID=...
は使いません。
今回は、
L0_UID
を使います。
L0_UID=$(
sqlite3 /var/lib/ronv/ronv.db \
"SELECT object_uid
FROM l0_beacons
ORDER BY id DESC
LIMIT 1;"
)
echo "L0_UID=$L0_UID"
101. Deliveryを登録する
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.delivery_worker \
--register-l0 "$L0_UID"
登録直後：
state pending
attempt_count 0
この時点ではまだMeshへ送っていません。
102. Workerを1回実行する
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.delivery_worker \
--run-once \
--wait-sec 10
実機では、
DELIVERY: 2 mesh_id: 9
STATE: inflight
STATE: acked
となりました。
つまり、
pending
↓
inflight
↓
acked
まで進みました。
103. real ACKを確認する
SQLite側：
state acked
attempt_count 1
ack_type real
ack_error NONE
ronv-mesh.service側でも、
MESH TX ... ack=real
を確認できました。
これは単に、
SQLiteへ送信済みと書いた
という意味ではありません。
Meshtastic側から、
real ACK
を取得しています。
104. DeliveryとMeshをSQLiteで確認する
sqlite3 /var/lib/ronv/ronv.db <<'SQL'
.headers on
.mode column
SELECT
id,
object_uid,
state,
attempt_count,
remote_id,
ack_token,
last_error
FROM sync_deliveries
ORDER BY id DESC
LIMIT 10;
SELECT
id,
timestamp_jst,
direction,
to_id,
state,
packet_id,
ack_type,
ack_error
FROM mesh_messages
WHERE direction='TX'
ORDER BY id DESC
LIMIT 10;
PRAGMA integrity_check;
SQL
確認ポイントは、
sync_deliveries
state = acked
attempt_count = 1
mesh_messages
state = acked
ack_type = real
ack_error = NONE
です。
105. ControllerからDelivery登録まで接続する
Delivery Worker単体が動いたので、
Integration Cycle
から自動的にL0 Deliveryを登録するようにします。
cycle.pyでは、
from app.integration.delivery_worker import (
register_l0_delivery,
)
を読み込み、
L0生成直後に、
l0_delivery_id = (
register_l0_delivery(
object_uid=
l0_result[
"object_uid"
],
config=cfg,
)
)
を実行します。
これにより、
Controller
↓
L0生成
↓
sync_objects
↓
sync_deliveries
state=pending
までが自動化されます。
106. Cycleから登録されたDeliveryを確認する
手動Cycleを実行します。
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.cycle
実機：
===== RON-V INTEGRATION CYCLE =====
l0_sequence : 15
l0_gps_status : no_fix
l0_crc_ok : True
l0_delivery_id : 3
vehicle_l1_count : 0
lte_l1_count : 0
cycle_sec : 600
Delivery：
delivery_id 3
state pending
attempt_count 0
つまり、
L0生成
↓
Delivery pending
までControllerだけで進みました。
107. Cycle生成DeliveryをWorkerで送る
/home/ronv/.venvs/ronv/bin/python \
-m app.integration.delivery_worker \
--run-once \
--wait-sec 10
実機：
DELIVERY: 3 mesh_id: 10
STATE: inflight
STATE: acked
再び、
real ACK
まで確認できました。
108. Delivery Workerをsystemd化する
ここまで手動実行していたDelivery Workerをsystemd化します。
systemd/ronv-delivery.serviceを作ります。
cd /home/ronv/ronv
cat > systemd/ronv-delivery.service <<'EOF'
[Unit]
Description=RON-V Delivery Worker
After=network.target ronv-mesh.service
Wants=ronv-mesh.service
[Service]
Type=oneshot
User=ronv
Group=ronv
WorkingDirectory=/home/ronv/ronv
Environment=RONV_CONFIG=/home/ronv/ronv/config/ronv.toml
ExecStart=/home/ronv/.venvs/ronv/bin/python -m app.integration.delivery_worker --run-once --wait-sec 10
TimeoutStartSec=30
EOF
Delivery Workerも、
Type=oneshot
です。
109. Delivery timerを作る
systemd/ronv-delivery.timer：
cat > systemd/ronv-delivery.timer <<'EOF'
[Unit]
Description=RON-V Delivery Worker Timer
[Timer]
OnActiveSec=1min
OnUnitInactiveSec=1min
AccuracySec=5s
Unit=ronv-delivery.service
[Install]
WantedBy=timers.target
EOF
110. Deliveryの1分は測定周期ではない
ここも重要です。
Controller 約10分
Delivery 約1分
ですが、
LTEを1分ごとに測定する
という意味ではありません。
Delivery timerの役割は、
配送待ちがあるか確認する
ことだけです。
つまり、
Controller
観測情報を統合
Delivery
配送待ちを処理
という違いがあります。
111. systemdファイルをverifyする
systemd-analyze verify \
/home/ronv/ronv/systemd/ronv-delivery.service \
/home/ronv/ronv/systemd/ronv-delivery.timer
何も表示されなければOKです。
112. Delivery service / timerを登録する
sudo install -m 0644 \
systemd/ronv-delivery.service \
/etc/systemd/system/ronv-delivery.service
sudo install -m 0644 \
systemd/ronv-delivery.timer \
/etc/systemd/system/ronv-delivery.timer
sudo systemctl daemon-reload
113. systemd経由でDeliveryを試験する
まずtimerを有効化する前に、service単体で確認します。
sudo systemctl start \
ronv-delivery.service
実機：
DELIVERY: 4 mesh_id: 11
STATE: inflight
STATE: acked
Mesh Gateway：
MESH TX id=11 ... ack=real
DB：
state acked
attempt_count 1
ack_error NONE
integrity ok
systemd経由でも、
Delivery Worker
↓
Mesh Gateway
↓
real ACK
まで正常に動きました。
114. Delivery timerを有効化する
sudo systemctl enable --now \
ronv-delivery.timer
確認：
systemctl is-enabled \
ronv-delivery.timer
systemctl is-active \
ronv-delivery.timer
期待値：
enabled
active
115. Controller timerとDelivery timerを確認する
systemctl list-timers \
ronv-controller.timer \
ronv-delivery.timer \
--no-pager
ここまでで、
ronv-controller.timer
と、
ronv-delivery.timer
の2つが自動実行されるようになりました。
構造：
ronv-controller.timer
│
▼
Integration Cycle
│
├─ L0
├─ L1
├─ L2
└─ Delivery pending
│
▼
ronv-delivery.timer
│
▼
Delivery Worker
│
▼
mesh_messages
│
▼
ronv-mesh.service
│
▼
Meshtastic
次はいよいよ、
一切手動実行せず
この全経路が動くか確認します。
116. 完全自動運転試験
ここまでで、
ronv-controller.timer
ronv-delivery.timer
を両方有効化しました。
次は、
手動cycleを実行しない
手動deliveryを実行しない
状態で、本当に全経路が動くか確認します。
まずtimer状態を確認します。
systemctl is-enabled \
ronv-controller.timer \
ronv-delivery.timer
systemctl is-active \
ronv-controller.timer \
ronv-delivery.timer
実機：
enabled
enabled
active
active
ここから、人が何も実行せず待ちます。
117. Controllerが自動起動した
最初にController timerが発火しました。
journal：
00:17:41
Starting ronv-controller.service
ControllerはIntegration Cycleを開始しました。
生成結果：
l0_sequence : 17
l0_delivery_id : 5
l0_crc_ok : True
cycle_sec : 600
ここで重要なのは、
l0_delivery_id = 5
まで自動で作られていることです。
つまり、
timer
↓
Controller
↓
L0生成
↓
Delivery pending登録
まで人手なしで進みました。
118. Delivery timerが自動起動した
その後、Delivery timerが発火しました。
00:17:56
Starting ronv-delivery.service
処理：
DELIVERY: 5 mesh_id: 12
STATE: inflight
STATE: acked
つまり、
Delivery pending
↓
mesh_messages queued
↓
inflight
↓
acked
まで自動で進みました。
119. Meshtastic real ACK
Mesh Gateway側では、
MESH TX id=12 ... ack=real
を確認しました。
つまり最終的に、
Raspberry Pi
↓
Meshtastic device
↓
920MHz
↓
Remote Node
↓
real ACK
まで実際の無線経路を通っています。
120. SQLiteで完全自動配送を確認する
SQLite側でも確認しました。
結果：
L0 sequence 17
delivery_id 5
delivery_state acked
attempt_count 1
mesh_id 12
ack_type real
ack_error NONE
integrity_check ok
つまり、
Controller timer
↓
L0
↓
sync_objects
↓
sync_deliveries
↓
Delivery timer
↓
mesh_messages
↓
Meshtastic
↓
real ACK
まで一周しています。
121. 完全自動経路
この時点で成立した経路をまとめると、
ronv-controller.timer
│
▼
ronv-controller.service
│
▼
Integration Cycle
│
├─ L0生成
│
├─ L1取込
│
├─ L2生成
│
└─ delivery=pending
│
▼
ronv-delivery.timer
│
▼
ronv-delivery.service
│
▼
Delivery Worker
│
▼
mesh_messages
│
▼
ronv-mesh.service
│
▼
Meshtastic 920MHz
│
▼
Remote Node
│
▼
real ACK
│
▼
sync_deliveries=acked
です。
ここまで、手動の、
cycle実行
delivery実行
はありません。
122. 一晩連続運転する
1回だけ成功しても、車載ノードとしては不十分です。
そこで、このまま一晩動作させました。
翌朝、まずtimerを確認します。
cd /home/ronv/ronv
echo "===== TIMERS ====="
systemctl is-enabled \
ronv-controller.timer \
ronv-delivery.timer
systemctl is-active \
ronv-controller.timer \
ronv-delivery.timer
続いて、主要サービスを確認します。
echo
echo "===== SERVICES ====="
systemctl is-active \
ronv-vehicle.service \
ronv-web.service \
ronv-mesh.service \
gpsd.socket \
grafana-server.service
123. 一晩運転後のsystemd状態
結果：
enabled
enabled
active
active
Controller timerとDelivery timerは、
enabled
active
を維持していました。
主要サービス：
active
active
active
active
active
つまり、
ronv-vehicle.service
ronv-web.service
ronv-mesh.service
gpsd.socket
grafana-server.service
もすべてactiveでした。
124. 一晩運転後のL0
最新L0を確認すると、
sequence 65
sequence 66
sequence 67
sequence 68
sequence 69
まで進んでいました。
前夜の自動運転確認では、
sequence 17
だったため、その後もControllerが繰り返し動作していることが分かります。
125. 最新5配送を確認する
最新5件のDeliveryも確認しました。
対応する最新5配送は、
state acked
attempt_count 1
last_error NULL
でした。
つまり最新5回については、
1回目の送信で
real ACKまで到達
しています。
ここでは、
全過去配送がすべて成功
とまでは断定せず、
翌朝確認した最新5件が
すべてacked
という実測結果として扱います。
126. 一晩運転後のSQLite integrity
SQLiteの整合性も確認します。
sqlite3 /var/lib/ronv/ronv.db \
'PRAGMA integrity_check;'
結果：
ok
一晩、
Controller
Delivery
Mesh Gateway
Vehicle
Web
Grafana
gpsd
が動作したあともSQLite integrityに問題はありませんでした。
127. 一晩運転で見つかった周期の問題
一晩運転によって、もう1つ気付いた点がありました。
L0の生成時刻を見ると、
08:17:32
08:27:42
08:37:52
08:48:02
08:58:12
のように、
約610秒
間隔になっていました。
設定は、
[controller]
cycle_sec = 600
です。
なぜ10秒程度ずれるのか確認しました。
128. 原因はOnUnitInactiveSec
初期のController timerでは、
OnUnitInactiveSec=10min
を使っていました。
Controllerは1回の処理に、
約5～6秒
かかります。
そのため、
Controller開始
↓
5～6秒処理
↓
Controller終了
↓
そこから600秒待機
↓
次回開始
になります。
結果として、
開始 → 開始
は、
約605～610秒
になります。
一方、L2 Summaryは、
過去600秒
を見ています。
つまり、
Controller周期 > L2窓
になっていました。
129. Controller timerを修正する
timerを、
OnUnitInactiveSec=10min
から、
OnUnitActiveSec=10min
へ変更しました。
最終版：
[Unit]
Description=RON-V Integration Controller Timer
[Timer]
OnActiveSec=10min
OnUnitActiveSec=10min
AccuracySec=1s
Unit=ronv-controller.service
[Install]
WantedBy=timers.target
OnUnitActiveSecは、
前回serviceが開始した時刻
を基準にします。
つまり、
前回開始
↓
600秒
↓
次回開始
です。
130. timer変更前にバックアップする
systemd timerを変更する前にバックアップしました。
cd /home/ronv/ronv
STAMP=$(date +%Y%m%d_%H%M%S)
cp -a \
systemd/ronv-controller.timer \
"systemd/ronv-controller.timer.bak_period_${STAMP}"
sudo cp -a \
/etc/systemd/system/ronv-controller.timer \
"/etc/systemd/system/ronv-controller.timer.bak_period_${STAMP}"
131. timerをverifyする
systemd-analyze verify \
/home/ronv/ronv/systemd/ronv-controller.timer
問題がないことを確認後、インストールします。
sudo install -m 0644 \
systemd/ronv-controller.timer \
/etc/systemd/system/ronv-controller.timer
sudo systemctl daemon-reload
Controller timerだけ再起動します。
sudo systemctl restart \
ronv-controller.timer
132. timer状態を確認する
systemctl status \
ronv-controller.timer \
--no-pager -l
echo
systemctl list-timers \
ronv-controller.timer \
ronv-delivery.timer \
--no-pager
実機では、
Loaded: loaded
Active: active (waiting)
となりました。
次回Triggerも表示されています。
133. start-to-startを実測する
timer変更後、Controller journalを確認しました。
journalctl \
-u ronv-controller.service \
--since "2026-09-05 09:07:00" \
--no-pager
実機：
Sep 05 09:08:16
Starting ronv-controller.service
Sep 05 09:08:22
l0_sequence : 70
次回：
Sep 05 09:18:17
Starting ronv-controller.service
Sep 05 09:18:23
l0_sequence : 71
開始時刻だけを見ると、
09:08:16
09:18:17
です。
差は、
約601秒
となりました。
134. 600秒窓との関係
systemdのタイマー精度や実行時刻を考えると、
600～601秒程度
で動作する形になりました。
L2の、
cycle_sec = 600
とほぼ一致しています。
なお、L2はあくまで、
直近600秒のrolling summary
です。
個々のFactそのものは、
L1
としてSQLiteへ保存されています。
したがって、Summary境界に数秒単位の差が発生しても、
元の観測データそのものが失われる
わけではありません。
135. 現時点のRON-V統合構成
ここまでのRON-Vをまとめます。
┌─────────────────┐
│ gpsd │
└────────┬────────┘
│
▼
L0 Beacon
│
▼
sync_objects
│
│
│
┌──────────────┐ │
│ AtomS3R │ │
└──────┬───────┘ │
│ │
▼ │
vehicle_events │
│ │
▼ │
L1 Vehicle Fact │
│ │
├──────────────┐ │
│ │ │
│ ▼ │
│ L2 Summary │
│ │ │
└──────────────┴──────────┘
│
▼
sync_objects
│
▼
sync_deliveries
│
▼
ronv-delivery.timer
│
▼
mesh_messages
│
▼
ronv-mesh.service
│
▼
Meshtastic 920MHz
│
▼
real ACK
LTEは、
RTL-SDR
↓
srsRAN cell_search
↓
lte_measurements
↓
L1 LTE Fact
↓
sync_objects
↓
L2 Summary
として統合されています。
136. 重要なのは「巨大Controller」にしなかったこと
今回の統合で最も重要だった設計判断は、
全機能をControllerへ入れない
ことです。
RON-Vでは、
Vehicle取得
ronv-vehicle.service
LTE観測
LTE module
GNSS
gpsd
統合
ronv-controller.service
配送
ronv-delivery.service
Mesh USB
ronv-mesh.service
表示
ronv-web.service
Grafana
と分離しました。
137. Meshtastic USBはGatewayだけが所有する
特にMeshについては、
ronv-controller.service
ronv-delivery.service
からMeshtastic USBを直接操作しません。
USBを所有するのは、
ronv-mesh.service
だけです。
他のサービスは、
SQLite
mesh_messages
を通して連携します。
これにより、
複数プロセスが同じSerialInterfaceを開く
問題を避けています。
138. SQLiteをサービス間連携にも使う
今回のSQLiteは単なるログ保存先ではありません。
役割は、
観測履歴
に加えて、
統合状態
配送状態
サービス間queue
も持っています。
具体的には、
vehicle_events
lte_measurements
l0_beacons
sync_objects
sync_deliveries
sync_cursors
mesh_messages
が連携します。
つまりRON-VではSQLiteが、
小型のpersistent message bus
のような役割も持っています。
139. L0 / L1 / L2の役割を再整理する
今回実装した3レベルを整理します。
L0
Node Beacon
小さい
定期的
ノード自身の基本状態
L1
Sensor Fact
button
shock
LTE Cell
など個々の観測事実
L2
Cycle Summary
直近600秒の
観測・通信・サービス状態
L0、L1、L2は、
順番に変換して最後にL2になる
わけではありません。
それぞれ独立した情報クラスです。
140. L3 / L4は今回まだ実装しない
将来的には、
L3
Evidence / Analysis
L4
Raw File / Raw Log
を追加する予定です。
例えば、
L3
SHOCKイベントの証拠情報
LTE解析結果
イベント前後のサマリー
L4
JPEG
LTE raw_log
長時間ログ
などです。
ただし今回の時点では、無理に旧システムのL3/L4をコピーしません。
141. LTE raw_logはL4候補として保持している
LTEの、
raw_log
はL1から除外しました。
しかし、
lte_measurements.raw_log
にはそのまま残っています。
つまり、
L1
軽量Fact
元テーブル
raw_log保持
です。
将来、
L4
を実装するときに再利用できます。
142. カメラも将来L3/L4へ統合する
USBカメラ自体は既にRON-Vで動作しています。
現在は、
JPEG保存
まで確認しています。
ただし、
SHOCK発生前
SHOCK発生後
のようなEvidence Bufferはまだ作っていません。
そのため今回の#15前編では、
カメラ画像をL3 Evidenceとして完成
とはしません。
今後、
イベント
↓
前後画像
↓
Evidence object
という仕組みを追加する余地を残します。
143. 前編で確認できた項目
ここまで実機で確認できた項目をまとめます。
[OK] sync_objects
[OK] sync_deliveries
[OK] sync_cursors
[OK] L0 23-byte packet
[OK] CRC16
[OK] GNSS no_fix処理
[OK] network_flags
[OK] Vehicle L1
[OK] LTE L1
[OK] LTE raw_logをL1から分離
[OK] L2 600秒 summary
[OK] Integration Cycle
[OK] Controller systemd化
[OK] Controller timer
[OK] Delivery Worker
[OK] Delivery systemd化
[OK] Delivery timer
[OK] Meshtastic real ACK
[OK] Controller → Delivery完全自動連携
[OK] 一晩連続運転
[OK] Controller start-to-start 約600秒化
[OK] SQLite integrity
144. 今回の統合で成立した経路
今回最終的に成立した経路は、
観測
↓
SQLite
↓
L0 / L1 / L2
↓
sync_objects
↓
sync_deliveries
↓
Delivery Worker
↓
mesh_messages
↓
Meshtastic Gateway
↓
920MHz
↓
real ACK
です。
これを、
systemd timer
だけで自動実行できるようになりました。
145. 前編終了時点のサービス構成
RON-Vのサービス構成は、
ronv-vehicle.service
Vehicleイベント取得
ronv-web.service
Web UI
ronv-mesh.service
Meshtastic Gateway
ronv-controller.service
Integration Cycle
ronv-controller.timer
約600秒周期
ronv-delivery.service
Delivery Worker
ronv-delivery.timer
配送queue確認
gpsd.socket
GNSS
grafana-server.service
可視化
となりました。
146. 前編ではまだreboot実証をしていない
ここまでの試験では、
サービス稼働中
一晩連続運転
までは確認しました。
しかし、
Raspberry Pi再起動
後に、
全service
全timer
Controller
Delivery
real ACK
が自動復帰するところまでは、まだ最終確認していません。
これは#15の後編で実施します。
147. 後編で確認すること
後編では、
統合状態確認コマンド
↓
L3/L4方針整理
↓
最終バックアップ
↓
Raspberry Pi reboot
↓
systemd自動復帰
↓
Controller自動実行
↓
Delivery自動実行
↓
920MHz real ACK
を確認します。
特に重要なのは、
reboot後に
人がserviceをstartしなくても
RON-Vが自動復帰するか
です。
148. 後編の完成条件
#15後編では、再起動後に、
ronv-vehicle.service
ronv-web.service
ronv-mesh.service
ronv-controller.timer
ronv-delivery.timer
gpsd.socket
grafana-server.service
が正常に復帰していることを確認します。
さらに、
reboot
↓
Controller timer
↓
新しいL0
↓
Delivery pending
↓
Delivery timer
↓
Mesh送信
↓
real ACK
まで自動で進めば、
RON-V Integration
の完成とします。
149. 今回のポイント
今回の統合では、
何でもControllerへ入れる
方式にはしませんでした。
代わりに、
センサーはセンサー
統合は統合
配送は配送
無線は無線
表示は表示
という責務分離にしました。
特に、
Meshtastic USB所有はGatewayのみ
というルールを維持したことで、既に安定していた#13のMesh通信を壊さずに統合できました。
150. RON-Vは何になったか
ここまでのRON-Vは、単なる、
複数センサーを載せたRaspberry Pi
ではなくなりました。
現在は、
観測
↓
構造化
↓
履歴保存
↓
同期管理
↓
配送管理
↓
無線送信
↓
ACK確認
を自動で実行するノードになっています。
これは、災害時や通信状態の悪い場所で、
重要な情報を
小さく整理して
利用可能な通信経路へ渡す
ための基礎になります。
151. 今後の拡張
今回実装した、
sync_objects
sync_deliveries
はMesh専用ではありません。
例えば将来的には、
transport = mesh
transport = cellular
transport = wifi
transport = satellite
のように配送経路を増やせます。
同じ、
object_uid
に対して、
Meshへ送る
Cellularへ送る
Wi-Fiで同期する
といった複数Deliveryを登録できます。
152. RON-Vの基本思想
RON-Vでは、
通信手段
と、
送るべき情報
を分離します。
例えば、
SHOCK
というL1 Factは、Meshtastic専用データではありません。
SHOCK Fact
│
├─ Mesh
├─ Cellular
├─ Wi-Fi
└─ 将来の衛星回線
へ展開できます。
今回の、
sync_objects
sync_deliveries
はその基礎となります。
153. 前編まとめ
#15前編では、
L0
L1
L2
という統合情報モデルを作り、
sync_objects
sync_deliveries
sync_cursors
をSQLiteへ追加しました。
さらに、
Controller
Delivery Worker
Mesh Gateway
を分離し、
ronv-controller.timer
↓
L0生成
L1取込
L2生成
Delivery登録
↓
ronv-delivery.timer
↓
mesh_messages
↓
ronv-mesh.service
↓
Meshtastic 920MHz
↓
real ACK
を完全自動で一周させました。
一晩連続運転でも、
Controller timer
Delivery timer
主要service
は稼働を継続しました。
翌朝確認した最新5配送は、
acked
attempt_count = 1
last_error = NULL
でした。
SQLiteも、
integrity_check
ok
を維持しました。
さらに、一晩運転で発見したControllerの約610秒周期を、
OnUnitActiveSec=10min
AccuracySec=1s
へ修正し、
09:08:16
09:18:17
の約601秒間隔まで改善しました。
次回
次回の#15後編では、最後にRaspberry Piを再起動します。
確認するのは、
reboot
↓
各service自動復帰
↓
各timer自動復帰
↓
Controller自動実行
↓
新しいL0生成
↓
Delivery自動処理
↓
920MHz送信
↓
real ACK
です。
ここまで確認できれば、
RON-V（Radio Observation Node – Vehicle）の統合処理を完成
とします。