【AWS】SDVのテレメトリ収集とOTA更新を模擬してみた

1. 概要

SDV（Software Defined Vehicle）で中核となる2つの機能――「車両からのテレメトリ収集」と「車両へのOTA（Over The Air）ファームウェア更新」――を、AWS IoT CoreとIoT Jobsを使って最小構成で模擬しました。

実車の代わりにPythonスクリプト（以下sim.py）を「車両」に見立て、以下の一連の流れを構築しています。

• 車両からAWSへ、5秒おきにテレメトリ（速度・バッテリー・温度など）を送信
• AWS側でそれを受け取り、DynamoDBに保存
• AWSから車両へ、ファームウェア更新の指示（ジョブ）を配信
• 車両側がジョブを検知し、ファイルをダウンロードして更新、結果をAWSに報告

起動時のポーリングと稼働中のプッシュ通知を両方実装し、ファームウェア本体は非公開のS3バケット＋署名付きURLで配布するところまで、実装してみました。

2. なぜやろうと思ったか

AWS認定資格（SAA-03）を取得できたので、興味ある分野がどのようにAWS上で構築されているか手を動かして理解を深めてみたいと考えたのがきっかけです。
以前から興味のあったSDVにテーマを決めたのですが、実車や専用のシミュレーション環境を用意するのはさすがにハードルが高いと考え、まずはSDV中核概念であるテレメトリ収集とOTA更新をAWSのマネージドサービスだけで再現してみることにしました。

3. アーキテクチャ図

[テレメトリ収集]
sim.py (awsiotsdk)
  │ MQTT publish
  ▼
vehicle/{thingName}/telemetry
  │
  ▼
AWS IoTルール (SQL: SELECT * FROM 'vehicle/+/telemetry')
  │ Lambda invoke
  ▼
Lambda (sdv-mock-telemetry-handler)
  │ put_item
  ▼
DynamoDB (sdv-mock-telemetry)


[OTA更新]
S3 (非公開バケット)
 ├─ firmware-v1.1.0.bin      … ファームウェア本体
 └─ job-doc.json             … ジョブドキュメント(署名付きURLを含む)
  │
  ▼
AWS IoT Jobs (ジョブ作成)
  │
  ├─ notify-next … 稼働中の車両へプッシュ通知
  └─ get/accepted … 起動時に車両からの問い合わせに応答
  │
  ▼
sim.py
  │ ダウンロード・バージョン更新
  ▼
jobs/{jobId}/update … 完了報告(SUCCEEDED)

学びのメモ
テレメトリ収集は車両→AWSの一方通行
OTAはAWS→車両への指示と、車両→AWSへの結果報告という双方向のやり取り

4. やったことの詳細ステップ

4-1. Thingの登録と証明書発行

AWS IoT Coreで車両を表す「モノ（Thing）」を1つ登録し、証明書を発行しました。

証明書を車両側（sim.py）に持たせることで、AWSとの通信を確立します。

4-2. テレメトリ送信の実装

sim.pyから5秒おきに、以下のようなJSONをMQTTでpublishするようにしました。

payload = {
    "vehicleId": THING,
    "speed": 42,
    "battery": 88,
    "temp": 31.5,
    "fwVersion": current_version,
    "ts": int(time.time()),
}
conn.publish(topic=TELEMETRY_TOPIC, payload=json.dumps(payload), qos=mqtt.QoS.AT_LEAST_ONCE)

4-3. IoTルール・Lambda・DynamoDBの連携

IoTルールでトピックを購読し、Lambdaを呼び出してDynamoDBに保存する構成にしました。

SELECT * FROM 'vehicle/+/telemetry'

Lambda側では、DynamoDBがfloat型を直接扱えない点に対応し、Decimal型に変換して保存しています。

item = json.loads(json.dumps(event), parse_float=Decimal)
table.put_item(Item=item)

4-4. OTA：S3とジョブドキュメントの準備

ファームウェア本体（ダミーファイル）と、更新内容を記述したジョブドキュメントをS3に配置しました。ジョブドキュメントには、boto3で発行した署名付きURLを埋め込んでいます。

url = s3.generate_presigned_url(
    'get_object',
    Params={'Bucket': BUCKET, 'Key': KEY},
    ExpiresIn=3600,
)

バケット自体は非公開のままにできるため、実運用でも通用する構成と考えています。

4-5. OTA：ジョブの検知と適用

sim.py側は、起動時と稼働中の2つの方法でジョブを検知するようにしました。

• 起動時：$aws/things/{thing}/jobs/get に問い合わせ、保留中のジョブを能動的に取得（プル型）
• 稼働中：$aws/things/{thing}/jobs/notify-next を購読し、新しいジョブが作成された瞬間に通知を受け取る（プッシュ型）

ジョブを検知すると、ジョブ詳細（ファームウェアURLとバージョン）を取得し、ファイルをダウンロードしてバージョン情報を更新、結果をAWSに報告します。

def process_job(job_id):
    # ジョブ詳細を取得
    conn.publish(topic=f"$aws/things/{THING}/jobs/{job_id}/get", payload="{}", ...)
    # firmwareUrl, version を受け取ったら
    success = apply_update(firmware_url, new_version)
    report_job_status(job_id, "SUCCEEDED" if success else "FAILED")

5. できたこと、詰まったこと

できたこと

• テレメトリの一連のパイプライン（MQTT→ルール→Lambda→DynamoDB）が動作し、DynamoDBに車両データが蓄積されることを確認
  

• OTAジョブを起動時のポーリングと稼働中のプッシュ通知の両方で検知し、ファームウェア更新（バージョン更新）と完了報告までを一連で実行

  • 起動時
    
  • 更新時（python起動中にジョブ更新し、自動的に更新をキャッチ）
    

• ファームウェア配布をS3の署名付きURLで行い、バケットを非公開に保ったまま安全に配信できる構成を実現

詰まったこと

DynamoDBにfloat型が保存できない（基本すぎてお恥ずかしい）
テレメトリのペイロードに含まれる温度などの小数値がそのままではDynamoDBに書き込めず、Lambda側でDecimal型に変換する処理が必要でした。

notify-nextは「サブスクライブした瞬間」には来ない
当初、「保留中のジョブがあればsubscribe時に通知が届く」と誤解していましたが、実際にはnotify-nextは「状態が変化した瞬間」にだけ送られるプッシュ型の通知でした。

デバイス不在のタイミングで作成されたジョブの通知は再送されず、jobs/getへの能動的な問い合わせ（プル型）が別途必要だと分かりました。実車が電源のON/OFFを繰り返すことを考えると、起動時はプル型、稼働中はプッシュ型という二段構えが実運用でも定石になっている理由が体感できました。

コールバックの中で同期待ちをするとデッドロックする
MQTTのコールバックはメッセージ受信専用のスレッドで動くのですが、その中で応答を待つ処理を書いてしまい、待っている間そのスレッド自体が塞がって、肝心の応答メッセージすら受け取れなくなるという状態に。コールバックから別スレッドを立てて処理を任せる形に直し、事なきを得ました。

S3の403 Forbidden
非公開バケットに対して素のURLでアクセスしようとして権限エラー。バケットポリシーで公開してみましたが、実運用に近づけるため、署名付きURLを使う方式に切り替え、非公開のまま安全に配布する構成にしました。

6. 気づき

• トピック設計そのものが実運用の設計思想を反映する
  本運用では、地域や車種をトピックの階層に埋め込み、IoTルールのtopic()関数で抽出してDynamoDBに列として保存する、という設計パターンがあることを学びました。データ規制がある国の場合は、トピックとテーブル自体を分ける運用も考えられます。今回のモックでは車両1台・1トピックのみですが、スケールする際にはこうした設計が必要になる点が構成を考える中で見えてきました。

• ジョブは「指示書」であって「データ本体」ではない
  OTAというと更新データそのものをイメージしていましたが、実際にはIoT Jobsが管理するのは「何を・どのURLから取得すべきか」という指示であり、実データの転送は別途S3などから行う、という役割が手を動かしてみて明確になりました

• プッシュ型とプル型は両方必要
  片方だけでは実運用のデバイスの挙動（起動・再起動・通信断）をカバーできないことも手を動かして初めて実感できた点です

7. まとめ

AWS IoT CoreとIoT Jobsだけで、SDVの中核機能であるテレメトリ収集とOTA更新を最小構成で再現できました。特にOTAについては、プッシュ型・プル型の使い分けや、ファームウェア配布の安全性（署名付きURL）まで踏み込んだことで、少し実運用に近い設計判断が体験できたのかなと思いました。

次のステップとして、地域別のトピック設計や、複数車両・複数ジョブへの対応、証明書ポリシーの最小権限化などに取り組み、より実運用に近い構成にブラッシュアップできればと思っています。