AWS Panorama

初心者から実務者向けの包括的解説

AWS Panorama は、既存の IP カメラネットワークにエッジ処理機能を追加し、オンプレミス環境でコンピュータービジョン推論をリアルタイムに実行するサービスです。Panorama Appliance（専用ハードウェア）または認定デバイスを導入することで、映像全体をクラウドに送信することなく、エッジデバイス上で物体検出・人数カウント・品質検査を実現します。

⚠️ End of Support Notice: AWS Panorama は 2026年5月31日をもってサポート終了予定です。詳細参照。

概要

AWS Panorama とは

AWS Panorama は 「既存 IP カメラを AI 対応にアップグレードするエッジコンピュータビジョンサービス」 です。

• Panorama Appliance: NVIDIA GPU 搭載の専用ハードウェア
• エッジ推論: ローカルネットワークで RTSP 映像を処理
• クラウド統合: 結果のみをクラウドに送信（低帯域）
• 複数カメラ対応: 単一アプライアンスで複数ストリーム処理
• SageMaker 統合: クラウド学習モデルをエッジにデプロイ

Panorama を選ぶべき状況

以下のいずれかに該当すれば Panorama は最適です：

• 既存の CCTV カメラ資産（投資）を有効活用したい
• エッジで低レイテンシ推論が必要（数十 ms レベル）
• 映像全体をクラウドに送信する帯域がない
• プライバシー規制（映像ローカル保持）に対応する必要がある
• 製造ラインの品質検査・工場安全管理を自動化したい

Panorama が解決する課題

主な特徴

✅ RTSP カメラとの直接互換性

• IP カメラからの RTSP ストリームを直接取り込み
• H.264 / H.265 対応
• 複数ストリームの同時処理

✅ NVIDIA GPU による高速推論

• Tegra Xavier / Orin などの SOM 搭載
• 複数並列推論エンジン
• FPS 対応（720p 30 FPS、4K 対応）

✅ SageMaker Neo 最適化

• PyTorch / TensorFlow モデルのエッジ最適化
• モデルサイズ削減・推論速度向上
• 精度の低下を最小化

✅ クラウド無しで動作

• インターネット断時も推論継続
• OTA 更新は接続時に自動

✅ 複数カメラの一元管理

• AWS Management Console でグローバル管理
• DeviceFleet で複数拠点の Appliance 管理

アーキテクチャ

既存 IP カメラネットワーク（RTSP）
    ↓ ローカルネットワーク（低遅延）
Panorama Appliance（エッジ）
    ├── NVIDIA GPU / CPU
    ├── 複数カメラストリーム入力
    ├── ML 推論エンジン
    └── 結果出力（アラート / メタデータ）

クラウド（AWS）
    ← 推論結果（CSV / JSON）
    ← CloudWatch ログ・メトリクス
    ← S3 上異常画像保存
    → OTA モデル配信

データフロー例（製造ライン欠陥検査）

工場ラインカメラ（RTSP）
    → Appliance で実時間フレーム取得
    → 欠陥検出モデル推論（20ms/フレーム）
    → IsAnomalous = true → S3 に欠陥画像保存 + SNS アラート
    → IsAnomalous = false → ログのみ記録

コアコンポーネント

1. Panorama Appliance

仕様

• NVIDIA Jetson Xavier NX / AGX Orin ベース
• GPU メモリ: 8-32 GB
• ストレージ: 256 GB NVMe SSD
• ネットワーク: ギガビットイーサネット + WiFi オプション
• 電源: AC 100-240V / DC PoE 対応
• IP-62 防塵・防滴対応

価格

• ハードウェア: 約 $4,000
• サービス料: $600/年/デバイス

2. Panorama Camera Input

class CameraInput:
    """RTSP カメラからストリーム取得"""
    def __init__(self):
        self.camera_url = "rtsp://192.168.1.100/stream"  # IP カメラ
        self.credentials = {"username": "admin", "password": "***"}

    def connect(self):
        """RTSP ストリーム確立"""
        pass

3. Panorama Model

モデル対応フォーマット

• PyTorch (.pth / .pt)
• TensorFlow 2.x (.h5 / SavedModel)
• ONNX (.onnx)

パッケージング例

panorama-cli model package \
    --model-dir ./models/defect_detection \
    --output-s3-location s3://my-bucket/models/

4. Panorama Application

Application は Python コードで構成：

import panoramasdk
import cv2
import numpy as np

class DefectDetectionApp(panoramasdk.node):
    def __init__(self):
        super.__init__
        self.model = panoramasdk.model({
            'ModelPackageName': 'defect-detector',
            'BatchSize': 1
        })

    def process_streams(self):
        """各ストリームを処理"""
        streams = self.inputs.video_in.get

        for stream in streams:
            frame = stream.image  # numpy (H, W, 3)

            # 前処理
            resized = cv2.resize(frame, (224, 224))
            normalized = resized.astype(np.float32) / 255.0

            # 推論
            output = self.model.predict(normalized)

            # ポスト処理
            if output[0] > 0.8:  # 欠陥スコア 80%
                stream.add_label(
                    text="DEFECT",
                    loc_x=0.1,
                    loc_y=0.1
                )
                # S3 に欠陥画像を保存
                self.outputs.defect_frames.put(stream)
            else:
                stream.add_label("OK", 0.1, 0.1)

        self.outputs.video_out.put(streams)

def main:
    app = DefectDetectionApp
    while True:
        app.process_streams

if __name__ == "__main__":
    main

5. Panorama Node

Node は Application 内の処理ユニット：

class DataOutput(panoramasdk.node):
    """推論結果をクラウドに送信"""

    def process_streams(self):
        results = self.inputs.data_in.get

        for result in results:
            # CloudWatch Metrics に記録
            self.metric({
                'Namespace': 'Panorama/Production',
                'MetricName': 'DefectCount',
                'Value': result['defect_count']
            })

            # S3 に JSON ログを保存
            self.s3({
                'Bucket': 'my-results',
                'Key': f"results/{result['timestamp']}.json",
                'Body': json.dumps(result)
            })

対応デバイス

公式 Panorama Appliance

• AWS 製造の専用ハードウェア
• NVIDIA GPU 統合
• 複数ストリーム対応
• IP-62 防塵・防滴

Panorama Enabled Device

認定パートナーデバイス：

• Lenovo ThinkStation Appliance
• その他産業用コンピュータ

Panorama SDK 統合可能デバイス

• NVIDIA Jetson シリーズ（Nano, Xavier, AGX Orin）
• x86 Linux マシン（Ubuntu 20.04+）
• Raspberry Pi（制限あり）

アプリケーション開発

開発フロー

1. ローカルマシンで開発・テスト

   python my_app.py --test-input /path/to/test.mp4
   

2. SageMaker Neo でモデル最適化

   import sagemaker
   
   sess = sagemaker.Session
   model_name = "defect-detector"
   
   # モデルを Neo にコンパイル
   compiled_model = sess.create_model_from_job(
       job_name="my-training-job"
   ).compile(
       target_instance_family="ml.eia2.medium",
       input_shape={'input': [1, 224, 224, 3]},
       framework='pytorch',
       framework_version='1.9'
   )
   

3. Application をパッケージング

   panorama-cli application package \
       --app-dir ./my_app \
       --output-s3-location s3://my-bucket/apps/
   

4. Appliance にデプロイ

   panorama-cli deployment deploy \
       --devices arn:aws:panorama:region:account:device/device-id \
       --application-s3-location s3://my-bucket/apps/my_app.tar.gz
   

開発ツール

# Panorama CLI インストール
pip install panorama-cli

# プロジェクト初期化
panorama-cli create --template-dir my_project

# ローカルテスト実行
panorama-cli test \
    --container-image-name my-app:latest \
    --input-video /path/to/sample.mp4

# Device Fleet デプロイ
panorama-cli deployment create \
    --name my-deployment \
    --devices device-1 device-2 device-3

推論フロー

リアルタイム推論パイプライン

入力: RTSP ストリーム（30 FPS）
    ↓
フレームキャプチャ（毎フレーム）
    ↓
前処理（リサイズ・正規化・カラー変換）
    ↓
バッチ作成（複数フレーム）
    ↓
GPU 推論（並列処理）
    ↓
ポスト処理（NMS・ラベリング・スコアリング）
    ↓
出力: メタデータ / 画像アノテーション

計算効率

バッチサイズ: 1 フレーム
入力解像度: 1920x1080 (Full HD)
モデル: ResNet-50 (SageMaker Neo 最適化)
推論時間: 20-50 ms（フレーム）
スループット: 20-50 FPS

エッジでのモデル最適化

SageMaker Neo の活用

SageMaker Neo はモデルを Panorama Appliance のハードウェアに最適化：

# TensorFlow モデルの最適化
optimized_model = sagemaker.model.neo.compile(
    model_name="defect-detector",
    input_shape={
        'input_1': [1, 224, 224, 3]
    },
    framework='tensorflow',
    framework_version='2.8',
    target_device='panorama',
    role=role,
    s3_location='s3://bucket/model.tar.gz'
)

最適化のメリット

• モデルサイズ削減: 90% 圧縮
• 推論速度向上: 5-10倍高速化
• 精度維持: 量子化で 1-2% 低下抑制
• メモリ使用量削減: GPU メモリ節約

クラウド統合

1. AWS IoT Core への接続

import json
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

class IoTPublisher(panoramasdk.node):
    def __init__(self):
        self.client = AWSIoTMQTTClient("panorama-device")
        self.client.configureEndpoint(
            "a1234567-ats.iot.region.amazonaws.com",
            8883
        )
        self.client.configureCredentials(
            caPath="/etc/panorama/ca-cert.pem",
            keyPath="/etc/panorama/private.key",
            certPath="/etc/panorama/cert.pem"
        )
        self.client.connect

    def publish_results(self, results):
        payload = {
            'timestamp': int(time.time),
            'device_id': 'arn:aws:panorama:...',
            'defect_count': results['defect_count'],
            'confidence': results['confidence']
        }
        self.client.publish(
            topic="panorama/detections",
            QoS=1,
            payload=json.dumps(payload)
        )

2. AWS Lambda トリガー

# Panorama → S3 → Lambda 自動トリガー

# S3 イベント通知
{
    "Records": [
        {
            "s3": {
                "bucket": {"name": "defects-bucket"},
                "object": {"key": "2025-04-27/001.jpg"}
            }
        }
    ]
}

# Lambda ハンドラー
def lambda_handler(event, context):
    """S3 に保存された欠陥画像を処理"""
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # 欠陥画像を分析
        response = rekognition.detect_text(
            Image={'S3Object': {'Bucket': bucket, 'Name': key}}
        )

        # 結果を SNS で通知
        sns.publish(
            TopicArn='arn:aws:sns:region:account:defect-alerts',
            Subject='Manufacturing Defect Detected',
            Message=json.dumps(response)
        )

3. CloudWatch Metrics への送信

class MetricsPublisher(panoramasdk.node):
    def __init__(self):
        self.cloudwatch = boto3.client('cloudwatch')

    def publish_metrics(self, results):
        self.cloudwatch.put_metric_data(
            Namespace='Panorama/Production',
            MetricData=[
                {
                    'MetricName': 'DefectRate',
                    'Value': results['defect_rate'],
                    'Unit': 'Percent',
                    'Timestamp': datetime.utcnow
                },
                {
                    'MetricName': 'ProcessingLatency',
                    'Value': results['latency_ms'],
                    'Unit': 'Milliseconds'
                }
            ]
        )

主要ユースケース

1. 製造ライン品質検査

シナリオ: プリント基板（PCB）の製造ラインで、ハンダ付け・部品実装の不良を検査

入力: PCB ラインカメラ（全数検査）
処理:
    - 基板を 224x224 リサイズ
    - 欠陥検出モデル推論（20ms）
    - IsAnomalous = true → 不良品排出機構に信号
出力: 不良率 KPI / 欠陥パターン分析

2. 小売店舗の来客分析

入力: 店舗入口カメラ
処理:
    - 人物検出・カウント
    - 客動線追跡
    - 滞留時間集計
出力:
    - リアルタイム来客数
    - 店舗レイアウト最適化
    - 実績連動ディスプレイ

3. 工場安全管理

入力: 工場フロアカメラ
処理:
    - 安全ベスト・ヘルメット検出
    - 危険エリア侵入検知
    - 正しい動作確認
出力:
    - ヒヤリハット検出
    - KPI ダッシュボード
    - アラート通知

4. 交通・道路監視

入力: 信号機カメラ / 駐車場カメラ
処理:
    - 車両検出・分類
    - 違法駐車検出
    - 交通流量監視
出力:
    - 交通量レポート
    - 駐車区画最適化
    - 違法駐車アラート

セキュリティ

デバイスセキュリティ

1. トラスト・アンカー

   • Appliance 内部に秘密鍵保存
   • AWS IoT 認証書による識別

2. 暗号化

   • TLS 1.2 による通信暗号化
   • S3 への送信データ暗号化

3. アクセス制御

   • IAM ロールで権限管理
   • デバイス別の S3 バケット制限

アプリケーション署名

# Application に署名（デプロイ前）
panorama-cli application sign \
    --app-dir ./my_app \
    --certificate /path/to/cert.pem \
    --private-key /path/to/key.pem

ネットワーク分離

• Appliance ←（プライベート VLAN）→ IP カメラ
• ↓（インターネット）
• AWS（結果通信のみ）

コスト構造

ハードウェアコスト

サービスコスト

年間総コスト例

単一拠点: Appliance x1 + サービス
年間: $600 + 電力（$50/年） = $650

5拠点展開: Appliance x5
年間: $600 x 5 + 電力 $250 = $3,250

ROI シミュレーション

従来: 検査員 2 名 × $40,000/年 = $80,000/年
Panorama: $3,250/年 + メンテナンス $500 = $3,750/年
年間削減: $76,250（初年度 Net ROI）

Kinesis Video Streams との比較

他エッジ CV プラットフォーム比較

マイグレーション戦略

SageMaker / Bedrock への移行

現在（Panorama）:
    既存 IP カメラ → Panorama Appliance → エッジ推論

将来（推奨）:
    既存 IP カメラ → AWS IoT Greengrass → SageMaker Edge Manager
        または
    既存 IP カメラ → Bedrock Vision API → クラウド推論

移行ステップ

1. モデルエクスポート

   # ONNX / SavedModel フォーマットで出力
   python export_model.py --format onnx
   

2. SageMaker へのインポート

   import sagemaker
   model = sagemaker.model.Model(
       image_uri='382416733822.dkr.ecr.region.amazonaws.com/onnx-inference',
       model_data='s3://bucket/model.onnx',
       role=role
   )
   

3. Greengrass デプロイ

   aws greengrass create-component-version \
       --inline-recipe "{...}"
   

ベストプラクティス

1. ネットワーク設計

推奨:
    カメラ（ローカル VLAN） → Appliance（有線 ギガビット）
    Appliance → クラウド（VPN または専用線）

非推奨:
    WiFi カメラ（遅延） → Appliance（WiFi）
    インターネット経由の低遅延信号送信

2. モデル選択

推奨:
    - 事前学習済み：YOLO v8, ResNet, MobileNet
    - Lightweight：MobileNet v2, SqueezeNet
    - 量子化対応

非推奨:
    - 大規模モデル：ResNet-152, Vision Transformer
    - 量子化非対応モデル

3. リソース管理

# メモリ効率的な推論
class OptimizedApp(panoramasdk.node):
    def __init__(self):
        # バッチサイズ 1（メモリ削減）
        self.model = panoramasdk.model({
            'ModelPackageName': 'detector',
            'BatchSize': 1
        })
        # フレームキャッシュ最小化
        self.frame_cache = []

    def process_streams(self):
        streams = self.inputs.video_in.get
        # スキップフレーム処理（毎フレーム推論しない）
        for i, stream in enumerate(streams):
            if i % 3 == 0:  # 3 フレームごと
                output = self.model.predict(stream.image)

4. エラーハンドリング

class RobustApp(panoramasdk.node):
    def process_streams(self):
        try:
            streams = self.inputs.video_in.get
            for stream in streams:
                try:
                    result = self.model.predict(stream.image)
                except Exception as e:
                    # スキップ（キープアライブ）
                    self.logger.error(f"Inference failed: {e}")
                    stream.add_label("ERROR", 0.1, 0.1)
                    continue
        except Exception as e:
            self.logger.critical(f"Stream error: {e}")
            # Fallback: 最後の成功フレーム再利用
            pass

トラブルシューティング

Appliance が起動しない

1. 電源確認

   # SSH で接続テスト
   ssh ubuntu@<appliance-ip>
   

2. ネットワーク確認

   # Appliance から AWS に到達可能か確認
   curl https://panorama.region.amazonaws.com
   

3. デバイス登録確認

   aws panorama describe-device --device-id <device-arn>
   

推論精度が低い

① モデルがハードウェアに最適化されているか確認
   → SageMaker Neo で再コンパイル

② 入力前処理が正しいか確認
   → テスト画像で前処理結果を検証

③ 推論ノードのスコア閾値調整
   → 応報スコア = 0.8 から 0.7 に下げる

デプロイが失敗する

エラー: "DeviceAlreadyHasApplication"
→ 既存アプリを削除: aws panorama delete-application

エラー: "ModelNotFound"
→ S3 パスを確認: s3://bucket/path/to/model.tar.gz

エラー: "InsufficientDeviceMemory"
→ バッチサイズを 1 に設定
→ モデルを軽量化（量子化）

End of Support Notice

重要なお知らせ

AWS Panorama は 2026年5月31日をもってサポート終了予定です。

• 新規利用: 代替手段（SageMaker / Bedrock）の使用を検討してください
• 既存利用者: サポート期間内に代替ソリューションへの移行を計画してください

推奨される移行経路

オプション 1: AWS IoT Greengrass + SageMaker

• Panorama Appliance
• → AWS IoT Greengrass エッジ処理
• • SageMaker Edge Manager でモデル管理

メリット

• より柔軟なエッジ処理
• SageMaker との深い統合
• 複数 AWS サービスの連携

オプション 2: Bedrock Vision API

IP カメラ → クラウド（Bedrock） → 推論結果

# コード例
import boto3
bedrock = boto3.client('bedrock-runtime')
response = bedrock.invoke_model(
    modelId='anthropic.claude-3-5-vision-20241022-v2:0',
    body=image_data,
    contentType='image/jpeg'
)

メリット

• セットアップが簡単
• モデル管理不要
• スケーラビリティ

まとめ

AWS Panorama は 「既存 IP カメラをエッジ AI 推論で最強にアップグレードするサービス」 です。

強み

• 既存カメラ資産を活用（初期投資最小化）
• 低レイテンシ・低帯域の運用（製造ライン向け）
• SageMaker との統合（モデル管理の一元化）

制約

• ハードウェア購入必須（$4K）
• 2026年5月31日でサポート終了
• 複雑なカスタマイズには限界

選択基準

• ✅ 既存 RTSP カメラを活用したい
• ✅ エッジでの低レイテンシ推論が必須
• ✅ SageMaker モデルを使いたい
• ❌ 2026年以降の長期運用を計画している

公式ドキュメント

• AWS Panorama Developer Guide
• AWS Panorama Application SDK
• AWS Panorama Pricing
• AWS Panorama End of Support

最終更新: 2025年4月27日