สร้างระบบเชื่อมต่อแอปมือถือกับระบบชาร์จไฟฟ้า 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
- คิดให้ดีกว่าไปกับโค้ด: ใช้คณิตศาสตร์ทางลัดเพื่อเข้าใจระบบซอฟต์แวร์ขนาดใหญ่
- สร้าง Macrohard แห่งวันนี้: แพลตฟอร์ม AI Agents สำหรับองค์กร
- พัฒนา Vue.js อย่างชาญฉลาดด้วย Aider + การเชื่อมต่อกับ IDE
- เฮ้! มาใช้ AI ผู้ช่วยโค้ดอย่าง Codex CLI กับ Aider กันเถอะ
- การทำงานร่วมกับ AI ในการเขียนโค้ดอย่างถูกวิธี
- วิธีเลือกโมเดล LLM ที่เหมาะสม: Instruct, MLX, 8-bit และ Embedding
- วิธีใช้โมเดล LLM แบบรันในเครื่อง (Local LLM) ในการทำงานประจำวัน
- วิธีใช้โมเดล Embedding ร่วมกับ LLM เพื่อสร้างแอป AI ที่ชาญฉลาดยิ่งขึ้น
- ระบบกล้องอัจฉริยะสำหรับตรวจหาข้อบกพร่องของวัสดุต่อเนื่อง
- สร้างระบบตรวจจับความเสียหายแบบเรียลไทม์ด้วยกล้อง Line-Scan + AI (แนวทางนำไปใช้ได้หลายอุตสาหกรรม)
- วิธีอ่านซอร์สโค้ด: ตัวอย่างจาก Frappe Framework
- Interface-Oriented Design: รากฐานของ Clean Architecture
- เข้าใจระบบต่อต้านโดรน (Anti-Drone System) – สถาปัตยกรรม ฮาร์ดแวร์ และซอฟต์แวร์
- RTOS vs Linux ในระบบโดรน: ออกแบบอย่างไรให้ทันสมัย ปลอดภัย และเขียนด้วย Rust ได้หรือไม่?
- ทำไม Spring ต้องใช้ Annotation เยอะ? เจาะลึกโลก Java และ Python สำหรับนักพัฒนาเว็บ
- จาก Django สู่ Spring Boot: คู่มือเปรียบเทียบฉบับเข้าใจง่ายสำหรับนักพัฒนาเว็บ
- สร้างระบบ Python ขนาดใหญ่แบบยั่งยืนด้วย Clean Architecture (พร้อมตัวอย่างและแผนภาพ)
- ทำไม Test-Driven Development (TDD) ถึงตอบโจทย์ธุรกิจยุคใหม่
- สร้างระบบ Continuous Delivery ให้ Django บน DigitalOcean ด้วย GitHub Actions และ Docker
- สร้างระบบแนะนำสินค้าในอีคอมเมิร์ซด้วย LangChain, Ollama และ Open-source Embedding แบบ Local
