Architettura DERMS: Aggregare Milioni di Risorse Distribuite

# Architettura DERMS - Vista ad alto livello

+================================================================+
|                    LAYER 4: MARKET                              |
|   Mercati Energia: MGP, MSD, MO, Capacity Market               |
|   DSO Flexibility Markets, Aggregatori terzi                   |
|   Revenue stacking, Portfolio optimization                     |
+================================================================+
          |                              |
    Bid/Offer API              Settlement data
          |                              |
+================================================================+
|                    LAYER 3: PLATFORM (Cloud DERMS)             |
|                                                                |
|  +------------------+    +------------------+                  |
|  | Forecasting      |    | Dispatch Engine  |                  |
|  | (ML: FV, load,   |    | (Optimization:   |                  |
|  |  EV availability)|    |  LP/MILP/MPC)    |                  |
|  +------------------+    +------------------+                  |
|                                                                |
|  +------------------+    +------------------+                  |
|  | Aggregation      |    | Market Interface |                  |
|  | Service (VPP     |    | (Bid builder,    |                  |
|  |  portfolio mgmt) |    |  settlement)     |                  |
|  +------------------+    +------------------+                  |
|                                                                |
|  +------------------+    +------------------+                  |
|  | Event Bus        |    | Time-Series DB   |                  |
|  | (Apache Kafka)   |    | (InfluxDB/        |                  |
|  |                  |    |  TimescaleDB)     |                  |
|  +------------------+    +------------------+                  |
|                                                                |
|  +------------------+    +------------------+                  |
|  | Device Registry  |    | API Gateway      |                  |
|  | (DER catalog,    |    | (REST, WebSocket,|                  |
|  |  metadata, caps) |    |  gRPC)           |                  |
|  +------------------+    +------------------+                  |
+================================================================+
          |                              |
   Commands (dispatch)         Telemetry (status)
          |                              |
+================================================================+
|                    LAYER 2: EDGE                               |
|                                                                |
|  +------------------+    +------------------+                  |
|  | Site Aggregator  |    | Protocol Gateway |                  |
|  | (building/plant  |    | (Modbus->MQTT,   |                  |
|  |  controller)     |    |  SunSpec->JSON)  |                  |
|  +------------------+    +------------------+                  |
|                                                                |
|  Local optimization, failsafe, buffering, compression          |
+================================================================+
          |                              |
  Device protocols (Modbus, SunSpec, OCPP, BACnet, OPC-UA)
          |                              |
+================================================================+
|                    LAYER 1: FIELD                              |
|   Inverter FV - BESS BMS - EV Charger - Smart Meter           |
|   Industrial Loads - CHP Controller - Heat Pump               |
+================================================================+

# models.py - Modello dati per le risorse distribuite

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import datetime

class DERType(Enum):
    SOLAR_PV = "solar_pv"
    BATTERY_STORAGE = "battery_storage"
    EV_CHARGER = "ev_charger"
    DEMAND_RESPONSE = "demand_response"
    CHP = "chp"
    HEAT_PUMP = "heat_pump"

class DERStatus(Enum):
    ONLINE = "online"
    OFFLINE = "offline"
    DISPATCHING = "dispatching"
    FAULT = "fault"
    STANDBY = "standby"

@dataclass
class DERCapabilities:
    """capacità fisiche di una DER"""
    max_power_kw: float          # Potenza massima erogabile (kW)
    min_power_kw: float          # Potenza minima (0 per curtailment FV)
    ramp_up_kw_per_sec: float    # Velocita rampa salita (kW/s)
    ramp_down_kw_per_sec: float  # Velocita rampa discesa (kW/s)
    # Solo per storage (BESS/EV)
    capacity_kwh: Optional[float] = None
    min_soc_pct: Optional[float] = None   # SoC minimo (es. 10%)
    max_soc_pct: Optional[float] = None   # SoC massimo (es. 95%)
    roundtrip_efficiency: Optional[float] = None  # Efficienza ciclo (es. 0.92)

@dataclass
class DERTelemetry:
    """Stato in tempo reale di una DER"""
    der_id: str
    timestamp: datetime.datetime
    active_power_kw: float         # Potenza attuale (positiva = generazione)
    reactive_power_kvar: float
    voltage_v: float
    current_a: float
    status: DERStatus
    # Solo per storage
    soc_pct: Optional[float] = None
    available_charge_kw: Optional[float] = None
    available_discharge_kw: Optional[float] = None
    # Solo per FV
    irradiance_wm2: Optional[float] = None
    temperature_c: Optional[float] = None

@dataclass
class DERAsset:
    """Registro completo di una DER nel portfolio"""
    id: str
    type: DERType
    name: str
    site_id: str                          # Sito fisico di appartenenza
    grid_node_id: str                     # Nodo di rete (per vincoli topologici)
    capabilities: DERCapabilities
    protocol: str                         # "sunspec", "openadr", "ocpp", "modbus"
    endpoint: str                         # URL/IP del dispositivo o del gateway
    owner_id: str                         # Proprietario (utente o azienda)
    aggregation_vpp_ids: list = field(default_factory=list)  # VPP di appartenenza
    # Telemetria più recente (aggiornata dal telemetry service)
    last_telemetry: Optional[DERTelemetry] = None

# aggregation_service.py - Servizio di aggregazione VPP

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict, List, Optional
import asyncio
import logging
from datetime import datetime, timezone

# Import interni
from models import DERAsset, DERTelemetry, DERType, DERStatus
from dispatch_optimizer import DispatchOptimizer
from telemetry_store import TelemetryStore

logger = logging.getLogger(__name__)
app = FastAPI(title="DERMS Aggregation Service", version="2.1.0")

# --- Pydantic schemas per la API ---

class VPPPortfolioResponse(BaseModel):
    vpp_id: str
    der_count: int
    total_capacity_kw: float
    available_capacity_kw: float
    current_dispatch_kw: float
    battery_soc_avg_pct: Optional[float]
    timestamp: str

class DispatchRequest(BaseModel):
    vpp_id: str
    target_power_kw: float       # Potenza target per la VPP (positiva = generazione)
    duration_minutes: int        # Durata del dispatch
    priority: str = "normal"     # "emergency" | "normal" | "economic"
    max_deviation_pct: float = 5.0  # Tolleranza deviazione dal target

class DispatchSetpoint(BaseModel):
    der_id: str
    power_kw: float
    duration_minutes: int
    timestamp: str

class DispatchResponse(BaseModel):
    dispatch_id: str
    vpp_id: str
    target_power_kw: float
    achieved_power_kw: float
    setpoints: List[DispatchSetpoint]
    feasibility_score: float     # 0.0 - 1.0 (1.0 = target pienamente raggiunto)
    timestamp: str

# --- Registro in-memory (in produzione: database PostgreSQL + cache Redis) ---
_vpp_registry: Dict[str, List[str]] = {}     # vpp_id -> [der_id]
_der_registry: Dict[str, DERAsset] = {}      # der_id -> DERAsset
_telemetry_store = TelemetryStore()
_optimizer = DispatchOptimizer()

# --- Endpoints ---

@app.get("/health")
async def health_check():
    return {"status": "ok", "version": "2.1.0", "timestamp": datetime.now(timezone.utc).isoformat()}

@app.get("/api/v1/vpp/{vpp_id}/portfolio", response_model=VPPPortfolioResponse)
async def get_vpp_portfolio(vpp_id: str):
    """
    Ritorna la snapshot aggregata dello stato corrente di una VPP.
    Consolida telemetria di tutti i DER nel portfolio.
    """
    if vpp_id not in _vpp_registry:
        raise HTTPException(status_code=404, detail=f"VPP '{vpp_id}' non trovata")

    der_ids = _vpp_registry[vpp_id]
    der_assets = [_der_registry[did] for did in der_ids if did in _der_registry]

    if not der_assets:
        raise HTTPException(status_code=503, detail="Nessun DER disponibile nel portfolio")

    # Aggregazione metriche
    total_capacity = sum(a.capabilities.max_power_kw for a in der_assets)
    current_dispatch = 0.0
    available_capacity = 0.0
    storage_assets = []

    for asset in der_assets:
        telemetry = _telemetry_store.get_latest(asset.id)
        if telemetry and telemetry.status in [DERStatus.ONLINE, DERStatus.DISPATCHING]:
            current_dispatch += telemetry.active_power_kw
            # capacità disponibile = differenza tra massimo e corrente
            if telemetry.available_discharge_kw is not None:
                available_capacity += telemetry.available_discharge_kw
            else:
                available_capacity += (asset.capabilities.max_power_kw - telemetry.active_power_kw)
            # Raccogli SoC per storage
            if telemetry.soc_pct is not None:
                storage_assets.append(telemetry.soc_pct)

    battery_soc_avg = (sum(storage_assets) / len(storage_assets)) if storage_assets else None

    return VPPPortfolioResponse(
        vpp_id=vpp_id,
        der_count=len(der_assets),
        total_capacity_kw=round(total_capacity, 2),
        available_capacity_kw=round(max(0, available_capacity), 2),
        current_dispatch_kw=round(current_dispatch, 2),
        battery_soc_avg_pct=round(battery_soc_avg, 1) if battery_soc_avg else None,
        timestamp=datetime.now(timezone.utc).isoformat()
    )

@app.post("/api/v1/vpp/dispatch", response_model=DispatchResponse)
async def dispatch_vpp(request: DispatchRequest, background_tasks: BackgroundTasks):
    """
    Esegue un dispatch ottimizzato della VPP verso il target di potenza richiesto.
    Usa Linear Programming per distribuire il carico tra i DER disponibili.
    """
    if request.vpp_id not in _vpp_registry:
        raise HTTPException(status_code=404, detail=f"VPP '{request.vpp_id}' non trovata")

    # Recupero DER disponibili e telemetria aggiornata
    der_ids = _vpp_registry[request.vpp_id]
    available_ders = []
    for der_id in der_ids:
        asset = _der_registry.get(der_id)
        telemetry = _telemetry_store.get_latest(der_id)
        if asset and telemetry and telemetry.status in [DERStatus.ONLINE, DERStatus.STANDBY]:
            available_ders.append((asset, telemetry))

    if not available_ders:
        raise HTTPException(status_code=503, detail="Nessun DER disponibile per il dispatch")

    # Ottimizzazione del dispatch tramite LP
    result = _optimizer.optimize_dispatch(
        ders=available_ders,
        target_power_kw=request.target_power_kw,
        duration_minutes=request.duration_minutes
    )

    # Invio setpoint in background (async, non bloccante)
    background_tasks.add_task(
        _send_setpoints_to_ders,
        setpoints=result.setpoints,
        dispatch_id=result.dispatch_id
    )

    return result

async def _send_setpoints_to_ders(setpoints: List[DispatchSetpoint], dispatch_id: str):
    """Invia i setpoint a ogni DER in parallelo tramite il rispettivo adapter."""
    logger.info(f"Inizio dispatch {dispatch_id} - {len(setpoints)} setpoint")
    tasks = [_send_single_setpoint(sp) for sp in setpoints]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    errors = [r for r in results if isinstance(r, Exception)]
    if errors:
        logger.warning(f"Dispatch {dispatch_id}: {len(errors)} errori su {len(setpoints)} setpoint")
    logger.info(f"Dispatch {dispatch_id} completato")

async def _send_single_setpoint(setpoint: DispatchSetpoint):
    """Stub: in produzione chiama l'adapter specifico del protocollo del DER."""
    asset = _der_registry.get(setpoint.der_id)
    if not asset:
        raise ValueError(f"DER {setpoint.der_id} non trovato nel registro")
    logger.debug(f"Setpoint {setpoint.der_id}: {setpoint.power_kw} kW per {setpoint.duration_minutes} min")
    # In produzione: chiamata all'adapter (SunSpec/OpenADR/OCPP)
    await asyncio.sleep(0.1)  # Simulazione latenza rete

# dispatch_optimizer.py - Ottimizzazione LP per dispatch DER

import pulp
import uuid
import logging
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import List, Tuple

from models import DERAsset, DERTelemetry, DERType, DERStatus

logger = logging.getLogger(__name__)

@dataclass
class OptimizationResult:
    dispatch_id: str
    vpp_id: str
    target_power_kw: float
    achieved_power_kw: float
    setpoints: list
    feasibility_score: float
    timestamp: str
    solver_status: str
    solve_time_ms: float

class DispatchOptimizer:
    """
    Ottimizzatore LP per dispatch di portfolio DER.

    Problema: data una richiesta di potenza target P_target,
    trovare i setpoint p_i per ogni DER i nel portfolio
    che minimizzino |sum(p_i) - P_target|, rispettando:
      - p_i_min <= p_i <= p_i_max per ogni DER
      - Vincoli SoC per storage (BESS/EV)
      - Vincoli ramp rate
      - Vincoli di topologia di rete (opzionale)
    """

    def optimize_dispatch(
        self,
        ders: List[Tuple[DERAsset, DERTelemetry]],
        target_power_kw: float,
        duration_minutes: int,
        vpp_id: str = "vpp-001"
    ) -> OptimizationResult:

        import time
        start_time = time.time()
        dispatch_id = f"disp-{uuid.uuid4().hex[:8]}"

        # Costruzione del problema LP con PuLP
        prob = pulp.LpProblem(
            name=f"DER_Dispatch_{dispatch_id}",
            sense=pulp.LpMinimize
        )

        # Variabili decisionali: setpoint per ogni DER (kW)
        # Vincolate tra min e max fisico del dispositivo
        p_vars = {}
        for asset, telemetry in ders:
            # Calcola limiti effettivi considerando SoC per storage
            p_min, p_max = self._compute_effective_limits(asset, telemetry, duration_minutes)
            var_name = f"p_{asset.id.replace('-', '_')}"
            p_vars[asset.id] = pulp.LpVariable(
                name=var_name,
                lowBound=p_min,
                upBound=p_max,
                cat=pulp.constants.LpContinuous
            )

        # Variabile di slack per la deviazione dal target (non-negativa)
        slack_pos = pulp.LpVariable("slack_pos", lowBound=0)  # Eccesso rispetto target
        slack_neg = pulp.LpVariable("slack_neg", lowBound=0)  # Deficit rispetto target

        # Funzione obiettivo: minimizzare la deviazione assoluta dal target
        # Pesi differenziati: penalizza di più il deficit (mancata fornitura)
        prob += 1.5 * slack_neg + 1.0 * slack_pos, "MinimizeDeviation"

        # Vincolo di bilanciamento della potenza
        total_power = pulp.lpSum(p_vars[asset.id] for asset, _ in ders)
        prob += (total_power - target_power_kw == slack_pos - slack_neg), "PowerBalance"

        # Risoluzione (COIN-BC solver, open source)
        solver = pulp.COIN_CMD(msg=False, timeLimit=5.0)
        status = prob.solve(solver)

        solve_time_ms = (time.time() - start_time) * 1000
        solver_status = pulp.LpStatus[prob.status]

        # Costruzione setpoint dal risultato
        setpoints = []
        achieved_power = 0.0

        if status in [pulp.LpStatusOptimal := 1, -1]:  # Optimal or Infeasible
            for asset, telemetry in ders:
                if asset.id in p_vars:
                    p_val = pulp.value(p_vars[asset.id]) or 0.0
                    achieved_power += p_val
                    if abs(p_val) > 0.1:  # Ignora setpoint trascurabili
                        setpoints.append({
                            "der_id": asset.id,
                            "power_kw": round(p_val, 2),
                            "duration_minutes": duration_minutes,
                            "timestamp": datetime.now(timezone.utc).isoformat()
                        })

        # Calcolo feasibility score: 1.0 se target raggiunto, < 1.0 se parziale
        if abs(target_power_kw) > 0.1:
            deviation = abs(achieved_power - target_power_kw) / abs(target_power_kw)
            feasibility_score = max(0.0, 1.0 - deviation)
        else:
            feasibility_score = 1.0

        logger.info(
            f"Dispatch {dispatch_id}: target={target_power_kw}kW, "
            f"achieved={achieved_power:.1f}kW, score={feasibility_score:.3f}, "
            f"solver={solver_status}, time={solve_time_ms:.0f}ms"
        )

        return OptimizationResult(
            dispatch_id=dispatch_id,
            vpp_id=vpp_id,
            target_power_kw=target_power_kw,
            achieved_power_kw=round(achieved_power, 2),
            setpoints=setpoints,
            feasibility_score=round(feasibility_score, 4),
            timestamp=datetime.now(timezone.utc).isoformat(),
            solver_status=solver_status,
            solve_time_ms=round(solve_time_ms, 1)
        )

    def _compute_effective_limits(
        self,
        asset: DERAsset,
        telemetry: DERTelemetry,
        duration_minutes: int
    ) -> Tuple[float, float]:
        """
        Calcola i limiti effettivi di potenza per un DER considerando:
        - Limiti fisici dichiarati
        - SoC corrente per storage (quanta energia residua disponibile)
        - Temperatura (semplificato)
        """
        p_min = asset.capabilities.min_power_kw
        p_max = asset.capabilities.max_power_kw

        # Vincoli aggiuntivi per storage (BESS o EV)
        if asset.capabilities.capacity_kwh and telemetry.soc_pct is not None:
            cap_kwh = asset.capabilities.capacity_kwh
            soc = telemetry.soc_pct / 100.0
            min_soc = (asset.capabilities.min_soc_pct or 10.0) / 100.0
            max_soc = (asset.capabilities.max_soc_pct or 95.0) / 100.0
            efficiency = asset.capabilities.roundtrip_efficiency or 0.92
            duration_h = duration_minutes / 60.0

            # Energia disponibile in scarica (discharge -> positivo)
            energy_available_discharge_kwh = (soc - min_soc) * cap_kwh * efficiency
            p_max_soc = energy_available_discharge_kwh / duration_h if duration_h > 0 else 0
            p_max = min(p_max, p_max_soc)

            # Energia disponibile in carica (charge -> negativo = consumo)
            energy_available_charge_kwh = (max_soc - soc) * cap_kwh / efficiency
            p_min_soc = -(energy_available_charge_kwh / duration_h) if duration_h > 0 else 0
            p_min = max(p_min, p_min_soc)

        return p_min, p_max

# openadr_adapter.py - Client OpenADR 2.0b (VEN - Virtual End Node)
# Implementazione semplificata per illustrare il flusso di comunicazione

import httpx
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import Optional
import asyncio
import logging

logger = logging.getLogger(__name__)

@dataclass
class DREvent:
    """Evento di Demand Response ricevuto dal VTN (utility/DERMS)"""
    event_id: str
    program_id: str
    signal_name: str           # "SIMPLE" | "ELECTRICITY_PRICE" | "LOAD_DISPATCH"
    signal_type: str           # "LEVEL" | "PRICE" | "X-LOAD_DISPATCH"
    signal_value: float        # Valore del segnale (es. livello 1/2/3 o prezzo EUR/MWh)
    dtstart: str               # ISO 8601 - inizio evento
    duration_minutes: int
    randomize_start_minutes: int = 0  # Randomizzazione per evitare picchi sincroni

class OpenADRVENClient:
    """
    Virtual End Node (VEN): rappresenta un sito/aggregatore che riceve
    eventi DR dal Virtual Top Node (VTN) del DERMS o della utility.

    Flusso tipico OpenADR 2.0b in modalità PULL:
    1. VEN -> VTN: oadrRequestEvent (richiede eventi disponibili)
    2. VTN -> VEN: oadrDistributeEvent (lista eventi attivi)
    3. VEN -> VTN: oadrCreatedEvent (conferma ricezione con optIn/optOut)
    4. VEN -> VTN: oadrUpdateReport (report su energia effettivamente modificata)
    """

    def __init__(self, vtn_url: str, ven_id: str, ven_name: str):
        self.vtn_url = vtn_url.rstrip("/")
        self.ven_id = ven_id
        self.ven_name = ven_name
        self._client = httpx.AsyncClient(timeout=30.0)
        self._registered = False
        self._active_events: dict = {}

    async def register(self) -> bool:
        """Registrazione del VEN sul VTN - obbligatoria prima di ricevere eventi."""
        payload = self._build_register_payload()
        try:
            response = await self._client.post(
                f"{self.vtn_url}/OpenADR2/Simple/2.0b/EiRegisterParty",
                content=payload,
                headers={"Content-Type": "application/xml"}
            )
            response.raise_for_status()
            root = ET.fromstring(response.text)
            # Parsing della risposta oadrCreatedParty
            registration_id = self._extract_registration_id(root)
            if registration_id:
                self._registered = True
                logger.info(f"VEN {self.ven_id} registrato con ID: {registration_id}")
                return True
        except Exception as e:
            logger.error(f"Errore registrazione VEN: {e}")
        return False

    async def poll_events(self) -> list:
        """Polling degli eventi DR disponibili sul VTN."""
        if not self._registered:
            raise RuntimeError("VEN non registrato. Chiamare register() prima.")

        payload = self._build_request_event_payload()
        try:
            response = await self._client.post(
                f"{self.vtn_url}/OpenADR2/Simple/2.0b/EiEvent",
                content=payload,
                headers={"Content-Type": "application/xml"}
            )
            response.raise_for_status()
            root = ET.fromstring(response.text)
            events = self._parse_distribute_event(root)
            logger.info(f"VEN {self.ven_id}: ricevuti {len(events)} eventi DR")
            return events
        except Exception as e:
            logger.error(f"Errore polling eventi: {e}")
            return []

    async def confirm_event(self, event_id: str, opt_in: bool = True) -> bool:
        """Conferma ricezione evento e comunicazione optIn/optOut."""
        status = "optIn" if opt_in else "optOut"
        payload = self._build_created_event_payload(event_id, status)
        try:
            response = await self._client.post(
                f"{self.vtn_url}/OpenADR2/Simple/2.0b/EiEvent",
                content=payload,
                headers={"Content-Type": "application/xml"}
            )
            response.raise_for_status()
            logger.info(f"Evento {event_id}: {status} confermato")
            return True
        except Exception as e:
            logger.error(f"Errore conferma evento {event_id}: {e}")
            return False

    async def run_polling_loop(self, poll_interval_seconds: int = 60):
        """Loop di polling continuo - eseguito come task asyncio."""
        logger.info(f"Avvio polling loop VEN {self.ven_id} ogni {poll_interval_seconds}s")
        while True:
            events = await self.poll_events()
            for event in events:
                if event.event_id not in self._active_events:
                    self._active_events[event.event_id] = event
                    # OptIn automatico (in produzione: logica di accettazione business)
                    await self.confirm_event(event.event_id, opt_in=True)
                    logger.info(
                        f"Nuovo evento DR: {event.event_id} | "
                        f"Segnale: {event.signal_name}={event.signal_value} | "
                        f"Durata: {event.duration_minutes} min"
                    )
            await asyncio.sleep(poll_interval_seconds)

    def _build_register_payload(self) -> str:
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<oadrPayload>
  <oadrSignedObject>
    <oadrRegisterReport specificationID="TELEMETRY_STATUS">
      <ei:venID>{self.ven_id}</ei:venID>
    </oadrRegisterReport>
  </oadrSignedObject>
</oadrPayload>"""

    def _build_request_event_payload(self) -> str:
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<oadrPayload>
  <oadrSignedObject>
    <oadrRequestEvent>
      <ei:eiRequestEvent>
        <ei:venID>{self.ven_id}</ei:venID>
        <ei:replyLimit>10</ei:replyLimit>
      </ei:eiRequestEvent>
    </oadrRequestEvent>
  </oadrSignedObject>
</oadrPayload>"""

    def _build_created_event_payload(self, event_id: str, status: str) -> str:
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<oadrPayload>
  <oadrSignedObject>
    <oadrCreatedEvent>
      <ei:eiCreatedEvent>
        <ei:venID>{self.ven_id}</ei:venID>
        <ei:eventResponses>
          <ei:eventResponse>
            <ei:responseCode>200</ei:responseCode>
            <ei:requestID>{event_id}</ei:requestID>
            <ei:qualifiedEventID>
              <ei:eventID>{event_id}</ei:eventID>
              <ei:modificationNumber>0</ei:modificationNumber>
            </ei:qualifiedEventID>
            <ei:optType>{status}</ei:optType>
          </ei:eventResponse>
        </ei:eventResponses>
      </ei:eiCreatedEvent>
    </oadrCreatedEvent>
  </oadrSignedObject>
</oadrPayload>"""

    def _parse_distribute_event(self, root: ET.Element) -> list:
        """Parser semplificato per oadrDistributeEvent."""
        events = []
        # In produzione: parsing completo con namespace XML OpenADR
        # Qui simuliamo la struttura per chiarezza
        return events

    def _extract_registration_id(self, root: ET.Element) -> Optional[str]:
        """Estrae l'ID di registrazione dalla risposta VTN."""
        return "reg-001"  # Semplificato

# kafka_derms_config.py - Configurazione Kafka per DERMS scalabile

from confluent_kafka import Producer, Consumer, KafkaError
from confluent_kafka.admin import AdminClient, NewTopic
import json
import logging
from dataclasses import asdict
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

# === TOPIC DESIGN ===
# Strategia: topic separati per tipo di dato, partitionati per DER ID
# La key del messaggio = der_id garantisce che tutti i messaggi dello stesso
# DER vadano alla stessa partizione (ordering garantito per dispositivo)

KAFKA_TOPICS = {
    # Telemetria (alta frequenza, alta velocità)
    "der.telemetry.raw": {
        "partitions": 48,      # 48 partizioni per parallelismo elevato
        "replication_factor": 3,
        "retention_ms": 86400000,   # 24 ore (poi su time-series DB)
        "compression_type": "lz4",  # LZ4 per compressione veloce
        "config": {"cleanup.policy": "delete"}
    },
    # Comandi di dispatch (bassa frequenza, alta affidabilità)
    "der.dispatch.commands": {
        "partitions": 12,
        "replication_factor": 3,
        "retention_ms": 604800000,  # 7 giorni
        "compression_type": "gzip",
        "config": {"cleanup.policy": "delete", "min.insync.replicas": "2"}
    },
    # Conferme dispatch (ack dai dispositivi)
    "der.dispatch.acks": {
        "partitions": 12,
        "replication_factor": 3,
        "retention_ms": 604800000,
        "config": {"cleanup.policy": "delete"}
    },
    # Aggregati VPP (output dell'aggregation service)
    "vpp.portfolio.snapshots": {
        "partitions": 4,
        "replication_factor": 3,
        "retention_ms": 2592000000, # 30 giorni
        "config": {"cleanup.policy": "compact"}  # Log compaction: mantieni ultima snapshot
    },
    # Alert e fault detection
    "der.alerts": {
        "partitions": 6,
        "replication_factor": 3,
        "retention_ms": 2592000000,
        "config": {"cleanup.policy": "delete"}
    }
}

class DERMSTelemetryProducer:
    """
    Producer Kafka per la telemetria DER.
    Usato dai Gateway Edge per pubblicare telemetria verso il cloud.
    """

    def __init__(self, bootstrap_servers: str):
        self._producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "client.id": "derms-telemetry-producer",
            # Affidabilità: ACK da tutti i broker in-sync
            "acks": "all",
            # Performance: batching aggressivo per throughput
            "linger.ms": 20,
            "batch.size": 65536,
            "compression.type": "lz4",
            # Retry per fault tolerance
            "retries": 5,
            "retry.backoff.ms": 200,
            "enable.idempotence": True  # Exactly-once semantics
        })

    def publish_telemetry(self, der_id: str, telemetry: dict) -> None:
        """
        Pubblica telemetria su Kafka.
        La key = der_id garantisce ordering per dispositivo.
        """
        payload = json.dumps({
            **telemetry,
            "_published_at": datetime.now(timezone.utc).isoformat()
        }).encode("utf-8")

        self._producer.produce(
            topic="der.telemetry.raw",
            key=der_id.encode("utf-8"),
            value=payload,
            on_delivery=self._delivery_callback
        )
        self._producer.poll(0)  # Non-blocking flush

    def flush(self, timeout: float = 10.0) -> None:
        """Flush dei messaggi in coda prima dello shutdown."""
        pending = self._producer.flush(timeout=timeout)
        if pending > 0:
            logger.warning(f"{pending} messaggi non ancora consegnati dopo flush")

    def _delivery_callback(self, err, msg):
        if err:
            logger.error(f"Errore consegna messaggio: {err}")
        else:
            logger.debug(f"Messaggio consegnato: topic={msg.topic()}, partition={msg.partition()}")


class DERMSDispatchConsumer:
    """
    Consumer Kafka per i comandi di dispatch.
    Ogni Site Aggregator consuma dal topic dispatch.commands
    i setpoint relativi ai propri DER.
    """

    def __init__(self, bootstrap_servers: str, group_id: str, site_id: str):
        self.site_id = site_id
        self._consumer = Consumer({
            "bootstrap.servers": bootstrap_servers,
            "group.id": group_id,
            "client.id": f"site-aggregator-{site_id}",
            "auto.offset.reset": "latest",   # Solo messaggi recenti (no backlog storico)
            "enable.auto.commit": False,      # Commit manuale dopo elaborazione
            "max.poll.interval.ms": 30000,
            "session.timeout.ms": 10000
        })
        self._consumer.subscribe(["der.dispatch.commands"])

    def process_commands(self, timeout_seconds: float = 1.0):
        """Poll e processa comandi di dispatch per questo sito."""
        msg = self._consumer.poll(timeout=timeout_seconds)
        if msg is None:
            return None
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                return None
            logger.error(f"Errore consumer: {msg.error()}")
            return None

        command = json.loads(msg.value().decode("utf-8"))
        # Filtra solo i comandi per i DER di questo sito
        if command.get("site_id") == self.site_id:
            logger.info(
                f"Sito {self.site_id}: ricevuto dispatch "
                f"der={command['der_id']} power={command['power_kw']}kW"
            )
            # Commit esplicito dopo elaborazione riuscita (at-least-once)
            self._consumer.commit(msg)
            return command
        return None

# mv_service.py - Measurement & Verification per Demand Response
# Calcola la riduzione effettiva di carico rispetto alla baseline

import numpy as np
from typing import List, Tuple
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

class MVService:
    """
    Measurement & Verification (M&V) per programmi Demand Response.

    Metodo: CBL (Customer Baseline Load) - approccio standard FERC/ENTSO-E
    La baseline e calcolata come media degli N giorni simili più recenti
    prima dell'evento, escludendo giorni con altri eventi DR.
    """

    def __init__(self, n_baseline_days: int = 10, exclude_top_bottom: bool = True):
        self.n_baseline_days = n_baseline_days
        self.exclude_top_bottom = exclude_top_bottom  # High-5 / Low-5 exclusion

    def calculate_baseline(
        self,
        site_id: str,
        event_date: datetime,
        event_hour: int,
        historical_consumption: dict  # {date_str: {hour: kw}}
    ) -> dict:
        """
        Calcola la Customer Baseline Load (CBL) per il sito.

        Algoritmo CBL con High-5/Low-5 exclusion (CAISO/PJM standard):
        1. Seleziona N giorni simili recenti (stessa tipologia giorno: lavorativo/festivo)
        2. Esclude il top 20% e il bottom 20% dei giorni per consumo nell'ora evento
        3. Media i restanti giorni
        """
        target_weekday = event_date.weekday()  # 0=Lunedi, 6=Domenica
        is_target_workday = target_weekday < 5  # Lavorativo vs weekend

        # Raccolta dati storici compatibili
        similar_days = []
        check_date = event_date - timedelta(days=1)

        while len(similar_days) < self.n_baseline_days and check_date > event_date - timedelta(days=60):
            date_str = check_date.strftime("%Y-%m-%d")
            is_check_workday = check_date.weekday() < 5

            # Stesso tipo di giorno (lavorativo/festivo)
            if is_check_workday == is_target_workday and date_str in historical_consumption:
                day_data = historical_consumption[date_str]
                if event_hour in day_data:
                    similar_days.append({
                        "date": date_str,
                        "consumption_kw": day_data[event_hour]
                    })

            check_date -= timedelta(days=1)

        if len(similar_days) < 3:
            logger.warning(f"Dati insufficienti per baseline sito {site_id}: solo {len(similar_days)} giorni")
            return {"baseline_kw": None, "method": "insufficient_data", "days_used": len(similar_days)}

        consumptions = [d["consumption_kw"] for d in similar_days]

        # High-5/Low-5 exclusion (standard CAISO)
        if self.exclude_top_bottom and len(consumptions) >= 10:
            n_exclude = max(1, len(consumptions) // 5)  # 20% per lato
            sorted_consumptions = sorted(consumptions)
            filtered_consumptions = sorted_consumptions[n_exclude:-n_exclude]
        else:
            filtered_consumptions = consumptions

        baseline_kw = np.mean(filtered_consumptions)

        return {
            "site_id": site_id,
            "baseline_kw": round(baseline_kw, 2),
            "method": "CBL_HighLow_Exclusion",
            "days_analyzed": len(similar_days),
            "days_used": len(filtered_consumptions),
            "std_dev_kw": round(np.std(filtered_consumptions), 2)
        }

    def calculate_demand_reduction(
        self,
        site_id: str,
        baseline_kw: float,
        actual_consumption_kw: float,
        event_duration_hours: float
    ) -> dict:
        """
        Calcola la riduzione di carico e l'energia risparmiata.
        Risultato usato per il settlement del programma DR.
        """
        reduction_kw = max(0, baseline_kw - actual_consumption_kw)
        reduction_pct = (reduction_kw / baseline_kw * 100) if baseline_kw > 0 else 0
        energy_reduced_kwh = reduction_kw * event_duration_hours

        return {
            "site_id": site_id,
            "baseline_kw": baseline_kw,
            "actual_kw": actual_consumption_kw,
            "reduction_kw": round(reduction_kw, 2),
            "reduction_pct": round(reduction_pct, 1),
            "energy_reduced_kwh": round(energy_reduced_kwh, 3),
            "verified": reduction_pct >= 5.0  # Soglia minima per settlement
        }

# docker-compose.yml - Stack DERMS per VPP regionale
# Configurazione di sviluppo/staging (produzione su Kubernetes)

version: "3.9"

services:
  # ========================
  # INGESTION LAYER
  # ========================

  # Broker MQTT per telemetria edge
  mosquitto:
    image: eclipse-mosquitto:2.0
    ports:
      - "1883:1883"   # MQTT non sicuro (solo LAN interna)
      - "8883:8883"   # MQTT over TLS (produzione)
    volumes:
      - ./config/mosquitto.conf:/mosquitto/config/mosquitto.conf
      - mosquitto-data:/mosquitto/data

  # Apache Kafka (broker eventi principale)
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_NUM_PARTITIONS: 12
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1  # 3 in produzione
      KAFKA_LOG_RETENTION_HOURS: 168        # 7 giorni
    depends_on:
      - zookeeper

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  # ========================
  # PROCESSING LAYER
  # ========================

  # DERMS Aggregation Service (FastAPI)
  aggregation-service:
    build: ./services/aggregation
    ports:
      - "8080:8080"
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      REDIS_URL: redis://redis:6379/0
      INFLUXDB_URL: http://influxdb:8086
      INFLUXDB_TOKEN: {{INFLUXDB_TOKEN}}
      INFLUXDB_ORG: sunflex-puglia
      INFLUXDB_BUCKET: der-telemetry
    depends_on:
      - kafka
      - redis
      - influxdb

  # Dispatch Optimizer Service
  dispatch-optimizer:
    build: ./services/dispatch
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      REDIS_URL: redis://redis:6379/0
      SOLVER: COIN_CMD  # o CPLEX in produzione per portfolio grandi
    depends_on:
      - kafka
      - redis

  # Forecasting Service (ML - produzione FV + carico)
  forecasting-service:
    build: ./services/forecasting
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      INFLUXDB_URL: http://influxdb:8086
      MODEL_REGISTRY_URL: http://mlflow:5000
      WEATHER_API_KEY: {{OPENMETEO_API_KEY}}
    depends_on:
      - kafka
      - influxdb
      - mlflow

  # OpenADR VTN (Virtual Top Node - invia eventi DR ai siti)
  openadr-vtn:
    build: ./services/openadr-vtn
    ports:
      - "8081:8081"
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      POSTGRES_URL: postgresql://postgres:5432/derms
    depends_on:
      - kafka
      - postgres

  # ========================
  # STORAGE LAYER
  # ========================

  # Time-Series Database per telemetria
  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    volumes:
      - influxdb-data:/var/lib/influxdb2
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_ORG: sunflex-puglia
      DOCKER_INFLUXDB_INIT_BUCKET: der-telemetry
      DOCKER_INFLUXDB_INIT_RETENTION: 30d

  # Cache per stato corrente DER (Device Shadow)
  redis:
    image: redis:7.2-alpine
    command: redis-server --maxmemory 4gb --maxmemory-policy allkeys-lru
    ports:
      - "6379:6379"

  # Database relazionale per asset registry, events, settlement
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: derms
      POSTGRES_USER: derms_user
      POSTGRES_PASSWORD: {{POSTGRES_PASSWORD}}
    volumes:
      - postgres-data:/var/lib/postgresql/data

  # ========================
  # OBSERVABILITY
  # ========================

  # MLflow per tracking modelli forecasting
  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.11.0
    ports:
      - "5000:5000"
    command: mlflow server --host 0.0.0.0 --backend-store-uri postgresql://mlflow:5432/mlflow

  # Grafana per dashboard operativo
  grafana:
    image: grafana/grafana:10.4.0
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
      - ./config/grafana/dashboards:/etc/grafana/provisioning/dashboards

volumes:
  influxdb-data:
  postgres-data:
  redis-data:
  grafana-data:
  mosquitto-data: