FastAPI を使ってモバイルアプリと OCPP EV 充電システムを連携する方法

電気自動車(EV)の普及に伴い、ユーザーはモバイルアプリから充電ステーションをリアルタイムで操作・監視することを求めています。そのためには、リアルタイム通信・高性能・拡張性のあるバックエンドが不可欠です。

そこで登場するのが FastAPI。Python 製の軽量で高速なフレームワークで、OCPP による EV 充電インフラとモバイルアプリをつなぐ理想的なソリューションです。


⚡ FastAPI とは?

FastAPI は、Python 3.7+ 向けの次世代 API フレームワークで、以下をベースに構築されています:

  • Starlette:Web サーバー/WebSocket の基盤
  • Pydantic:データ検証と型チェック
  • Uvicorn:超高速 ASGI サーバー

🚀 なぜ FastAPI が EV システムに最適なのか?

  • 非同期(async)対応:同時に多数の充電器と通信可能
  • WebSocket サポート:OCPP 1.6 に最適
  • 自動 API ドキュメント:Swagger / ReDoc 対応
  • 型安全でシンプルな構文:保守性・拡張性が高い
  • 高速処理:Node.js や Go に匹敵するパフォーマンス

✅ FastAPI vs Flask

特徴 Flask FastAPI
非同期(async) ⚠️ 限定的対応 ✅ 完全対応
WebSocket 対応 ❌ 外部ライブラリ必須 ✅ 標準搭載
型安全・データ検証 ❌ 手動で実装 ✅ Pydantic による自動化
OpenAPI ドキュメント生成 ❌ 拡張が必要 ✅ 自動生成
パフォーマンス(サーバー) 🐢 WSGI(同期) ⚡ ASGI(非同期)

🚦 今回構築するシステム

  • ✅ OCPP 1.6 に対応した WebSocket サーバー
  • ✅ モバイルアプリ用の REST API
  • ✅ アプリから充電ステーションをリモートで起動

⚙️ システム構成図

graph TD
    Charger["EV 充電器 (OCPP 1.6)"] -->|WebSocket| FastAPI_WS["FastAPI WebSocket"]
    FastAPI_REST["FastAPI REST API (モバイルアプリ)"] -->|API 呼び出し| Dispatcher
    Mobile["モバイルアプリ (Flutter / React Native)"] -->|HTTP| FastAPI_REST
    Dispatcher --> FastAPI_WS

📁 プロジェクト構成

ev-system/
├── main.py              # FastAPI メインアプリ
├── charge_point.py      # OCPP クライアントロジック
├── dispatcher.py        # RemoteStartTransaction 呼び出し
├── registry.py          # 充電器管理
└── requirements.txt

🔧 Step 1: 必要パッケージのインストール

pip install fastapi uvicorn ocpp

🧠 Step 2: charge_point.py

from ocpp.v16 import ChargePoint as CP
from ocpp.v16.enums import RegistrationStatus
from ocpp.v16 import call_result
from ocpp.routing import on

class ChargePoint(CP):
    @on('BootNotification')
    async def on_boot_notification(self, charge_point_model, charge_point_vendor, **kwargs):
        return call_result.BootNotificationPayload(
            current_time="2025-06-08T00:00:00Z",
            interval=10,
            status=RegistrationStatus.accepted
        )

🔌 Step 3: 接続管理 registry.py

class ConnectionRegistry:
    def __init__(self):
        self._registry = {}

    def add(self, cp_id, cp):
        self._registry[cp_id] = cp

    def get(self, cp_id):
        return self._registry.get(cp_id)

    def remove(self, cp_id):
        self._registry.pop(cp_id, None)

registry = ConnectionRegistry()

⚡ Step 4: リモート起動ロジック dispatcher.py

from ocpp.v16 import call

async def send_remote_start(cp, id_tag="MOBILE_USER"):
    request = call.RemoteStartTransaction(
        id_tag=id_tag,
        connector_id=1
    )
    response = await cp.call(request)
    return response.status

🚀 Step 5: アプリ本体 main.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from charge_point import ChargePoint
from dispatcher import send_remote_start
from registry import registry

app = FastAPI()

@app.websocket("/ws/{cp_id}")
async def websocket_handler(websocket: WebSocket, cp_id: str):
    await websocket.accept(subprotocol="ocpp1.6")
    cp = ChargePoint(cp_id, websocket)
    registry.add(cp_id, cp)

    try:
        await cp.start()
    except WebSocketDisconnect:
        registry.remove(cp_id)

@app.post("/api/remote-start/{cp_id}")
async def remote_start(cp_id: str):
    cp = registry.get(cp_id)
    if not cp:
        raise HTTPException(404, "充電器が接続されていません")
    status = await send_remote_start(cp)
    return {"status": status}

📲 Step 6: API テスト例

POST /api/remote-start/CP001
Content-Type: application/json

{}

/ws/CP001 に接続している OCPP 充電器に RemoteStartTransaction が送信されます。


🔐 実運用での対策

  • ✅ JWT 認証の導入
  • ✅ Redis / MongoDB で接続管理
  • ✅ Rate Limiting の実装
  • ✅ HTTPS を使った通信保護

🎯 このシステムでできること

  • モバイルアプリから充電開始/停止
  • 接続中の充電器ステータスの表示
  • ファームウェアの更新
  • 複数台の EV 充電器を一括管理

💬 まとめ

FastAPI を使うことで、EV 充電システムとモバイルアプリを簡単・高速・安全に接続できます。

  • WebSocket + REST API を 1 つのアプリで提供
  • ドキュメント生成も自動
  • スケーラブルで将来性のある設計

📦 次は何を作りたいですか?

  • Docker + Redis + MongoDB の構成?
  • Flutter クライアントアプリ?
  • 管理者向けダッシュボード?

ご要望があればぜひ教えてください。未来の EV プラットフォームを一緒に作りましょう ⚡



Get in Touch with us

Chat with Us on LINE

iiitum1984

Speak to Us or Whatsapp

(+66) 83001 0222

Related Posts

Our Products