EV Charging Analytics Pipelines

By Michael Samuel Naeem · August 6, 2025 · 5 min read

You have 200 chargers across 40 sites. Site managers ask: which chargers earn money, which sit idle, and where should we install the next ones? The OCPP server knows every session. The billing system knows every invoice. The charger logs know every fault. But until you pipe all of that into an analytics pipeline, the answer lives in three disconnected systems and nobody can query it. Analytics turns operational data into decisions.

Pipeline architecture

Chargers (OCPP) → OCPP Server → Event bus (Kafka) → Stream processor → Storage → Dashboards
                                     │                    │
                                     │                    ├─ TimescaleDB (meter values)
                                     │                    ├─ PostgreSQL (sessions)
                                     │                    └─ BigQuery (analytics)
                                     │
                                     └─ Real-time alerts (faults, offline)

Events flow through Kafka for durability and replay. Stream processors normalize and enrich before writing to purpose-built stores.

Event ingestion from OCPP

Every OCPP message becomes a Kafka event:

def on_ocpp_message(charger_id: str, action: str, payload: dict):
    event = {
        "event_id": str(uuid4()),
        "charger_id": charger_id,
        "action": action,
        "payload": payload,
        "timestamp": datetime.utcnow().isoformat(),
        "source": "ocpp-server",
    }
    kafka_producer.send("ocpp-events", key=charger_id, value=json.dumps(event))

Key topics:

• ocpp-events — raw OCPP messages
• charging-sessions — session lifecycle (started, updated, completed)
• charger-status — StatusNotification changes
• meter-values — periodic MeterValues

Session enrichment

A stream processor builds session records from raw events:

def process_session_events(events: list) -> Session:
    start = next(e for e in events if e["action"] == "StartTransaction")
    stop = next((e for e in events if e["action"] == "StopTransaction"), None)
    meters = [e for e in events if e["action"] == "MeterValues"]

    session = Session(
        transaction_id=start["payload"]["transactionId"],
        charger_id=start["charger_id"],
        connector_id=start["payload"]["connectorId"],
        id_tag=start["payload"]["idTag"],
        meter_start=start["payload"]["meterStart"],
        started_at=start["timestamp"],
    )

    if stop:
        session.meter_stop = stop["payload"]["meterStop"]
        session.stopped_at = stop["timestamp"]
        session.energy_kwh = (session.meter_stop - session.meter_start) / 1000
        session.duration_minutes = (session.stopped_at - session.started_at).total_seconds() / 60
        session.status = "completed"
    else:
        session.status = "active"

    return session

Storage schema

Sessions table (PostgreSQL):

CREATE TABLE charging_sessions (
    transaction_id INTEGER PRIMARY KEY,
    charger_id TEXT NOT NULL,
    site_id TEXT NOT NULL,
    connector_id INTEGER NOT NULL,
    id_tag TEXT,
    meter_start INTEGER,
    meter_stop INTEGER,
    energy_kwh NUMERIC(10,3),
    duration_minutes NUMERIC(10,2),
    cost NUMERIC(10,2),
    status TEXT CHECK (status IN ('active', 'completed', 'reconciled')),
    started_at TIMESTAMPTZ NOT NULL,
    stopped_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_sessions_charger_time ON charging_sessions(charger_id, started_at);
CREATE INDEX idx_sessions_site_time ON charging_sessions(site_id, started_at);

Meter values (TimescaleDB):

CREATE TABLE meter_values (
    time TIMESTAMPTZ NOT NULL,
    charger_id TEXT NOT NULL,
    connector_id INTEGER,
    transaction_id INTEGER,
    measurand TEXT NOT NULL,
    value NUMERIC,
    unit TEXT
);
SELECT create_hypertable('meter_values', 'time');

Key metrics and queries

Utilization by site (last 30 days):

SELECT
    site_id,
    COUNT(DISTINCT charger_id) AS chargers,
    SUM(duration_minutes) / (COUNT(DISTINCT charger_id) * 30 * 24 * 60.0) AS utilization_rate,
    SUM(energy_kwh) AS total_energy,
    SUM(cost) AS total_revenue
FROM charging_sessions
WHERE status = 'completed'
  AND started_at > NOW() - INTERVAL '30 days'
GROUP BY site_id
ORDER BY utilization_rate DESC;

Peak demand by hour:

SELECT
    date_trunc('hour', started_at) AS hour,
    COUNT(*) AS sessions_started,
    SUM(energy_kwh) AS energy_delivered
FROM charging_sessions
WHERE started_at > NOW() - INTERVAL '7 days'
GROUP BY hour
ORDER BY hour;

Fault rate:

SELECT
    charger_id,
    COUNT(*) FILTER (WHERE status = 'Faulted') AS fault_count,
    COUNT(*) AS total_status_changes,
    COUNT(*) FILTER (WHERE status = 'Faulted')::float / COUNT(*) AS fault_rate
FROM charger_status_events
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY charger_id
HAVING COUNT(*) FILTER (WHERE status = 'Faulted') > 0
ORDER BY fault_rate DESC;

Real-time dashboards

Grafana panels that matter to operators:

• Live map — charger status (Available/Charging/Faulted/Offline) by site
• Today's revenue — running total with comparison to yesterday
• Utilization heatmap — hour-of-day × day-of-week
• Active sessions — count, total power draw
• Fault feed — recent StatusNotification faults with auto-refresh

Connect Grafana to TimescaleDB for time-series panels and PostgreSQL for session tables.

Data quality checks

Run daily validation jobs:

def daily_quality_checks():
    orphans = db.query("SELECT COUNT(*) FROM charging_sessions WHERE status = 'active' AND started_at < NOW() - INTERVAL '24 hours'")
    assert orphans < 10, f"{orphans} orphaned sessions"

    energy_gaps = db.query("""
        SELECT transaction_id FROM charging_sessions
        WHERE status = 'completed' AND energy_kwh IS NULL
    """)
    assert len(energy_gaps) == 0, f"{len(energy_gaps)} sessions missing energy"

    meter_consistency = db.query("""
        SELECT s.transaction_id FROM charging_sessions s
        JOIN meter_values m ON s.transaction_id = m.transaction_id
        WHERE s.status = 'completed'
        GROUP BY s.transaction_id, s.meter_start, s.meter_stop
        HAVING ABS(s.meter_stop - s.meter_start - MAX(m.value) + MIN(m.value)) > 100
    """)
    alert_if_nonzero("meter_inconsistency", meter_consistency)

Common production mistakes

Teams get ev charging analytics pipeline wrong in predictable ways:

• Skipping failure-mode rehearsal — run a game day or fault injection exercise before peak traffic, not after the first outage.
• Missing correlation context — every error path should carry request, trace, or tenant identifiers so incidents are debuggable.
• Optimizing for demo, not steady state — load tests, cache warm-up, and cold-start paths matter more than local dev latency.
• Undocumented trade-offs — if you chose speed over strict correctness (or vice versa), write that down for the next engineer.

IoT deployments of ev charging analytics pipeline fail in the field when firmware assumes stable Wi-Fi, OTA rollback is untested, and device certificates expire without automated renewal.

Debugging and triage workflow

When ev charging analytics pipeline misbehaves in production, work top-down instead of guessing:

1. Confirm scope — one tenant, region, or deployment stage? Narrow blast radius before deep diving.
2. Check recent changes — deploys, flag flips, config pushes, and schema migrations in the last 24 hours.
3. Compare golden signals — latency, error rate, saturation, and traffic for the affected surface vs. baseline.
4. Reproduce minimally — smallest input or scenario that triggers the failure; capture traces/logs with correlation IDs.
5. Fix forward or rollback — if rollback is faster than root-cause during incident, rollback first, postmortem second.
6. Add a guard — alert, integration test, or circuit breaker so the same class of failure is caught earlier next time.

Document the timeline during triage. Future you (and on-call) will need timestamps, not just conclusions.

Resources

• TimescaleDB documentation — time-series extension for PostgreSQL
• Apache Kafka documentation — event streaming for OCPP message ingestion
• OCPP 1.6 specification — event types and payload formats
• Grafana time-series panels — dashboard configuration for operational metrics