สร้างระบบจัดการ EV Charging OCPP 1.6 ด้วย Flask[async], WebSocket และ MongoDB

การบริหารจัดการสถานีชาร์จรถยนต์ไฟฟ้า (EV Charging Station) ต้องใช้ระบบหลังบ้านที่สามารถสื่อสารแบบเรียลไทม์ สั่งงานจากระยะไกล และจัดเก็บข้อมูลได้อย่างปลอดภัย เราได้พัฒนาระบบ OCPP 1.6 CSMS ที่ออกแบบมาให้ทำงานร่วมกับสถานีชาร์จหลากหลายยี่ห้อได้อย่างมีประสิทธิภาพ ด้วยเทคโนโลยี:

🐍 Python + Flask แบบ Async
📡 WebSocket สำหรับโปรโตคอล OCPP
🗃 MongoDB สำหรับเก็บประวัติคำสั่ง
💻 Alpine.js สำหรับ UI Dashboard
🐳 Docker สำหรับใช้งานแบบ container

🧠 สถาปัตยกรรมของระบบ

graph TD
    UI["แดชบอร์ด UI\n(Alpine.js)\nพอร์ต 5050"] -->|"เรียกใช้ API"| FlaskAPI["Flask API\n(ควบคุมคำสั่ง)\nพอร์ต 6000"]
    FlaskAPI -->|"ส่งคำสั่ง OCPP\nผ่าน main_loop"| WebSocket["WebSocket Server\n(OCPP 1.6)\nพอร์ต 9000"]
    FlaskAPI -->|"บันทึก Log"| MongoDB["MongoDB\ncollection: messages"]
    WebSocket -->|"รับ/ส่งข้อมูล"| CP["Charge Point\n(OCPP 1.6 Client)"]

⚙️ เทคโนโลยีที่ใช้

ส่วนประกอบ	เทคโนโลยี
WebSocket	`python-ocpp`, `websockets`
API Backend	`Flask[async]`, `asyncio`
ฐานข้อมูล	`MongoDB`
ส่วนแสดงผล	`Alpine.js`, `fetch()`
ระบบรัน	`Docker`, `docker-compose`

🚀 การทำงานร่วมกันของ Flask[async] และ WebSocket

เรารวม WebSocket และ Flask API ไว้ในแอปเดียว โดยใช้ asyncio event loop ร่วมกัน ทั้ง WebSocket และ API ต่างใช้ loop เดียวกันเพื่อประสิทธิภาพสูงสุด โดยไม่ต้องแยกเป็นหลาย thread

✅ วิธีการใช้งานร่วมกัน

async def main():
    global main_loop
    main_loop = asyncio.get_running_loop()
    ...

แล้วในแต่ละ API route ที่จะสื่อสารกับ Charge Point:

future = asyncio.run_coroutine_threadsafe(cp.call(...), main_loop)
result = future.result(timeout=10)

🛰 OCPP WebSocket Server

เราใช้ python-ocpp เพื่อรับคำสั่ง OCPP 1.6 จากสถานีชาร์จ เช่น:

Authorize
StartTransaction
StopTransaction
DiagnosticsStatusNotification

ทุกคำขอที่เข้ามาจะถูกเก็บไว้ใน MongoDB เพื่อให้สามารถตรวจสอบย้อนหลังได้

@on(Action.heartbeat)
async def on_heartbeat(self, **kwargs):
    log_client_message(self.id, "2", "Heartbeat", **kwargs)
    return call_result.HeartbeatPayload(current_time=iso_now())

🧾 การบันทึกข้อมูลใน MongoDB

เราเก็บข้อมูลของทุกคำขอ/คำตอบ OCPP ใน collection messages:

def log_client_message(cp_id, message_type, action, **payload):
    db.messages.insert_one({
        "cp_id": cp_id,
        "type": message_type,
        "action": action,
        "payload": payload,
        "timestamp": datetime.utcnow()
    })

🛠 API สำหรับควบคุมจากระยะไกล

Flask API รองรับคำสั่งต่างๆ เช่น:

/api/remote_start_transaction
/api/remote_stop_transaction
/api/unlock_connector
/api/get_diag
/api/messages/latest

ผู้ใช้สามารถเรียกใช้งานผ่าน Dashboard หรือ HTTP client อย่าง Postman ได้ทันที

💻 Dashboard UI ด้วย Alpine.js

เราสร้าง UI เบาๆ ด้วย Alpine.js เพื่อให้ผู้ใช้งานสามารถสั่งงานสถานีชาร์จได้จากหน้าเว็บโดยตรง

ตัวอย่าง:

async remoteStart(cp_id) {
  const idTag = prompt("กรอก ID Tag:");
  const connectorId = prompt("กรอกหมายเลข Connector:");
  const res = await fetch("/api/remote_start_transaction", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ cp_id, id_tag: idTag, connector_id: parseInt(connectorId) })
  });
  alert(await res.text());
}

🐳 Docker Compose เพื่อการใช้งานที่ง่าย

ไฟล์ docker-compose.yml ของเราประกอบด้วย 3 บริการหลัก:

services:
  mongo:
    image: mongo
    ports:
      - "27017:27017"

  dashboard:
    build: ./dashboard
    ports:
      - "5050:80"

  ocpp:
    build: .
    ports:
      - "9000:9000"     # WebSocket
      - "6030:6000"     # Flask API
    volumes:
      - .:/app
    depends_on:
      - mongo

✅ จุดเด่น:

สามารถใช้งานทั้ง WebSocket และ REST API พร้อมกัน
เชื่อมต่อ MongoDB ได้ง่ายโดยไม่ต้องตั้งค่าซับซ้อน
แชร์ volume /app เพื่อให้แก้ไขโค้ดจากเครื่องนอก container ได้ทันที

📈 ตัวอย่างการเรียกดูข้อมูลล่าสุด

API /api/messages/latest จะคืนข้อมูล 100 รายการล่าสุด:

@app.route("/api/messages/latest", methods=["GET"])
def get_latest_messages():
    cp_id = request.args.get("cp_id")
    query = {"cp_id": cp_id} if cp_id else {}
    results = messages_collection.find(query).sort("_id", -1).limit(100)
    return dumps(list(results))