สร้างระบบเชื่อมต่อแอปมือถือกับระบบชาร์จไฟฟ้า 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
- ซอฟต์แวร์โรงงานรีไซเคิล: ระบบจัดการครบวงจรสำหรับธุรกิจรีไซเคิลไทย
- คืนทุนจากซอฟต์แวร์พลังงาน: ลดต้นทุนค่าไฟได้ 15–40% จริงหรือ?
- วิธีสร้าง SOC แบบ Lightweight ด้วย Wazuh + Open Source
- วิธีเชื่อมต่อร้านค้าออนไลน์กับระบบ ERP อย่างถูกต้อง: คู่มือปฏิบัติจริง (2026)
- AI Coding Assistant ใช้เครื่องมืออะไรอยู่เบื้องหลัง? (Claude Code, Codex CLI, Aider)
- ประหยัดน้ำมันอย่างได้ผล: ฟิสิกส์ของการขับด้วยโหลดสูง รอบต่ำ
- ระบบบริหารคลังทุเรียนและผลไม้ — WMS เชื่อมบัญชี สร้างเอกสารส่งออกอัตโนมัติ
- ล้งทุเรียนยุคใหม่: หยุดนับสต็อกด้วยกระดาษ เริ่มควบคุมธุรกิจด้วยระบบ
- AI System Reverse Engineering: ใช้ AI ทำความเข้าใจระบบซอฟต์แวร์ Legacy (Architecture, Code และ Data)
- ความได้เปรียบของมนุษย์: บริการพัฒนาซอฟต์แวร์ที่ AI ไม่อาจทดแทนได้
- จาก Zero สู่ OCPP: สร้างแพลตฟอร์มชาร์จ EV แบบ White-Label
- Wazuh Decoders & Rules: โมเดลความเข้าใจที่หายไป
- การสร้างระบบติดตาม OEE แบบเรียลไทม์สำหรับโรงงานอุตสาหกรรม
- ความเชื่อเรื่อง Enterprise Software ราคาเป็นล้านกำลังจะจบลง มื่อ Open‑Source + AI กำลังแทนที่ระบบองค์กรราคาแพง
- วิธี Cache ข้อมูล Ecommerce โดยไม่แสดงราคาหรือสต็อกที่ล้าสมัย
- การนำ AI เข้าสู่ระบบ Legacy: บูรณาการ ERP, SCADA และระบบ On-Premise ด้วย Machine Learning
- ราคาของความฉลาด: AI ต้องใช้เงินเท่าไหร่กันแน่
- ทำไม RAG App ของคุณถึงพังใน Production (และวิธีแก้ไข)
- AI-Assisted Programming ในยุค AI: บทเรียนจาก *The Elements of Style* ที่ช่วยให้คุณเขียนโค้ดได้ดีกว่าด้วย Copilot
- มายาคติ AI แทนที่มนุษย์: ทำไมองค์กรยังต้องการวิศวกรและระบบซอฟต์แวร์จริงในปี 2026
