Nabíjení EV Vyrovnávání zátěže: Algoritmy v reálném čase pro inteligentní nabíjení
Je úterý večer v 18:30. Tisíce italských motoristů se po práci vrací domů, zaparkují své elektrické vozidlo a připojí nabíjecí kabel. Za pár minut otázka výkon na sousedním transformátoru vystřelí nahoru. Bez inteligentního řízení, tento scénář – který se každým dnem opakuje stále intenzivněji, jak park roste EV – vede k přetížení sítě, lokalizovaným výpadkům a nákladům na upgrade infrastruktury v řádu miliard eur.
V roce 2025 budou dále obíhat 17 milionů elektromobilů v Evropě, z toho cca 230 000 v Itálii s 94 230 novými registracemi jen v roce 2025 (+46 % ve srovnání s rokem 2024). Projekce do roku 2030 hovoří o více než 50 milionech EV v Evropě: každé vozidlo potřebuje v průměru mezi 7 a 22 kW výkonu při nabíjení. Vynásobeno miliony uživatelů, kteří dobíjejí ve stejných večerních hodinách dostaneme bezprecedentní problém se stabilitou sítě.
Řešením není budovat další síť: je to příliš drahé a pomalé. Řešením je inteligentní vyvažování zátěže: Algoritmy v reálném čase, které distribuují energii dostupné mezi nabíjecími vozidly, respektující síťová omezení, minimalizující náklady na energii a maximalizaci spokojenosti uživatelů. V tomto článku prozkoumáme celý zásobník technologické: od protokolu OCPP 2.0.1 k optimalizačním algoritmům, od učení výztuže k integraci V2G, s fungujícím kódem Python a italským regulačním kontextem.
Co se dozvíte v tomto článku
- Kachní křivky a nezvládnutý dopad EV na distribuční síť
- Taxonomie algoritmů pro vyrovnávání zátěže: Statický, Dynamický, Prediktivní
- Chytré nabíjení OCPP 2.0.1: SetChargingProfile, ChargingSchedule, StackLevel
- Implementace Pythonu: Equal Share s dynamickým přepočtem v reálném čase
- Algoritmus založený na prioritě s frontou haldy (SoC, termín, rychlost)
- Posílení učení (PPO) pro optimalizaci nákladů na energii
- Vehicle-to-Grid (V2G): obousměrný, ISO 15118-20, regulace frekvence
- Fotovoltaická integrace: solární přebytek a směrování do EV
- Architektura mikroslužeb: Charge Controller, Energy Manager, Forecaster
- Italský kontext: dvouhodinové tarify, CER s EV, předpisy ARERA
Řada EnergyTech – 10 článků
| # | Položka | Stát |
|---|---|---|
| 1 | Smart Grid a IoT: Architektura pro elektrickou síť budoucnosti | Publikováno |
| 2 | Architektura DERMS: Agregace milionů distribuovaných zdrojů | Publikováno |
| 3 | Battery Management System: Řídicí algoritmy pro BESS | Publikováno |
| 4 | Digitální dvojče elektrické sítě s Pythonem a Pandapower | Publikováno |
| 5 | Prognóza obnovitelné energie: ML pro FV a větrnou energii | Publikováno |
| 6 | EV Charging Load Balancing: Real-Time Algorithms (jste zde) | Proud |
| 7 | MQTT a InfluxDB pro energetickou telemetrii v reálném čase | Již brzy |
| 8 | IEC 61850: Komunikace v elektrické rozvodně | Již brzy |
| 9 | Software pro uhlíkové účetnictví: Měření a snižování emisí | Již brzy |
| 10 | Blockchain pro P2P obchodování s energií v CER | Již brzy |
Problém vrcholu: Duck Curve a Unmanaged EVs
Abychom pochopili, proč není vyvažování zátěže EV možností, ale nutností, musíme začít z fyziky elektrické sítě a z konceptu, kterého se manažeři sítě každý den obávají: a kachní křivky.
Kachní křivka a její zhoršování
Kachní křivka popisuje tvar křivky čisté spotřeby elektřiny během typického dne v systému s vysokou penetrací fotovoltaiky. Tak se tomu říká protože křivka připomíná profil kachny: nízké břicho uprostřed dne (když solární energie vyrábí hojně a čistá a nízká poptávka) a večer s vysokým hrbem (když slunce zapadá, výroba FV kolabuje a poptávka po bydlení raketově stoupá).
Italský start bez elektromobilů tento problém řeší každý den. S růstem EV pokud se problém nezvládne, drasticky se zesílí. Výzkum MDPI (2025) ukazuje, že bez optimalizace špičková poptávka přechází z 22 000 MW základní úrovně na 35 000 MW s 5 miliony EV. S chytrým nabíjením stejná flotila přidává pouze 2 000 až 3 000 MW distribuovaných během nočních hodin.
| Scénář | Špičkový čas | Dodatečná zátěž | Síťové riziko |
|---|---|---|---|
| Základní linie (bez EV) | 19:00–20:00 | 0 MW | Ovladatelné |
| 500 000 neřízených EV | 18:30–19:30 | +3 500 MW | Lokální namáhání transformátorů |
| 1,5 milionu neřízených EV | 18:00–20:00 | +10 500 MW | Rozsáhlé přetížení |
| 5 milionů neřízených EV | 18:00–21:00 | +35 000 MW | Systémové výpadky proudu |
| 1,5M EV s chytrým nabíjením | Doručeno 22:00-6:00 | +2 100 MW zředěno | Zanedbatelný |
Optimalizace nabíjecího profilu řízená umělou inteligencí může snížit špičkovou poptávku 16 % s 1,5 milionu EVo 21 % se 3 miliony a o 34 % s 5 miliony (zdroj: MDPI Electronics, 2025). Rozdíl mezi řízeným a neřízeným scénářem Není to pár procentních bodů: je to rozdíl mezi stabilní sítí a kolapsem.
Vliv na distribuční transformátory
Problém není pouze na úrovni přenosové soustavy: je především na úrovni sítě místní distribuce. 400 kVA sousedící transformátor typicky slouží 80-120 rodiny. Pokud 15 z nich má elektromobil, který se začne nabíjet při 11 kW dohromady v 18:30, přídavné zatížení je 165 kW - téměř 41 % jmenovitého výkonu transformátoru, který už by to mohlo být na 60-70 % pro domácí spotřebu. Výsledek: přehřátí, snížení životnosti a v nejhorších případech otevření ochran.
Náklady na nečinnost
Podle Terna, náklady na posílení italské distribuční sítě na podporu přechod EV bez chytrého nabíjení se odhaduje na 10-15 miliard eur do roku 2030. Díky rozšířenému chytrému nabíjení si samotný přechod vyžaduje investice náklady na infrastrukturu o 60–70 % nižší, jednoduše přesunem nákladu během hodin, ve kterých síť má dostupnou kapacitu.
Taxonomie algoritmů vyvažování zátěže
Neexistuje jediný optimální algoritmus vyvažování zátěže: výběr závisí na velikosti instalace, dostupnost historických dat, požadavky na latenci a složitost, kterou je manažer ochoten zachovat. Přístupy klasifikujeme do tří hlavní rodiny.
Statické algoritmy
Statické algoritmy definují pravidla distribuce energie a priori, aniž by se dynamicky přizpůsobovaly stavu systému. Jejich implementace je jednoduchá a ladění, ideální pro malé, homogenní instalace.
- Round Robin: každý konektor dostává stejný maximální výkon na otáčku. Jednoduché, ale neefektivní: téměř nabitý elektromobil spotřebuje stejnou energii jako prázdný.
- Rovný podíl: celkový dostupný výkon je rozdělen rovným dílem mezi všechny aktivní konektory. Efektivní pro homogenní instalace, ale nebere v úvahu individuální potřeby.
- Kdo dřív přijde, je dřív na řadě: první připojené EV získávají maximální výkon, následující rozdělí zbytek. Penalizovat ty, kteří přijdou pozdě.
Dynamické algoritmy
Dynamické algoritmy přizpůsobují distribuci v reálném čase na základě stavu systémový proud: počet připojených EV, úroveň nabití (SoC), dostupný výkon, uživatelské priority.
- Úměrný: výkon je distribuován úměrně energii požaduje každý EV. Kdo má nejnižší baterii, dostane nejvíce energie.
- Na základě priority: každá EV má prioritní skóre vypočítané z více faktory (SoC, termín, předplatné). Síla proudí k vyšším prioritám.
- Fuzzy logika: lingvistická pravidla jako „pokud SoC a low E deadline a zavřít PAK je priorita vysoká." Dobře zvládá nejistotu ve vstupních datech.
Prediktivní algoritmy (založené na ML)
Prediktivní algoritmy využívají modely strojového učení k předvídání budoucích událostí (příjezdy, odjezdy, kolísání cen energií) a optimalizovat plánování v předem, což dlouhodobě minimalizuje celkové náklady.
- Model Predictive Control (MPC): řeší optimalizační problém v budoucím časovém horizontu (např. 4 hodiny) v každém kontrolním intervalu (např. 15 min), použití pouze prvního kroku a přeformulování v dalším kroku.
- Deep Reinforcement Learning (DRL): agent se naučí optimální politiku interakce se simulátorem. Ve výrobě jsou nejpoužívanější PPO a SAC.
- Stochastické programování: explicitně zahrnuje nejistotu do Příjezdy a odjezdy elektromobilů, generování plánů odolných vůči nejhorším scénářům.
| Algoritmus | Složitost | Latence | Optimalita | Údaje jsou požadovány | Ideální případ použití |
|---|---|---|---|---|---|
| Rovný podíl | Nízký | <1 ms | Nízký | Nikdo | Rezidenční, <10 prodejen |
| Na základě priority | Průměrný | <10 ms | Středně vysoká | SoC, uživatelská lhůta | Kanceláře, hotely, flotily |
| MPC | Vysoký | 100 ms-1s | Vysoký | Ceny energií, prognózy | Hub >50 zásuvek, C&I |
| Deep RL (PPO) | Velmi vysoká | <50 ms (inference) | Velmi vysoká | Historie 6+ měsíců | Skvělé rozbočovače, vestavěný V2G |
OCPP 2.0.1 Smart Charging: Protokol inteligentního nabíjení
OCPP (Open Charge Point Protocol) a univerzální jazyk mezi nabíjecími body (CP) a centrální systém (CSMS - Charging Station Management System). Ve verzi 2.0.1 Chytré nabíjení bylo hluboce vylepšeno ve srovnání s OCPP 1.6 zavedením podrobnější a spolehlivější správa profilů nabíjení.
Mechanismus SetChargingProfile
CSMS odešle zprávu do nabíjecího bodu SetChargingProfileRequest obsahující
a ChargingProfile který definuje, jak musí stanice dodávat energii v průběhu času.
A ChargingProfile a skládá se z:
-
nabíjecí profil Účel:
ChargePointMaxProfile(omezit vše nádraží),TxDefaultProfile(výchozí pro nové transakce),TxProfile(specifické pro probíhající transakci). - stackLevel: celé číslo, které určuje prioritu. V případě profilů překrývající se, vyhrává ten, kdo má nejvyšší stackLevel.
- Harmonogram nabíjení: seznam období (startPeriod v sekundách, limit v A nebo W), které definují nabíjecí křivku v čase.
Kritická sekvence: Snížit před zvýšením
Při odesílání aktualizovaných profilů na více stanic je nezbytné nejprve odeslat příkazy pro snížení výkonu a poté příkazy pro zvýšení. Pokud nejprve zvýšíte, riskujete překročení limitu sítě na krátký interval a spuštění elektrické ochrany nebo generování odběrových špiček s následnými smluvními pokutami. Toto pravidlo je nesmlouvavé ve výrobě.
Implementace Pythonu: generování profilu OCPP 2.0.1
"""
OCPP 2.0.1 Smart Charging Profile Generator
Genera ChargingProfile compatibili con la specifica OCPP 2.0.1.
"""
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum
import json
from datetime import datetime, timezone
class ChargingRateUnit(str, Enum):
WATTS = "W"
AMPERES = "A"
class ChargingProfilePurpose(str, Enum):
CHARGE_POINT_MAX = "ChargePointMaxProfile"
TX_DEFAULT = "TxDefaultProfile"
TX = "TxProfile"
class ChargingProfileKind(str, Enum):
ABSOLUTE = "Absolute"
RECURRING = "Recurring"
RELATIVE = "Relative"
@dataclass
class ChargingSchedulePeriod:
"""Singolo intervallo nella schedule di ricarica."""
start_period: int # secondi dall'inizio della schedule
limit: float # limite in W o A
number_phases: Optional[int] = None
@dataclass
class ChargingSchedule:
"""Schedule completa di un profilo di ricarica."""
id: int
charging_rate_unit: ChargingRateUnit
charging_schedule_period: List[ChargingSchedulePeriod]
duration: Optional[int] = None
start_schedule: Optional[str] = None
min_charging_rate: Optional[float] = None
@dataclass
class ChargingProfile:
"""Profilo di ricarica OCPP 2.0.1 completo."""
id: int
stack_level: int
charging_profile_purpose: ChargingProfilePurpose
charging_profile_kind: ChargingProfileKind
charging_schedule: List[ChargingSchedule]
transaction_id: Optional[str] = None
valid_from: Optional[str] = None
valid_to: Optional[str] = None
def build_equal_share_profile(
profile_id: int,
power_limit_watts: float,
duration_seconds: int = 3600,
stack_level: int = 1
) -> ChargingProfile:
"""
Costruisce un profilo OCPP per distribuzione equa (potenza fissa).
Usato dall'algoritmo Equal Share per applicare il limite calcolato.
"""
schedule_period = ChargingSchedulePeriod(
start_period=0,
limit=power_limit_watts,
number_phases=3
)
schedule = ChargingSchedule(
id=profile_id,
charging_rate_unit=ChargingRateUnit.WATTS,
charging_schedule_period=[schedule_period],
duration=duration_seconds,
start_schedule=datetime.now(timezone.utc).isoformat()
)
return ChargingProfile(
id=profile_id,
stack_level=stack_level,
charging_profile_purpose=ChargingProfilePurpose.TX,
charging_profile_kind=ChargingProfileKind.ABSOLUTE,
charging_schedule=[schedule]
)
def build_time_of_use_profile(
profile_id: int,
schedule_periods: List[tuple],
stack_level: int = 2
) -> ChargingProfile:
"""
Costruisce un profilo Time-of-Use con potenza variabile nel tempo.
Esempio: alta potenza di notte (energia economica), bassa di giorno.
Args:
schedule_periods: lista di (start_period_sec, limit_watts)
"""
periods = [
ChargingSchedulePeriod(start_period=start, limit=limit)
for start, limit in schedule_periods
]
schedule = ChargingSchedule(
id=profile_id,
charging_rate_unit=ChargingRateUnit.WATTS,
charging_schedule_period=periods,
start_schedule=datetime.now(timezone.utc).isoformat()
)
return ChargingProfile(
id=profile_id,
stack_level=stack_level,
charging_profile_purpose=ChargingProfilePurpose.TX_DEFAULT,
charging_profile_kind=ChargingProfileKind.ABSOLUTE,
charging_schedule=[schedule]
)
def profile_to_ocpp_dict(profile: ChargingProfile) -> dict:
"""Serializza il profilo nel formato JSON OCPP 2.0.1."""
return {
"id": profile.id,
"stackLevel": profile.stack_level,
"chargingProfilePurpose": profile.charging_profile_purpose.value,
"chargingProfileKind": profile.charging_profile_kind.value,
"chargingSchedule": [
{
"id": sched.id,
"chargingRateUnit": sched.charging_rate_unit.value,
"chargingSchedulePeriod": [
{
"startPeriod": p.start_period,
"limit": p.limit,
}
for p in sched.charging_schedule_period
],
**({"duration": sched.duration} if sched.duration else {}),
**({"startSchedule": sched.start_schedule} if sched.start_schedule else {}),
}
for sched in profile.charging_schedule
],
}
if __name__ == "__main__":
# Profilo Equal Share: 3.3 kW per 2 ore
profile = build_equal_share_profile(
profile_id=101,
power_limit_watts=3300.0,
duration_seconds=7200
)
print(json.dumps(profile_to_ocpp_dict(profile), indent=2))
# Profilo Time-of-Use: 11 kW da adesso, 3.3 kW dopo 4 ore
tou = build_time_of_use_profile(
profile_id=102,
schedule_periods=[
(0, 11000), # da ora: 11 kW (fascia fuori punta)
(4 * 3600, 3300), # dopo 4 ore: 3.3 kW (fascia F1 inizia)
],
stack_level=2
)
print(json.dumps(profile_to_ocpp_dict(tou), indent=2))Algoritmus Equal Share s dynamickým přepočítáním
Equal Share je nejrozšířenější algoritmus vyvažování zátěže v reálných instalacích. Jeho síla nespočívá v matematické eleganci, ale v provozní robustnosti: je to jednoduché vysvětlit uživatelům, snadno se ladí a jsou dostatečně účinné většina instalací do 30-40 zásuvek. Skutečná složitost spočívá v spravovat dynamické přepočty: pokaždé, když se EV připojí nebo odpojí, nebo když dostupné změny výkonu pro signál DSO, systém musí přepočítat a redistribuovat v milisekundách, respektovat kritickou sekvenci snížení-potom-zvýšení.
"""
Equal Share Load Balancer
Distribuzione equa con ricalcolo dinamico event-driven.
Thread-safe per deployment in produzione con OCPP WebSocket server.
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Dict, Optional, Callable, Awaitable
from enum import Enum
import time
logger = logging.getLogger(__name__)
class ConnectorStatus(str, Enum):
AVAILABLE = "Available"
CHARGING = "Charging"
SUSPENDED_EV = "SuspendedEV"
FINISHING = "Finishing"
UNAVAILABLE = "Unavailable"
FAULTED = "Faulted"
@dataclass
class ConnectorState:
"""Stato di un singolo connettore nella stazione."""
connector_id: str
status: ConnectorStatus = ConnectorStatus.AVAILABLE
transaction_id: Optional[str] = None
allocated_power_w: float = 0.0
max_power_w: float = 22000.0 # max fisico del connettore (22 kW AC trifase)
min_power_w: float = 1380.0 # min per mantenere la carica (6A x 230V)
soc_percent: Optional[float] = None
connected_at: Optional[float] = None
@property
def is_charging(self) -> bool:
return self.status == ConnectorStatus.CHARGING
@property
def session_duration_min(self) -> float:
if self.connected_at is None:
return 0.0
return (time.time() - self.connected_at) / 60.0
@dataclass
class StationConfig:
"""Configurazione della stazione di ricarica."""
station_id: str
max_station_power_w: float
dynamic_limit_w: Optional[float] = None
rebalance_interval_s: float = 30.0
@property
def available_power_w(self) -> float:
if self.dynamic_limit_w is not None:
return min(self.max_station_power_w, self.dynamic_limit_w)
return self.max_station_power_w
ProfileSendCallback = Callable[[str, str, float], Awaitable[bool]]
class EqualShareBalancer:
"""
Load balancer Equal Share thread-safe e asincrono.
Gestisce la distribuzione equa della potenza disponibile
tra tutti i connettori attivamente in carica.
"""
def __init__(
self,
config: StationConfig,
profile_sender: ProfileSendCallback
):
self._config = config
self._connectors: Dict[str, ConnectorState] = {}
self._send_profile = profile_sender
self._lock = asyncio.Lock()
self._last_allocation: Dict[str, float] = {}
async def register_connector(
self,
connector_id: str,
max_power_w: float = 22000.0,
min_power_w: float = 1380.0
) -> None:
async with self._lock:
self._connectors[connector_id] = ConnectorState(
connector_id=connector_id,
max_power_w=max_power_w,
min_power_w=min_power_w
)
logger.info("Connettore %s registrato (max=%.0f W)", connector_id, max_power_w)
async def on_ev_connected(self, connector_id: str, transaction_id: str) -> None:
"""Chiamato quando un EV si connette. Triggera ricalcolo immediato."""
async with self._lock:
if connector_id not in self._connectors:
logger.warning("Connettore sconosciuto: %s", connector_id)
return
self._connectors[connector_id].status = ConnectorStatus.CHARGING
self._connectors[connector_id].transaction_id = transaction_id
self._connectors[connector_id].connected_at = time.time()
await self._rebalance()
async def on_ev_disconnected(self, connector_id: str) -> None:
"""Chiamato quando un EV si disconnette. Libera la potenza e redistribuisce."""
async with self._lock:
if connector_id not in self._connectors:
return
connector = self._connectors[connector_id]
connector.status = ConnectorStatus.AVAILABLE
connector.transaction_id = None
connector.allocated_power_w = 0.0
connector.connected_at = None
await self._rebalance()
async def update_dynamic_limit(self, limit_w: float) -> None:
"""Aggiorna il limite DSO in tempo reale. Triggera ricalcolo immediato."""
async with self._lock:
self._config.dynamic_limit_w = limit_w
logger.info("Limite DSO aggiornato: %.0f W (stazione %s)", limit_w, self._config.station_id)
await self._rebalance()
async def _rebalance(self) -> None:
"""
Calcola la nuova allocazione Equal Share e invia i profili OCPP.
REGOLA CRITICA: invia prima le riduzioni, poi gli aumenti.
"""
async with self._lock:
active = [c for c in self._connectors.values() if c.is_charging]
if not active:
return
available = self._config.available_power_w
n = len(active)
raw_share = available / n
new_allocations: Dict[str, float] = {}
# Prima passata: applica limiti min/max per connettore
constrained = []
free = []
for connector in active:
if raw_share <= connector.min_power_w:
new_allocations[connector.connector_id] = connector.min_power_w
constrained.append(connector)
elif raw_share >= connector.max_power_w:
new_allocations[connector.connector_id] = connector.max_power_w
constrained.append(connector)
else:
free.append(connector)
# Seconda passata: redistribuisce il residuo tra i connettori liberi
if free:
allocated_constrained = sum(
new_allocations[c.connector_id] for c in constrained
)
residual = available - allocated_constrained
share_free = residual / len(free)
for connector in free:
new_allocations[connector.connector_id] = max(
connector.min_power_w,
min(connector.max_power_w, share_free)
)
# Ordina: prima le riduzioni, poi gli aumenti (regola OCPP)
reductions = []
increases = []
for connector_id, new_power in new_allocations.items():
old_power = self._last_allocation.get(connector_id, float('inf'))
if new_power < old_power:
reductions.append((connector_id, new_power))
else:
increases.append((connector_id, new_power))
for connector_id, power in reductions:
success = await self._send_profile(self._config.station_id, connector_id, power)
if success:
self._last_allocation[connector_id] = power
logger.info("Riduzione: %s -> %.0f W", connector_id, power)
for connector_id, power in increases:
success = await self._send_profile(self._config.station_id, connector_id, power)
if success:
self._last_allocation[connector_id] = power
logger.info("Aumento: %s -> %.0f W", connector_id, power)
logger.info(
"Ribilanciamento: %d EV attivi, %.0f W disponibili, stazione %s",
len(active), self._config.available_power_w, self._config.station_id
)
async def start_periodic_rebalance(self) -> None:
"""Avvia il ricalcolo periodico come safety net per deriva dello stato."""
async def _loop():
while True:
await asyncio.sleep(self._config.rebalance_interval_s)
await self._rebalance()
asyncio.create_task(_loop())
def get_status(self) -> dict:
return {
"station_id": self._config.station_id,
"available_power_w": self._config.available_power_w,
"connectors": [
{
"connector_id": c.connector_id,
"status": c.status.value,
"allocated_power_w": c.allocated_power_w,
"soc_percent": c.soc_percent,
"session_duration_min": round(c.session_duration_min, 1)
}
for c in self._connectors.values()
]
}
async def demo_equal_share():
async def mock_send(station_id, connector_id, power_w) -> bool:
print(f" OCPP -> {connector_id}: {power_w:.0f} W")
return True
config = StationConfig(station_id="STATION_001", max_station_power_w=44000.0)
balancer = EqualShareBalancer(config, mock_send)
for i in range(1, 5):
await balancer.register_connector(f"C0{i}", max_power_w=22000.0, min_power_w=1380.0)
print("\n[EV1 connesso]")
await balancer.on_ev_connected("C01", "TX001") # C01: 44000 W
print("\n[EV2 connesso]")
await balancer.on_ev_connected("C02", "TX002") # C01,C02: 22000 W ciascuno
print("\n[EV3 e EV4 connessi]")
await balancer.on_ev_connected("C03", "TX003")
await balancer.on_ev_connected("C04", "TX004") # tutti: 11000 W
print("\n[DSO riduce limite a 22 kW]")
await balancer.update_dynamic_limit(22000.0) # tutti: 5500 W
print("\n[EV2 si disconnette]")
await balancer.on_ev_disconnected("C02") # C01,C03,C04: 7333 W
if __name__ == "__main__":
asyncio.run(demo_equal_share())Algoritmus založený na prioritách s vícefaktorovým hodnocením
Rovný podíl je spravedlivý, ale není optimální z hlediska uživatelů. EV s baterií těch 5 %, jejichž majitel musí odejít do 30 minut, má mnohem naléhavější potřebu ve srovnání s 60% EV, které zůstane zaparkované 8 hodin. Vyvažování podle priority reaguje na tuto potřebu výpočtem multifaktoriálního skóre naléhavosti; rozdělování moci úměrně prioritám.
Vzorec hodnocení priority
Skóre kombinuje čtyři rozměry s konfigurovatelnými hmotnostmi:
- SoC naléhavost (35 %): kolik a vybití baterie. Konkávní funkce pro zesílení kritických případů (např. SoC 5 % = velmi vysoká naléhavost).
- Časový tlak (35 %): jak blízko je odjezd. Exponenciální funkce, která rychle roste s blížícím se termínem.
- Uživatelská úroveň (15 %): úroveň předplatného (základní/prémiové/prioritní). Vytvořte obchodní model s diferencovanými smlouvami SLA.
- Energetická účinnost (15 %): odměňuje ty, kteří skutečně dokážou absorbovat moc. Vyhněte se přidělování energie téměř plným bateriím.
"""
Priority-Based Load Balancer.
Distribuisce potenza proporzionalmente ai punteggi di urgenza,
con algoritmo iterativo per rispettare i vincoli min/max.
"""
import math
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
logger = logging.getLogger(__name__)
@dataclass
class ChargingSession:
"""Sessione con tutti i parametri per il calcolo della priorità."""
connector_id: str
transaction_id: str
soc_percent: float
target_soc_percent: float
battery_capacity_kwh: float
departure_time: float # timestamp unix
max_power_w: float
min_power_w: float
user_tier: int = 1 # 1=base, 2=premium, 3=priority
connected_at: float = field(default_factory=time.time)
@property
def energy_needed_kwh(self) -> float:
delta = max(0.0, self.target_soc_percent - self.soc_percent) / 100.0
return self.battery_capacity_kwh * delta
@property
def time_remaining_h(self) -> float:
return max(0.1, (self.departure_time - time.time()) / 3600.0)
class PriorityScoreCalculator:
"""Calcola punteggio di urgenza normalizzato [0, 100]."""
W_SOC = 0.35
W_TIME = 0.35
W_TIER = 0.15
W_EFFICIENCY = 0.15
def calculate(self, session: ChargingSession) -> float:
# 1. Urgenza SoC: concava, amplifica i SoC bassissimi
soc_ratio = session.soc_percent / 100.0
soc_score = (1.0 - soc_ratio) ** 1.5 * 100
# 2. Pressione temporale: esponenziale, cresce rapido vicino alla deadline
time_score = 100.0 * math.exp(-0.3 * session.time_remaining_h)
# 3. Tier: bonus per abbonamenti premium
tier_score = {1: 33.0, 2: 66.0, 3: 100.0}.get(session.user_tier, 33.0)
# 4. Efficienza: premia chi ha batteria da riempire
eff = min(100.0, (session.energy_needed_kwh / max(0.1, session.battery_capacity_kwh)) * 100)
return round(
self.W_SOC * soc_score +
self.W_TIME * time_score +
self.W_TIER * tier_score +
self.W_EFFICIENCY * eff,
2
)
class PriorityBalancer:
"""
Distribuisce potenza proporzionalmente ai punteggi con vincoli min/max.
Algoritmo iterativo: separa i connettori con vincoli attivi e redistribuisce
il residuo tra quelli liberi, ripetendo fino a convergenza.
"""
def __init__(self, station_id: str, max_station_power_w: float, profile_sender):
self._station_id = station_id
self._max_power = max_station_power_w
self._dynamic_limit: Optional[float] = None
self._sessions: Dict[str, ChargingSession] = {}
self._send_profile = profile_sender
self._score_calc = PriorityScoreCalculator()
self._lock = asyncio.Lock()
@property
def available_power_w(self) -> float:
if self._dynamic_limit is not None:
return min(self._max_power, self._dynamic_limit)
return self._max_power
async def add_session(self, session: ChargingSession) -> None:
async with self._lock:
self._sessions[session.connector_id] = session
await self._rebalance()
async def remove_session(self, connector_id: str) -> None:
async with self._lock:
self._sessions.pop(connector_id, None)
await self._rebalance()
async def update_soc(self, connector_id: str, soc_percent: float) -> None:
async with self._lock:
if connector_id in self._sessions:
self._sessions[connector_id].soc_percent = soc_percent
await self._rebalance()
def _compute_allocations(self) -> Dict[str, float]:
"""Algoritmo iterativo per allocazione con vincoli min/max."""
if not self._sessions:
return {}
sessions = list(self._sessions.values())
scores = {s.connector_id: self._score_calc.calculate(s) for s in sessions}
allocations: Dict[str, float] = {}
remaining_power = self.available_power_w
remaining = sessions[:]
for _ in range(10): # max 10 iterazioni per convergenza
if not remaining:
break
total_score = sum(scores[s.connector_id] for s in remaining)
if total_score == 0:
share = remaining_power / len(remaining)
for s in remaining:
allocations[s.connector_id] = max(s.min_power_w, min(s.max_power_w, share))
break
constrained, free_sessions = [], []
for s in remaining:
ratio = scores[s.connector_id] / total_score
proposed = remaining_power * ratio
if proposed <= s.min_power_w:
allocations[s.connector_id] = s.min_power_w
constrained.append(s)
elif proposed >= s.max_power_w:
allocations[s.connector_id] = s.max_power_w
constrained.append(s)
else:
free_sessions.append(s)
if not constrained:
# Nessun vincolo: allocazione finale proporzionale
for s in remaining:
ratio = scores[s.connector_id] / total_score
allocations[s.connector_id] = remaining_power * ratio
break
for s in constrained:
remaining_power -= allocations[s.connector_id]
remaining = free_sessions
return allocations
async def _rebalance(self) -> None:
async with self._lock:
allocs = self._compute_allocations()
# Stampa riepilogo per logging
for cid, power in sorted(allocs.items()):
s = self._sessions.get(cid)
score = self._score_calc.calculate(s) if s else 0
logger.info(
"Priority: %s score=%.1f -> %.0f W (SoC=%.0f%%, %.1fh)",
cid, score, power,
s.soc_percent if s else 0,
s.time_remaining_h if s else 0
)
await self._send_profile(self._station_id, cid, power)
# Demo
async def priority_demo():
async def mock_send(station_id, cid, power_w) -> bool:
print(f" SET {cid}: {power_w:.0f} W")
return True
balancer = PriorityBalancer("HUB_01", 44000.0, mock_send)
now = time.time()
sessions = [
# Urgente: 5% SoC, parte tra 45 min
ChargingSession("C1", "T1", 5.0, 80.0, 77.0, now + 45*60, 22000.0, 1380.0, user_tier=2),
# Normale: 60% SoC, parte tra 8 ore
ChargingSession("C2", "T2", 60.0, 80.0, 40.0, now + 8*3600, 11000.0, 1380.0, user_tier=1),
# Priority tier: 30% SoC, parte tra 2 ore
ChargingSession("C3", "T3", 30.0, 90.0, 100.0, now + 2*3600, 22000.0, 1380.0, user_tier=3),
]
print("\n--- Sessioni aggiunte (44 kW disponibili) ---")
for s in sessions:
await balancer.add_session(s)
print("\n--- SoC C1 sale al 40% (meno urgente) ---")
await balancer.update_soc("C1", 40.0)
if __name__ == "__main__":
asyncio.run(priority_demo())Posílení Naučte se minimalizovat náklady na energii
Předchozí algoritmy reagují na aktuální stav. Posilovací učení přesahuje: agent se prostřednictvím milionů simulací naučí optimální politiku, kterou minimalizuje celkové náklady na energii s ohledem na proměnlivé ceny v čase, předpovědi příjezdu a odchod EV a omezení sítě. Ve výrobě v roce 2025 hlavní operátoři nabíjecí rozbočovače používají varianty Proximální optimalizace zásad (PPO) pro snížit náklady o 15–25 % ve srovnání s reaktivními algoritmy (zdroj: ScienceDirect, 2025).
Formulace jako Markovův rozhodovací proces
- Stavový prostor: pro každý konektor (SoC, zbývající čas, aktuální výkon) plus stav sítě (dostupný výkon, cena energie, denní doba, předpověď FV).
- Akční prostor: spojitý vektor normalizovaných úrovní výkonu [0,1] pro každý konektor.
- odměny: minimalizuje náklady na energii, penalizuje EV, která nedosáhnou cílový SoC do termínu, penalizuje porušení síťových limitů.
"""
Ambiente Gymnasium per EV Charging RL.
Compatibile con Stable-Baselines3 (PPO, SAC).
Ogni step = 15 minuti. Un episodio = 24 ore (96 step).
Installazione: pip install gymnasium stable-baselines3 numpy
"""
import numpy as np
import gymnasium as gym
from gymnasium import spaces
from dataclasses import dataclass
from typing import List, Optional, Tuple
@dataclass
class EVSession:
soc: float # [0, 1] corrente
target_soc: float # [0, 1] obiettivo
battery_kwh: float
time_remaining_h: float
max_power_kw: float
min_power_kw: float
class EVChargingEnv(gym.Env):
"""
Ambiente Gymnasium per scheduling EV charging.
Obiettivo: minimizzare costo energia rispettando deadline SoC.
"""
metadata = {"render_modes": ["human"]}
def __init__(
self,
n_connectors: int = 4,
max_station_power_kw: float = 44.0,
episode_steps: int = 96,
electricity_prices: Optional[np.ndarray] = None
):
super().__init__()
self.n = n_connectors
self.max_power = max_station_power_kw
self.episode_steps = episode_steps
self.prices = electricity_prices if electricity_prices is not None \
else self._italian_biorario_prices()
# Action: potenza normalizzata [0,1] per ogni connettore
self.action_space = spaces.Box(
low=0.0, high=1.0, shape=(n_connectors,), dtype=np.float32
)
# Observation: 5 feature per connettore + 3 globali
obs_dim = n_connectors * 5 + 3
self.observation_space = spaces.Box(
low=-1.0, high=1.0, shape=(obs_dim,), dtype=np.float32
)
self.sessions: List[Optional[EVSession]] = [None] * n_connectors
self.step_count = 0
def _italian_biorario_prices(self) -> np.ndarray:
"""Prezzi biorari italiani simulati: F1 (08-19) = 0.28, F2/F3 = 0.18 EUR/kWh."""
prices = np.zeros(96)
for i in range(96):
hour = (i * 15) // 60
prices[i] = 0.28 if 8 <= hour < 19 else 0.18
# Variabilità mercato spot
return (prices + np.random.normal(0, 0.02, 96)).clip(0.10, 0.40)
def _get_obs(self) -> np.ndarray:
obs = []
for s in self.sessions:
if s is not None:
obs.extend([
s.soc,
s.target_soc,
min(1.0, s.time_remaining_h / 12.0),
1.0,
s.max_power_kw / self.max_power
])
else:
obs.extend([0.0, 0.0, 0.0, 0.0, 0.0])
step_idx = self.step_count % len(self.prices)
hour = (self.step_count * 15 // 60) % 24
obs.extend([
self.prices[step_idx] / 0.40,
np.sin(2 * np.pi * hour / 24),
np.cos(2 * np.pi * hour / 24)
])
return np.array(obs, dtype=np.float32)
def step(self, action: np.ndarray) -> Tuple:
dt_h = 15 / 60
price = self.prices[self.step_count % len(self.prices)]
# Converti azioni in potenze reali con vincoli
actual_powers = np.zeros(self.n)
for i, s in enumerate(self.sessions):
if s is not None:
power = action[i] * s.max_power_kw
actual_powers[i] = np.clip(power, s.min_power_kw, s.max_power_kw)
# Rispetta limite stazione
total = actual_powers.sum()
if total > self.max_power:
actual_powers *= self.max_power / total
# Calcola reward
energy_kwh = actual_powers.sum() * dt_h
reward = -energy_kwh * price # costo negativo (vogliamo minimizzare)
# Aggiorna SoC e penalita deadline
for i, s in enumerate(self.sessions):
if s is None:
continue
delta_soc = (actual_powers[i] * dt_h * 0.92) / s.battery_kwh
s.soc = min(1.0, s.soc + delta_soc)
s.time_remaining_h -= dt_h
if s.time_remaining_h <= 0:
gap = max(0, s.target_soc - s.soc)
reward -= gap * 50.0 # penalita pesante per deadline mancata
# Penalita violazione rete
if total > self.max_power * 1.01:
reward -= (total - self.max_power) * 5.0
# Bonus ricarica off-peak
hour = (self.step_count * 15 // 60) % 24
if not (8 <= hour < 19) and total > 0:
reward += total * 0.015
self.step_count += 1
obs = self._get_obs()
terminated = self.step_count >= self.episode_steps
return obs, float(reward), terminated, False, {"price": price, "total_kw": total}
def reset(self, seed=None, options=None) -> Tuple:
super().reset(seed=seed)
self.step_count = 0
self._spawn_random_sessions()
return self._get_obs(), {}
def _spawn_random_sessions(self) -> None:
"""Genera sessioni EV casuali realistiche per il training."""
for i in range(self.n):
if np.random.random() > 0.3:
self.sessions[i] = EVSession(
soc=float(np.random.uniform(0.05, 0.7)),
target_soc=float(np.random.uniform(0.7, 0.95)),
battery_kwh=float(np.random.choice([40.0, 60.0, 77.0, 100.0])),
time_remaining_h=float(np.random.uniform(1.0, 10.0)),
max_power_kw=float(np.random.choice([7.4, 11.0, 22.0])),
min_power_kw=1.38
)
else:
self.sessions[i] = None
def train_ppo_agent(total_timesteps: int = 500_000):
"""
Addestra un agente PPO sull'ambiente di EV charging.
Richiede: pip install stable-baselines3
"""
try:
from stable_baselines3 import PPO
from stable_baselines3.common.env_checker import check_env
env = EVChargingEnv(n_connectors=4, max_station_power_kw=44.0)
check_env(env)
model = PPO(
"MlpPolicy", env,
learning_rate=3e-4,
n_steps=2048,
batch_size=64,
n_epochs=10,
gamma=0.99,
gae_lambda=0.95,
clip_range=0.2,
verbose=1,
)
model.learn(total_timesteps=total_timesteps, progress_bar=True)
model.save("./models/ppo_ev_charging")
print("Agente PPO salvato in ./models/ppo_ev_charging")
return model
except ImportError:
print("Installa: pip install stable-baselines3")
return NoneVehicle-to-Grid (V2G): Vozidlo jako síťový zdroj
V2G představuje kvalitativní skok v paradigmatu EV: vozidlo již není samo spotřebitelem energie, ale stává se a obousměrný zdroj. Když je cena energie vysoká nebo je síť ve stresu, baterie Elektromobily mohou posílat energii zpět do sítě a vydělávat peníze pro majitele a stabilizaci elektrizační soustavy.
Standard V2G v roce 2025
| Norma | Košťata | Novinky V2G | Postavení |
|---|---|---|---|
| ISO 15118-2 | Komunikace EV-EVSE AC/DC | Plug & Charge, chytré nabíjení | Používá se |
| ISO 15118-20 | Plná V2G druhé generace | Obousměrná BPT, integrace DER, dynamické plánování | Adopce 2025-2026 |
| OCPP 2.0.1 | Komunikace CP-CSMS | Integrace ISO 15118, nabíjecí profily V2G | Používá se |
| EU AFIR Reg. | Regulace infrastruktury | ISO 15118 povinná pro nové systémy od ledna 2026 | V platnosti |
V Evropě má Utrecht (Nizozemsko) největší komerční nasazení V2G s více než 500 Obousměrné Renaulty ve sdílení aut. Každé vozidlo vydělává peníze 600-1 500 EUR/rok poskytování frekvenční rezervy (FCR) nizozemskému TSO TenneT. V2G v Itálii a stále hlavně experimentální, ale nové modely Nissan Leaf e2+, Volkswagen ID.4 Pro a některé verze Tesla Model 3 Highland již podporují obousměrnost.
Špičkové oholení a frekvenční odezva s bateriemi EV
"""
V2G Controller: peak shaving e frequency response.
Gestisce la bidirezionalita delle sessioni EV compatibili V2G.
"""
from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Optional
import asyncio
import logging
logger = logging.getLogger(__name__)
class ChargingDirection(str, Enum):
G2V = "G2V" # Grid-to-Vehicle (ricarica normale)
V2G = "V2G" # Vehicle-to-Grid (scarica)
IDLE = "IDLE"
@dataclass
class V2GSession:
"""Sessione V2G con capacità bidirezionale."""
connector_id: str
transaction_id: str
soc_percent: float
battery_capacity_kwh: float
max_charge_kw: float # potenza max di ricarica G2V
max_discharge_kw: float # potenza max di scarica V2G
min_soc_percent: float = 20.0 # SoC minimo garantito (non scarica sotto)
departure_soc_target: float = 80.0
@property
def available_discharge_kwh(self) -> float:
usable = max(0.0, self.soc_percent - self.min_soc_percent) / 100.0
return self.battery_capacity_kwh * usable
class V2GController:
"""Controller V2G per peak shaving e frequency response."""
FREQ_NOMINAL = 50.0
FREQ_DEADBAND = 0.05 # +/- 50 mHz: zona morta
FREQ_FULL_ACTIVATION = 0.5 # +/- 500 mHz: attivazione completa FCR
def __init__(
self,
station_id: str,
max_export_kw: float,
profile_sender
):
self._station_id = station_id
self._max_export = max_export_kw
self._send = profile_sender
self._sessions: Dict[str, V2GSession] = {}
self._lock = asyncio.Lock()
async def add_session(self, session: V2GSession) -> None:
async with self._lock:
self._sessions[session.connector_id] = session
async def peak_shaving(
self,
building_load_kw: float,
peak_threshold_kw: float
) -> dict:
"""
Quando il carico supera la soglia, usa le batterie EV per ridurre l'import.
Restituisce le azioni intraprese per ogni connettore V2G.
"""
if building_load_kw <= peak_threshold_kw:
return {"action": "none", "reason": "under_threshold"}
excess_kw = building_load_kw - peak_threshold_kw
logger.info(
"Peak shaving: %.1f kW > soglia %.1f kW (eccesso %.1f kW)",
building_load_kw, peak_threshold_kw, excess_kw
)
actions = {}
async with self._lock:
sessions = sorted(
self._sessions.values(),
key=lambda s: s.available_discharge_kwh,
reverse=True
)
remaining = excess_kw
for s in sessions:
if remaining <= 0:
break
if s.available_discharge_kwh < 1.0:
continue
discharge = min(s.max_discharge_kw, remaining)
actions[s.connector_id] = {
"direction": ChargingDirection.V2G.value,
"power_kw": discharge
}
# Potenza negativa = V2G discharge in OCPP
await self._send(self._station_id, s.connector_id, -discharge * 1000)
remaining -= discharge
logger.info("V2G: %s scarica %.1f kW (peak shaving)", s.connector_id, discharge)
return {"action": "peak_shaving", "actions": actions, "residual_excess_kw": remaining}
async def frequency_response(self, freq_hz: float) -> dict:
"""
FCR (Frequency Containment Reserve).
Risposta lineare alla deviazione di frequenza, entro 500ms.
Frequenza bassa (< 50 Hz) -> V2G discharge.
Frequenza alta (> 50 Hz) -> aumento ricarica G2V.
"""
deviation = freq_hz - self.FREQ_NOMINAL
abs_dev = abs(deviation)
if abs_dev < self.FREQ_DEADBAND:
return {"action": "none", "frequency_hz": freq_hz}
activation = min(1.0,
(abs_dev - self.FREQ_DEADBAND) / (self.FREQ_FULL_ACTIVATION - self.FREQ_DEADBAND)
)
actions = {}
async with self._lock:
sessions = list(self._sessions.values())
for s in sessions:
if deviation < 0 and s.available_discharge_kwh > 0.5:
# Frequenza bassa: scarica per supportare la rete
power_kw = s.max_discharge_kw * activation
actions[s.connector_id] = {"direction": "V2G", "power_kw": power_kw}
await self._send(self._station_id, s.connector_id, -power_kw * 1000)
elif deviation > 0 and s.soc_percent < 90:
# Frequenza alta: aumenta ricarica per assorbire eccesso
power_kw = s.max_charge_kw * activation
actions[s.connector_id] = {"direction": "G2V", "power_kw": power_kw}
await self._send(self._station_id, s.connector_id, power_kw * 1000)
return {
"frequency_hz": freq_hz,
"deviation_hz": deviation,
"activation_factor": activation,
"actions": actions
}Fotovoltaická integrace: Solární přebytek směrem k EV
Integrace mezi FV systémem a nabíjecí stanicí pro elektromobily je jedním z případů použití Rychlejší návratnost investic do energetického managementu. Když fotovoltaika vyrábí víc než co spotřebovává budovu, přebytek je směrován do elektromobilů místo toho, aby se do nich vkládal síť (odměňována ~0,07-0,10 EUR/kWh GSE) – umožňující využití energie, která jinak by to stálo 0,28 EUR/kWh ze sítě. Čistá úspora e 0,18-0,21 EUR/kWh za každou kWh vlastní spotřeby prostřednictvím EV.
| Strategie | Logika | Prospěch | Omezení |
|---|---|---|---|
| Excess-Only | Nabíjejte EV POUZE přebytkem FV | Maximální vlastní spotřeba, nulové náklady na síť | Variabilní výkon, pomalé nabíjení |
| Solar-First | Přednost solárnímu, neporušené ze sítě | Stabilní nabíjení se solárním maximem | Část energie ze sítě |
| Zelený Maximizer | Optimalizujte na 4h horizontu s předpovědí počasí | Maximalizujte % obnovitelné energie | Vyžaduje to přesnou předpověď |
| Optimalizátor nákladů | Solární + spotové ceny + V2G dohromady | Absolutní minimální náklady | Vysoká algoritmická složitost |
"""
Solar-EV Integration Controller.
Implementa strategie Excess-Only e Solar-First
con isteresi per evitare oscillazioni frequenti.
"""
from dataclasses import dataclass
from typing import Optional
import asyncio
import logging
logger = logging.getLogger(__name__)
@dataclass
class PowerState:
"""Stato istantaneo del sistema energetico."""
pv_production_kw: float
building_load_kw: float # consumo edificio escluso EV
electricity_price_eur: float = 0.25
feed_in_tariff_eur: float = 0.07
@property
def net_surplus_kw(self) -> float:
return max(0.0, self.pv_production_kw - self.building_load_kw)
@property
def savings_per_kwh(self) -> float:
"""Risparmio per kWh autoconsumato invece di cedere+comprare."""
return self.electricity_price_eur - self.feed_in_tariff_eur
class SolarEVController:
"""
Controlla la ricarica EV in funzione del surplus fotovoltaico.
Supporta Excess-Only e Solar-First con isteresi configurabile.
"""
HYSTERESIS_KW = 0.3 # evita comandi continui per piccole oscillazioni
def __init__(
self,
balancer,
strategy: str = "solar_first",
min_ev_power_kw: float = 1.38,
max_grid_supplement_kw: float = 5.0
):
self._balancer = balancer
self._strategy = strategy
self._min_ev_power = min_ev_power_kw
self._max_grid_supp = max_grid_supplement_kw
self._last_limit_kw = 0.0
async def update(self, power_state: PowerState, n_active_evs: int) -> dict:
"""
Ricalcola e applica il limite di potenza EV.
Chiamato ogni 5-15 secondi dal loop di controllo.
"""
if n_active_evs == 0:
return {"action": "no_ev", "allocated_kw": 0}
if self._strategy == "excess_only":
target_kw = self._excess_only(power_state, n_active_evs)
else:
target_kw = self._solar_first(power_state, n_active_evs)
# Applica isteresi
if abs(target_kw - self._last_limit_kw) > self.HYSTERESIS_KW:
await self._balancer.update_dynamic_limit(target_kw * 1000)
self._last_limit_kw = target_kw
logger.info("Solar routing [%s]: %.1f kW -> EV", self._strategy, target_kw)
hourly_savings = target_kw * power_state.savings_per_kwh
return {
"strategy": self._strategy,
"pv_kw": power_state.pv_production_kw,
"surplus_kw": power_state.net_surplus_kw,
"allocated_ev_kw": target_kw,
"savings_eur_h": round(hourly_savings, 3)
}
def _excess_only(self, ps: PowerState, n_evs: int) -> float:
surplus = ps.net_surplus_kw
min_total = self._min_ev_power * n_evs
return surplus if surplus >= min_total else 0.0
def _solar_first(self, ps: PowerState, n_evs: int) -> float:
surplus = ps.net_surplus_kw
min_total = self._min_ev_power * n_evs
if surplus < min_total:
grid_needed = min_total - surplus
return min_total if grid_needed <= self._max_grid_supp else surplus
return min(surplus + self._max_grid_supp, surplus + self._max_grid_supp)
async def solar_routing_demo():
"""Simula una giornata con produzione FV tipica di aprile in Italia."""
hourly_pv = [0, 0, 0, 0, 0, 0.5, 2, 5, 9, 14, 18, 21,
22, 20, 16, 10, 6, 2, 0.5, 0, 0, 0, 0, 0]
print("Ora | PV kW | Bldg kW | Surplus | EV kW | Risparmio €/h")
print("-" * 62)
for h in range(24):
pv = hourly_pv[h]
bldg = 5.0 + (2.0 if 8 <= h < 18 else 0)
surplus = max(0, pv - bldg)
price = 0.28 if 8 <= h < 19 else 0.18
feed_in = 0.07
ev_alloc = min(surplus, 11.0)
savings = ev_alloc * (price - feed_in)
print(
f"{h:02d}:00 | {pv:5.1f} | {bldg:7.1f} | "
f"{surplus:7.1f} | {ev_alloc:5.1f} | {savings:.3f}"
)
if __name__ == "__main__":
asyncio.run(solar_routing_demo())Architektura regulátoru nabíjení Microservices
Produkční systém vyvažování zátěže EV a distribuovaná architektura s specializované komponenty, které komunikují prostřednictvím událostí. Zde jsou hlavní služby:
| Servis | Odpovědnost | Hromady | ALS |
|---|---|---|---|
| Brána OCPP | Server WebSocket, překlad OCPP 1.6/2.0.1 | Python ocpp, asyncio | 99,9 %, <100 ms |
| Ovladač nabíjení | Algoritmus vyvažování zátěže, SetChargingProfile | FastAPI, Redis | 99,9 %, <500 ms |
| Energetický manažer | Změřte PV/BESS/síť, vypočítejte dostupnost | Modbus TCP, MQTT, Python | 99,5 %, průzkum 1s |
| Prognostici | Předpověď příjezdu/odjezdu, ceny, FV | LightGBM/LSTM, FastAPI | 99%, aktualizace 15min |
| Brána DSO | Signály OpenADR, dynamické limity | Python OpenADR 2.0b | 99 %, <2s |
| Fakturace | Měření, účtování, relace | FastAPI, PostgreSQL | 99,5 % |
Charge Controller API s FastAPI
"""
Charge Controller Service - FastAPI REST API.
Endpoints per load balancing, monitoring e configurazione.
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from datetime import datetime, timezone
import logging
logger = logging.getLogger(__name__)
app = FastAPI(
title="EV Charge Controller API",
version="2.1.0",
description="Load balancing real-time per stazioni di ricarica EV"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "PUT"],
allow_headers=["*"]
)
# -------------------------------------------------------
# Modelli Pydantic
# -------------------------------------------------------
class ConnectorInfo(BaseModel):
connector_id: str
status: str
allocated_power_w: float
soc_percent: Optional[float]
session_duration_min: float
class StationStatus(BaseModel):
station_id: str
available_power_w: float
algorithm: str
connectors: List[ConnectorInfo]
last_updated: str
class EVConnectedEvent(BaseModel):
connector_id: str
transaction_id: str
initial_soc: Optional[float] = None
battery_capacity_kwh: Optional[float] = None
departure_timestamp: Optional[int] = None
user_tier: int = Field(default=1, ge=1, le=3)
class PowerLimitUpdate(BaseModel):
limit_watts: float = Field(gt=0, le=200_000)
source: str = "manual" # "manual" | "dso" | "solar" | "openadr"
valid_until_ts: Optional[int] = None
class AlgorithmSwitch(BaseModel):
algorithm: str = Field(pattern="^(equal_share|priority|rl_ppo)$")
params: Dict = {}
# In-memory registry (in produzione: Redis)
_station_registry: Dict = {}
_algorithm_map: Dict[str, str] = {}
# -------------------------------------------------------
# Endpoints
# -------------------------------------------------------
@app.get("/health")
async def health():
return {
"status": "healthy",
"ts": datetime.now(timezone.utc).isoformat(),
"version": "2.1.0"
}
@app.get("/stations/{station_id}/status", response_model=StationStatus)
async def get_status(station_id: str):
if station_id not in _station_registry:
raise HTTPException(404, f"Stazione {station_id} non trovata")
balancer = _station_registry[station_id]
raw = balancer.get_status()
return StationStatus(
station_id=station_id,
available_power_w=raw["available_power_w"],
algorithm=_algorithm_map.get(station_id, "equal_share"),
connectors=[ConnectorInfo(**c) for c in raw["connectors"]],
last_updated=datetime.now(timezone.utc).isoformat()
)
@app.post("/stations/{station_id}/events/connected")
async def ev_connected(station_id: str, event: EVConnectedEvent, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(
_station_registry[station_id].on_ev_connected,
event.connector_id,
event.transaction_id
)
return {"status": "accepted", "rebalancing": True}
@app.delete("/stations/{station_id}/connectors/{connector_id}/session")
async def ev_disconnected(station_id: str, connector_id: str, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(_station_registry[station_id].on_ev_disconnected, connector_id)
return {"status": "accepted"}
@app.put("/stations/{station_id}/power-limit")
async def set_power_limit(station_id: str, limit: PowerLimitUpdate, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(_station_registry[station_id].update_dynamic_limit, limit.limit_watts)
logger.info("Limite potenza: %s -> %.0f W (source=%s)", station_id, limit.limit_watts, limit.source)
return {"status": "accepted", "new_limit_w": limit.limit_watts, "source": limit.source}
@app.put("/stations/{station_id}/algorithm")
async def set_algorithm(station_id: str, config: AlgorithmSwitch):
_algorithm_map[station_id] = config.algorithm
return {"status": "ok", "algorithm": config.algorithm}
@app.get("/stations/{station_id}/metrics")
async def get_metrics(station_id: str):
"""KPI della stazione (in produzione: query InfluxDB/TimescaleDB)."""
return {
"station_id": station_id,
"period": "last_24h",
"charging_satisfaction_rate": 0.94,
"peak_reduction_percent": 31.2,
"energy_cost_savings_eur": 47.80,
"solar_share_percent": 42.3,
"grid_stress_index": 0.23,
"total_energy_kwh": 201.6,
"avg_power_kw": 8.4,
"rebalance_latency_ms_p95": 180
}Metriky a KPI pro vyhodnocování zátěže
K vyhodnocení účinnosti systému vyvažování zátěže jsou zapotřebí kvantifikovatelné KPI které měří jak spokojenost uživatelů, tak energetickou optimalizaci.
| KPI | Definice | Cíl | Vzorec |
|---|---|---|---|
| Míra spokojenosti nabíjení (CSR) | % relací dosahujících cílů SoC v termínu | >90 % | ok_sessions / total_sessions |
| Maximální snížení % | Špičkové snížení vs. žádná správa | >25 % | (peak_unmanaged - peak_managed) / peak_unmanaged |
| Úspora nákladů na energii % | Úspora nákladů na kWh oproti okamžitému poplatku | >15 % | (základní_náklady - spravované_náklady) / základní_náklady |
| Solární vlastní spotřeba % | % EV energie z místní FV | >40 % | solar_energy_ev / total_energy_ev |
| Index napětí sítě (GSI) | Průměrný tlak na transformátoru [0-1] | <0,3 | avg(P_aktuální / P_nominální_trafo) |
| Rebalance Latency P95 | Byl použit čas od události k profilu OCPP | <2 s | t_ocpp_conf – t_event_received (str. 95) |
Italský kontext: ARERA, CER s EV a Incentives 2025
Implementace systémů vyvažování zátěže EV v Itálii představuje regulační problémy a trh, které přímo ovlivňují architektonické volby.
Dvouhodinové tarify ARERA a Chytré nabíjení
Italský tarifní systém vytváří přirozenou pobídku pro zpoplatnění mimo špičku:
- pásmo F1 (Po-Pá 8:00-19:00): přibližně 0,26-0,31 EUR/kWh. Pásek pro nabíjení EV.
- Pásmo F2 (Po-Pá 7-8, 19-23; So 7-23): přibližně 0,20-0,24 EUR/kWh.
- Pásmo F3 (noc 23:00–7:00; neděle a svátky celý den): přibližně 0,16-0,19 EUR/kWh. Ideální pro noční nabíjení.
Algoritmus doby používání, který přesune 60 % nabíjení z F1 na F3, snižuje roční náklady na nabíjení pro správce vozového parku 18–25 %.
Společenství obnovitelné energie (CER) s EV
Legislativní dekret 199/2021 a ministerský dekret MASE ze 7. prosince 2023 otevřely konkrétní příležitosti pro integraci EV do italských CER:
- Elektromobily, které nabíjejí komunitní přebytek fotovoltaiky, generují pobídku GSE na podílu sdílená energie (náhrada jízdného + motivace až 11 centů/kWh).
- U V2G fungují baterie EV jako virtuální úložiště CER a vytlačují se sluneční energie od poledne do večera (špičkový posun).
- GSE měří sdílenou energii MACSEvčetně obousměrné toky elektromobilů V2G v komunitním účetnictví.
Italská infrastruktura elektromobilů do roku 2025
Itálie má více než 58 000 veřejných nabíjecích stanic (Motus-E, konec roku 2024), z toho 22 % rychlý DC (oproti 14 % v roce 2023). V roce 2025 bylo registrováno 94 230 BEV (+46 %), s tržním podílem 6,2 %. Podle Terna je rozšířeno 7 % populace bez chytrého nabíjení Distribuční přivaděč bude do roku 2027 přetížen v nejhustších oblastech Severní Itálie. PNRR vyčlenila na infrastrukturu více než 740 milionů eur dobití s cílem 21 000 nových veřejných bodů do roku 2026.
Věnujte pozornost pobídkám
Daňové kredity za chytré nabíjení v rámci Přechodu 5.0 podléhají pravidelně kontrolovány. Aktualizované podmínky vždy zkontrolujte na webu GSE a CSEA před plánováním pobídkových investic. Ceny a podmínky přístupu se mohou lišit.
Závěry a plán implementace
Vyvažování zátěže EV není problém algoritmické elegance: je to nutnost kritická infrastruktura pro energetickou transformaci. Údaje jsou jednoznačné: s 5 miliony neřízených EV se večerní špička zvyšuje o 35 000 MW; s chytrým Samotné nabíjení vozového parku EV se stává flexibilním zdrojem, který snižuje špičku o 34 %.
Pro vývojový tým implementující tyto systémy do výroby, cestovní mapa optimální pro následující fáze a:
- Fáze 1 – Rovný podíl: jednoduché, robustní, dostačující pro instalace až 30-40 triků. Nejvyšší priorita: správně implementovat přepočet událostmi řízený sekvencí snížení a poté zvýšení. Předpokládaná doba: 2-3 týdny.
-
Fáze 2 – Integrace PV: Okamžitá ROI bez ML. Vyžaduje pouze
měření solárního přebytku přes Modbus/MQTT a integraci s balancerem přes
update_dynamic_limit(). Rychlá výhra s návratností za 6–18 měsíců. - Fáze 3 – podle priority: přidejte, až budete mít mobilní aplikaci ke sbírání SoC a termín zahájení. Zisk v CSR (Charging Satisfaction Rate) je měřitelný a odlišuje službu od konkurence.
- Fáze 4 – Infrastruktura V2G: navrhnout obousměrnou architekturu nyní, i když ještě nemáte V2G EV. OCPP 2.0.1 s ISO 15118-20 se stává povinným pro nové závody od roku 2026 (EU AFIR). Dodatečné vybavení stojí 5-10x více.
- Fáze 5 - RL ve výrobě: investovat do hlubokého RL až po 6+ měsících kvalitní data a se specializovaným týmem MLOps. Zisk nákladů je skutečný (15–25 %) ale provozní náročnost je vysoká.
Související články ze série EnergyTech
- Článek 4: Network Digital Twin – simulace dopadu EV na síť před nasazením
- Článek 5: Prognóza obnovitelné energie – prognóza výroby FV pro solární směrování
- Článek 7: MQTT a InfluxDB - telemetrie parametrů nabíjení v reálném čase
- Článek 2: Architektura DERMS – integrujte EV jako DER do distribuovaného systému správy
Statistiky napříč řadami
- Série MLOps (art. 306-315): nasazení modelu PPO s MLflow, monitorováním driftu a rekvalifikací
- AI Engineering Series (čl. 316-325): RAG na dokumentaci OCPP a ISO 15118 s podnikem LLM
- Data & AI Business Series (čl. 267-280): ROI, obchodní metriky a plán investic do chytrého nabíjení
- Řada PostgreSQL AI (čl. 356-361): časové řady opětovného načítání dat pomocí TimescaleDB na PostgreSQL