Building an OCPP 1.6 Central System with Flask async, WebSockets, and MongoDB

Managing EV chargers requires real-time, asynchronous communication, remote control, and reliable data logging. In this post, we share how we built a production-ready, OCPP 1.6-compatible central system (CSMS) using:

🐍 Python + Flask[async]
📡 WebSocket (via python-ocpp)
🗃 MongoDB for message storage
💻 Alpine.js dashboard
🐳 Docker for modular deployment

This system allows us to control charge points, receive their status and transaction updates, and visualize system activity through a dashboard.

🧠 Architecture Overview

graph TD
    UI["Dashboard UI\n(Alpine.js)\nPort 5050"] -->|Fetch API| FlaskAPI["Flask API\n(Remote Control)\nPort 6000"]
    FlaskAPI -->|Send OCPP call\nvia main_loop| WebSocket["OCPP WebSocket Server\n(websockets)\nPort 9000"]
    FlaskAPI -->|Log Messages| MongoDB["MongoDB\nCollection: messages"]
    WebSocket -->|Receive / Respond| CP["Charge Point\n(OCPP 1.6 Client)"]

⚙️ Tech Stack

Layer	Tool / Tech
WebSocket	`python-ocpp`, `websockets`
API Server	`Flask[async]`, `asyncio`
Database	`MongoDB`
UI Frontend	`Alpine.js` + `fetch()`
Deployment	`Docker`, `docker-compose`

🚀 Flask[async] + WebSockets: Working Together

Running both a WebSocket server and Flask together in one app requires careful coordination of the asyncio event loop. Flask 2.0 introduced native async view support, making it possible to:

Serve async REST APIs using Flask[async]
Share a common event loop with the OCPP WebSocket server
await charge point communication in Flask routes using asyncio.run_coroutine_threadsafe(...)

✅ Example

We capture the main loop like this:

async def main():
    global main_loop
    main_loop = asyncio.get_running_loop()
    ...

Then in any Flask API, we safely send OCPP commands like:

future = asyncio.run_coroutine_threadsafe(cp.call(...), main_loop)
result = future.result(timeout=10)

This lets us keep everything in a single asyncio context — no threading, no blocking, and full event-driven behavior.

🛰 OCPP WebSocket Server

We use websockets.serve(...) and python-ocpp to handle standard OCPP 1.6 actions like:

Authorize
StartTransaction
StopTransaction
StatusNotification
DataTransfer

Each connected charge point is tracked in a global connected_charge_points dictionary. We also log every incoming OCPP payload into MongoDB:

@on(Action.heartbeat)
async def on_heartbeat(self, **kwargs):
    log_client_message(self.id, "2", "Heartbeat", **kwargs)
    return call_result.HeartbeatPayload(current_time=iso_now())

🧾 MongoDB Logging

We store all client payloads in the ocpp.messages collection using a generic logger:

def log_client_message(cp_id, message_type, action, **payload):
    db.messages.insert_one({
        "cp_id": cp_id,
        "type": message_type,
        "action": action,
        "payload": payload,
        "timestamp": datetime.utcnow()
    })

This makes it easy to query historical activity, debug field installations, or integrate with reporting systems.

🛠 Flask API for Remote Control

We expose REST APIs to send remote commands:

POST /api/remote_start_transaction
POST /api/remote_stop_transaction
POST /api/unlock_connector
POST /api/get_diag
GET /api/messages/latest

Each API sends a command to the corresponding charge point using cp.call(...).

💻 Alpine.js Dashboard

The frontend is powered by Alpine.js and talks to Flask using simple fetch() calls.

Each charge point shows controls to:

Start / stop a transaction
Unlock the connector
Run diagnostics
View logs via /api/messages/latest?cp_id=...

Example:

async remoteStart(cp_id) {
  const idTag = prompt("Enter ID tag:");
  const connectorId = prompt("Connector ID:");
  const res = await fetch("/api/remote_start_transaction", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ cp_id, id_tag: idTag, connector_id: parseInt(connectorId) })
  });
  alert(await res.text());
}

🐳 Docker Compose Explained

We use docker-compose.yml to orchestrate the following services:

services:
  mongo:
    image: mongo
    volumes:
      - mongo-data:/data/db
    ports:
      - "27017:27017"

  dashboard:
    build: ./dashboard
    ports:
      - "5050:80"

  ocpp:
    build: .
    volumes:
      - .:/app
    ports:
      - "9000:9000"     # WebSocket server
      - "6030:6000"     # Flask API
    depends_on:
      - mongo

volumes:
  mongo-data:

🔍 Breakdown:

mongo stores all messages.
ocpp runs both the WebSocket server and Flask async API.
dashboard serves the Alpine.js frontend.
Port 6030 (host) is mapped to Flask’s port 6000 (container).
All services share a network and MongoDB is persistent across restarts.

This setup makes it easy to scale, debug, or run locally or in the cloud.

📈 Example Query: Show Last 100 Logs

We expose an API to query messages:

@app.route("/api/messages/latest", methods=["GET"])
def get_latest_messages():
    cp_id = request.args.get("cp_id")
    query = {"cp_id": cp_id} if cp_id else {}
    results = messages_collection.find(query).sort("_id", -1).limit(100)
    return dumps(list(results))