สร้างระบบเชื่อมต่อแอปมือถือกับระบบชาร์จไฟฟ้า 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
- การจำลองการรบกวนทางแม่เหล็กไฟฟ้า (EMC/EMI) บนดาดฟ้าเรือรบด้วย MEEP และ Python
- ระบบ TAK ทำงานอย่างไร: คู่มือฉบับสมบูรณ์สำหรับการรับรู้สถานการณ์แบบเรียลไทม์
- สร้างเว็บไซต์และแอปขายของออนไลน์ พร้อมระบบ AI แชทบอทอัจฉริยะ – ครบจบในที่เดียว
- ระบบแนะนำสินค้าอัจฉริยะมาแล้ว — พร้อมใช้งานในร้านของคุณ
- ปรียบเทียบ Rasa vs LangChain vs Rasa + LangChain
- เข้าใจ Wazuh ด้วยการสำรวจโครงการโอเพ่นซอร์สที่อยู่เบื้องหลัง
- วิธีเชื่อมต่อระบบยืนยันตัวตนจากแอปกับ OCPP Central System
- คู่มือสำหรับผู้เริ่มต้น: แอปชาร์จรถ EV ทำงานอย่างไร ติดต่อกับสถานีชาร์จ และคำนวณค่าใช้จ่ายอย่างไร
- สร้างระบบจัดการ EV Charging OCPP 1.6 ด้วย Flask[async], WebSocket และ MongoDB
- AI ยกระดับระบบบัญชีและคลังสินค้าใน Odoo อย่างไร (พร้อมแนวทางพัฒนา)
- พัฒนา E-commerce แบบ Fullstack ด้วย JavaScript
- สร้าง Agentic AI ด้วย Python, Langchain และ Ollama สำหรับระบบอีคอมเมิร์ซและโรงงานอัตโนมัติ
- วิเคราะห์หาสาเหตุของโค้ด P0420 ด้วย Python และข้อมูลสดจาก OBD-II
- วิธีนำแนวคิดจากหนังสือ The Mom Test มาใช้ตรวจสอบไอเดียสตาร์ทอัพของคุณ
- ควรเลือกใช้ Rasa หรือ Langchain สร้างแชทบอทเมื่อไหร่?
- แนะนำ OCR Document Manager: แปลงเอกสารเป็นข้อความได้ง่ายๆ บนเว็บ
- ผมกำลังทดสอบเครื่องมือ AI ที่ช่วยหาสินค้ามาแรงก่อนใคร — คุณสนใจไหม?
- เว็บไซต์ของคุณกำลังเสียโอกาส — เพราะมัน “เงียบเกินไป”
- Agentic AI คืออะไร? ทำไมฟาร์มของคุณถึงควรใช้ตั้งแต่วันนี้
- วิธีสร้าง RAG Chatbot ด้วย LangChain + Ollama
