【Python初心者向け】EV充電器の通信プロトコル「OCPP」とmobilityhouse/ocppライブラリを使ってOCPPを実装する方法

[2, "abc123", "BootNotification", {
  "chargingStation": {
    "model": "Model X",
    "vendorName": "ABC社"
  },
  "reason": "PowerUp"
}]

[3, "abc123", {
  "currentTime": "2025-11-26T12:00:00Z",
  "interval": 300,
  "status": "Accepted"
}]

[4, "abc123", "NotImplemented", "この機能は未実装です", {}]

# ライブラリなしの場合（とても大変！）
import json

async def handle_message(raw_message):
    # JSONをパース
    data = json.loads(raw_message)
    
    # メッセージタイプを判定
    if data[0] == 2:  # CALL
        message_id = data[1]
        action = data[2]
        payload = data[3]
        
        # アクションごとに分岐
        if action == "BootNotification":
            # BootNotificationの処理
            vendor = payload.get("chargingStation", {}).get("vendorName")
            model = payload.get("chargingStation", {}).get("model")
            
            # 返事を作成
            response = {
                "currentTime": datetime.now().isoformat(),
                "interval": 300,
                "status": "Accepted"
            }
            # JSONに変換して返送
            await websocket.send(json.dumps([3, message_id, response]))
        
        elif action == "Heartbeat":
            # Heartbeatの処理...
            response = {"currentTime": datetime.now().isoformat()}
            await websocket.send(json.dumps([3, message_id, response]))
        
        elif action == "StatusNotification":
            # StatusNotificationの処理...
            pass
        
        elif action == "Authorize":
            # Authorizeの処理...
            pass
        
        # ... OCPPには50種類以上のアクションがある！
        # 全部自分で書く必要がある！
    
    elif data[0] == 3:  # CALLRESULT
        # レスポンスの処理...
        pass
    
    elif data[0] == 4:  # CALLERROR
        # エラーの処理...
        pass

# ライブラリありの場合（シンプル！）
from ocpp.routing import on
from ocpp.v201 import ChargePoint, call_result

class MyChargePoint(ChargePoint):
    
    @on('BootNotification')
    async def on_boot_notification(self, charging_station, reason):
        return call_result.BootNotification(
            current_time=datetime.now().isoformat(),
            interval=300,
            status="Accepted"
        )
    
    @on('Heartbeat')
    async def on_heartbeat(self):
        return call_result.Heartbeat(
            current_time=datetime.now().isoformat()
        )
    
    @on('StatusNotification')
    async def on_status_notification(self, timestamp, connector_status, evse_id, connector_id):
        return call_result.StatusNotification()

from ocpp.v201 import ChargePoint

class MyChargePoint(ChargePoint):
    # ここに自分のロジックを書く
    pass

from ocpp.routing import on

@on('BootNotification')  # ← BootNotificationが来たら↓を呼ぶ
async def on_boot_notification(self, charging_station, reason):
    # 処理を書く
    pass

from ocpp.v201 import call_result

return call_result.BootNotification(
    current_time="2025-11-26T12:00:00Z",
    interval=300,
    status="Accepted"
)

await charge_point.route_message(message)

充電器からメッセージが来た
         ↓
route_message() がJSONをパース
         ↓
アクション名を見る（例：BootNotification）
         ↓
@on('BootNotification') がついた関数を探す
         ↓
見つけた関数を実行
         ↓
return した内容を自動でJSONに変換
         ↓
充電器に返送

あなた：「パスタください」
店員：「はい、作りますね」

（10分間、店員はあなたの前で立ち尽くしている）
（この間、他のお客さんは注文できない）

店員：「パスタできました！」

あなた：「パスタください」
店員：「はい、作りますね。できたら呼びます」← await

（店員は他のお客さんの対応へ）
（あなたはスマホを見て待つ）

店員：「パスタできました！」

import time

def make_pasta():
    print("パスタ作り始め...")
    time.sleep(3)  # 3秒待つ（この間、何もできない！）
    print("パスタ完成！")
    return "🍝"

def make_coffee():
    print("コーヒー作り始め...")
    time.sleep(2)  # 2秒待つ
    print("コーヒー完成！")
    return "☕"

# 実行
pasta = make_pasta()   # 3秒待つ
coffee = make_coffee() # さらに2秒待つ
# 合計: 5秒かかる

import asyncio

async def make_pasta():
    print("パスタ作り始め...")
    await asyncio.sleep(3)  # 3秒待つ（でも他の処理OK）
    print("パスタ完成！")
    return "🍝"

async def make_coffee():
    print("コーヒー作り始め...")
    await asyncio.sleep(2)  # 2秒待つ
    print("コーヒー完成！")
    return "☕"

# 実行（同時に作る！）
async def main():
    results = await asyncio.gather(
        make_pasta(),
        make_coffee()
    )
    print(results)  # ['🍝', '☕']

asyncio.run(main())
# 合計: 3秒で両方完成！（並行処理）

async def my_function():      # async = この関数は非同期
    result = await some_task() # await = ここで待つ（他の処理可能）
    return result

# ✕ 間違い
result = some_async_function()  # awaitがない！
print(result)  
# → <coroutine object some_async_function at 0x...> と表示される
# （実行されていない！）

# ◯ 正しい
result = await some_async_function()
print(result)  
# → ちゃんと結果が返る

# ✕ 間違い
def my_function():
    result = await some_task()  # エラー！asyncがないのにawaitは使えない

# ◯ 正しい
async def my_function():
    result = await some_task()  # OK

from ocpp.routing import on
from ocpp.v201 import ChargePoint as OCPPChargePoint
from ocpp.v201 import call_result

class CSMSChargePoint(OCPPChargePoint):
    """CSMSとして動作するChargePointクラス"""
    
    def __init__(self, cs_id, connection):
        super().__init__(cs_id, connection)
        self.cs_id = cs_id

    @on('BootNotification')
    async def on_boot_notification(self, charging_station, reason):
        """充電器の起動通知を処理"""
        print(f"充電器 {self.cs_id} が起動しました")
        print(f"メーカー: {charging_station.get('vendorName')}")
        print(f"理由: {reason}")
        
        return call_result.BootNotification(
            current_time=datetime.now(timezone.utc).isoformat(),
            interval=300,  # 5分ごとにHeartbeat送ってね
            status="Accepted"
        )
    
    @on('Heartbeat')
    async def on_heartbeat(self):
        """定期的な死活確認を処理"""
        print(f"充電器 {self.cs_id} からHeartbeat")
        
        return call_result.Heartbeat(
            current_time=datetime.now(timezone.utc).isoformat()
        )

@router.websocket("/ocpp/{cs_id}")
async def ocpp_websocket_endpoint(websocket: WebSocket, cs_id: str):
    """充電器からのWebSocket接続を受け付ける"""
    
    # 接続を受け入れる
    await websocket.accept(subprotocol="ocpp2.0.1")
    
    # ChargePointインスタンスを作成
    charge_point = CSMSChargePoint(cs_id, websocket)
    
    # メッセージを待ち続ける
    while True:
        # メッセージを受け取る（awaitで待つ）
        message = await websocket.receive_text()
        
        # ライブラリにルーティングを任せる
        await charge_point.route_message(message)

時間軸 →
────────────────────────────────────────────────
充電器A: [メッセージ待ち...........][処理][メッセージ待ち...]
充電器B:     [メッセージ待ち..][処理][メッセージ待ち........]
充電器C:         [メッセージ待ち....][処理][メッセージ待ち..]

# 返事を返す（受け身）
return call_result.BootNotification(...)

# リクエストを送る（能動的）
await self.call(call.RequestStartTransaction(...))

@on('BootNotification')      # ◯ 正しい
@on('boot_notification')     # ✕ 間違い（大文字小文字が違う）
@on('MyCustomAction')        # ✕ 間違い（存在しないアクション）

充電器 ─── WebSocket ───► FastAPI ───► route_message() ───► @on ハンドラ
                                              ↑
                                    mobilityhouse/ocpp ライブラリ
                                    （JSONパース、ルーティング、バリデーション）