สร้างระบบเชื่อมต่อแอปมือถือกับระบบชาร์จไฟฟ้า 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
- ข้อเสนอระบบ SimpliMES Lite — โซลูชัน MES แบบเบาสำหรับโรงงานไทย
- ทำไมร้านค้าออนไลน์ที่ประสบความสำเร็จถึงเลือกใช้ SimpliShop: สร้าง เติบโต และชนะตลาดของคุณ
- Vertical Integration of AI: อนาคตใหม่ของธุรกิจยุคดิจิทัล
- ระบบ AI Prediction — เปลี่ยนการตัดสินใจของคุณให้ทรงพลังยิ่งกว่าเดิม
- ถ้า AI Bubble แตก จะเกิดอะไรขึ้น? (วิเคราะห์จริง ไม่อิงกระแส)
- ใช้ Deep Learning + วิเคราะห์ข่าว (News Sentiment) ทำนายราคาหุ้น – คู่มือฉบับสมบูรณ์
- เปลี่ยนงาน COI ให้ง่ายขึ้นด้วย AI: ตัวอย่างใช้งานจริงในโรงงาน (Hybrid Rasa + LangChain)
- SimpliAgentic — อนาคตของโรงงานอัตโนมัติอัจฉริยะมาถึงแล้ว
- ทำไม “Android Internals” จึงสำคัญ — และบริการระดับสูงที่ธุรกิจของคุณสามารถสร้างได้จากความรู้นี้
- ทำไมธุรกิจควรพัฒนาระบบอีคอมเมิร์ซของตัวเอง (แทนการเช่าแพลตฟอร์มสำเร็จรูป)
- Upstream, Downstream และ Fork คืออะไร? คู่มือเข้าใจง่ายสำหรับนักพัฒนา Android & Linux
- บิ๊กเทคกำลังก่อ “ฟองสบู่ AI” อย่างไร? วิเคราะห์ NVIDIA, Microsoft, OpenAI, Google, Oracle และบทบาทของ AMD
- Deep Learning ในงานพัฒนาอสังหาริมทรัพย์
- บริการแก้โค้ดและดูแลระบบ Legacy — ทำให้ระบบธุรกิจของคุณเสถียร พร้อมใช้งานตลอดเวลา
- Python Deep Learning สำหรับโรงงานอัตโนมัติ: คู่มือฉบับสมบูรณ์ (อัปเดตปี 2025)
- บริการพัฒนาและฝึกอบรม Python สำหรับโรงงานอุตสาหกรรม (Factory Systems)
- ทำไม Python + Django คือ Tech Stack ที่ดีที่สุดในการสร้างระบบ eCommerce สมัยใหม่ (คู่มือฉบับสมบูรณ์ + แผนราคา)
- กลยุทธ์ซานซือหลิ่วจี (三十六计): คู่มือกลยุทธ์ธุรกิจจีนยุคใหม่ เข้าใจวิธีคิด การเจรจา และการแข่งขันแบบจีน
- เข้าใจ Training, Validation และ Testing ใน Machine Learning
- เข้าใจ Neural Network ให้ลึกจริง — ทำไมต้อง Convolution, ทำไม ReLU ต้องตามหลัง Conv2d และทำไมเลเยอร์ลึกขึ้นถึงเรียนรู้ฟีเจอร์ซับซ้อนขึ้น
