สร้างระบบเชื่อมต่อแอปมือถือกับระบบชาร์จไฟฟ้า OCPP ด้วย FastAPI
ในยุคที่รถยนต์ไฟฟ้า (EV) กลายเป็นเรื่องปกติในชีวิตประจำวัน ผู้ใช้งานคาดหวังว่าจะสามารถควบคุมและตรวจสอบสถานีชาร์จผ่านมือถือได้แบบเรียลไทม์ และเพื่อให้สิ่งนี้เป็นไปได้ ระบบหลังบ้านจะต้อง รองรับ WebSocket, เร็ว, ขยายได้ และใช้งานง่าย
หนึ่งในทางเลือกที่ดีที่สุดในการสร้างระบบแบบนี้คือ FastAPI
⚡ FastAPI คืออะไร?
FastAPI คือเฟรมเวิร์กสำหรับการพัฒนา API ด้วยภาษา Python ที่ออกแบบมาให้ทันสมัย เร็ว และรองรับ asynchronous โดยตรง
พัฒนาโดยใช้พื้นฐานจาก:
- Starlette สำหรับ web/websocket
- Pydantic สำหรับตรวจสอบข้อมูลและพิมพ์ตัวแปร
- Uvicorn สำหรับรันบน ASGI server (เร็วมาก)
✅ ทำไม FastAPI เหมาะกับระบบชาร์จ EV?
- รองรับ async และ WebSocket แบบ Native เหมาะกับโปรโตคอล OCPP ที่ต้องการการสื่อสารแบบ real-time
- มีระบบ Swagger Docs อัตโนมัติ สำหรับนักพัฒนาแอปมือถือ
- ตรวจสอบข้อมูลด้วย Pydantic ทำให้เขียนโค้ดได้เร็วและปลอดภัย
- ทำงานได้เร็วเทียบเท่า Node.js หรือ Go
- โค้ดสวย อ่านง่าย และใช้งานง่าย
🤔 เปรียบเทียบ FastAPI กับ Flask
| คุณสมบัติ | Flask | FastAPI |
|---|---|---|
| Async (asynchronous) | ❌ รองรับบางส่วน | ✅ รองรับสมบูรณ์ |
| WebSocket | ❌ ต้องใช้ lib เพิ่ม | ✅ มีในตัว |
| ตรวจสอบข้อมูล (Validation) | ❌ เขียนเอง | ✅ ใช้ Pydantic |
| เอกสาร API (Swagger) | ❌ ต้องติดตั้งเพิ่ม | ✅ อัตโนมัติ |
| ความเร็ว | 🐢 sync WSGI | ⚡ async ASGI |
🚦 เราจะสร้างอะไร?
- ✅ เซิร์ฟเวอร์ WebSocket ที่รับ OCPP 1.6 จากสถานีชาร์จ
- ✅ REST API สำหรับให้แอปมือถือเรียกใช้งาน
- ✅ คำสั่ง Remote Start Transaction ผ่านแอป
🧠 โครงสร้างระบบ
graph TD
Charger["สถานีชาร์จไฟฟ้า (OCPP 1.6)"] -->|WebSocket| FastAPI_WS["FastAPI WebSocket"]
FastAPI_REST["FastAPI REST API (มือถือ)"] -->|"เรียกใช้คำสั่ง"| Dispatcher
Mobile["แอปมือถือ (Flutter/React Native)"] -->|HTTP API| FastAPI_REST
Dispatcher --> FastAPI_WS📁 โครงสร้างโปรเจกต์
ev-system/
├── main.py # FastAPI: WebSocket + REST API
├── charge_point.py # ตอบสนองคำสั่ง OCPP
├── dispatcher.py # เรียกใช้งาน RemoteStartTransaction
├── registry.py # จัดการการเชื่อมต่อ
└── requirements.txt🧪 ติดตั้งไลบรารีที่ใช้
pip install fastapi uvicorn ocpp🧠 เขียนคลาส ChargePoint – 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
)🔌 สร้าง Registry สำหรับเก็บสถานี – 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()⚡ ส่งคำสั่ง Remote Start – 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🚀 เขียนแอปหลัก – 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}🧪 ทดสอบจากแอปหรือ Postman
ตัวอย่างคำสั่ง HTTP:
POST /api/remote-start/CP001
Content-Type: application/json
{}สถานีที่เชื่อมต่อผ่าน /ws/CP001 จะได้รับคำสั่ง RemoteStartTransaction
🔐 ความปลอดภัยในระบบจริง
- 🔑 ใช้ JWT สำหรับยืนยันตัวตน
- 📦 ใช้ Redis หรือ MongoDB เก็บสถานะ
- 📈 Rate-limiting เพื่อป้องกันการโดนโจมตี
- 🛡️ เชื่อมต่อผ่าน HTTPS เท่านั้น
✅ ใช้กรณีนี้กับอะไรได้บ้าง?
- เริ่ม/หยุดชาร์จจากมือถือ
- แสดงสถานะสถานีชาร์จ
- สั่งอัปเดตเฟิร์มแวร์
- จัดการสถานีได้หลายเครื่องพร้อมกัน
🎯 สรุป
การใช้ FastAPI สร้างระบบควบคุมสถานีชาร์จผ่านแอปมือถือที่รองรับ OCPP เป็นทางเลือกที่ เร็ว ทันสมัย และขยายได้ดี
ไม่ว่าจะเป็น WebSocket หรือ REST API — FastAPI จัดการได้ทั้งหมดในตัวเดียว
📦 อยากได้เพิ่ม?
- 🐳 Docker + Redis + MongoDB?
- 📱 Flutter หรือ React Native ตัวอย่าง?
- 🔐 หน้าแอดมิน Login/Role?
บอกมาได้เลยครับ พร้อมช่วยให้ระบบ EV ของคุณไปไกล ⚡
Get in Touch with us
Chat with Us on LINE
iiitum1984
Related Posts
- สถาปัตยกรรมอ้างอิงสำหรับระบบดิจิทัลระดับจังหวัด / เทศบาล
- สถาปัตยกรรม GovTech เชิงปฏิบัติ: ERP, GIS, ระบบบริการประชาชน และแพลตฟอร์มข้อมูล
- เหตุใดระบบรับมือเหตุฉุกเฉินจึงต้องออกแบบแบบ Offline First (บทเรียนจาก ATAK)
- เหตุใดโครงการซอฟต์แวร์ภาครัฐจึงล้มเหลว — และจะป้องกันได้อย่างไรก่อนเริ่มเขียนโค้ด
- หลัง AI Hype ซาลง: อะไรจะเกิดขึ้นต่อไป (และทำไมธุรกิจไทยต้องสนใจ)
- ทำไม AI ในธุรกิจรีไซเคิลจึงล้มเหลว หากไม่มี System Integration
- ISA-95 vs RAMI 4.0: โรงงานไทยควรใช้แบบไหน (และทำไมควรใช้ทั้งสอง)
- ทำไม Low-Code ถึงกำลังตกเทรนด์ (และอะไรมาแทนที่)
- ผลิตภัณฑ์ที่ล้มเหลวมากที่สุดในปี 2025 — และเหตุผลที่แท้จริงเบื้องหลังความล้มเหลว
- Agentic AI Explained: Manus vs OpenAI vs Google — ทางเลือกที่องค์กรไทยควรรู้
- AI กับการทำ Vertical Integration ของระบบโรงพยาบาล
- AI Accelerators ในระบบ Industrial AI ทำไม Software Framework จึงสำคัญกว่าแค่ชิปประมวลผล
- พัฒนาระบบสำหรับประเทศไทย: เชื่อมต่อ EC–ERP ด้วย AI และ Workflow ที่เชื่อถือได้
- ต้นทุนที่ซ่อนอยู่ของระบบ ‘อัจฉริยะ’ ที่ทำงานไม่เสถียร
- GPU vs LPU vs TPU: เลือก AI Accelerator ให้เหมาะกับงาน
- LPU คืออะไร? บทนำเชิงปฏิบัติและการใช้งานจริงในบริบทองค์กรไทย
- แปลคำศัพท์ Cybersecurity ให้เข้าใจแบบนักพัฒนา Software
- การออกแบบระบบ Cybersecurity Monitoring & Incident Response สมัยใหม่ สถาปัตยกรรมเชิงปฏิบัติ ด้วย Wazuh, SOAR และ Threat Intelligence
- แนวคิดการเขียนโปรแกรมแบบคลาสสิกในยุค AI
- SimpliPOSFlex. POS สำหรับธุรกิจที่อยู่บนความจริงของหน้างาน
