Protocollo OCPP 2.x: Costruire Sistemi di Ricarica EV

# Formato URL
wss://csms.example.com/ocpp/{chargingStationId}

# Esempi reali
wss://csms.example.com/ocpp/IT-MIL-STATION-001
wss://csms.example.com/ocpp/EVSE-PARK-NORD-042
wss://csms.example.com/ocpp/CPO-AUTOGRILL-A7-01

# Headers WebSocket obbligatori
Sec-WebSocket-Protocol: ocpp2.0.1
Authorization: Basic {base64(stationId:password)}  # Security Profile 1-2

# Con Security Profile 3 (mTLS): nessun header Authorization,
# l'autenticazione avviene tramite certificato client TLS

// Formato: [MessageTypeId, MessageId, Action, Payload]

// Esempio: BootNotification dalla Charging Station
[2, "19223201", "BootNotification", {
  "chargingStation": {
    "model": "SuperCharger-500",
    "vendorName": "EVPower Inc.",
    "serialNumber": "SN-2025-00142",
    "firmwareVersion": "3.2.1",
    "modem": {
      "iccid": "8939100000000000001",
      "imsi": "310260000000001"
    }
  },
  "reason": "PowerUp"
}]

// Esempio: TransactionEvent dalla Charging Station
[2, "tx-evt-001", "TransactionEvent", {
  "eventType": "Started",
  "timestamp": "2026-03-09T10:30:00Z",
  "triggerReason": "CablePluggedIn",
  "seqNo": 0,
  "transactionInfo": {
    "transactionId": "TXN-2026-0309-001",
    "chargingState": "EVConnected"
  },
  "evse": { "id": 1, "connectorId": 1 },
  "idToken": {
    "idToken": "RFID-04A2B3C4D5",
    "type": "ISO14443"
  }
}]

// Formato: [MessageTypeId, MessageId, Payload]

// Risposta a BootNotification
[3, "19223201", {
  "currentTime": "2026-03-09T10:00:00Z",
  "interval": 300,
  "status": "Accepted"
}]

// Risposta a TransactionEvent (Started)
[3, "tx-evt-001", {
  "totalCost": 0,
  "chargingPriority": 0,
  "idTokenInfo": {
    "status": "Accepted",
    "groupIdToken": {
      "idToken": "GROUP-FLEET-01",
      "type": "Central"
    }
  }
}]

// Formato: [MessageTypeId, MessageId, ErrorCode, ErrorDescription, ErrorDetails]

[4, "19223201", "FormatViolation",
  "Il campo 'vendorName' supera la lunghezza massima di 50 caratteri",
  {
    "field": "chargingStation.vendorName",
    "maxLength": 50,
    "actualLength": 67
  }
]

// Codici di errore OCPP 2.0.1 standardizzati:
// FormatViolation            - messaggio JSON malformato
// GenericError               - errore generico non classificabile
// InternalError              - errore interno del ricevente
// MessageTypeNotSupported    - tipo di messaggio non supportato
// NotImplemented             - azione riconosciuta ma non implementata
// NotSupported               - azione non supportata dall'implementazione
// OccurrenceConstraintViolation - violazione cardinalita elementi
// PropertyConstraintViolation   - vincolo su una proprietà violato
// ProtocolError              - violazione del protocollo OCPP
// RpcFrameworkError          - errore nel framework RPC di base
// SecurityError              - errore di sicurezza o autenticazione
// TypeConstraintViolation    - tipo di dato non corretto

// CALL dal CSMS: legge stato EVSE e tipo connettore
[2, "get-var-001", "GetVariables", {
  "getVariableData": [
    {
      "component": { "name": "EVSE", "evse": { "id": 1 } },
      "variable": { "name": "AvailabilityState" },
      "attributeType": "Actual"
    },
    {
      "component": { "name": "Connector", "evse": { "id": 1, "connectorId": 1 } },
      "variable": { "name": "ConnectorType" },
      "attributeType": "Actual"
    },
    {
      "component": { "name": "SmartChargingCtrlr" },
      "variable": { "name": "ChargingProfileMaxStackLevel" },
      "attributeType": "Actual"
    }
  ]
}]

// CALLRESULT dalla Charging Station
[3, "get-var-001", {
  "getVariableResult": [
    {
      "attributeStatus": "Accepted",
      "component": { "name": "EVSE", "evse": { "id": 1 } },
      "variable": { "name": "AvailabilityState" },
      "attributeValue": "Available"
    },
    {
      "attributeStatus": "Accepted",
      "component": { "name": "Connector", "evse": { "id": 1, "connectorId": 1 } },
      "variable": { "name": "ConnectorType" },
      "attributeValue": "cCCS2"
    },
    {
      "attributeStatus": "Accepted",
      "component": { "name": "SmartChargingCtrlr" },
      "variable": { "name": "ChargingProfileMaxStackLevel" },
      "attributeValue": "5"
    }
  ]
}]

// Strategia: integra produzione solare + peak shaving ore serali
// Scenario: sito con 50kW fotovoltaico, trasformatore 100A, picco 19-21h

// 1. Limite massimo stazione (rispetta contratto di fornitura)
[2, "smart-max-001", "SetChargingProfile", {
  "evseId": 0,
  "chargingProfile": {
    "id": 1,
    "stackLevel": 1,
    "chargingProfilePurpose": "ChargingStationMaxProfile",
    "chargingProfileKind": "Absolute",
    "chargingSchedule": [{
      "id": 1,
      "chargingRateUnit": "A",
      "chargingSchedulePeriod": [
        { "startPeriod": 0, "limit": 100.0 }
      ]
    }]
  }
}]

// 2. Profilo solare + peak shaving per un singolo EVSE
[2, "smart-solar-001", "SetChargingProfile", {
  "evseId": 1,
  "chargingProfile": {
    "id": 100,
    "stackLevel": 0,
    "chargingProfilePurpose": "TxDefaultProfile",
    "chargingProfileKind": "Absolute",
    "validFrom": "2026-03-09T00:00:00Z",
    "validTo": "2026-03-10T00:00:00Z",
    "chargingSchedule": [{
      "id": 1,
      "chargingRateUnit": "A",
      "startSchedule": "2026-03-09T06:00:00Z",
      "chargingSchedulePeriod": [
        { "startPeriod": 0,     "limit": 8.0,  "numberPhases": 3 },
        { "startPeriod": 7200,  "limit": 32.0, "numberPhases": 3 },
        { "startPeriod": 14400, "limit": 32.0, "numberPhases": 3 },
        { "startPeriod": 43200, "limit": 16.0, "numberPhases": 3 },
        { "startPeriod": 46800, "limit": 8.0,  "numberPhases": 3 },
        { "startPeriod": 54000, "limit": 24.0, "numberPhases": 3 }
      ]
    }]
  }
}]
// Orario potenza:  06-08h: 8A  (offpeak, bassa produzione solare)
//                 08-12h: 32A (piena produzione FV, massima potenza)
//                 12-18h: 32A (picco solare, alta produzione)
//                 18-19h: 16A (calo solare, riduzione)
//                 19-21h: 8A  (picco domanda residenziale, min potenza)
//                 21-24h: 24A (fine picco, potenza media)

interface ChargingSession {
  readonly stationId: string;
  readonly evseId: number;
  readonly transactionId: string;
  readonly priority: number;           // 0-9 (9 = massima)
  readonly minChargingRate: number;    // A minimi per caricare
  readonly maxChargingRate: number;    // A massimi del connettore
  readonly currentChargingRate: number;
  readonly energyDelivered: number;    // Wh totali erogati
  readonly targetEnergy?: number;      // Wh target (se specificato dall'utente)
  readonly isEV3Phase: boolean;        // Veicolo trifase
}

interface LoadBalancerConfig {
  readonly maxSitePowerAmps: number;   // A max del trasformatore
  readonly reservedBuildingAmps: number; // A riservati per l'edificio
  readonly minSessionAmps: number;     // A minimi per sessione (tipico: 6A)
  readonly rebalanceIntervalSec: number; // Secondi tra ricalcoli (tipico: 30s)
}

interface Allocation {
  readonly stationId: string;
  readonly evseId: number;
  readonly allocatedAmps: number;
  readonly phases: number;
}

function calculateChargingAllocations(
  sessions: ReadonlyArray<ChargingSession>,
  config: LoadBalancerConfig
): ReadonlyArray<Allocation> {
  if (sessions.length === 0) return [];

  const availableAmps = config.maxSitePowerAmps
    - config.reservedBuildingAmps;

  // Step 1: ordina per priorità (desc), poi energia erogata (asc = meno carico prima)
  const sorted = [...sessions].sort((a, b) => {
    if (b.priority !== a.priority) return b.priority - a.priority;
    return a.energyDelivered - b.energyDelivered;
  });

  // Step 2: garantisci potenza minima a tutti
  const minRequired = sorted.length * config.minSessionAmps;
  if (minRequired > availableAmps) {
    // Caso critico: potenza insufficiente, sospendi sessioni a bassa priorità
    return sorted
      .slice(0, Math.floor(availableAmps / config.minSessionAmps))
      .map((s) => ({
        stationId: s.stationId,
        evseId: s.evseId,
        allocatedAmps: config.minSessionAmps,
        phases: s.isEV3Phase ? 3 : 1,
      }));
  }

  // Step 3: distribuzione proporzionale ai pesi di priorità
  const totalWeight = sorted.reduce(
    (sum, s) => sum + (1 + s.priority), 0
  );
  const remainingAmps = availableAmps - minRequired;

  const allocations = sorted.map((session) => {
    const weight = (1 + session.priority) / totalWeight;
    const bonus = remainingAmps * weight;
    const raw = config.minSessionAmps + bonus;

    // Applica vincoli min/max del connettore
    const allocatedAmps = Math.max(
      config.minSessionAmps,
      Math.min(session.maxChargingRate, Math.round(raw * 10) / 10)
    );

    return {
      stationId: session.stationId,
      evseId: session.evseId,
      allocatedAmps,
      phases: session.isEV3Phase ? 3 : 1,
    };
  });

  // Step 4: verifica finale che il totale non superi il limite
  const total = allocations.reduce((s, a) => s + a.allocatedAmps, 0);
  if (total <= availableAmps) return allocations;

  // Riscaling proporzionale
  const scale = availableAmps / total;
  return allocations.map((a) => ({
    ...a,
    allocatedAmps: Math.max(
      config.minSessionAmps,
      Math.round(a.allocatedAmps * scale * 10) / 10
    ),
  }));
}

-- Registro stazioni di ricarica
CREATE TABLE charging_stations (
    station_id        TEXT PRIMARY KEY,
    vendor_name       TEXT NOT NULL,
    model             TEXT NOT NULL,
    serial_number     TEXT,
    firmware_version  TEXT,
    security_profile  SMALLINT NOT NULL DEFAULT 1,
    last_boot_reason  TEXT,
    last_seen_at      TIMESTAMPTZ,
    is_online         BOOLEAN NOT NULL DEFAULT FALSE,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Stato EVSE e connettori
CREATE TABLE evse_status (
    station_id        TEXT NOT NULL REFERENCES charging_stations(station_id),
    evse_id           SMALLINT NOT NULL,
    connector_id      SMALLINT NOT NULL,
    connector_type    TEXT,   -- cCCS2, cCHAdeMO, cType2, sType3
    status            TEXT NOT NULL DEFAULT 'Unknown',
    error_code        TEXT,
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (station_id, evse_id, connector_id)
);

-- Transazioni di ricarica
CREATE TABLE transactions (
    transaction_id    TEXT PRIMARY KEY,
    station_id        TEXT NOT NULL REFERENCES charging_stations(station_id),
    evse_id           SMALLINT NOT NULL,
    connector_id      SMALLINT,
    id_token          TEXT NOT NULL,
    id_token_type     TEXT NOT NULL,
    state             TEXT NOT NULL DEFAULT 'Started',
    started_at        TIMESTAMPTZ NOT NULL,
    ended_at          TIMESTAMPTZ,
    meter_start_wh    NUMERIC(12, 3),
    meter_end_wh      NUMERIC(12, 3),
    energy_wh         NUMERIC(12, 3) GENERATED ALWAYS AS (meter_end_wh - meter_start_wh) STORED,
    stop_reason       TEXT,
    total_cost_cents  INTEGER,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Meter values (time series - usare TimescaleDB in produzione)
CREATE TABLE meter_values (
    id                BIGSERIAL PRIMARY KEY,
    transaction_id    TEXT REFERENCES transactions(transaction_id),
    station_id        TEXT NOT NULL,
    evse_id           SMALLINT NOT NULL,
    sampled_at        TIMESTAMPTZ NOT NULL,
    energy_wh         NUMERIC(12, 3),
    power_w           NUMERIC(10, 2),
    current_a         NUMERIC(8, 3),
    voltage_v         NUMERIC(8, 2),
    soc_pct           SMALLINT   -- State of Charge da ISO 15118
);

CREATE INDEX idx_meter_values_station_time
    ON meter_values(station_id, sampled_at DESC);

CREATE INDEX idx_transactions_station_id
    ON transactions(station_id, started_at DESC);

-- Token di autorizzazione (lista locale cache)
CREATE TABLE authorization_cache (
    id_token          TEXT NOT NULL,
    id_token_type     TEXT NOT NULL,
    status            TEXT NOT NULL,  -- Accepted, Invalid, Blocked, Expired
    group_id          TEXT,
    expiry_date       TIMESTAMPTZ,
    cached_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id_token, id_token_type)
);

import asyncio
import logging
from datetime import datetime, timezone
from typing import Optional, Any

import asyncpg
import websockets
from ocpp.routing import on
from ocpp.v201 import ChargePoint as Cp
from ocpp.v201 import call, call_result
from ocpp.v201.enums import (
    Action, RegistrationStatusType, AuthorizationStatusType,
    ConnectorStatusType
)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(name)s %(levelname)s %(message)s'
)
log = logging.getLogger('csms')

# Pool globale connessioni PostgreSQL
_db_pool: Optional[asyncpg.Pool] = None


async def get_db() -> asyncpg.Pool:
    global _db_pool
    if _db_pool is None:
        _db_pool = await asyncpg.create_pool(
            dsn='postgresql://csms:password@localhost/csms_db',
            min_size=5,
            max_size=20,
        )
    return _db_pool


class ChargePointHandler(Cp):
    """
    Handler OCPP 2.0.1 per una singola Charging Station.
    Un'istanza per ogni connessione WebSocket attiva.
    """

    @on(Action.boot_notification)
    async def on_boot_notification(
        self, charging_station: dict, reason: str, **kwargs
    ) -> call_result.BootNotification:
        log.info(
            f"Boot: {self.id} | "
            f"{charging_station['vendor_name']} {charging_station['model']} "
            f"| reason={reason}"
        )
        db = await get_db()
        await db.execute(
            """
            INSERT INTO charging_stations
                (station_id, vendor_name, model, serial_number,
                 firmware_version, last_boot_reason, last_seen_at, is_online)
            VALUES ($1, $2, $3, $4, $5, $6, NOW(), TRUE)
            ON CONFLICT (station_id) DO UPDATE SET
                vendor_name = EXCLUDED.vendor_name,
                model = EXCLUDED.model,
                firmware_version = EXCLUDED.firmware_version,
                last_boot_reason = EXCLUDED.last_boot_reason,
                last_seen_at = NOW(),
                is_online = TRUE
            """,
            self.id,
            charging_station['vendor_name'],
            charging_station['model'],
            charging_station.get('serial_number'),
            charging_station.get('firmware_version'),
            reason,
        )
        return call_result.BootNotification(
            current_time=datetime.now(timezone.utc).isoformat(),
            interval=300,
            status=RegistrationStatusType.accepted,
        )

    @on(Action.heartbeat)
    async def on_heartbeat(self) -> call_result.Heartbeat:
        db = await get_db()
        await db.execute(
            "UPDATE charging_stations SET last_seen_at = NOW() WHERE station_id = $1",
            self.id
        )
        return call_result.Heartbeat(
            current_time=datetime.now(timezone.utc).isoformat()
        )

    @on(Action.status_notification)
    async def on_status_notification(
        self, timestamp: str, connector_status: str,
        evse_id: int, connector_id: int, **kwargs
    ) -> call_result.StatusNotification:
        log.info(
            f"Status: {self.id} EVSE[{evse_id}]"
            f"Connector[{connector_id}] = {connector_status}"
        )
        db = await get_db()
        await db.execute(
            """
            INSERT INTO evse_status
                (station_id, evse_id, connector_id, status, updated_at)
            VALUES ($1, $2, $3, $4, NOW())
            ON CONFLICT (station_id, evse_id, connector_id) DO UPDATE SET
                status = EXCLUDED.status,
                updated_at = NOW()
            """,
            self.id, evse_id, connector_id, connector_status
        )
        return call_result.StatusNotification()

    @on(Action.authorize)
    async def on_authorize(
        self, id_token: dict, **kwargs
    ) -> call_result.Authorize:
        token = id_token['id_token']
        token_type = id_token['type']
        log.info(f"Authorize: {self.id} token={token} type={token_type}")

        db = await get_db()
        # Controlla prima la cache locale
        row = await db.fetchrow(
            """
            SELECT status, group_id, expiry_date
            FROM authorization_cache
            WHERE id_token = $1 AND id_token_type = $2
              AND (expiry_date IS NULL OR expiry_date > NOW())
            """,
            token, token_type
        )
        status = AuthorizationStatusType.invalid
        if row and row['status'] == 'Accepted':
            status = AuthorizationStatusType.accepted

        return call_result.Authorize(
            id_token_info={'status': status}
        )

    @on(Action.transaction_event)
    async def on_transaction_event(
        self,
        event_type: str,
        timestamp: str,
        trigger_reason: str,
        seq_no: int,
        transaction_info: dict,
        evse: Optional[dict] = None,
        id_token: Optional[dict] = None,
        meter_value: Optional[list] = None,
        **kwargs
    ) -> call_result.TransactionEvent:
        tx_id = transaction_info['transaction_id']
        log.info(
            f"TransactionEvent: {self.id} {event_type} "
            f"tx={tx_id} trigger={trigger_reason}"
        )
        db = await get_db()

        if event_type == 'Started':
            await self._handle_tx_started(
                db, tx_id, evse, id_token, timestamp, meter_value
            )
        elif event_type == 'Updated' and meter_value:
            await self._handle_tx_updated(db, tx_id, evse, meter_value)
        elif event_type == 'Ended':
            await self._handle_tx_ended(
                db, tx_id, timestamp, transaction_info, meter_value
            )

        return call_result.TransactionEvent(
            total_cost=0,
            charging_priority=0,
            id_token_info={'status': AuthorizationStatusType.accepted},
        )

    async def _handle_tx_started(
        self, db, tx_id, evse, id_token, timestamp, meter_value
    ):
        evse_id = evse['id'] if evse else 0
        connector_id = evse.get('connector_id') if evse else None
        token = id_token['id_token'] if id_token else 'unknown'
        token_type = id_token['type'] if id_token else 'Local'
        meter_start = self._extract_energy(meter_value)

        await db.execute(
            """
            INSERT INTO transactions
                (transaction_id, station_id, evse_id, connector_id,
                 id_token, id_token_type, state, started_at, meter_start_wh)
            VALUES ($1, $2, $3, $4, $5, $6, 'Started', $7, $8)
            ON CONFLICT (transaction_id) DO NOTHING
            """,
            tx_id, self.id, evse_id, connector_id,
            token, token_type, timestamp, meter_start
        )

    async def _handle_tx_updated(self, db, tx_id, evse, meter_value):
        evse_id = evse['id'] if evse else 0
        energy = self._extract_energy(meter_value)
        power = self._extract_power(meter_value)

        if energy is not None:
            await db.execute(
                """
                INSERT INTO meter_values
                    (transaction_id, station_id, evse_id, sampled_at, energy_wh, power_w)
                VALUES ($1, $2, $3, NOW(), $4, $5)
                """,
                tx_id, self.id, evse_id, energy, power
            )

    async def _handle_tx_ended(
        self, db, tx_id, timestamp, transaction_info, meter_value
    ):
        meter_end = self._extract_energy(meter_value)
        stop_reason = transaction_info.get('stopped_reason')

        await db.execute(
            """
            UPDATE transactions SET
                state = 'Ended',
                ended_at = $1,
                meter_end_wh = $2,
                stop_reason = $3
            WHERE transaction_id = $4
            """,
            timestamp, meter_end, stop_reason, tx_id
        )

    def _extract_energy(self, meter_values: Optional[list]) -> Optional[float]:
        if not meter_values:
            return None
        for mv in meter_values:
            for sv in mv.get('sampled_value', []):
                if sv.get('measurand', '') == 'Energy.Active.Import.Register':
                    return float(sv['value'])
        return None

    def _extract_power(self, meter_values: Optional[list]) -> Optional[float]:
        if not meter_values:
            return None
        for mv in meter_values:
            for sv in mv.get('sampled_value', []):
                if sv.get('measurand', '') == 'Power.Active.Import':
                    return float(sv['value'])
        return None

    # === Comandi CSMS -> Stazione ===

    async def send_remote_start(
        self, evse_id: int, id_token: str, limit_amps: float = 32.0
    ) -> str:
        """Avvia una sessione da remoto su EVSE specificato."""
        request = call.RequestStartTransaction(
            id_token={'id_token': id_token, 'type': 'Central'},
            evse_id=evse_id,
            charging_profile={
                'id': 999,
                'stack_level': 0,
                'charging_profile_purpose': 'TxProfile',
                'charging_profile_kind': 'Relative',
                'charging_schedule': [{
                    'id': 1,
                    'charging_rate_unit': 'A',
                    'charging_schedule_period': [
                        {'start_period': 0, 'limit': limit_amps}
                    ],
                }],
            },
        )
        response = await self.call(request)
        log.info(f"RemoteStart {self.id} EVSE{evse_id}: {response.status}")
        return response.status

    async def send_charging_profile(
        self, evse_id: int, profile: dict
    ) -> str:
        """Imposta un profilo di carica per smart charging."""
        request = call.SetChargingProfile(
            evse_id=evse_id,
            charging_profile=profile
        )
        response = await self.call(request)
        log.info(f"ChargingProfile {self.id} EVSE{evse_id}: {response.status}")
        return response.status


# Registry globale delle connessioni attive
_connected_stations: dict[str, ChargePointHandler] = {}


async def on_connect(websocket, path: str):
    """Callback per nuove connessioni WebSocket OCPP."""
    station_id = path.strip('/').split('/')[-1]

    if not station_id:
        await websocket.close(1008, 'Missing station ID')
        return

    log.info(f"Connessione da: {station_id} | path={path}")

    cp = ChargePointHandler(station_id, websocket)
    _connected_stations[station_id] = cp

    try:
        await cp.start()
    except websockets.exceptions.ConnectionClosed as e:
        log.info(f"Disconnesso: {station_id} code={e.code}")
    except Exception as e:
        log.error(f"Errore: {station_id} - {e}")
    finally:
        _connected_stations.pop(station_id, None)
        db = await get_db()
        await db.execute(
            "UPDATE charging_stations SET is_online = FALSE WHERE station_id = $1",
            station_id
        )


async def main():
    await get_db()  # Inizializza pool DB
    log.info("CSMS OCPP 2.0.1 avviato")

    server = await websockets.serve(
        on_connect,
        '0.0.0.0',
        9000,
        subprotocols=['ocpp2.0.1'],
        # TLS: aggiungere ssl=ssl_context per Security Profile 2-3
        ping_interval=60,
        ping_timeout=30,
        max_size=1_048_576,  # 1MB max message size
    )
    log.info("In ascolto su ws://0.0.0.0:9000/ocpp/{stationId}")
    await server.wait_closed()


if __name__ == '__main__':
    asyncio.run(main())

import json
import asyncio
import redis.asyncio as aioredis

redis_client = aioredis.from_url(
    'redis://redis-cluster:6379',
    encoding='utf-8',
    decode_responses=True
)

# Registra il nodo della connessione
async def register_connection(station_id: str, gateway_id: str):
    await redis_client.setex(
        f"csms:gateway:{station_id}",
        value=gateway_id,
        time=600  # TTL: 10 minuti, rinnovato a ogni heartbeat
    )

# Pubblica un comando verso una stazione (qualunque nodo sia)
async def publish_command(station_id: str, action: str, payload: dict):
    channel = f"csms:commands:{station_id}"
    await redis_client.publish(channel, json.dumps({
        'action': action,
        'payload': payload
    }))

# Su ogni nodo gateway: ascolta i comandi per le stazioni connesse
async def listen_for_commands(connected_stations: dict):
    pubsub = redis_client.pubsub()

    # Sottoscrivi ai canali delle stazioni connesse a questo nodo
    async def subscribe_station(station_id: str):
        await pubsub.subscribe(f"csms:commands:{station_id}")

    async for message in pubsub.listen():
        if message['type'] != 'message':
            continue

        station_id = message['channel'].split(':')[-1]
        cp = connected_stations.get(station_id)
        if not cp:
            continue  # Stazione non su questo nodo, ignora

        cmd = json.loads(message['data'])
        try:
            if cmd['action'] == 'SetChargingProfile':
                await cp.send_charging_profile(
                    cmd['payload']['evse_id'],
                    cmd['payload']['profile']
                )
            elif cmd['action'] == 'RemoteStart':
                await cp.send_remote_start(
                    cmd['payload']['evse_id'],
                    cmd['payload']['id_token'],
                )
        except Exception as e:
            log.error(f"Errore esecuzione comando {cmd['action']}: {e}")

from prometheus_client import (
    Counter, Gauge, Histogram, start_http_server
)

# Metriche Prometheus
OCPP_MESSAGES_TOTAL = Counter(
    'ocpp_messages_total',
    'Numero totale messaggi OCPP processati',
    ['action', 'direction', 'status']  # direction: inbound/outbound
)

OCPP_MESSAGE_DURATION = Histogram(
    'ocpp_message_duration_seconds',
    'Latenza elaborazione messaggi OCPP',
    ['action'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

STATIONS_CONNECTED = Gauge(
    'csms_stations_connected_total',
    'Numero stazioni connesse al CSMS'
)

ACTIVE_TRANSACTIONS = Gauge(
    'csms_active_transactions_total',
    'Numero transazioni di ricarica attive'
)

ENERGY_DELIVERED_WH = Counter(
    'csms_energy_delivered_wh_total',
    'Energia totale erogata in Wh',
    ['station_id']
)

AUTH_RESULTS = Counter(
    'csms_authorization_results_total',
    'Risultati delle autorizzazioni OCPP',
    ['status']  # Accepted, Invalid, Blocked, Expired
)

SMART_CHARGING_EVENTS = Counter(
    'csms_smart_charging_events_total',
    'Operazioni smart charging',
    ['action', 'result']
)


def start_metrics_server(port: int = 8001):
    """Avvia il server HTTP Prometheus su porta specificata."""
    start_http_server(port)
    log.info(f"Prometheus metrics su http://0.0.0.0:{port}/metrics")


# Decorator per misurare latenza handler
import time
import functools

def track_ocpp_handler(action: str):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.monotonic()
            try:
                result = await func(*args, **kwargs)
                OCPP_MESSAGES_TOTAL.labels(
                    action=action, direction='inbound', status='success'
                ).inc()
                return result
            except Exception as e:
                OCPP_MESSAGES_TOTAL.labels(
                    action=action, direction='inbound', status='error'
                ).inc()
                raise
            finally:
                OCPP_MESSAGE_DURATION.labels(action=action).observe(
                    time.monotonic() - start
                )
        return wrapper
    return decorator

// Pannelli principali per dashboard Grafana CSMS

// 1. Stazioni online (Gauge)
{
  "title": "Stazioni Connesse",
  "type": "gauge",
  "fieldConfig": {
    "defaults": {
      "thresholds": {
        "steps": [
          { "color": "red", "value": 0 },
          { "color": "yellow", "value": 90 },
          { "color": "green", "value": 99 }
        ]
      }
    }
  },
  "targets": [{
    "expr": "csms_stations_connected_total / csms_stations_registered_total * 100",
    "legendFormat": "Availability %"
  }]
}

// 2. Latenza messaggi OCPP P99 (Time Series)
{
  "title": "OCPP Message Latency P99",
  "type": "timeseries",
  "targets": [{
    "expr": "histogram_quantile(0.99, rate(ocpp_message_duration_seconds_bucket[5m]))",
    "legendFormat": "P99 - {{action}}"
  }]
}

// 3. Energia erogata (Stat panel)
{
  "title": "Energia Totale Erogata oggi (kWh)",
  "type": "stat",
  "targets": [{
    "expr": "increase(csms_energy_delivered_wh_total[24h]) / 1000",
    "legendFormat": "kWh"
  }]
}

// 4. Auth rejection rate - alert su attacchi
{
  "title": "Rejection Rate Autorizzazioni (%)",
  "type": "timeseries",
  "targets": [{
    "expr": "rate(csms_authorization_results_total{status='Invalid'}[5m]) / rate(csms_authorization_results_total[5m]) * 100"
  }]
}

import ssl
import re
from functools import wraps

# 1. Configurazione TLS sicura (Security Profile 2-3)
def create_tls_context(
    certfile: str,
    keyfile: str,
    cafile: str,
    require_client_cert: bool = False
) -> ssl.SSLContext:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    ctx.load_cert_chain(certfile=certfile, keyfile=keyfile)

    if require_client_cert:  # Security Profile 3 (mTLS)
        ctx.verify_mode = ssl.CERT_REQUIRED
        ctx.load_verify_locations(cafile=cafile)

    # Disabilita cipher suite deboli
    ctx.set_ciphers(
        'ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:!aNULL:!eNULL:!LOW:!EXPORT'
    )
    return ctx


# 2. Rate limiting per connessioni WebSocket
from collections import defaultdict
import time

_connection_attempts: dict[str, list[float]] = defaultdict(list)
MAX_CONN_PER_MINUTE = 10


def check_rate_limit(client_ip: str) -> bool:
    """Ritorna True se il client può connettersi, False se throttled."""
    now = time.monotonic()
    window = _connection_attempts[client_ip]

    # Rimuovi tentativi più vecchi di 60 secondi
    _connection_attempts[client_ip] = [
        t for t in window if now - t < 60
    ]

    if len(_connection_attempts[client_ip]) >= MAX_CONN_PER_MINUTE:
        log.warning(f"Rate limit superato per IP: {client_ip}")
        return False

    _connection_attempts[client_ip].append(now)
    return True


# 3. Validazione MessageId per prevenire replay attack
_seen_message_ids: set[str] = set()
_message_id_pattern = re.compile(r'^[a-zA-Z0-9\-_\.]{1,36}$')


def validate_message_id(msg_id: str) -> bool:
    if not _message_id_pattern.match(msg_id):
        log.warning(f"MessageId formato invalido: {msg_id}")
        return False
    if msg_id in _seen_message_ids:
        log.warning(f"Possibile replay attack: MessageId duplicato {msg_id}")
        return False
    _seen_message_ids.add(msg_id)
    # In produzione: usa Redis con TTL per pulizia automatica
    return True


# 4. Validazione e sanitizzazione payload OCPP
def validate_station_id(station_id: str) -> bool:
    """Valida che station_id sia sicuro (prevenire path traversal)."""
    pattern = re.compile(r'^[a-zA-Z0-9\-_\.]{1,50}$')
    return bool(pattern.match(station_id))


def validate_id_token(id_token: str) -> bool:
    """Valida formato token RFID/app (max 255 chars, alfanumerico)."""
    if not id_token or len(id_token) > 255:
        return False
    return bool(re.match(r'^[a-zA-Z0-9\-_\.\:]{1,255}$', id_token))


# 5. Audit log di sicurezza (OCPP Security Block A)
async def log_security_event(
    station_id: str,
    event_type: str,
    severity: str,
    details: str
):
    db = await get_db()
    await db.execute(
        """
        INSERT INTO security_events
            (station_id, event_type, severity, details, logged_at)
        VALUES ($1, $2, $3, $4, NOW())
        """,
        station_id, event_type, severity, details
    )
    if severity == 'Critical':
        log.critical(f"SECURITY: {station_id} | {event_type} | {details}")